From 70d9bb0d34bc4b0e1e48df1589905c8b02d5f4d7 Mon Sep 17 00:00:00 2001 From: Valentin Haudiquet Date: Sun, 8 Nov 2026 00:50:09 +0100 Subject: [PATCH] fix(claw): a resumed stream numbers its live frames from the cursor The frame pump counted the live seqs from the last replayed event, which is right only when the replay carried something. A client whose cursor already sat at the record's head replays nothing, so the pump restarted at zero: the client's next resume then named seq 0 and replayed the whole record, and every event it printed in between carried a number that named something else. Base the count on the cursor when the gap is empty. The test resumes at the head of a finished turn and asserts the next live frame names the record's next seq. --- crates/rhema-claw/src/http.rs | 10 +++- crates/rhema-claw/tests/http.rs | 82 +++++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 1 deletion(-) diff --git a/crates/rhema-claw/src/http.rs b/crates/rhema-claw/src/http.rs index 5c35865..2cb2e4b 100644 --- a/crates/rhema-claw/src/http.rs +++ b/crates/rhema-claw/src/http.rs @@ -771,7 +771,15 @@ async fn events( // 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); + let mut next_seq = match history.last() { + Some((seq, _)) => seq + 1, + // A cursor replay with an empty gap names no seq to count + // from, but the record's next one is the cursor's successor + // — zero would hand the client a number that names nothing + // and make its next resume replay the whole record. + None if !stale => query.since.map_or(0, |cursor| cursor + 1), + None => 0, + }; if stale { // The cursor named events the record no longer holds: the // replay above is the tail, not the gap. Say so before it, diff --git a/crates/rhema-claw/tests/http.rs b/crates/rhema-claw/tests/http.rs index 120f2f0..5e68397 100644 --- a/crates/rhema-claw/tests/http.rs +++ b/crates/rhema-claw/tests/http.rs @@ -777,6 +777,88 @@ async fn a_since_cursor_replays_only_the_events_past_it() { // 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 live turn after the replay continues the record's numbering: a + // pump that counted from the replay alone would restart at zero and + // hand the client a cursor that names nothing. + let accepted = client + .post(format!("{base}/v1/sessions/interactive:web/messages")) + .json(&json!({"text": "again"})) + .send() + .await + .expect("the message posts"); + assert_eq!(accepted.status(), 202); + read_until(&mut stream, &mut seen, r#""text":"again""#).await; + assert!( + seen.contains("id: 3"), + "the live frame continues the seqs: {seen}" + ); +} + +/// A cursor already at the record's head replays nothing, and the live +/// feed that follows still names the record's next seq — not zero, which +/// would leave the client's next resume replaying the whole record. +#[tokio::test] +async fn a_cursor_at_the_head_leaves_the_live_frames_numbered() { + let dir = tempfile::tempdir().expect("tempdir"); + let config = config(&dir, None); + let daemon = Daemon::build( + &config, + mock_registry(Arc::new(MockProvider::new([reply("one"), reply("two")]))), + rhema_rsd::DispatcherConfig::default(), + ) + .expect("the daemon assembles"); + let base = serve(&daemon).await; + let client = reqwest::Client::new(); + let accepted = client + .post(format!("{base}/v1/sessions/interactive:web/messages")) + .json(&json!({"text": "hi"})) + .send() + .await + .expect("the message posts"); + assert_eq!(accepted.status(), 202); + + // Watch the first turn land, and note the seq it ended at. + let first = client + .get(format!("{base}/v1/sessions/interactive:web/events")) + .send() + .await + .expect("the events request goes out"); + let mut first_stream = sse_body(first).await; + let mut first_seen = String::new(); + read_until(&mut first_stream, &mut first_seen, r#""type":"turn_done""#).await; + drop(first_stream); + let head: i64 = first_seen + .lines() + .filter_map(|line| line.strip_prefix("id: ")) + .filter_map(|id| id.trim().parse::().ok()) + .max() + .expect("the frames name their seqs"); + + // Resume from the turn's head: nothing to replay, and the next live + // frame must still name the seq after it. + let events = client + .get(format!( + "{base}/v1/sessions/interactive:web/events?since={head}" + )) + .send() + .await + .expect("the events request goes out"); + let mut stream = sse_body(events).await; + let mut seen = String::new(); + let accepted = client + .post(format!("{base}/v1/sessions/interactive:web/messages")) + .json(&json!({"text": "second"})) + .send() + .await + .expect("the message posts"); + assert_eq!(accepted.status(), 202); + read_until(&mut stream, &mut seen, r#""text":"second""#).await; + assert!( + seen.contains(&format!("id: {}", head + 1)), + "the live frame names the record's next seq: {seen}" + ); + assert!(!seen.contains("id: 0"), "not a restarted count: {seen}"); } /// A client with nothing cached opens on the newest turn, not on the