From 189d85184ac9a3838f4a75184e9fb0d15dd0b65e Mon Sep 17 00:00:00 2001 From: William Casarin Date: Tue, 4 Aug 2026 07:55:42 -0700 Subject: [PATCH 1/6] headway: wait for the seed's rename in seed_demo_materialises_cards Same class of seed-race flake as board_cache_finalizes_once_per_fold: the test asserts the first backlog card's title is the *amended* value ("Define nostr event model for boards"), but only waited for total_cards == 7. seed_demo_board seeds that card as "Nostr event model" and renames it with a subject edit ingested after all seven cards land, so the count predicate can be satisfied a frame before the rename folds in, leaving the card at its original title and failing the assert. Wait for the renamed title itself. The amendment folds after every card, so its presence also implies all seven cards are here; the count assertions stay as-is. Fixes: 1209d1160b2a ("headway: derive a Linear-style activity timeline from board events") Changelog-None: Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/headway/src/store.rs | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/crates/headway/src/store.rs b/crates/headway/src/store.rs index 728680345..d3966f4b0 100644 --- a/crates/headway/src/store.rs +++ b/crates/headway/src/store.rs @@ -1084,15 +1084,22 @@ mod tests { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns.iter().map(|c| c.cards.len()).sum::() == 7); + // seed_demo_board renames the first backlog card ("Nostr event model" → + // "Define nostr event model for boards") via a subject edit ingested + // *after* all seven cards land. Waiting only for the card count would + // race that amendment, so wait for the renamed title itself — the + // amendment folds after every card, so its presence also implies all + // seven cards are here. + let view = t.wait(|v| { + v.columns[0] + .cards + .first() + .is_some_and(|c| c.title == "Define nostr event model for boards") + }); + assert_eq!(view.columns.iter().map(|c| c.cards.len()).sum::(), 7); assert_eq!(view.columns[0].cards.len(), 3); // Done is the last column; the seeded "done" card lands there. assert_eq!(view.columns.last().unwrap().cards.len(), 1); - // Seeded order is preserved by increasing ranks. - assert_eq!( - view.columns[0].cards[0].title, - "Define nostr event model for boards" - ); assert!(!view.columns[0].cards[0].description.is_empty()); } From 646e0d17b4d67fc46eff06f73c39a7832aa4cfa6 Mon Sep 17 00:00:00 2001 From: William Casarin Date: Tue, 4 Aug 2026 11:15:46 -0700 Subject: [PATCH 2/6] headway: await the writer via SubscriptionStream in the test harnesses MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The async-seed test harnesses (TestSync in notedeck_headway, TestNdb in headway::store) polled the writer against a fixed wall-clock deadline, so a starved-but-live writer on a loaded CI runner could blow past it — the intermittent "predicate never held" failures (e.g. poll_folds_changes_as_a_delta). A wall-clock budget can't reliably wait on an async ingest. Drive the fold loops off nostrdb's own subscription notification instead: `wait`/`poll_board` open a SubscriptionStream and, between folds, `await` the writer delivering the next note. No deadline, no sleep — the loop wakes exactly when there's more to fold and blocks otherwise. The tests become `#[tokio::test]`. Also replaces the old `drain()` quiescence guess (which returned on the first empty poll, mid-materialisation) with `demo_seed_complete`: the demo seed's last event moves the drag card into In Progress, so that card in column 2 is a deterministic "fully folded" signal. Verified under full CPU saturation (which failed the wall-clock versions 40/40): poll_folds_changes_as_a_delta and seed_demo_materialises_cards now pass 15/15. Fixes: b79bff4f62a3 ("notedeck_headway: memoize the folded board per frame") Changelog-None: Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 3 + crates/headway/Cargo.toml | 2 + crates/headway/src/store.rs | 314 +++++++++++++++++------------ crates/notedeck_headway/Cargo.toml | 1 + crates/notedeck_headway/src/lib.rs | 198 ++++++++++-------- 5 files changed, 299 insertions(+), 219 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d2fc0f87c..4f5941af2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3577,10 +3577,12 @@ version = "0.10.0-beta.4" dependencies = [ "bip39", "enostr", + "futures-util", "nostrdb", "profiling", "serde_json", "tempfile", + "tokio", ] [[package]] @@ -5548,6 +5550,7 @@ dependencies = [ "egui", "egui_kittest", "enostr", + "futures-util", "headway", "nostr 0.37.0", "nostrdb", diff --git a/crates/headway/Cargo.toml b/crates/headway/Cargo.toml index 4093ecd16..f647084c2 100644 --- a/crates/headway/Cargo.toml +++ b/crates/headway/Cargo.toml @@ -12,3 +12,5 @@ profiling = { workspace = true } [dev-dependencies] tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt"] } +futures-util = { workspace = true } diff --git a/crates/headway/src/store.rs b/crates/headway/src/store.rs index d3966f4b0..1f7ae0815 100644 --- a/crates/headway/src/store.rs +++ b/crates/headway/src/store.rs @@ -983,8 +983,8 @@ pub use event::load_board; mod tests { use super::*; use enostr::FullKeypair; - use nostrdb::{Config, Ndb, Transaction}; - use std::time::{Duration, Instant}; + use futures_util::StreamExt; + use nostrdb::{Config, Ndb, SubscriptionStream, Transaction}; struct TestNdb { ndb: Ndb, @@ -1007,21 +1007,26 @@ mod tests { self.kp.secret_key.secret_bytes() } - /// Poll the board out of ndb until `pred` holds (ingest is async). - fn wait(&self, pred: F) -> BoardView + /// Fold the board out of ndb until `pred` holds. Ingest is async, so + /// between folds this awaits the writer's own subscription notification + /// (see [`await_ingest`]) rather than polling against a wall-clock + /// deadline — the only way to wait for an async ingest that a loaded CI + /// runner can't race. + async fn wait(&self, pred: F) -> BoardView where F: Fn(&BoardView) -> bool, { - let deadline = Instant::now() + Duration::from_secs(5); + let mut stream = ingest_stream(&self.ndb, &self.kp.pubkey); loop { - let txn = Transaction::new(&self.ndb).unwrap(); - if let Some(view) = load_board(&self.ndb, &txn, &self.kp.pubkey, BOARD_ID) - && pred(&view) { - return view; + let txn = Transaction::new(&self.ndb).unwrap(); + if let Some(view) = load_board(&self.ndb, &txn, &self.kp.pubkey, BOARD_ID) + && pred(&view) + { + return view; + } } - assert!(Instant::now() < deadline, "board predicate never held"); - std::thread::sleep(Duration::from_millis(20)); + await_ingest(&mut stream).await; } } @@ -1038,6 +1043,25 @@ mod tests { } } + /// Open a live await-handle on `author`'s headway events. Subscribing before + /// a wait means every note the async writer ingests *after* this point wakes + /// [`await_ingest`], so the fold loops advance on the writer's own + /// notification instead of a wall-clock sleep. + fn ingest_stream(ndb: &Ndb, author: &Pubkey) -> SubscriptionStream { + let sub = ndb.subscribe(&[event::headway_filter(author)]).unwrap(); + SubscriptionStream::new(ndb.clone(), sub) + } + + /// Await the next batch of ingested notes on `stream`. Panics if the + /// subscription closes first, so a predicate that never holds surfaces as a + /// test-timeout hang rather than a silent spin. + async fn await_ingest(stream: &mut SubscriptionStream) { + stream + .next() + .await + .expect("subscription closed before predicate held"); + } + fn col_titles(view: &BoardView) -> Vec { view.columns.iter().map(|c| c.name.clone()).collect() } @@ -1065,13 +1089,13 @@ mod tests { ); } - #[test] - fn seed_materialises_default_board() { + #[tokio::test] + async fn seed_materialises_default_board() { let t = TestNdb::new(); seed_default_board(&t.ndb, &t.kp.pubkey, &t.secret(), BOARD_ID, &mut NoPublish); // The default board is card-less: just the five columns. - let view = t.wait(|v| v.columns.len() == 5); + let view = t.wait(|v| v.columns.len() == 5).await; assert_eq!( col_titles(&view), ["Backlog", "Todo", "In Progress", "In Review", "Done"] @@ -1079,8 +1103,8 @@ mod tests { assert!(view.columns.iter().all(|c| c.cards.is_empty())); } - #[test] - fn seed_demo_materialises_cards() { + #[tokio::test] + async fn seed_demo_materialises_cards() { let t = TestNdb::new(); seed_demo(&t); @@ -1090,12 +1114,14 @@ mod tests { // race that amendment, so wait for the renamed title itself — the // amendment folds after every card, so its presence also implies all // seven cards are here. - let view = t.wait(|v| { - v.columns[0] - .cards - .first() - .is_some_and(|c| c.title == "Define nostr event model for boards") - }); + let view = t + .wait(|v| { + v.columns[0] + .cards + .first() + .is_some_and(|c| c.title == "Define nostr event model for boards") + }) + .await; assert_eq!(view.columns.iter().map(|c| c.cards.len()).sum::(), 7); assert_eq!(view.columns[0].cards.len(), 3); // Done is the last column; the seeded "done" card lands there. @@ -1103,11 +1129,11 @@ mod tests { assert!(!view.columns[0].cards[0].description.is_empty()); } - #[test] - fn add_card_appends_to_column() { + #[tokio::test] + async fn add_card_appends_to_column() { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns[1].cards.len() == 2); + let view = t.wait(|v| v.columns[1].cards.len() == 2).await; t.apply( &view, @@ -1119,15 +1145,15 @@ mod tests { }, ); - let view = t.wait(|v| v.columns[1].cards.len() == 3); + let view = t.wait(|v| v.columns[1].cards.len() == 3).await; assert_eq!(card_titles(&view, 1).last().unwrap(), "New idea"); } - #[test] - fn add_card_with_labels_tags_the_new_card() { + #[tokio::test] + async fn add_card_with_labels_tags_the_new_card() { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns[1].cards.len() == 2); + let view = t.wait(|v| v.columns[1].cards.len() == 2).await; t.apply( &view, @@ -1139,12 +1165,14 @@ mod tests { }, ); - let view = t.wait(|v| { - v.columns[1] - .cards - .iter() - .any(|c| c.title == "Tagged idea" && c.labels.len() == 2) - }); + let view = t + .wait(|v| { + v.columns[1] + .cards + .iter() + .any(|c| c.title == "Tagged idea" && c.labels.len() == 2) + }) + .await; let card = view.columns[1] .cards .iter() @@ -1153,8 +1181,8 @@ mod tests { assert_eq!(card.labels, vec!["bug".to_string(), "ux".to_string()]); } - #[test] - fn publisher_receives_a_frame_per_ingested_event() { + #[tokio::test] + async fn publisher_receives_a_frame_per_ingested_event() { #[derive(Default)] struct Collect(Vec); impl Publisher for Collect { @@ -1165,7 +1193,7 @@ mod tests { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns[1].cards.len() == 2); + let view = t.wait(|v| v.columns[1].cards.len() == 2).await; // AddCard ingests two events — the issue and its placement — so the // publisher should see exactly two ready-to-send EVENT frames. @@ -1194,11 +1222,11 @@ mod tests { } } - #[test] - fn move_card_changes_column() { + #[tokio::test] + async fn move_card_changes_column() { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns[0].cards.len() == 3); + let view = t.wait(|v| v.columns[0].cards.len() == 3).await; // Move a Backlog card into Done (the last column, which seeds one card). let done = view.columns.len() - 1; @@ -1212,16 +1240,16 @@ mod tests { }, ); - let view = t.wait(|v| v.columns[done].cards.len() == 2); + let view = t.wait(|v| v.columns[done].cards.len() == 2).await; assert_eq!(view.columns[0].cards.len(), 2); assert!(view.columns[done].cards.iter().any(|c| c.id == card)); } - #[test] - fn edit_title_description_and_labels() { + #[tokio::test] + async fn edit_title_description_and_labels() { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns[1].cards.len() == 2); + let view = t.wait(|v| v.columns[1].cards.len() == 2).await; // The second Todo card ("Column reordering") is seeded without labels, // so the SetLabels union below is exactly the two we add. let card = view.columns[1].cards[1].id; @@ -1248,25 +1276,27 @@ mod tests { }, ); - let view = t.wait(|v| { - v.columns[1].cards.iter().any(|c| { - c.id == card - && c.title == "Renamed" - && c.description == "the details" - && c.labels.len() == 2 + let view = t + .wait(|v| { + v.columns[1].cards.iter().any(|c| { + c.id == card + && c.title == "Renamed" + && c.description == "the details" + && c.labels.len() == 2 + }) }) - }); + .await; let edited = view.columns[1].cards.iter().find(|c| c.id == card).unwrap(); assert_eq!(edited.title, "Renamed"); assert_eq!(edited.description, "the details"); assert_eq!(edited.labels, vec!["bug".to_string(), "ux".to_string()]); } - #[test] - fn add_comment_and_reply_fold_onto_the_card() { + #[tokio::test] + async fn add_comment_and_reply_fold_onto_the_card() { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns[1].cards.len() == 2); + let view = t.wait(|v| v.columns[1].cards.len() == 2).await; let card = view.columns[1].cards[0].id; // Top-level comment. @@ -1278,12 +1308,14 @@ mod tests { reply_to: None, }, ); - let view = t.wait(|v| { - v.columns[1] - .cards - .iter() - .any(|c| c.id == card && c.comments.len() == 1) - }); + let view = t + .wait(|v| { + v.columns[1] + .cards + .iter() + .any(|c| c.id == card && c.comments.len() == 1) + }) + .await; let parent = view.columns[1] .cards .iter() @@ -1301,12 +1333,14 @@ mod tests { reply_to: Some(parent), }, ); - let view = t.wait(|v| { - v.columns[1] - .cards - .iter() - .any(|c| c.id == card && c.comments.len() == 2) - }); + let view = t + .wait(|v| { + v.columns[1] + .cards + .iter() + .any(|c| c.id == card && c.comments.len() == 2) + }) + .await; let comments = &view.columns[1] .cards @@ -1320,32 +1354,32 @@ mod tests { assert_eq!(comments[1].parent, Some(parent)); } - #[test] - fn delete_card_removes_it() { + #[tokio::test] + async fn delete_card_removes_it() { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns[0].cards.len() == 3); + let view = t.wait(|v| v.columns[0].cards.len() == 3).await; let card = view.columns[0].cards[0].id; t.apply(&view, BoardAction::DeleteCard { card }); - let view = t.wait(|v| v.columns[0].cards.len() == 2); + let view = t.wait(|v| v.columns[0].cards.len() == 2).await; assert!(!view.columns[0].cards.iter().any(|c| c.id == card)); } - #[test] - fn archive_then_restore_round_trips_to_origin() { + #[tokio::test] + async fn archive_then_restore_round_trips_to_origin() { let t = TestNdb::new(); seed_demo(&t); // Pick a card out of "In Progress" (column 2), not the first column, so a // restore that ignored the origin would land it somewhere else. - let view = t.wait(|v| v.columns[2].cards.len() == 1); + let view = t.wait(|v| v.columns[2].cards.len() == 1).await; let card = view.columns[2].cards[0].id; t.apply(&view, BoardAction::ArchiveCard { card }); // It leaves the columns and shows up in the archived list, with origin. - let view = t.wait(|v| !v.archived.is_empty()); + let view = t.wait(|v| !v.archived.is_empty()).await; assert!( view.columns .iter() @@ -1358,15 +1392,17 @@ mod tests { t.apply(&view, BoardAction::RestoreCard { card }); // Restored back into the exact column it came from, and unarchived. - let view = t.wait(|v| v.archived.is_empty() && v.columns[2].cards.len() == 1); + let view = t + .wait(|v| v.archived.is_empty() && v.columns[2].cards.len() == 1) + .await; assert_eq!(view.columns[2].cards[0].id, card); } - #[test] - fn column_ops_round_trip() { + #[tokio::test] + async fn column_ops_round_trip() { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns.len() == 5); + let view = t.wait(|v| v.columns.len() == 5).await; t.apply( &view, @@ -1374,7 +1410,7 @@ mod tests { name: "Review".to_string(), }, ); - let view = t.wait(|v| v.columns.len() == 6); + let view = t.wait(|v| v.columns.len() == 6).await; assert_eq!(view.columns[5].name, "Review"); t.apply( @@ -1384,22 +1420,26 @@ mod tests { name: "Inbox".to_string(), }, ); - let view = t.wait(|v| v.columns[0].name == "Inbox"); + let view = t.wait(|v| v.columns[0].name == "Inbox").await; t.apply(&view, BoardAction::MoveColumn { from: 0, to: 1 }); - let view = t.wait(|v| v.columns[1].name == "Inbox"); + let view = t.wait(|v| v.columns[1].name == "Inbox").await; t.apply(&view, BoardAction::RemoveColumn { col: 1 }); - let view = t.wait(|v| !v.columns.iter().any(|c| c.name == "Inbox")); + let view = t + .wait(|v| !v.columns.iter().any(|c| c.name == "Inbox")) + .await; // The removed column's cards aren't lost; they fall back to column 0. assert!(view.columns.iter().map(|c| c.cards.len()).sum::() >= 7); } - #[test] - fn rename_board_changes_title_preserving_columns_and_cards() { + #[tokio::test] + async fn rename_board_changes_title_preserving_columns_and_cards() { let t = TestNdb::new(); seed_demo(&t); - let view = t.wait(|v| v.columns.iter().map(|c| c.cards.len()).sum::() == 7); + let view = t + .wait(|v| v.columns.iter().map(|c| c.cards.len()).sum::() == 7) + .await; let cols_before = col_titles(&view); t.apply( @@ -1409,7 +1449,7 @@ mod tests { }, ); - let view = t.wait(|v| v.title == "Renamed Board"); + let view = t.wait(|v| v.title == "Renamed Board").await; // Slug (the addressable `d`-tag) is untouched, so refs still resolve. assert_eq!(view.id, BOARD_ID); // Columns and cards ride along the republished definition unchanged. @@ -1445,33 +1485,35 @@ mod tests { assert_eq!(board_slug("", taken_board), "board-2"); } - /// Load an arbitrary board (the [`TestNdb`] helpers are pinned to `BOARD_ID`). - fn poll_board(t: &TestNdb, board_id: &str, pred: impl Fn(&BoardView) -> bool) -> BoardView { - let deadline = Instant::now() + Duration::from_secs(5); + /// Load an arbitrary board (the [`TestNdb`] helpers are pinned to `BOARD_ID`), + /// awaiting the writer's ingest notifications until `pred` holds. + async fn poll_board( + t: &TestNdb, + board_id: &str, + pred: impl Fn(&BoardView) -> bool, + ) -> BoardView { + let mut stream = ingest_stream(&t.ndb, &t.kp.pubkey); loop { - let txn = Transaction::new(&t.ndb).unwrap(); - if let Some(view) = load_board(&t.ndb, &txn, &t.kp.pubkey, board_id) - && pred(&view) { - return view; + let txn = Transaction::new(&t.ndb).unwrap(); + if let Some(view) = load_board(&t.ndb, &txn, &t.kp.pubkey, board_id) + && pred(&view) + { + return view; + } } - drop(txn); - assert!( - Instant::now() < deadline, - "board '{board_id}' predicate never held" - ); - std::thread::sleep(Duration::from_millis(20)); + await_ingest(&mut stream).await; } } /// Seed two boards, add a card to one, and add a card we can relocate. - fn two_boards_with_a_card(t: &TestNdb) -> NoteId { + async fn two_boards_with_a_card(t: &TestNdb) -> NoteId { seed_default_board(&t.ndb, &t.kp.pubkey, &t.secret(), "src", &mut NoPublish); seed_default_board(&t.ndb, &t.kp.pubkey, &t.secret(), "dst", &mut NoPublish); - poll_board(t, "src", |v| v.columns.len() == 5); - poll_board(t, "dst", |v| v.columns.len() == 5); + poll_board(t, "src", |v| v.columns.len() == 5).await; + poll_board(t, "dst", |v| v.columns.len() == 5).await; - let src = poll_board(t, "src", |v| v.columns.len() == 5); + let src = poll_board(t, "src", |v| v.columns.len() == 5).await; super::apply( &t.ndb, "src", @@ -1486,16 +1528,20 @@ mod tests { }, &mut NoPublish, ); - poll_board(t, "src", |v| v.columns[0].cards.len() == 1).columns[0].cards[0].id + poll_board(t, "src", |v| v.columns[0].cards.len() == 1) + .await + .columns[0] + .cards[0] + .id } - #[test] - fn link_card_places_on_both_boards() { + #[tokio::test] + async fn link_card_places_on_both_boards() { let t = TestNdb::new(); - let card = two_boards_with_a_card(&t); + let card = two_boards_with_a_card(&t).await; - let src = poll_board(&t, "src", |v| v.columns[0].cards.len() == 1); - let dst = poll_board(&t, "dst", |v| v.columns.len() == 5); + let src = poll_board(&t, "src", |v| v.columns[0].cards.len() == 1).await; + let dst = poll_board(&t, "dst", |v| v.columns.len() == 5).await; link_card( &t.ndb, BoardRef { @@ -1513,8 +1559,8 @@ mod tests { ); // Same card on both boards, with its labels intact (it's shared, not copied). - let src = poll_board(&t, "src", |v| v.columns[0].cards.len() == 1); - let dst = poll_board(&t, "dst", |v| v.columns[0].cards.len() == 1); + let src = poll_board(&t, "src", |v| v.columns[0].cards.len() == 1).await; + let dst = poll_board(&t, "dst", |v| v.columns[0].cards.len() == 1).await; assert_eq!(src.columns[0].cards[0].id, card); assert_eq!(dst.columns[0].cards[0].id, card); assert_eq!( @@ -1523,13 +1569,13 @@ mod tests { ); } - #[test] - fn move_card_between_boards_relocates_it() { + #[tokio::test] + async fn move_card_between_boards_relocates_it() { let t = TestNdb::new(); - let card = two_boards_with_a_card(&t); + let card = two_boards_with_a_card(&t).await; - let src = poll_board(&t, "src", |v| v.columns[0].cards.len() == 1); - let dst = poll_board(&t, "dst", |v| v.columns.len() == 5); + let src = poll_board(&t, "src", |v| v.columns[0].cards.len() == 1).await; + let dst = poll_board(&t, "dst", |v| v.columns.len() == 5).await; move_card_between_boards( &t.ndb, BoardRef { @@ -1547,19 +1593,19 @@ mod tests { ); // Leaves src, lands on dst — same id, same overlays. - let dst = poll_board(&t, "dst", |v| v.columns[0].cards.len() == 1); - poll_board(&t, "src", |v| v.columns[0].cards.is_empty()); + let dst = poll_board(&t, "dst", |v| v.columns[0].cards.len() == 1).await; + poll_board(&t, "src", |v| v.columns[0].cards.is_empty()).await; assert_eq!(dst.columns[0].cards[0].id, card); assert_eq!(dst.columns[0].cards[0].title, "Roamer"); } - #[test] - fn move_card_preserves_column_when_target_has_it() { + #[tokio::test] + async fn move_card_preserves_column_when_target_has_it() { let t = TestNdb::new(); - let card = two_boards_with_a_card(&t); + let card = two_boards_with_a_card(&t).await; // Push the card into In Progress on src (both default boards share this column). - let src = poll_board(&t, "src", |v| v.columns[0].cards.len() == 1); + let src = poll_board(&t, "src", |v| v.columns[0].cards.len() == 1).await; super::apply( &t.ndb, "src", @@ -1574,8 +1620,8 @@ mod tests { &mut NoPublish, ); - let src = poll_board(&t, "src", |v| v.columns[2].cards.len() == 1); - let dst = poll_board(&t, "dst", |v| v.columns.len() == 5); + let src = poll_board(&t, "src", |v| v.columns[2].cards.len() == 1).await; + let dst = poll_board(&t, "dst", |v| v.columns.len() == 5).await; move_card_between_boards( &t.ndb, BoardRef { @@ -1593,13 +1639,13 @@ mod tests { ); // Lands in the same-id column (In Progress), not the first column. - let dst = poll_board(&t, "dst", |v| v.columns[2].cards.len() == 1); + let dst = poll_board(&t, "dst", |v| v.columns[2].cards.len() == 1).await; assert_eq!(dst.columns[2].cards[0].id, card); assert!(dst.columns[0].cards.is_empty()); } - #[test] - fn move_card_falls_back_to_first_column_when_target_lacks_it() { + #[tokio::test] + async fn move_card_falls_back_to_first_column_when_target_lacks_it() { let t = TestNdb::new(); seed_default_board(&t.ndb, &t.kp.pubkey, &t.secret(), "src", &mut NoPublish); // A target board whose columns don't include "in-progress". @@ -1617,11 +1663,11 @@ mod tests { &t.secret(), &mut NoPublish, ); - poll_board(&t, "src", |v| v.columns.len() == 5); - poll_board(&t, "dst", |v| v.columns.len() == 2); + poll_board(&t, "src", |v| v.columns.len() == 5).await; + poll_board(&t, "dst", |v| v.columns.len() == 2).await; // Add a card and move it into In Progress on src. - let src = poll_board(&t, "src", |v| v.columns.len() == 5); + let src = poll_board(&t, "src", |v| v.columns.len() == 5).await; super::apply( &t.ndb, "src", @@ -1637,9 +1683,9 @@ mod tests { &mut NoPublish, ); - let src = poll_board(&t, "src", |v| v.columns[2].cards.len() == 1); + let src = poll_board(&t, "src", |v| v.columns[2].cards.len() == 1).await; let card = src.columns[2].cards[0].id; - let dst = poll_board(&t, "dst", |v| v.columns.len() == 2); + let dst = poll_board(&t, "dst", |v| v.columns.len() == 2).await; move_card_between_boards( &t.ndb, BoardRef { @@ -1657,7 +1703,7 @@ mod tests { ); // No "in-progress" on dst, so it falls back to the first column (Inbox). - let dst = poll_board(&t, "dst", |v| v.columns[0].cards.len() == 1); + let dst = poll_board(&t, "dst", |v| v.columns[0].cards.len() == 1).await; assert_eq!(dst.columns[0].id, "inbox"); assert_eq!(dst.columns[0].cards[0].id, card); } diff --git a/crates/notedeck_headway/Cargo.toml b/crates/notedeck_headway/Cargo.toml index c8b022a0d..8412bad7f 100644 --- a/crates/notedeck_headway/Cargo.toml +++ b/crates/notedeck_headway/Cargo.toml @@ -21,6 +21,7 @@ notedeck = { workspace = true, features = ["snapshot-testing"] } notedeck_testing = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true, features = ["macros", "rt"] } +futures-util = { workspace = true } [features] default = [] diff --git a/crates/notedeck_headway/src/lib.rs b/crates/notedeck_headway/src/lib.rs index bf6132f68..52adbe0ed 100644 --- a/crates/notedeck_headway/src/lib.rs +++ b/crates/notedeck_headway/src/lib.rs @@ -995,8 +995,8 @@ impl notedeck::ReferenceParser for HeadwayRefParser { mod tests { use super::*; use enostr::FullKeypair; - use nostrdb::{Config, Filter, Ndb}; - use std::time::{Duration, Instant}; + use futures_util::StreamExt; + use nostrdb::{Config, Filter, Ndb, SubscriptionStream}; /// A headless harness driving a [`BoardCache`] against a bare `Ndb` — the /// subscription / poll / refold logic with no egui in sight. Mirrors the @@ -1053,35 +1053,69 @@ mod tests { board_summaries(&self.cache.all_boards(&self.ndb, &txn, &self.kp.pubkey)) } - /// Poll until the folded default board satisfies `pred` (ingest is async). - /// Fails the test if it never holds. - fn wait bool>(&mut self, pred: F) { - let deadline = Instant::now() + Duration::from_secs(5); - loop { - self.poll(); - if self.view().is_some_and(|v| pred(&v)) { - return; - } - assert!(Instant::now() < deadline, "sync predicate never held"); - std::thread::sleep(Duration::from_millis(20)); - } + /// Fold until the default board satisfies `pred` (ingest is async). + async fn wait bool>(&mut self, pred: F) { + self.wait_until(|t| t.view().is_some_and(|v| pred(&v))) + .await; } - /// Poll until the subscription stops reporting new notes, so the cache - /// is quiescent (the async writer has drained). - fn drain(&mut self) { - let deadline = Instant::now() + Duration::from_secs(5); - while self.poll() { - assert!(Instant::now() < deadline, "sync never quiesced"); - std::thread::sleep(Duration::from_millis(20)); + /// Like [`wait`](Self::wait) but keyed on the switcher's board list — + /// used when the quiescence signal is a *second* board appearing rather + /// than a change to the default board. + async fn wait_boards bool>(&mut self, pred: F) { + self.wait_until(|t| pred(&t.boards())).await; + } + + /// Shared loop for [`wait`](Self::wait) / [`wait_boards`](Self::wait_boards): + /// pump the reducer until `done` holds, awaiting the writer's own ingest + /// notification (see [`await_ingest`]) between folds rather than polling + /// against a wall-clock deadline — the only race-free way to wait on an + /// async ingest. Replaces the old fixed deadline and `drain()` quiescence + /// guess (see `demo_seed_complete`). + async fn wait_until(&mut self, done: impl Fn(&mut Self) -> bool) { + let mut stream = ingest_stream(&self.ndb, &self.kp.pubkey); + while !done(self) { + await_ingest(&mut stream).await; } } } + /// Open a live await-handle on `author`'s headway events. Subscribing before + /// a wait means every note the async writer ingests *after* this point wakes + /// [`await_ingest`], so the fold loops advance on the writer's own + /// notification instead of a wall-clock sleep. + fn ingest_stream(ndb: &Ndb, author: &Pubkey) -> SubscriptionStream { + let sub = ndb.subscribe(&[event::headway_filter(author)]).unwrap(); + SubscriptionStream::new(ndb.clone(), sub) + } + + /// Await the next batch of ingested notes on `stream`. Panics if the + /// subscription closes first, so a predicate that never holds surfaces as a + /// test-timeout hang rather than a silent spin. + async fn await_ingest(stream: &mut SubscriptionStream) { + stream + .next() + .await + .expect("subscription closed before predicate held"); + } + fn total_cards(view: &BoardView) -> usize { view.columns.iter().map(|c| c.cards.len()).sum() } + /// The demo seed's terminal state: [`seed_demo`] moves the drag card into + /// In Progress with the *last* event it ingests, so the drag card sitting in + /// column 2 means every earlier seed event has folded and the async writer is + /// quiescent. A deterministic "seed fully materialised" signal that — unlike a + /// card-count check plus a `drain()` — can't be satisfied mid-materialisation. + fn demo_seed_complete(view: &BoardView) -> bool { + view.columns.get(2).is_some_and(|col| { + col.cards + .iter() + .any(|c| c.title == "Drag-and-drop between columns") + }) + } + /// Seed the populated demo board for the sync tests to fold and act on. The /// production seed is card-less; the fixture lives in [`store::seed_demo_board`]. /// Seeded in the past so follow-up edits (stamped with the wall clock) @@ -1099,14 +1133,14 @@ mod tests { /// Subscribing before seeding, then polling, materialises the whole board /// from events already in ndb. - #[test] - fn poll_materialises_the_board() { + #[tokio::test] + async fn poll_materialises_the_board() { let mut t = TestSync::new(); // Subscribe first so the seed's ingests are reported as new notes. t.poll(); t.seed(); - t.wait(|v| total_cards(v) == 7); + t.wait(|v| total_cards(v) == 7).await; let view = t.view().expect("board loaded"); assert_eq!( view.columns @@ -1121,12 +1155,12 @@ mod tests { /// A click on an inline widget resolves to the app's navigation target /// (see [`resolve_open_target`]): a board opens itself with no card detail, /// while an issue opens its owning board *and* its own card detail. - #[test] - fn resolve_open_target_board_and_issue() { + #[tokio::test] + async fn resolve_open_target_board_and_issue() { let mut t = TestSync::new(); t.poll(); t.seed(); - t.wait(|v| total_cards(v) == 7); + t.wait(|v| total_cards(v) == 7).await; // Pull a board (kind 30619) and an issue (kind 1621) note id out of the db. let (board_id, issue_id, issue_board) = { @@ -1179,12 +1213,14 @@ mod tests { /// An edit ingested after the initial load is picked up on a later poll — /// the cache reflects the change, not a stale snapshot. - #[test] - fn poll_reloads_on_change() { + #[tokio::test] + async fn poll_reloads_on_change() { let mut t = TestSync::new(); t.poll(); t.seed(); - t.wait(|v| v.columns[1].cards.len() == 2); + // Fully materialise the seed first (Todo settles at 2 once the drag card + // moves out), so the edit below folds against a stable board. + t.wait(demo_seed_complete).await; // Apply against the cached pre-edit view (as render does). { @@ -1206,7 +1242,13 @@ mod tests { } // The new card only appears if a later poll re-folded the board. - t.wait(|v| v.columns[1].cards.len() == 3); + t.wait(|v| { + v.columns[1] + .cards + .last() + .is_some_and(|c| c.title == "Fresh card") + }) + .await; let view = t.view().expect("board loaded"); assert_eq!(view.columns[1].cards.last().unwrap().title, "Fresh card"); } @@ -1214,8 +1256,8 @@ mod tests { /// Two boards under one account are both discoverable, and switching the /// active board re-picks from the existing reducer — no full re-fold, since /// the target board's events are already folded in. - #[test] - fn switching_board_repicks_without_refold() { + #[tokio::test] + async fn switching_board_repicks_without_refold() { let mut t = TestSync::new(); // Subscribe first so the seeds' ingests arrive as subscription deltas. t.poll(); @@ -1229,10 +1271,10 @@ mod tests { &mut store::NoPublish, ); - // Materialise on the default board, then quiesce so the 'work' events are - // folded in too. - t.wait(|v| total_cards(v) == 7); - t.drain(); + // The 'work' board is seeded after the whole demo board, so its event is + // the last one ingested: waiting for it to appear means every demo event + // has folded in too, with no quiescence guess. + t.wait_boards(|bs| bs.iter().any(|b| b.id == "work")).await; let folds = t.cache.full_reloads; // Both boards are discoverable from the one reducer. @@ -1258,13 +1300,12 @@ mod tests { /// Once quiescent, polling with nothing new must NOT re-fold — this is the /// whole point of the cache (no per-frame walk of the event history). - #[test] - fn poll_does_not_refold_when_idle() { + #[tokio::test] + async fn poll_does_not_refold_when_idle() { let mut t = TestSync::new(); t.poll(); t.seed(); - t.wait(|v| total_cards(v) == 7); - t.drain(); + t.wait(demo_seed_complete).await; assert!( !t.poll(), @@ -1275,13 +1316,12 @@ mod tests { /// A change after the initial load is absorbed incrementally: the live /// reducer folds the delta, with no additional full-history re-fold. Guards /// against a regression to reload-on-every-change. - #[test] - fn poll_folds_changes_as_a_delta() { + #[tokio::test] + async fn poll_folds_changes_as_a_delta() { let mut t = TestSync::new(); t.poll(); t.seed(); - t.wait(|v| v.columns[1].cards.len() == 2); - t.drain(); + t.wait(demo_seed_complete).await; // Seeding does exactly one full fold; everything since is incremental. assert_eq!( @@ -1306,7 +1346,7 @@ mod tests { &mut store::NoPublish, ); } - t.wait(|v| v.columns[1].cards.len() == 3); + t.wait(|v| v.columns[1].cards.len() == 3).await; assert_eq!( t.cache.full_reloads, 1, @@ -1318,8 +1358,8 @@ mod tests { /// edits as deltas via its subscription — never re-walking the history per /// frame. The render-path counterpart to [`poll_folds_changes_as_a_delta`], /// exercising the `&Ndb` fold-on-read the inline widgets use. - #[test] - fn board_cache_folds_once_then_deltas() { + #[tokio::test] + async fn board_cache_folds_once_then_deltas() { let dir = tempfile::TempDir::new().unwrap(); let ndb = Ndb::new(dir.path().to_str().unwrap(), &Config::new()).unwrap(); let kp = FullKeypair::generate(); @@ -1335,16 +1375,13 @@ mod tests { // Subscribe (seeding an empty reducer) before the board exists, so the // seed's ingests arrive as subscription deltas rather than a re-fold. fold(&mut cache, &ndb); + let mut stream = ingest_stream(&ndb, &kp.pubkey); seed_demo(&ndb, &kp); - // Poll until the board materialises (ingest is async on a writer thread). - let deadline = Instant::now() + Duration::from_secs(5); - loop { - if fold(&mut cache, &ndb).is_some_and(|v| total_cards(&v) == 7) { - break; - } - assert!(Instant::now() < deadline, "board never materialised"); - std::thread::sleep(Duration::from_millis(20)); + // Fold each ingest in as the writer delivers it (ingest is async on a + // writer thread), until the whole board has materialised. + while fold(&mut cache, &ndb).is_none_or(|v| total_cards(&v) != 7) { + await_ingest(&mut stream).await; } // Exactly one full fold — the initial empty seed; every event since @@ -1366,40 +1403,32 @@ mod tests { let kp = FullKeypair::generate(); let mut cache = BoardCache::default(); - // A read subscription of our own to await the writer thread deterministically - // (the cache's own subscription is internal). Subscribe before seeding so - // every seeded event is reported. - let sub = ndb.subscribe(&[event::headway_filter(&kp.pubkey)]).unwrap(); - // Prime the cache's subscription too, then seed. - let txn = Transaction::new(&ndb).unwrap(); - cache.board(&ndb, &txn, &kp.pubkey, store::BOARD_ID); - drop(txn); + // Prime the cache's own (internal) subscription and open a read stream to + // await the writer, both before seeding so every seeded event is reported. + { + let txn = Transaction::new(&ndb).unwrap(); + cache.board(&ndb, &txn, &kp.pubkey, store::BOARD_ID); + } + let mut stream = ingest_stream(&ndb, &kp.pubkey); seed_demo(&ndb, &kp); - // Fold in the seed as it arrives — awaiting notes rather than sleeping — - // until the board has fully materialised *and quiesced*. `seed_demo` - // keeps ingesting after the 7th card lands (parent relations, card - // amendments, and finally the drag-card move into "In Progress"), so - // stopping at `total_cards == 7` would leave those trailing notes to - // arrive mid-frame below — each folds and re-finalizes, breaking the - // "no fold ⇒ no finalize" invariant this test measures. The drag move - // is the *last* event seeded, so the drag card landing in In Progress - // means every prior seed event has folded too and the writer is done. + // Fold in the seed as the writer delivers it, until the board has fully + // materialised *and quiesced*. `seed_demo` keeps ingesting after the 7th + // card lands (parent relations, card amendments, and finally the drag-card + // move into In Progress), so stopping at `total_cards == 7` would leave + // those trailing notes to arrive mid-frame below — each folds and + // re-finalizes, breaking the "no fold ⇒ no finalize" invariant this test + // measures. The drag move is the *last* seed event, so `demo_seed_complete` + // (drag card in In Progress) means every prior event has folded too. while { let txn = Transaction::new(&ndb).unwrap(); let ready = cache .board(&ndb, &txn, &kp.pubkey, store::BOARD_ID) - .is_some_and(|v| { - total_cards(&v) == 7 - && v.columns[2] - .cards - .iter() - .any(|c| c.title == "Drag-and-drop between columns") - }); + .is_some_and(|v| demo_seed_complete(&v)); drop(txn); !ready } { - ndb.wait_for_notes(sub, 1).await.unwrap(); + await_ingest(&mut stream).await; } // A steady frame with no new notes: many reads (what N inline references @@ -1493,14 +1522,13 @@ mod tests { /// [`BoardCache`]) and re-encodes its cards to match the word id, /// yielding the card's kind-1621 issue note. It resolves relative to the /// selected account (the author gap), so no account means no resolution. - #[test] - fn ref_parser_resolves_word_id_to_card() { + #[tokio::test] + async fn ref_parser_resolves_word_id_to_card() { use notedeck::{ReferenceParser, ReferenceResolveCtx}; let mut t = TestSync::new(); t.poll(); t.seed(); - t.wait(|v| total_cards(v) == 7); - t.drain(); + t.wait(demo_seed_complete).await; // Take a real card and its word id off the folded board. let (card_id, words) = { From 9545a20339b67e409f44af56ba881e7469802d71 Mon Sep 17 00:00:00 2001 From: William Casarin Date: Tue, 4 Aug 2026 13:05:53 -0700 Subject: [PATCH 3/6] notedeck: make a cleared private relay list authoritative against stale re-delivery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Clearing (or retargeting) a private relay marker republishes an empty kind-10013 NIP-37 list, but `poll_private_for_updates` harvested whatever kind-10013 note ndb had just ingested. An out-of-order re-delivery of the older AddPrivate note therefore resurrected the private set, flipping dave's PNS sync back on and emitting a second NEG-OPEN — the deterministic `dave_pns_clears_configured_relay_when_private_marker_removed_e2e` / `dave_pns_retargets_when_configured_relay_changes_e2e` failures. Three composing parts: - Poll re-resolves the canonical latest replaceable note instead of trusting the arrived note keys. - Republishes stamp a strictly monotonic `created_at` (`unix_time_secs().max( private_updated_at + 1)`) so a same-second remove supersedes the add rather than relying on NIP-01's non-deterministic lowest-id tie-break. - `private_updated_at` becomes an authoritative read-side floor: a poll that resolves a note at a `created_at` below our last local write is a stale re-delivery and is ignored, closing the window where local ndb still holds only the older note until the empty one round-trips back. Fixes: c7c14be118c9 ("relay: store private-sync relays as a kind-10013 NIP-37 list") Changelog-Fixed: Fix a cleared private (NIP-37) relay being resurrected by a stale replaceable-list re-delivery Co-Authored-By: Claude Opus 4.8 (1M context) --- crates/notedeck/src/account/relay.rs | 111 +++++++++++++++++++++------ 1 file changed, 88 insertions(+), 23 deletions(-) diff --git a/crates/notedeck/src/account/relay.rs b/crates/notedeck/src/account/relay.rs index 43f6dc633..d3608988d 100644 --- a/crates/notedeck/src/account/relay.rs +++ b/crates/notedeck/src/account/relay.rs @@ -17,6 +17,21 @@ pub(crate) struct AccountRelayData { /// dave/headway/notebook to sync private state across the user's own /// devices. Empty for read-only accounts (can't decrypt the list). pub private: BTreeSet, + /// `created_at` of the last kind-10013 list this device authoritatively + /// resolved — either by publishing it or by adopting a strictly-newer note. + /// Republishes stamp a strictly monotonic `created_at` above this, and polls + /// reject any note at or below it, so a stale re-delivery of a superseded + /// list can never resurrect a private relay the user already cleared (NIP-01 + /// otherwise breaks a same-`created_at` tie by note id, non-deterministically). + private_updated_at: u64, +} + +/// The account's canonical kind-10013 private relay list resolved from ndb: the +/// decrypted relay set plus the `created_at` of the note it came from (`None` +/// when the account has no kind-10013 note yet). +struct PrivateRelayList { + relays: BTreeSet, + created_at: Option, } impl AccountRelayData { @@ -41,6 +56,7 @@ impl AccountRelayData { local: BTreeSet::new(), advertised: BTreeSet::new(), private: BTreeSet::new(), + private_updated_at: 0, } } @@ -60,26 +76,43 @@ impl AccountRelayData { debug!("initial relays {:?}", relays); self.advertised = relays.into_iter().collect(); - self.private = self.query_private_relays(ndb, txn, keypair); + + // Seed the private list from the canonical note and remember its + // `created_at` as the authoritative floor, so a later stale re-delivery + // (and republishes) resolve against the newest list we've seen. + let list = self.query_private_relays(ndb, txn, keypair); + if let Some(created_at) = list.created_at { + self.private_updated_at = created_at; + } + self.private = list.relays; } - /// Query the ndb for the account's current kind-10013 private relay list and - /// return the decrypted relay set. + /// Query the ndb for the account's current kind-10013 private relay list. + /// + /// Returns the decrypted relay set alongside the `created_at` of the + /// canonical (latest replaceable) note it came from. Callers use that + /// `created_at` to reject a stale re-delivery of a list this device already + /// superseded (see [`Self::poll_private_for_updates`]). fn query_private_relays( &self, ndb: &Ndb, txn: &Transaction, keypair: &Keypair, - ) -> BTreeSet { + ) -> PrivateRelayList { let nks = ndb .query(txn, std::slice::from_ref(&self.private_filter), 1) .expect("query private relays results") .iter() .map(|qr| qr.note_key) .collect::>(); - Self::harvest_private_relays(ndb, txn, &nks, keypair) + let created_at = nks + .first() + .and_then(|nk| ndb.get_note_by_key(txn, *nk).ok()) + .map(|note| note.created_at()); + let relays = Self::harvest_private_relays(ndb, txn, &nks, keypair) .into_iter() - .collect() + .collect(); + PrivateRelayList { relays, created_at } } pub(crate) fn harvest_private_relays( @@ -97,8 +130,12 @@ impl AccountRelayData { relays } - pub fn new_private_relay_list_note(&'_ self, keypair: &Keypair) -> Option> { - construct_private_relay_list_note(self.private.iter(), keypair) + pub fn new_private_relay_list_note( + &'_ self, + keypair: &Keypair, + created_at: u64, + ) -> Option> { + construct_private_relay_list_note(self.private.iter(), keypair, created_at) } pub(crate) fn harvest_nip65_relays( @@ -159,19 +196,34 @@ impl AccountRelayData { sub: Subscription, keypair: &Keypair, ) { - let nks = ndb.poll_for_notes(sub, 1); - if nks.is_empty() { + // A poll hit only signals that *some* kind-10013 note landed; it may be + // an out-of-order re-delivery of a list the account already superseded. + // Re-resolve the canonical latest replaceable note instead of trusting + // the arrived note, so a stale re-delivery can't resurrect a private + // relay the user just cleared. + if ndb.poll_for_notes(sub, 1).is_empty() { return; } - let private: BTreeSet = - AccountRelayData::harvest_private_relays(ndb, txn, &nks, keypair) - .into_iter() - .collect(); + let list = self.query_private_relays(ndb, txn, keypair); + + // The canonical note lost to an authoritative list this device already + // resolved (a local clear/edit, or a newer note we adopted). Local ndb + // may still hold only the older note until the newer one round-trips, so + // treat our floor as authoritative and ignore this stale resolution. + if list + .created_at + .is_some_and(|at| at < self.private_updated_at) + { + return; + } - if private != self.private { - debug!("updated private relays {:?}", private); - self.private = private; + if let Some(created_at) = list.created_at { + self.private_updated_at = created_at; + } + if list.relays != self.private { + debug!("updated private relays {:?}", list.relays); + self.private = list.relays; } } } @@ -291,6 +343,7 @@ pub(crate) fn parse_private_relay_list_note( pub fn construct_private_relay_list_note<'a>( relays: impl IntoIterator, keypair: &Keypair, + created_at: u64, ) -> Option> { let secret_key = keypair.secret_key.as_ref()?; let tags: Vec> = relays @@ -301,6 +354,7 @@ pub fn construct_private_relay_list_note<'a>( let content = nip44_self_encrypt(keypair, &plaintext)?; NoteBuilder::new() .kind(PRIVATE_RELAY_LIST_KIND) + .created_at(created_at) .content(&content) .sign(&secret_key.to_secret_bytes()) .build() @@ -505,8 +559,17 @@ fn modify_private_relays( RelayAction::Add(_) | RelayAction::Remove(_) => unreachable!(), } + // Keep republishes strictly newer than the last one this device wrote so an + // add immediately followed by a remove (same wall-clock second) still + // resolves to the remove, rather than relying on NIP-01's id tie-break. + let created_at = crate::unix_time_secs().max(account_data.relay.private_updated_at + 1); + account_data.relay.private_updated_at = created_at; + // Encrypt + sign the kind-10013 list. None for a read-only account. - let Some(note) = account_data.relay.new_private_relay_list_note(kp) else { + let Some(note) = account_data + .relay + .new_private_relay_list_note(kp, created_at) + else { return; }; @@ -598,7 +661,7 @@ mod tests { NormRelayUrl::new("wss://private-b.example.com").expect("relay"), ]; - let note = construct_private_relay_list_note(relays.iter(), &owner) + let note = construct_private_relay_list_note(relays.iter(), &owner, 1_700_000_000) .expect("private relay list note"); assert_eq!(note.kind(), PRIVATE_RELAY_LIST_KIND); @@ -616,7 +679,7 @@ mod tests { NormRelayUrl::new("wss://private-b.example.com").expect("relay"), ]; - let note = construct_private_relay_list_note(relays.iter(), &owner) + let note = construct_private_relay_list_note(relays.iter(), &owner, 1_700_000_000) .expect("private relay list note"); let mut parsed = Vec::new(); @@ -633,7 +696,7 @@ mod tests { let other = FullKeypair::generate().to_keypair(); let relays = [NormRelayUrl::new("wss://private-a.example.com").expect("relay")]; - let note = construct_private_relay_list_note(relays.iter(), &owner) + let note = construct_private_relay_list_note(relays.iter(), &owner, 1_700_000_000) .expect("private relay list note"); let mut parsed = Vec::new(); @@ -647,11 +710,13 @@ mod tests { fn private_relay_list_read_only_account_is_noop() { let owner = FullKeypair::generate().to_keypair(); let relays = [NormRelayUrl::new("wss://private-a.example.com").expect("relay")]; - let note = construct_private_relay_list_note(relays.iter(), &owner) + let note = construct_private_relay_list_note(relays.iter(), &owner, 1_700_000_000) .expect("private relay list note"); let read_only = Keypair::only_pubkey(owner.pubkey); - assert!(construct_private_relay_list_note(relays.iter(), &read_only).is_none()); + assert!( + construct_private_relay_list_note(relays.iter(), &read_only, 1_700_000_000).is_none() + ); let mut parsed = Vec::new(); parse_private_relay_list_note(¬e, &read_only, &mut parsed); From a239586ba27457f1347ca94d79524c3afd4b1edc Mon Sep 17 00:00:00 2001 From: William Casarin Date: Tue, 4 Aug 2026 14:24:37 -0700 Subject: [PATCH 4/6] dave: share one per-account conversation subscription across all sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each live/remote session opened its own nostrdb subscription for kind-1988 conversation events (filtered by kind + author + the session's `d`-tag). nostrdb caps subscriptions at 256 per Ndb, so once an account accumulated that many sessions the subscribe calls started failing ("too many subscriptions"), silently dropping live conversation and permission events for every session past the cap. Replace the per-session subscriptions with two shared per-account cursors over all kind-1988 events (`conversation_sub` for chat sync, `conversation_action_sub` for permission responses / mode commands — two cursors because they poll at different points in the frame). Each consumer polls its cursor once and demuxes the returned notes to the owning session by `d`-tag via `conversation_session_index`. One pair of subscriptions now serves any number of sessions. Removes the per-session `live_conversation_sub`/`conversation_action_sub` fields, the `setup_conversation_subscription`/`setup_conversation_action_subscription` helpers, and the now-unused `ConversationSubscriptionScope` plumbing through `create_session_with_cwd`. The 500-session `pns_outbox_e2e` stress binary now runs with 0 "too many subscriptions" warnings (was 252) and all 19 tests pass. Changelog-Fixed: Dave no longer drops live conversation events once an account has many sessions (nostrdb subscription cap) --- crates/notedeck_dave/src/lib.rs | 409 ++++++++---------- crates/notedeck_dave/src/session.rs | 8 - crates/notedeck_dave/src/update.rs | 30 +- .../notedeck_dave/tests/model_picker_tests.rs | 2 - 4 files changed, 191 insertions(+), 258 deletions(-) diff --git a/crates/notedeck_dave/src/lib.rs b/crates/notedeck_dave/src/lib.rs index 7cf0e7e92..8405c3fa8 100644 --- a/crates/notedeck_dave/src/lib.rs +++ b/crates/notedeck_dave/src/lib.rs @@ -123,6 +123,13 @@ struct PnsLocalRuntime { pending_relay_events: Vec, session_state_sub: Option, session_command_sub: Option, + /// One shared per-account subscription for live conversation events across + /// every session (demuxed by `d`-tag in `poll_remote_conversation_events`), + /// so the session count is not bounded by nostrdb's per-db subscription cap. + conversation_sub: Option, + /// Independent shared cursor over the same conversation events, consumed by + /// `poll_remote_conversation_actions` at a different point in the frame. + conversation_action_sub: Option, processed_commands: std::collections::HashSet, pending_spawn_commands: Vec, pending_perm_responses: Vec, @@ -137,21 +144,6 @@ struct PnsLocalRuntime { pending_reap: Vec, } -/// Account-scoped ndb context for live Dave conversation subscriptions. -#[derive(Clone, Copy)] -pub struct ConversationSubscriptionScope<'a> { - pub(crate) account: enostr::Pubkey, - pub(crate) ndb: &'a nostrdb::Ndb, -} - -impl<'a> ConversationSubscriptionScope<'a> { - /// Pair an ndb handle with the selected account author for Dave live - /// conversation subscriptions. - pub fn new(account: enostr::Pubkey, ndb: &'a nostrdb::Ndb) -> Self { - Self { account, ndb } - } -} - impl PnsLocalRuntime { fn empty_agentic() -> Self { Self { @@ -171,6 +163,8 @@ impl PnsLocalRuntime { pending_relay_events: Vec::new(), session_state_sub: None, session_command_sub: None, + conversation_sub: None, + conversation_action_sub: None, processed_commands: std::collections::HashSet::new(), pending_spawn_commands: Vec::new(), pending_perm_responses: Vec::new(), @@ -338,6 +332,16 @@ pub struct Dave { session_state_sub: Option, /// Local ndb subscription for kind-31989 session command events. session_command_sub: Option, + /// One shared per-account subscription for kind-1988 live conversation + /// events across every session. Notes are demuxed by their `d`-tag + /// (`event_session_id`) to the owning session in + /// `poll_remote_conversation_events`, so the number of live sessions is no + /// longer bounded by nostrdb's per-db subscription cap. + conversation_sub: Option, + /// Independent shared cursor over the same kind-1988 events, consumed by + /// `poll_remote_conversation_actions` (permission responses / mode commands) + /// at a different point in the frame than `conversation_sub`. + conversation_action_sub: Option, /// Command UUIDs already processed (dedup for spawn commands). processed_commands: std::collections::HashSet, /// Spawn commands waiting to be built+published in update() where secret key is available. @@ -736,6 +740,8 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr pending_relay_events: Vec::new(), session_state_sub: None, session_command_sub: None, + conversation_sub: None, + conversation_action_sub: None, processed_commands: std::collections::HashSet::new(), pending_spawn_commands: Vec::new(), pending_perm_responses: Vec::new(), @@ -847,7 +853,6 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr cwd, &self.hostname, self.model_config.backend, - None, Model::Default, ); @@ -881,7 +886,6 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr // Extract secret key once for live event generation let secret_key = secret_key_bytes(app_ctx.accounts.get_selected_account().keypair()); - let account = *app_ctx.accounts.selected_account_pubkey(); // Get all session IDs to process let session_ids = self.session_manager.session_ids(); @@ -997,11 +1001,7 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr handle_tool_result(session, result); } DaveApiResponse::SessionInfo(info) => { - handle_session_info( - session, - info, - ConversationSubscriptionScope::new(account, app_ctx.ndb), - ); + handle_session_info(session, info); } DaveApiResponse::SubagentSpawned(subagent) => { handle_subagent_spawned(session, subagent); @@ -1584,7 +1584,6 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr cwd, &self.hostname, backend_type, - None, model, ); } @@ -1695,78 +1694,100 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr let Some(account) = self.pns_local_state.as_ref().map(|state| state.account) else { return mode_applies; }; - let session_ids = self.session_manager.session_ids(); - for session_id in session_ids { - let Some(session) = self.session_manager.get_mut(session_id) else { + let Some(sub) = self.conversation_action_sub else { + return mode_applies; + }; + + let note_keys = ndb.poll_for_notes(sub, 256); + if note_keys.is_empty() { + return mode_applies; + } + + // Route each conversation event to its session by `d`-tag. Only local + // sessions process remote actions, so the index excludes remote ones. + let by_dtag = self.conversation_session_index(true); + + let txn = match Transaction::new(ndb) { + Ok(txn) => txn, + Err(_) => return mode_applies, + }; + + for key in note_keys { + let Ok(note) = ndb.get_note_by_key(&txn, key) else { continue; }; - // Only local sessions poll for remote actions - if session.is_remote() { + if *note.pubkey() != *account.bytes() { continue; } - let Some(agentic) = &mut session.agentic else { + let Some(session_id) = session_events::get_tag_value(¬e, "d") + .and_then(|dtag| by_dtag.get(dtag).copied()) + else { continue; }; - let Some(sub) = agentic.conversation_action_sub else { + let Some(session) = self.session_manager.get_mut(session_id) else { continue; }; - - let note_keys = ndb.poll_for_notes(sub, 64); - if note_keys.is_empty() { + let Some(agentic) = &mut session.agentic else { continue; - } - - let txn = match Transaction::new(ndb) { - Ok(txn) => txn, - Err(_) => continue, }; - for key in note_keys { - let Ok(note) = ndb.get_note_by_key(&txn, key) else { - continue; - }; - if *note.pubkey() != *account.bytes() { - continue; + match session_events::get_tag_value(¬e, "role") { + Some("permission_response") => { + handle_remote_permission_response(¬e, agentic, &mut session.chat); } + Some("set_permission_mode") => { + let content = note.content(); + let mode_str = match serde_json::from_str::(content) { + Ok(v) => v + .get("mode") + .and_then(|m| m.as_str()) + .unwrap_or("default") + .to_string(), + Err(_) => continue, + }; - match session_events::get_tag_value(¬e, "role") { - Some("permission_response") => { - handle_remote_permission_response(¬e, agentic, &mut session.chat); - } - Some("set_permission_mode") => { - let content = note.content(); - let mode_str = match serde_json::from_str::(content) { - Ok(v) => v - .get("mode") - .and_then(|m| m.as_str()) - .unwrap_or("default") - .to_string(), - Err(_) => continue, - }; - - let new_mode = crate::session::permission_mode_from_str(&mode_str); - agentic.permission_mode = new_mode; - session.state_dirty = true; - - mode_applies.push(( - format!("dave-session-{}", session_id), - session.backend_type, - new_mode, - )); + let new_mode = crate::session::permission_mode_from_str(&mode_str); + agentic.permission_mode = new_mode; + session.state_dirty = true; - tracing::info!( - "remote command: set permission mode to {:?} for session {}", - new_mode, - session_id, - ); - } - _ => {} + mode_applies.push(( + format!("dave-session-{}", session_id), + session.backend_type, + new_mode, + )); + + tracing::info!( + "remote command: set permission mode to {:?} for session {}", + new_mode, + session_id, + ); } + _ => {} } } mode_applies } + /// Map each session's live-event `d`-tag (its `event_session_id`) to the + /// session id, so a shared conversation subscription can route polled notes + /// to the right session. `local_only` drops remote sessions (used by the + /// action consumer, which only applies actions to local sessions). + fn conversation_session_index(&self, local_only: bool) -> HashMap { + let mut index = HashMap::new(); + for session_id in self.session_manager.session_ids() { + let Some(session) = self.session_manager.get(session_id) else { + continue; + }; + if local_only && session.is_remote() { + continue; + } + if let Some(agentic) = session.agentic.as_ref() { + index.insert(agentic.event_session_id().to_string(), session_id); + } + } + index + } + /// Publish kind-31988 state events for sessions whose status changed. fn publish_dirty_session_states(&mut self, ctx: &mut AppContext<'_>) { let Some(sk) = secret_key_bytes(ctx.accounts.get_selected_account().keypair()) else { @@ -2083,13 +2104,8 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr if let Some(ref pm) = state.permission_mode { agentic.permission_mode = crate::session::permission_mode_from_str(pm); } - - setup_conversation_subscription( - agentic, - &state.claude_session_id, - account, - ctx.ndb, - ); + // Live conversation events flow through the shared per-account + // subscription; no per-session subscription needed here. } } existing_ids.insert(state.claude_session_id.clone()); @@ -2362,8 +2378,8 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr if let Some(ref pm) = state.permission_mode { agentic.permission_mode = crate::session::permission_mode_from_str(pm); } - - setup_conversation_subscription(agentic, claude_sid, account, ctx.ndb); + // Live conversation events flow through the shared per-account + // subscription; no per-session subscription needed here. } } @@ -2450,7 +2466,6 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr PathBuf::from(cwd), &self.hostname, backend, - Some(ConversationSubscriptionScope::new(account, ctx.ndb)), Model::Default, ); @@ -2482,37 +2497,52 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr let Some(account) = self.pns_local_state.as_ref().map(|state| state.account) else { return (remote_user_messages, events_to_publish); }; - let session_ids = self.session_manager.session_ids(); - for session_id in session_ids { - let Some(session) = self.session_manager.get_mut(session_id) else { - continue; - }; - let is_remote = session.is_remote(); + let Some(sub) = self.conversation_sub else { + return (remote_user_messages, events_to_publish); + }; - // Get sub without holding agentic borrow - let sub = match session - .agentic - .as_ref() - .and_then(|a| a.live_conversation_sub) - { - Some(s) => s, - None => continue, - }; + let note_keys = ndb.poll_for_notes(sub, 256); + if note_keys.is_empty() { + return (remote_user_messages, events_to_publish); + } + + // Route each polled conversation event to its session by `d`-tag. Both + // local and remote sessions consume conversation events, so the index + // keeps remote sessions too. + let by_dtag = self.conversation_session_index(false); - let note_keys = ndb.poll_for_notes(sub, 128); - if note_keys.is_empty() { + let txn = match Transaction::new(ndb) { + Ok(txn) => txn, + Err(_) => return (remote_user_messages, events_to_publish), + }; + + // Group polled notes by their target session, preserving arrival order + // within each session so `process_conversation_notes` sees a coherent + // batch. + let mut by_session: HashMap> = HashMap::new(); + for key in note_keys { + let Ok(note) = ndb.get_note_by_key(&txn, key) else { + continue; + }; + if *note.pubkey() != *account.bytes() { continue; } - - let txn = match Transaction::new(ndb) { - Ok(txn) => txn, - Err(_) => continue, + let Some(session_id) = session_events::get_tag_value(¬e, "d") + .and_then(|dtag| by_dtag.get(dtag).copied()) + else { + continue; }; + by_session.entry(session_id).or_default().push(key); + } - let notes: Vec<_> = note_keys + for (session_id, keys) in by_session { + let Some(session) = self.session_manager.get_mut(session_id) else { + continue; + }; + let is_remote = session.is_remote(); + let notes: Vec<_> = keys .iter() .filter_map(|key| ndb.get_note_by_key(&txn, *key).ok()) - .filter(|note| *note.pubkey() == *account.bytes()) .collect(); let result = @@ -2524,6 +2554,10 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr } } + // Drop the read txn before the reorder pass, which opens its own fresh + // transaction per session (avoids nested transactions). + drop(txn); + // Out-of-order relay delivery was detected for these remote sessions: // rebuild each chat from ndb in `seq` order. Done after the poll loop // so each rebuild uses a fresh transaction (no nested txns). @@ -3642,6 +3676,8 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr pending_relay_events: std::mem::take(&mut self.pending_relay_events), session_state_sub: self.session_state_sub.take(), session_command_sub: self.session_command_sub.take(), + conversation_sub: self.conversation_sub.take(), + conversation_action_sub: self.conversation_action_sub.take(), processed_commands: std::mem::take(&mut self.processed_commands), pending_spawn_commands: std::mem::take(&mut self.pending_spawn_commands), pending_perm_responses: std::mem::take(&mut self.pending_perm_responses), @@ -3684,6 +3720,8 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr self.pending_relay_events = runtime.pending_relay_events; self.session_state_sub = runtime.session_state_sub; self.session_command_sub = runtime.session_command_sub; + self.conversation_sub = runtime.conversation_sub; + self.conversation_action_sub = runtime.conversation_action_sub; self.processed_commands = runtime.processed_commands; self.pending_spawn_commands = runtime.pending_spawn_commands; self.pending_perm_responses = runtime.pending_perm_responses; @@ -3726,6 +3764,15 @@ You are an AI agent for the nostr protocol called Dave, created by Damus. nostr tracing::warn!("failed to subscribe for session command events: {:?}", e); } } + + // Two shared cursors over all kind-1988 conversation events for this + // account. One drives `poll_remote_conversation_events` (chat sync), the + // other `poll_remote_conversation_actions` (permission responses / mode + // commands); they poll at different points in the frame, so each needs + // its own cursor. Notes are demuxed by `d`-tag to the owning session, so + // one pair of subscriptions serves any number of sessions. + self.conversation_sub = subscribe_conversation_events(ndb, account); + self.conversation_action_sub = subscribe_conversation_events(ndb, account); } fn subscribe_pns_run_configs(&mut self, ndb: &nostrdb::Ndb, account: enostr::Pubkey) { @@ -4055,61 +4102,26 @@ impl notedeck::App for Dave { /// single-window mode is particularly aggressive, so we use both /// NSRunningApplication::activateWithOptions and orderFrontRegardless /// on the key window. -/// Set up a live conversation subscription for a session if not already subscribed. +/// Subscribe to every kind-1988 conversation event authored by `account`. /// -/// Subscribes to kind-1988 events tagged with the session's claude ID so we -/// receive messages from remote clients (phone) even before the local backend starts. -pub(crate) fn setup_conversation_subscription( - agentic: &mut session::AgenticSessionData, - claude_session_id: &str, - account: enostr::Pubkey, +/// This is the shared, session-independent subscription that replaces the old +/// per-session (kind + author + `d`-tag) subscriptions: callers poll it once and +/// demux notes to the owning session by their `d`-tag. Returns `None` if nostrdb +/// refuses the subscription (e.g. cap reached), matching the old warn-and-skip +/// behavior. +pub(crate) fn subscribe_conversation_events( ndb: &nostrdb::Ndb, -) { - if agentic.live_conversation_sub.is_some() { - return; - } - let filter = nostrdb::Filter::new() - .kinds([session_events::AI_CONVERSATION_KIND as u64]) - .authors([account.bytes()]) - .tags([claude_session_id], 'd') - .build(); - match ndb.subscribe(&[filter]) { - Ok(sub) => { - agentic.live_conversation_sub = Some(sub); - tracing::info!( - "subscribed for live conversation events (session {})", - claude_session_id, - ); - } - Err(e) => { - tracing::warn!("failed to subscribe for conversation events: {:?}", e,); - } - } -} - -/// Subscribe for kind-1988 conversation action events (permission responses, -/// mode commands) for the given session d-tag. -pub(crate) fn setup_conversation_action_subscription( - agentic: &mut session::AgenticSessionData, - event_id: &str, account: enostr::Pubkey, - ndb: &nostrdb::Ndb, -) { - if agentic.conversation_action_sub.is_some() { - return; - } +) -> Option { let filter = nostrdb::Filter::new() .kinds([session_events::AI_CONVERSATION_KIND as u64]) .authors([account.bytes()]) - .tags([event_id], 'd') .build(); match ndb.subscribe(&[filter]) { - Ok(sub) => { - agentic.conversation_action_sub = Some(sub); - tracing::info!("subscribed for conversation actions (session {})", event_id,); - } + Ok(sub) => Some(sub), Err(e) => { - tracing::warn!("failed to subscribe for conversation actions: {:?}", e); + tracing::warn!("failed to subscribe for conversation events: {:?}", e); + None } } } @@ -4681,14 +4693,7 @@ fn handle_query_complete(session: &mut session::ChatSession, info: messages::Usa } /// Handle a SessionInfo response from the AI backend. -/// -/// Sets up ndb subscriptions for permission responses and conversation events -/// when we first learn the claude session ID. -fn handle_session_info( - session: &mut session::ChatSession, - info: SessionInfo, - subscription_scope: ConversationSubscriptionScope<'_>, -) { +fn handle_session_info(session: &mut session::ChatSession, info: SessionInfo) { // Propagate the runtime model for header display only. // Keep the original requested override intact so duplicate/clear // can reuse the user's intent instead of the backend's resolved model. @@ -4697,22 +4702,9 @@ fn handle_session_info( } if let Some(agentic) = &mut session.agentic { - // Use the stable event_id (not the CLI session ID) for subscriptions, - // since all live events are tagged with event_id as the d-tag. - let event_id = agentic.event_session_id().to_string(); - setup_conversation_action_subscription( - agentic, - &event_id, - subscription_scope.account, - subscription_scope.ndb, - ); - setup_conversation_subscription( - agentic, - &event_id, - subscription_scope.account, - subscription_scope.ndb, - ); - + // Live conversation and action events flow through the shared + // per-account subscriptions (see `subscribe_conversation_events`); no + // per-session subscription is created here. agentic.session_info = Some(info); } // Persist initial session state now that we know the claude_session_id @@ -4902,10 +4894,14 @@ mod tests { key } - async fn same_d_live_subscription_pubkeys( - setup: impl FnOnce(&mut session::AgenticSessionData, &str, enostr::Pubkey, &Ndb), - sub: impl FnOnce(&session::AgenticSessionData) -> Option, - ) -> ([u8; 32], Vec<[u8; 32]>) { + /// The selected account's pubkey alongside the author pubkeys of every + /// note the shared conversation subscription matched. + struct ConversationSubAuthors { + account: [u8; 32], + matched: Vec<[u8; 32]>, + } + + async fn conversation_subscription_author_pubkeys() -> ConversationSubAuthors { let account = enostr::FullKeypair::generate(); let other_account = enostr::FullKeypair::generate(); let account_pubkey = *account.pubkey.bytes(); @@ -4938,15 +4934,8 @@ mod tests { let tmp_dir = TempDir::new().unwrap(); let ndb = Ndb::new(tmp_dir.path().to_str().unwrap(), &test_config()).unwrap(); - let mut session = session::ChatSession::new( - 1, - PathBuf::from("/tmp"), - AiMode::Agentic, - BackendType::Claude, - ); - let agentic = session.agentic.as_mut().expect("agentic session"); - setup(agentic, session_id_str, account.pubkey, &ndb); - let sub = sub(agentic).expect("live subscription"); + let sub = + subscribe_conversation_events(&ndb, account.pubkey).expect("conversation subscription"); ndb.process_event_with( &other_event.to_event_json(), @@ -4969,7 +4958,10 @@ mod tests { .iter() .map(|key| *ndb.get_note_by_key(&txn, *key).expect("note").pubkey()) .collect(); - (account_pubkey, pubkeys) + ConversationSubAuthors { + account: account_pubkey, + matched: pubkeys, + } } fn test_dave(data_path: &DataPath) -> Dave { @@ -4979,36 +4971,13 @@ mod tests { } #[tokio::test] - async fn live_conversation_subscription_filters_selected_account_author() { - let (account_pubkey, pubkeys) = same_d_live_subscription_pubkeys( - |agentic, session_id, account, ndb| { - setup_conversation_subscription(agentic, session_id, account, ndb); - }, - |agentic| agentic.live_conversation_sub, - ) - .await; - - assert_eq!( - pubkeys, - vec![account_pubkey], - "same-d events from another account must not match conversation subscription" - ); - } - - #[tokio::test] - async fn live_action_subscription_filters_selected_account_author() { - let (account_pubkey, pubkeys) = same_d_live_subscription_pubkeys( - |agentic, session_id, account, ndb| { - setup_conversation_action_subscription(agentic, session_id, account, ndb); - }, - |agentic| agentic.conversation_action_sub, - ) - .await; + async fn conversation_subscription_filters_selected_account_author() { + let authors = conversation_subscription_author_pubkeys().await; assert_eq!( - pubkeys, - vec![account_pubkey], - "same-d events from another account must not match action subscription" + authors.matched, + vec![authors.account], + "same-d events from another account must not match the conversation subscription" ); } diff --git a/crates/notedeck_dave/src/session.rs b/crates/notedeck_dave/src/session.rs index 9fd45dda9..c4a51f6f8 100644 --- a/crates/notedeck_dave/src/session.rs +++ b/crates/notedeck_dave/src/session.rs @@ -276,18 +276,12 @@ pub struct AgenticSessionData { pub git_status: GitStatusCache, /// Threading state for live kind-1988 event generation. pub live_threading: ThreadingState, - /// Subscription for remote kind-1988 events (permission responses, commands). - /// Set up once when the session's claude_session_id becomes known. - pub conversation_action_sub: Option, /// Status as reported by the remote desktop's kind-31988 event. /// Only meaningful when session source is Remote. pub remote_status: Option, /// Timestamp of the kind-31988 event that last set `remote_status`. /// Used to ignore older replaceable event revisions that arrive out of order. pub remote_status_ts: u64, - /// Subscription for live kind-1988 conversation events from relays. - /// Used by remote sessions to receive new messages in real-time. - pub live_conversation_sub: Option, /// Note IDs we've already processed from live conversation polling. /// Prevents duplicate messages when events are loaded during restore /// and then appear again via the subscription. @@ -338,10 +332,8 @@ impl AgenticSessionData { resume_session_id: None, git_status, live_threading: ThreadingState::new(), - conversation_action_sub: None, remote_status: None, remote_status_ts: 0, - live_conversation_sub: None, seen_note_ids: HashSet::new(), max_seen_seq: None, usage: Default::default(), diff --git a/crates/notedeck_dave/src/update.rs b/crates/notedeck_dave/src/update.rs index cd310682b..d1c9be922 100644 --- a/crates/notedeck_dave/src/update.rs +++ b/crates/notedeck_dave/src/update.rs @@ -1239,7 +1239,6 @@ pub fn create_session_with_cwd( cwd: PathBuf, hostname: &str, backend_type: BackendType, - conversation_subscription: Option>, model: Model, ) -> SessionId { directory_picker.add_recent(cwd.clone()); @@ -1257,26 +1256,8 @@ pub fn create_session_with_cwd( scene.focus_on(agentic.scene_position); } } - - // Set up ndb subscriptions so remote clients can send messages - // to this session (e.g. to kickstart the backend remotely). - if let (Some(subscription), Some(agentic)) = - (conversation_subscription, &mut session.agentic) - { - let event_id = agentic.event_session_id().to_string(); - crate::setup_conversation_subscription( - agentic, - &event_id, - subscription.account, - subscription.ndb, - ); - crate::setup_conversation_action_subscription( - agentic, - &event_id, - subscription.account, - subscription.ndb, - ); - } + // Remote clients reach this session's live conversation events through + // the shared per-account subscription; nothing per-session to wire up. } session_manager.rebuild_cwd_groups(); id @@ -1360,7 +1341,6 @@ pub fn clone_session( cwd, hostname, backend_type, - None, model, ); None @@ -1488,7 +1468,6 @@ mod tests { PathBuf::from(cwd), hostname, BackendType::Claude, - None, Model::Default, ); @@ -1515,7 +1494,6 @@ mod tests { PathBuf::from("/tmp"), "remote-a", BackendType::Claude, - None, Model::Default, ); let session = sm.get_mut(id).expect("session should exist"); @@ -1607,7 +1585,6 @@ mod tests { PathBuf::from("/tmp"), "localhost", BackendType::Claude, - None, Model::Opus, ); @@ -1654,7 +1631,6 @@ mod tests { PathBuf::from("/tmp"), "localhost", BackendType::Claude, - None, Model::Default, ); @@ -1694,7 +1670,6 @@ mod tests { PathBuf::from("/tmp"), "localhost", BackendType::Codex, - None, Model::Custom("gpt-5.2-codex".to_string()), ); @@ -2197,7 +2172,6 @@ mod tests { PathBuf::from("/tmp"), "localhost", BackendType::Claude, - None, Model::Default, ); diff --git a/crates/notedeck_dave/tests/model_picker_tests.rs b/crates/notedeck_dave/tests/model_picker_tests.rs index 88a12a10c..9bc01f15c 100644 --- a/crates/notedeck_dave/tests/model_picker_tests.rs +++ b/crates/notedeck_dave/tests/model_picker_tests.rs @@ -101,7 +101,6 @@ fn test_picker_selection_flows_to_session() { PathBuf::from("/tmp"), "localhost", picked_backend, - None, picked_model, ); @@ -126,7 +125,6 @@ fn test_picker_selection_flows_to_session() { PathBuf::from("/tmp"), "localhost", picked_backend, - None, picked_model, ); From 1ec3f6bc289948ad9a9e85962a2eb1b6a426aba3 Mon Sep 17 00:00:00 2001 From: William Casarin Date: Tue, 4 Aug 2026 16:39:05 -0700 Subject: [PATCH 5/6] headway: wait on the seed's terminal event in poll_materialises_the_board MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `total_cards(v) == 7` can hold mid-fold, before the demo seed's last event (the drag card landing in In Progress) settles, so the follow-up layout assertion occasionally ran against a half-materialised board — the flake seen on macOS/Windows CI (`left: 5, right: 3`). Wait on `demo_seed_complete` instead: the drag card sitting in column 2 is the seed's terminal state, so every earlier event has already folded. This matches the deterministic signal the other sync tests already use. Changelog-None: --- crates/notedeck_headway/src/lib.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/notedeck_headway/src/lib.rs b/crates/notedeck_headway/src/lib.rs index 52adbe0ed..81fdcc39d 100644 --- a/crates/notedeck_headway/src/lib.rs +++ b/crates/notedeck_headway/src/lib.rs @@ -1140,7 +1140,10 @@ mod tests { t.poll(); t.seed(); - t.wait(|v| total_cards(v) == 7).await; + // Wait on the seed's *terminal* event (the drag card landing in In + // Progress), not a card count: `total_cards == 7` can hold mid-fold + // before the drag move settles, asserting a half-materialised layout. + t.wait(demo_seed_complete).await; let view = t.view().expect("board loaded"); assert_eq!( view.columns From 1f70228710b36d46815e95062fa36ae57be99966 Mon Sep 17 00:00:00 2001 From: William Casarin Date: Tue, 4 Aug 2026 16:39:52 -0700 Subject: [PATCH 6/6] dave: skip the real-codex tests when the codex binary is absent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The `test_real_codex_*` integration tests are `#[ignore]`d, but the snapshot CI job runs every ignored test in the notedeck_dave lib via `--ignored` (see scripts/snapshot-test), which sweeps these real-binary tests up alongside the lavapipe snapshots. On runners without `codex` on PATH they panicked at spawn ("Failed to spawn codex app-server — is codex installed?"), reddening the job. Have `setup_real_codex_test` return `None` when `spawn_codex` fails so the tests skip rather than fail. The real path still runs locally when codex is installed; a missing binary is now a visible skip, not a hard failure. Changelog-None: --- crates/notedeck_dave/src/backend/codex.rs | 35 +++++++++++++++++------ 1 file changed, 27 insertions(+), 8 deletions(-) diff --git a/crates/notedeck_dave/src/backend/codex.rs b/crates/notedeck_dave/src/backend/codex.rs index b10f04f31..46ac2c673 100644 --- a/crates/notedeck_dave/src/backend/codex.rs +++ b/crates/notedeck_dave/src/backend/codex.rs @@ -4634,16 +4634,31 @@ mod tests { /// Helper: spawn a real codex app-server process and wire it into /// `session_actor_loop`. Returns the command sender, response receiver, - /// and join handle. - fn setup_real_codex_test() -> ( + /// and join handle, or `None` when the `codex` binary isn't installed so + /// the caller can skip rather than fail. + /// + /// These tests are `#[ignore]`d, but the snapshot CI job runs every ignored + /// test in this binary via `--ignored` (see `scripts/snapshot-test`), which + /// sweeps these real-binary tests up too. Skipping on a missing binary keeps + /// that sweep green on runners without codex while still exercising the real + /// path locally when codex is present. + fn setup_real_codex_test() -> Option<( tokio_mpsc::Sender, mpsc::Receiver, tokio::task::JoinHandle<()>, - ) { + )> { let codex_binary = std::env::var("CODEX_BINARY").unwrap_or_else(|_| "codex".to_string()); - let mut child = spawn_codex(&codex_binary, &None) - .expect("Failed to spawn codex app-server — is codex installed?"); + let mut child = match spawn_codex(&codex_binary, &None) { + Ok(child) => child, + Err(e) => { + eprintln!( + "[test] skipping real codex test: cannot spawn `{codex_binary}`: {e} \ + (is codex installed?)" + ); + return None; + } + }; let stdin = child.stdin.take().expect("stdin piped"); let stdout = child.stdout.take().expect("stdout piped"); @@ -4693,13 +4708,15 @@ mod tests { .unwrap(); }); - (command_tx, response_rx, handle) + Some((command_tx, response_rx, handle)) } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore] // Requires `codex` binary on PATH async fn test_real_codex_streaming() { - let (command_tx, response_rx, handle) = setup_real_codex_test(); + let Some((command_tx, response_rx, handle)) = setup_real_codex_test() else { + return; + }; // Wait for at least one token (with a generous timeout for API calls) let mut got_token = false; @@ -4748,7 +4765,9 @@ mod tests { #[tokio::test(flavor = "multi_thread", worker_threads = 2)] #[ignore] // Requires `codex` binary on PATH async fn test_real_codex_turn_completes() { - let (command_tx, response_rx, handle) = setup_real_codex_test(); + let Some((command_tx, response_rx, handle)) = setup_real_codex_test() else { + return; + }; // Wait for turn to complete let mut got_turn_done = false;