diff --git a/crates/rhema-claw/README.md b/crates/rhema-claw/README.md index 1cc0094..c5d8654 100644 --- a/crates/rhema-claw/README.md +++ b/crates/rhema-claw/README.md @@ -124,7 +124,8 @@ The native surface: GET /healthz GET /v1/sessions POST /v1/sessions {"name"} — create an empty interactive session POST /v1/sessions/{key}/messages {"text": "..."} -> 202 -GET /v1/sessions/{key}/events SSE turn events (see below) +GET /v1/sessions/{key}/events SSE turn events (see below); ?since= resumes a cursor +GET /v1/sessions/{key}/history one page of older turns: ?before=&limit= GET /v1/watches POST /v1/watches DELETE /v1/watches/{forge}/{repo} GET /v1/bootstrap {"soul": bool, "user": bool} POST /v1/ingest {"source","key","payload"} -> 202 @@ -165,18 +166,29 @@ and it simply keeps offering, never interrogating. Both files stay hand-editable forever; `/v1/bootstrap` lets a UI badge setup state. -SSE frames are JSON, `data: {..}`: `user_message`, `thinking_delta`, -`text_delta`, `tool_start`, `tool_result`, `turn_done`. A subscription -opens with the session's recorded turns replayed first and the live feed -after, cut atomically daemon-side — every streamed event is persisted as -it streams, so both a reload and a daemon restart rebuild the whole -transcript, not a recent window; a session that runs for weeks scrolls -back to its first message. A chat UI subscribes once, -POSTs messages, and resets its mirror whenever the stream (re)opens, -letting the replay rebuild it; the daemon owns the conversation, so the -client mirrors the stream rather than managing history. The OpenAI shim -and the message POST read the live feed only — replay never reaches a -turn that is starting. +SSE frames are `id: ` plus `data: {..}`: `user_message`, +`thinking_delta`, `text_delta`, `tool_start`, `tool_result`, +`turn_done`. Every streamed event is persisted as it streams, so both a +reload and a daemon restart rebuild the transcript, and a session that +runs for weeks keeps all of it. + +The seq is the record's dense per-key number for the event, and it is +the client's cursor. A subscription opens with a replay cut atomically +against the live feed, and the frames of both carry their seqs, so a +client that reconnects sends `?since=` and receives +exactly the gap. With no cursor the replay is the tail — the newest 400 +events, cut back to the first turn boundary in them — so a weeks-old +session paints its latest turn at once instead of replaying everything; +the turns above it come from the history route, one whole turn per +page, each ending exactly where the previous page began. A cursor the +record can no longer serve (a wiped table) answers with a `reset;` +comment and the tail, and the client rebuilds rather than splice. + +A chat UI subscribes once, POSTs messages, and resets its mirror +whenever the stream (re)opens, letting the replay rebuild it; the daemon +owns the conversation, so the client mirrors the stream rather than +managing history. The OpenAI shim and the message POST read the live +feed only — replay never reaches a turn that is starting. A message into a session whose turn is running **steers** it, the same interrupt rhema-code's UI owns: the running turn is cut at its current diff --git a/crates/rhema-claw/src/http.rs b/crates/rhema-claw/src/http.rs index 74b0088..5c35865 100644 --- a/crates/rhema-claw/src/http.rs +++ b/crates/rhema-claw/src/http.rs @@ -261,6 +261,7 @@ pub fn router(state: ClawState) -> Router { post(messages).layer(axum::extract::DefaultBodyLimit::max(MAX_MESSAGE_BODY_BYTES)), ) .route("/sessions/{key}/events", get(events)) + .route("/sessions/{key}/history", get(session_history)) .route("/bootstrap", get(bootstrap_status)) .route("/watches", get(watches_list).post(watches_add)) .route("/watches/{forge}/{*repo}", delete(watches_remove)) @@ -696,15 +697,29 @@ where } } -/// The events route's query: `since` counts how many of the record's -/// oldest events the client already holds in its local transcript cache, -/// and those are skipped in the replay. Absent (the web UI keeps no -/// cache) the replay stays whole; a blank `since` is absent, see +/// How much of the record a client with nothing cached opens with: the +/// newest events, cut back to the first turn boundary in the window. A +/// resident session runs for weeks, so replaying it whole to paint a +/// screen would cost more the longer the session lived; the tail paints +/// the latest turn at once and the client walks back through history as +/// the reader scrolls up. +const TAIL_WINDOW: usize = 400; + +/// One backward history page: the size asked for by default, and the +/// most one request may ask for. +const HISTORY_PAGE: usize = 400; +const HISTORY_PAGE_MAX: usize = 2000; + +/// The events route's query: `since` is the seq of the last event the +/// client already holds, from the `id:` line of the frame that carried +/// it. The replay starts after it, so a reconnecting client receives +/// exactly the gap it missed, whatever its size. Absent, the stream +/// opens with the tail ([`TAIL_WINDOW`]); a blank `since` is absent, see /// [`blank_as_none`]. #[derive(Debug, Deserialize)] struct EventStreamQuery { #[serde(default, deserialize_with = "blank_as_none")] - since: Option, + since: Option, } async fn events( @@ -723,13 +738,26 @@ async fn events( } // History first, live second, cut atomically: the snapshot and the // subscription leave the hub under one lock, so a client folding the - // replay and then the stream sees every event exactly once. A reload - // rebuilds its transcript from this — nothing else backfills it. A - // client that cached the transcript opens with `since` and receives - // only the events it has not seen. - let (history, mut receiver) = state - .hub - .subscribe_with_history_from(&ResourceKey::from(key), query.since.unwrap_or(0)); + // replay and then the stream sees every event exactly once. A + // client holding a cursor opens with `since` and receives only the + // gap; one with nothing cached opens with the tail, then walks back + // through `history` for the turns above it. + let key = ResourceKey::from(key); + let (replay, mut receiver) = match query.since { + Some(cursor) => state.hub.replay_after(&key, cursor, TAIL_WINDOW), + None => state.hub.replay_tail(&key, TAIL_WINDOW), + }; + let stale = replay.stale; + // A tail window can open mid-turn (the oldest events in it belong + // to a turn that started before the window), and a transcript that + // opens on a reply with no question above it reads as a bug. Cut + // back to the window's first question. A cursor replay is exact by + // construction and is never cut. + let history = if query.since.is_none() { + snap_to_turn(replay.events) + } else { + replay.events + }; // The pump decouples the broadcast channel from the SSE body: the // client's disconnect closes the body, which fails the send, which // ends the pump. The stream never ends on its own — a chat UI @@ -737,16 +765,41 @@ async fn events( let (sender, body) = tokio::sync::mpsc::channel::>(64); tokio::spawn(async move { - for turn in history { - if sender.send(Ok(sse_event(turn))).await.is_err() { + // Every frame carries its seq as the SSE `id:`, so the client's + // cursor is read off the wire and a reconnect names the exact + // event it stopped at. The record numbers events densely and + // the hub appends and broadcasts under one lock, so the seq the + // pump counts matches the record's: a lagged drop is announced + // with how many were skipped, and the count steps over them. + let mut next_seq = history.last().map_or(0, |(seq, _)| seq + 1); + if stale { + // The cursor named events the record no longer holds: the + // replay above is the tail, not the gap. Say so before it, + // so a client that kept its transcript drops it instead of + // splicing a reset onto the end of a stale one. + let reset = SseEvent::default() + .comment("reset; the record no longer holds the events your cursor named"); + if sender.send(Ok(reset)).await.is_err() { + return; + } + } + for (seq, turn) in history { + if sender.send(Ok(sse_event(seq, turn))).await.is_err() { return; } } loop { let item = match receiver.recv().await { - Ok(turn) => sse_event(turn), - Err(RecvError::Lagged(dropped)) => SseEvent::default() - .comment(format!("lagged; {dropped} earlier event(s) dropped")), + Ok(turn) => { + let seq = next_seq; + next_seq += 1; + sse_event(seq, turn) + } + Err(RecvError::Lagged(dropped)) => { + next_seq += dropped as i64; + SseEvent::default() + .comment(format!("lagged; {dropped} earlier event(s) dropped")) + } Err(RecvError::Closed) => break, }; if sender.send(Ok(item)).await.is_err() { @@ -824,14 +877,16 @@ async fn calendars_list(State(state): State) -> Response { } } -/// One `TurnEvent` as an SSE frame: `data: {json}`. A lagged subscriber -/// gets a comment naming the drop — live delivery is best-effort; the -/// record replays what completed. A serialization failure is logged with -/// the event kind instead of vanishing: a dropped frame a client can -/// never see must at least reach the operator. -fn sse_event(turn: TurnEvent) -> SseEvent { +/// One `TurnEvent` as an SSE frame: `id: {seq}` then `data: {json}`. +/// The id is the record's seq for the event, which is what a client +/// names as `since` when it reconnects. A lagged subscriber gets a +/// comment naming the drop — live delivery is best-effort; the record +/// replays what completed. A serialization failure is logged with the +/// event kind instead of vanishing: a dropped frame a client can never +/// see must at least reach the operator. +fn sse_event(seq: i64, turn: TurnEvent) -> SseEvent { match serde_json::to_string(&turn) { - Ok(data) => SseEvent::default().data(data), + Ok(data) => SseEvent::default().id(seq.to_string()).data(data), Err(err) => { eprintln!( "[claw] warning: dropped unserializable {} event: {err}", @@ -842,6 +897,69 @@ fn sse_event(turn: TurnEvent) -> SseEvent { } } +/// A replay window cut back to its first question: the events before +/// it belong to a turn whose start the window did not reach, and half a +/// turn reads as a bug. A window with no question in it (one turn +/// longer than the window) is returned whole — showing it beats showing +/// nothing, and the walk back through history continues above it. +fn snap_to_turn(mut events: Vec) -> Vec { + match events + .iter() + .position(|(_, event)| matches!(event, TurnEvent::UserMessage { .. })) + { + Some(index) if index > 0 => events.split_off(index), + _ => events, + } +} + +/// The history route's query: the page ends at `before` (the seq of the +/// oldest event the client holds; absent reads from the newest), and +/// `limit` is the most events it may hold before the page is widened +/// back to a turn boundary — see [`EventHub::replay_turn_before`]. +#[derive(Debug, Deserialize)] +struct HistoryQuery { + #[serde(default, deserialize_with = "blank_as_none")] + before: Option, + #[serde(default, deserialize_with = "blank_as_none")] + limit: Option, +} + +/// `GET /v1/sessions/{key}/history?before=&limit=`: one page of +/// turns older than the client's oldest event, for the walk back +/// through a session's history. The page starts on a turn boundary and +/// ends exactly at `before`, so pages stack without gaps or overlaps. +/// The response's `more` says whether anything precedes it: `false` +/// means the client has reached the session's first event. +async fn session_history( + State(state): State, + Path(key): Path, + Query(query): Query, +) -> Response { + if !crate::interactive::is_resident_key(&key) { + return ( + StatusCode::BAD_REQUEST, + Json(json!({"error": "key must start with interactive:, task:, or phone:"})), + ) + .into_response(); + } + let key = ResourceKey::from(key); + let limit = query + .limit + .unwrap_or(HISTORY_PAGE) + .clamp(1, HISTORY_PAGE_MAX); + let page = state.hub.replay_turn_before(&key, query.before, limit); + // Whether anything sits below the page: a lookup, not a read. + let more = match page.first().map(|(seq, _)| *seq) { + Some(first) => state.hub.has_record_before(&key, first), + None => false, + }; + let events: Vec = page + .into_iter() + .map(|(seq, event)| json!({"seq": seq, "event": event})) + .collect(); + Json(json!({"events": events, "more": more})).into_response() +} + async fn watches_list(State(state): State) -> Result, (StatusCode, String)> { let watches = state .watches diff --git a/crates/rhema-claw/src/hub.rs b/crates/rhema-claw/src/hub.rs index 853ddf8..4f646a4 100644 --- a/crates/rhema-claw/src/hub.rs +++ b/crates/rhema-claw/src/hub.rs @@ -39,6 +39,12 @@ use crate::turn_store::TurnStore; /// past four digits. pub const TURN_CHANNEL_CAPACITY: usize = 1024; +/// The furthest a bare history page widens while hunting for the turn +/// boundary it must start on, in events. A turn longer than this has +/// no boundary to find; the page then starts wherever the scan stops, +/// so the walk back through history always advances. +pub const HISTORY_SCAN_CAP: usize = 8192; + /// How many events one key's in-memory record holds when no turn store /// is attached. Wide enough that a reload rebuilds whole recent turns — /// thinking traces and tool rows are deltas, and they add up. @@ -173,11 +179,30 @@ impl TurnEvent { } } +/// One recorded event with the seq the record carries for it: dense +/// per key from zero, assigned as the event is appended, never reused. +/// The seq is the wire frame's `id:`, so a client's cursor names an +/// exact place in the stream even across a reconnect. +pub type SeqEvent = (i64, TurnEvent); + +/// What a replay hands back with its live receiver. +pub struct Replay { + /// The events the caller asked for, oldest first, each with its seq. + pub events: Vec, + /// The caller's cursor names events this record no longer holds — + /// a wiped table, or a memory window that has already dropped them. + /// Its cache is invalid: it must drop what it has and rebuild from + /// the tail, or the seqs it prints past this point would be lies. + pub stale: bool, +} + /// One key's entry: the live channel, plus the in-memory record a /// replay reads when no turn store is attached. Both live behind one /// mutex, which is what makes a history snapshot + subscription atomic. struct KeyEntry { - history: VecDeque, + history: VecDeque, + /// The seq the next memory event takes; dense from zero. + next_seq: i64, sender: broadcast::Sender, } @@ -239,32 +264,214 @@ impl EventHub { &self, key: &ResourceKey, ) -> (Vec, broadcast::Receiver) { - Self::subscribe_with_history_from(self, key, 0) + let (Replay { events, .. }, receiver) = self.replay_after(key, -1, 0); + ( + events.into_iter().map(|(_, event)| event).collect(), + receiver, + ) } - /// [`Self::subscribe_with_history`] with a cursor: `skip` names how - /// many of the record's oldest events the caller already holds (the - /// count a client tracks for its local transcript cache), and the - /// replay drops them, so a reconnecting client that kept its - /// transcript receives only what is new. A `skip` past the record's - /// end clamps to it — the client cannot skip into events that do not - /// exist, and the live feed after the cut is unaffected either way. - pub fn subscribe_with_history_from( + /// The record past a cursor, cut at the same instant as the live + /// receiver: every event with a seq greater than `after` is in + /// `events`, everything emitted after the cut is on the receiver. + /// `after` is `-1` for a client with nothing cached, which reads the + /// whole record. + /// + /// A cursor the record can no longer serve (a wiped table, a pruned + /// memory window) comes back `stale`, and the replay carries the + /// newest `tail` events in place of the unreachable gap: the client + /// drops what it has and rebuilds from those, instead of printing + /// seqs that no longer name anything. The tail is read under the + /// same lock as the receiver, so the reset has no hole either. + pub fn replay_after( &self, key: &ResourceKey, - skip: usize, - ) -> (Vec, broadcast::Receiver) { + after: i64, + tail: usize, + ) -> (Replay, broadcast::Receiver) { let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner); let entry = Self::entry(&mut entries, key, self.capacity); - let mut history = match self.record_store.as_ref() { - Some(store) => store.events(key.as_str()).unwrap_or_else(|err| { - eprintln!("[claw] warning: replay for {key} failed: {err}"); - Vec::new() - }), - None => entry.history.iter().cloned().collect(), - }; - history.drain(..skip.min(history.len())); - (history, entry.sender.subscribe()) + let store = self.record_store.as_deref(); + let (mut events, stale) = Self::record_after(store, entry, key, after); + if stale { + events = Self::record_tail(store, entry, key, tail); + } + (Replay { events, stale }, entry.sender.subscribe()) + } + + /// The record's newest `limit` events, cut with the live receiver — + /// what a client with no cache opens with, so a session that ran + /// for weeks paints its latest turn before anything older arrives. + pub fn replay_tail( + &self, + key: &ResourceKey, + limit: usize, + ) -> (Replay, broadcast::Receiver) { + let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner); + let entry = Self::entry(&mut entries, key, self.capacity); + let events = Self::record_tail(self.record_store.as_deref(), entry, key, limit); + ( + Replay { + events, + stale: false, + }, + entry.sender.subscribe(), + ) + } + + /// One page of the record older than `before` (`None` reads from the + /// newest), ending exactly at `before` and starting at a turn's + /// first event — the page a client walking backward asks for, so a + /// transcript never grows a fragment of a turn at its top. Reading + /// back far enough to find a turn boundary widens the window; a + /// record with no boundary in it (one turn longer than + /// [`HISTORY_SCAN_CAP`] events) returns the widest window it read + /// instead of nothing, so the walk always makes progress. + pub fn replay_turn_before( + &self, + key: &ResourceKey, + before: Option, + limit: usize, + ) -> Vec { + let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner); + let entry = Self::entry(&mut entries, key, self.capacity); + let store = self.record_store.as_deref(); + let mut want = limit.max(1); + loop { + let Some(window) = Self::record_before(store, entry, key, before, want) else { + return Vec::new(); + }; + if window.is_empty() { + return window; + } + if let Some(index) = window + .iter() + .position(|(_, event)| matches!(event, TurnEvent::UserMessage { .. })) + { + return window[index..].to_vec(); + } + // The whole window sits inside one turn: read further back, + // unless there is nothing further back to read. + if window.len() < want || want >= HISTORY_SCAN_CAP { + return window; + } + want = (want * 2).min(HISTORY_SCAN_CAP); + } + } + + /// Whether the record holds anything older than `seq` — the + /// history walk's `more` flag, a lookup rather than a read. + pub fn has_record_before(&self, key: &ResourceKey, seq: i64) -> bool { + let mut entries = self.entries.lock().unwrap_or_else(PoisonError::into_inner); + let entry = Self::entry(&mut entries, key, self.capacity); + Self::record_before(self.record_store.as_deref(), entry, key, Some(seq), 1) + .is_some_and(|window| !window.is_empty()) + } + + /// `after`'s events and whether the cursor outran the record. + fn record_after( + store: Option<&TurnStore>, + entry: &KeyEntry, + key: &ResourceKey, + after: i64, + ) -> (Vec, bool) { + match store { + Some(store) => match store.events_from(key.as_str(), after, usize::MAX) { + Ok(rows) => { + // The table never drops rows, so a cursor past its + // newest seq can only mean the table was reset + // under a client that kept a cache of the old one. + let last = store.last_seq(key.as_str()).ok().flatten(); + let stale = match last { + Some(last) => after > last, + None => after >= 0, + }; + (rows, stale) + } + Err(err) => { + eprintln!("[claw] warning: replay for {key} failed: {err}"); + // A record that cannot be read cannot be vouched + // for: the client rebuilds rather than trust a gap. + (Vec::new(), true) + } + }, + None => { + let first = entry.history.front().map(|(seq, _)| *seq); + let last = entry.history.back().map(|(seq, _)| *seq); + let stale = match (first, last) { + // A pruned window it has passed, or a cursor ahead + // of the record it holds. + (Some(first), Some(last)) => { + after > last || (after >= 0 && after.saturating_add(1) < first) + } + _ => after >= 0, + }; + let events = entry + .history + .iter() + .filter(|(seq, _)| *seq > after) + .cloned() + .collect(); + (events, stale) + } + } + } + + /// The record's newest `limit` events. + fn record_tail( + store: Option<&TurnStore>, + entry: &KeyEntry, + key: &ResourceKey, + limit: usize, + ) -> Vec { + match store { + Some(store) => store + .events_last(key.as_str(), limit) + .unwrap_or_else(|err| { + eprintln!("[claw] warning: tail replay for {key} failed: {err}"); + Vec::new() + }), + None => { + let skip = entry.history.len().saturating_sub(limit); + entry.history.iter().skip(skip).cloned().collect() + } + } + } + + /// The newest `want` events below `before`, oldest first; `None` + /// when the source could not be read. + fn record_before( + store: Option<&TurnStore>, + entry: &KeyEntry, + key: &ResourceKey, + before: Option, + want: usize, + ) -> Option> { + match store { + Some(store) => store + .events_before(key.as_str(), before.unwrap_or(i64::MAX), want) + .map_err(|err| eprintln!("[claw] warning: history page for {key} failed: {err}")) + .ok(), + None => { + let end = match before { + Some(before) => entry + .history + .iter() + .position(|(seq, _)| *seq >= before) + .unwrap_or(entry.history.len()), + None => entry.history.len(), + }; + Some( + entry + .history + .iter() + .take(end) + .skip(end.saturating_sub(want)) + .cloned() + .collect(), + ) + } + } } /// Publishes one event to `key`: the record takes it, and every @@ -284,7 +491,8 @@ impl EventHub { } } None => { - entry.history.push_back(event.clone()); + entry.history.push_back((entry.next_seq, event.clone())); + entry.next_seq += 1; if entry.history.len() > RECORD_CAPACITY { entry.history.pop_front(); } @@ -305,6 +513,7 @@ impl EventHub { ) -> &'a mut KeyEntry { entries.entry(key.clone()).or_insert_with(|| KeyEntry { history: VecDeque::new(), + next_seq: 0, sender: broadcast::channel(capacity).0, }) } @@ -577,28 +786,119 @@ mod tests { assert_eq!(receiver.try_recv().unwrap(), text("in this life")); } + /// The seq cursor: a client names the last event it holds and gets + /// exactly what came after — no replay of what it has, no gap. #[test] - fn a_cursor_skips_exactly_the_events_the_caller_already_holds() { + fn a_seq_cursor_replays_exactly_the_events_past_it() { let hub = EventHub::default(); let key = key("cached"); for n in 0..4 { hub.emit(&key, text(&n.to_string())); } - // A client that cached the first three receives only the fourth; - // the live cut is untouched by the skip. - let (history, mut receiver) = hub.subscribe_with_history_from(&key, 3); - assert_eq!(history, vec![text("3")], "only the unseen part replays"); + // Seqs are dense from zero: the fourth event is seq 3. + let (replay, mut receiver) = hub.replay_after(&key, 2, 0); + assert!(!replay.stale); + assert_eq!( + replay.events, + vec![(3, text("3"))], + "only the unseen part replays" + ); hub.emit(&key, done("hi back")); assert_eq!(receiver.try_recv().unwrap(), done("hi back")); - // A cursor past the record's end clamps to it: nothing replays, - // and nothing the record still holds is skipped twice. - let (history, _) = hub.subscribe_with_history_from(&key, 100); - assert!(history.is_empty()); + // A cursor at the record's newest replays nothing and is not + // stale: the client is simply up to date. + let (replay, _) = hub.replay_after(&key, 4, 0); + assert!(replay.events.is_empty()); + assert!(!replay.stale); - // Zero behaves as the plain replay. - let (history, _) = hub.subscribe_with_history_from(&key, 0); - assert_eq!(history.len(), 5); + // Nothing cached behaves as the plain replay. + let (replay, _) = hub.replay_after(&key, -1, 0); + assert_eq!(replay.events.len(), 5); + assert_eq!(replay.events.first().map(|(seq, _)| *seq), Some(0)); + } + + /// A cursor the record cannot serve is reported, and the tail comes + /// back in place of the gap: a memory window that dropped the events, + /// or a record that was wiped under the client. + #[test] + fn a_cursor_past_the_record_is_stale_and_gets_the_tail() { + let hub = EventHub::with_capacity(16); + let key = key("wiped"); + for n in 0..3 { + hub.emit(&key, text(&n.to_string())); + } + // A cursor from a record this one never had (seq 40): stale, and + // the tail is what replaces the gap. + let (replay, _) = hub.replay_after(&key, 40, 2); + assert!(replay.stale, "a cursor ahead of the record is stale"); + assert_eq!(replay.events, vec![(1, text("1")), (2, text("2"))]); + } + + /// A memory window that pruned events the client still names: the + /// gap is unreachable, so it is stale too. + #[test] + fn a_cursor_below_a_pruned_window_is_stale() { + let hub = EventHub::default(); + let key = key("pruned"); + for n in 0..RECORD_CAPACITY + 5 { + hub.emit(&key, text(&n.to_string())); + } + // The window now starts at seq 5; a client parked at seq 0 can + // never be served events 1..5 again. + let (replay, _) = hub.replay_after(&key, 0, 1); + assert!(replay.stale); + assert_eq!(replay.events.len(), 1, "the tail replaces the gap"); + } + + /// The tail is the newest events, with their seqs intact. + #[test] + fn the_tail_is_the_newest_events() { + let hub = EventHub::default(); + let key = key("tail"); + for n in 0..6 { + hub.emit(&key, text(&n.to_string())); + } + let (replay, _) = hub.replay_tail(&key, 2); + assert!(!replay.stale); + assert_eq!(replay.events, vec![(4, text("4")), (5, text("5"))]); + } + + /// Backward paging: each page ends where the last began and starts + /// on a turn boundary, so a transcript stacks pages without gaps, + /// overlaps, or a fragment at its top. + #[test] + fn a_backward_page_starting_on_a_turn_boundary_stacks_exactly() { + let hub = EventHub::default(); + let key = key("paged"); + // Three turns of three events each: question, reply, done. + for turn in 0..3 { + hub.emit( + &key, + TurnEvent::UserMessage { + text: format!("q{turn}"), + images: Vec::new(), + }, + ); + hub.emit(&key, text(&format!("a{turn}"))); + hub.emit(&key, done(&format!("a{turn}"))); + } + // A window of 4 from the end (seqs 5..8) opens mid-turn; the + // page cuts back to the question at seq 6. + let page = hub.replay_turn_before(&key, None, 4); + assert_eq!(page.len(), 3, "one whole turn"); + assert_eq!(page.first().map(|(seq, _)| *seq), Some(6)); + assert!(hub.has_record_before(&key, 6)); + + // The next page ends exactly at that first seq. + let page = hub.replay_turn_before(&key, Some(6), 4); + assert_eq!(page.first().map(|(seq, _)| *seq), Some(3)); + assert_eq!(page.len(), 3); + + // The first page reaches the record's start and says so. + let page = hub.replay_turn_before(&key, Some(3), 4); + assert_eq!(page.first().map(|(seq, _)| *seq), Some(0)); + assert!(!hub.has_record_before(&key, 0), "nothing below the first"); } } diff --git a/crates/rhema-claw/src/turn_store.rs b/crates/rhema-claw/src/turn_store.rs index c5cf14f..fa7e712 100644 --- a/crates/rhema-claw/src/turn_store.rs +++ b/crates/rhema-claw/src/turn_store.rs @@ -122,19 +122,104 @@ impl TurnStore { Ok(()) } - /// `key`'s whole recorded stream, oldest first — the replay a fresh - /// subscriber folds before the live feed. - pub fn events(&self, key: &str) -> Result, TurnStoreError> { + /// `key`'s recorded stream oldest first, as `(seq, event)` pairs — + /// the replay a subscriber folds before the live feed. `after` + /// bounds it below: only events with a seq past `after` come back, + /// so a client holding a cursor reads exactly the gap it missed; + /// `-1` (or any negative) reads whole. `limit` caps the page, + /// [`usize::MAX`] reads all of it. + pub fn events_from( + &self, + key: &str, + after: i64, + limit: usize, + ) -> Result, TurnStoreError> { let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner); - let mut stmt = conn.prepare("SELECT event FROM turn_events WHERE key = ?1 ORDER BY seq")?; - let rows = stmt.query_map(params![key], |row| row.get::<_, String>(0))?; + let mut stmt = conn.prepare( + "SELECT seq, event FROM turn_events + WHERE key = ?1 AND seq > ?2 + ORDER BY seq + LIMIT ?3", + )?; + let rows = stmt.query_map(params![key, after, limit as i64], |row| { + Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)) + })?; let mut out = Vec::new(); for row in rows { - let json = row.map_err(TurnStoreError::from)?; - out.push(parse(key, &json)?); + let (seq, json) = row.map_err(TurnStoreError::from)?; + out.push((seq, parse(key, &json)?)); } Ok(out) } + + /// The record's last `limit` events, oldest first — the tail a + /// client without a transcript cache opens with instead of the + /// whole history. (Whether that tail opens on a turn boundary is + /// the caller's concern; the store only reads rows.) + pub fn events_last( + &self, + key: &str, + limit: usize, + ) -> Result, TurnStoreError> { + let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner); + let mut stmt = conn.prepare( + "SELECT seq, event FROM turn_events + WHERE key = ?1 + ORDER BY seq DESC + LIMIT ?2", + )?; + let rows = stmt.query_map(params![key, limit as i64], |row| { + Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)) + })?; + let mut out = Vec::new(); + for row in rows { + let (seq, json) = row.map_err(TurnStoreError::from)?; + out.push((seq, parse(key, &json)?)); + } + out.reverse(); + Ok(out) + } + + /// The `limit` events just below `before` (exclusive), oldest first + /// — the page a client walking backward through history asks for. + /// The newest of them has the largest seq under `before`. + pub fn events_before( + &self, + key: &str, + before: i64, + limit: usize, + ) -> Result, TurnStoreError> { + let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner); + let mut stmt = conn.prepare( + "SELECT seq, event FROM turn_events + WHERE key = ?1 AND seq < ?2 + ORDER BY seq DESC + LIMIT ?3", + )?; + let rows = stmt.query_map(params![key, before, limit as i64], |row| { + Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)) + })?; + let mut out = Vec::new(); + for row in rows { + let (seq, json) = row.map_err(TurnStoreError::from)?; + out.push((seq, parse(key, &json)?)); + } + out.reverse(); + Ok(out) + } + + /// The record's newest seq, or `None` when the key has no history — + /// the anchor a cursor conversation starts from, and the value a + /// reconnect's `since` is checked against. + pub fn last_seq(&self, key: &str) -> Result, TurnStoreError> { + let conn = self.conn.lock().unwrap_or_else(PoisonError::into_inner); + conn.query_row( + "SELECT MAX(seq) FROM turn_events WHERE key = ?1", + params![key], + |row| row.get::<_, Option>(0), + ) + .map_err(TurnStoreError::from) + } } /// Moves a legacy `turn_history` table (one JSON-array blob per key, @@ -214,7 +299,12 @@ mod tests { #[test] fn an_appended_history_round_trips_through_events() { let store = TurnStore::in_memory().expect("store opens"); - assert_eq!(store.events("interactive:web").expect("reads"), vec![]); + assert_eq!( + store + .events_from("interactive:web", -1, usize::MAX) + .expect("reads"), + vec![] + ); store .append( @@ -238,19 +328,27 @@ mod tests { ) .expect("appends again"); assert_eq!( - store.events("interactive:web").expect("reads"), + store + .events_from("interactive:web", -1, usize::MAX) + .expect("reads"), vec![ - TurnEvent::UserMessage { - text: "hello".into(), - images: Vec::new() - }, - text_delta("hi"), - TurnEvent::TurnDone { - reply: Some("hi".into()), - outcome: "handled".into(), - }, + ( + 0, + TurnEvent::UserMessage { + text: "hello".into(), + images: Vec::new() + } + ), + (1, text_delta("hi")), + ( + 2, + TurnEvent::TurnDone { + reply: Some("hi".into()), + outcome: "handled".into(), + } + ), ], - "appends extend the stream in order" + "appends extend the stream in order, seqs dense from zero" ); } @@ -260,7 +358,12 @@ mod tests { store .append("interactive:web", &[]) .expect("an empty append is a no-op"); - assert_eq!(store.events("interactive:web").expect("reads"), vec![]); + assert_eq!( + store + .events_from("interactive:web", -1, usize::MAX) + .expect("reads"), + vec![] + ); } #[test] @@ -271,11 +374,17 @@ mod tests { .append("interactive:web", &[text_delta("a")]) .expect("appends"); assert_eq!( - store.events("task:2").expect("reads"), - vec![text_delta("b")], + store.events_from("task:2", -1, usize::MAX).expect("reads"), + vec![(0, text_delta("b"))], "one key's stream never leaks into another's" ); - assert_eq!(store.events("interactive:web").expect("reads").len(), 1); + assert_eq!( + store + .events_from("interactive:web", -1, usize::MAX) + .expect("reads") + .len(), + 1 + ); } #[test] @@ -290,7 +399,7 @@ mod tests { .expect("writes raw"); } let err = store - .events("interactive:web") + .events_from("interactive:web", -1, usize::MAX) .expect_err("corruption is loud"); assert!(err.to_string().contains("interactive:web"), "{err}"); } @@ -317,11 +426,18 @@ mod tests { } let store = TurnStore::open(&path).expect("opens and migrates"); assert_eq!( - store.events("interactive:web").expect("reads"), - vec![text_delta("old days")], + store + .events_from("interactive:web", -1, usize::MAX) + .expect("reads"), + vec![(0, text_delta("old days"))], "the readable blob migrated" ); - assert_eq!(store.events("interactive:broke").expect("reads"), vec![]); + assert_eq!( + store + .events_from("interactive:broke", -1, usize::MAX) + .expect("reads"), + vec![] + ); { let conn = store.conn.lock().unwrap_or_else(PoisonError::into_inner); let legacy: i64 = conn @@ -337,6 +453,12 @@ mod tests { store .append("interactive:web", &[text_delta("new days")]) .expect("appends"); - assert_eq!(store.events("interactive:web").expect("reads").len(), 2); + assert_eq!( + store + .events_from("interactive:web", -1, usize::MAX) + .expect("reads") + .len(), + 2 + ); } } diff --git a/crates/rhema-claw/tests/http.rs b/crates/rhema-claw/tests/http.rs index b5dacbe..120f2f0 100644 --- a/crates/rhema-claw/tests/http.rs +++ b/crates/rhema-claw/tests/http.rs @@ -723,11 +723,12 @@ async fn a_cold_subscription_replays_the_turns_it_missed_even_after_a_restart() ); } -/// The `since` cursor: a client that already holds the record's first -/// event receives the record minus it — the phone client's transcript -/// cache, so reopening a session is not a rebuild from zero. +/// The `since` cursor is the seq of the last frame the client saw (the +/// SSE `id:`), so a reconnecting client receives exactly the gap past +/// it — its transcript cache, so reopening a session is not a rebuild +/// from zero. #[tokio::test] -async fn a_since_cursor_skips_the_replays_oldest_events() { +async fn a_since_cursor_replays_only_the_events_past_it() { let dir = tempfile::tempdir().expect("tempdir"); let config = config(&dir, None); { @@ -760,7 +761,7 @@ async fn a_since_cursor_skips_the_replays_oldest_events() { let base = serve(&daemon).await; let client = reqwest::Client::new(); let events = client - .get(format!("{base}/v1/sessions/interactive:web/events?since=1")) + .get(format!("{base}/v1/sessions/interactive:web/events?since=0")) .send() .await .expect("the events request goes out"); @@ -773,6 +774,133 @@ async fn a_since_cursor_skips_the_replays_oldest_events() { !seen.contains(r#""type":"user_message""#), "the event the cursor names does not replay: {seen}" ); + // The frames carry the record's seq: the delta is seq 1, the end + // marker seq 2 — the user_message the cursor named was seq 0. + assert!(seen.contains("id: 1"), "the frame names its seq: {seen}"); +} + +/// A client with nothing cached opens on the newest turn, not on the +/// whole record: a session that ran for weeks paints its latest +/// conversation at once, and the turns above it come from the history +/// route, one whole turn per page, ending exactly where the last page +/// began. This is the wiring the web UI and the phone use. +#[tokio::test] +async fn a_fresh_stream_opens_on_the_tail_and_history_walks_back_turns() { + let dir = tempfile::tempdir().expect("tempdir"); + let config = config(&dir, None); + { + let daemon = Daemon::build( + &config, + mock_registry(Arc::new(MockProvider::new([ + reply("first"), + reply("second"), + reply("third"), + ]))), + rhema_rsd::DispatcherConfig::default(), + ) + .expect("the daemon assembles"); + let base = serve(&daemon).await; + let client = reqwest::Client::new(); + for text in ["one", "two", "three"] { + let accepted = client + .post(format!("{base}/v1/sessions/interactive:web/messages")) + .json(&json!({"text": text})) + .send() + .await + .expect("the message posts"); + assert_eq!(accepted.status(), 202); + } + } + + let daemon = Daemon::build( + &config, + mock_registry(Arc::new(MockProvider::new([]))), + rhema_rsd::DispatcherConfig::default(), + ) + .expect("the restarted daemon assembles"); + let base = serve(&daemon).await; + let client = reqwest::Client::new(); + + // Nothing cached: the tail replays everything here (the record is + // under the window), and the frames name their seqs. + let mut stream = sse_body( + client + .get(format!("{base}/v1/sessions/interactive:web/events")) + .send() + .await + .expect("the events request goes out"), + ) + .await; + let mut seen = String::new(); + read_until(&mut stream, &mut seen, r#""type":"turn_done""#).await; + assert!( + seen.contains(r#""text":"one""#), + "the tail is whole: {seen}" + ); + + // The history route pages backward from a seq, one whole turn at a + // time, and says when there is nothing older. + let newest = client + .get(format!( + "{base}/v1/sessions/interactive:web/history?before=9999&limit=1" + )) + .send() + .await + .expect("the history request goes out") + .json::() + .await + .expect("the history body parses"); + let newest = newest["events"] + .as_array() + .expect("events is a list") + .clone(); + assert_eq!(newest.len(), 3, "one whole turn per page: {newest:?}"); + assert_eq!( + newest[0]["event"]["type"], "user_message", + "a page starts on a turn boundary: {newest:?}" + ); + assert_eq!(newest[0]["event"]["text"], "three"); + assert_eq!(newest[0]["seq"], 6, "seqs are dense across turns"); + assert_eq!(newest[2]["event"]["type"], "turn_done"); + + let older = client + .get(format!( + "{base}/v1/sessions/interactive:web/history?before=6&limit=1" + )) + .send() + .await + .expect("the history request goes out") + .json::() + .await + .expect("the history body parses"); + assert_eq!(older["events"][0]["event"]["text"], "two"); + assert_eq!(older["more"], true); + + let first = client + .get(format!( + "{base}/v1/sessions/interactive:web/history?before=3&limit=1" + )) + .send() + .await + .expect("the history request goes out") + .json::() + .await + .expect("the history body parses"); + assert_eq!(first["events"][0]["event"]["text"], "one"); + assert_eq!(first["more"], false, "the walk reached the first turn"); + + let none = client + .get(format!( + "{base}/v1/sessions/interactive:web/history?before=0" + )) + .send() + .await + .expect("the history request goes out") + .json::() + .await + .expect("the history body parses"); + assert_eq!(none["events"].as_array().map(Vec::len), Some(0)); + assert_eq!(none["more"], false); } /// The other two resident prefixes take chat messages, and their turns'