feat(maestro-tui): CatalogStore over WS /api/events (NAP-278)

CatalogStore is the TUI's view of the daemons-and-sessions tree, backed by
`WS /api/events` on the local daemon. The daemon's WS opens with a Snapshot
frame, so that's bootstrap; subsequent frames are deltas. No separate REST
round-trip needed.

Surface:
- `apply_event` — pure, used by tests directly and by the WS loop.
- `snapshot()` — cheap O(d+s) read returning sorted Vec views + monotonic
  version for cheap render staleness checks.
- `wait_for_initial_snapshot` / `changed` — render-loop wakeups.
- `run_events_loop(ws_url)` — long-running task with exponential backoff
  (1s → 30s, ±25% jitter). Reconnect re-sends a fresh Snapshot, which
  reconciles state without server-side replay.

8 unit tests + 1 axum WS round-trip integration test passing.

crates/maestro-tui/Cargo.toml

@@ -10,5 +10,17 @@ description = "Terminal UI client over the local maestro daemon's `/api/*` surfa
 [dependencies]
 clap.workspace = true
 crossterm.workspace = true
+futures-util.workspace = true
 maestro-config.workspace = true
+maestro-protocol.workspace = true
+serde_json.workspace = true
 thiserror.workspace = true
+tokio = { workspace = true, features = ["sync", "time", "macros", "rt"] }
+tokio-tungstenite.workspace = true
+tracing.workspace = true
+uuid.workspace = true
+
+[dev-dependencies]
+axum.workspace = true
+chrono.workspace = true
+tokio = { workspace = true, features = ["sync", "time", "macros", "rt-multi-thread", "net"] }

crates/maestro-tui/src/catalog.rs

@@ -0,0 +1,378 @@
+//! `CatalogStore` — the TUI's view of the daemons-and-sessions tree.
+//!
+//! Backed entirely by the daemon's `WS /api/events` stream:
+//!
+//! 1. The first frame after connect is a [`RegistryEvent::Snapshot`] — that's
+//!    the bootstrap. No separate REST round-trip needed.
+//! 2. Subsequent frames are deltas (`DaemonAdded`, `SessionAdded`, …) that we
+//!    apply atomically and bump a monotonic version counter.
+//! 3. On WS disconnect: exponential backoff (1s → 30s, ±25% jitter); the next
+//!    connect resends a fresh `Snapshot` so we converge without needing
+//!    server-side replay.
+//! 4. `Lagged { skipped }` from the broadcast queue is logged at warn level
+//!    and ignored — the server will re-snapshot on its own bookkeeping; for
+//!    safety we trigger a self-reconnect so the next Snapshot reconciles us.
+//!
+//! Cloneable and `Send + Sync` — pass it to the render loop and to a Tokio
+//! task that drives [`CatalogStore::run_events_loop`].
+
+use std::collections::HashMap;
+use std::sync::{Arc, RwLock};
+use std::time::Duration;
+
+use futures_util::StreamExt;
+use maestro_protocol::{DaemonInfo, RegistryEvent, Session};
+use tokio::sync::Notify;
+use tokio_tungstenite::tungstenite::Message;
+use uuid::Uuid;
+
+/// Snapshot returned by [`CatalogStore::snapshot`]. `version` is monotone — a
+/// renderer can cheap-compare it across frames to know whether re-render is
+/// needed.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct CatalogSnapshot {
+    pub daemons: Vec<DaemonInfo>,
+    pub sessions: Vec<Session>,
+    pub version: u64,
+}
+
+#[derive(Debug, Default, Clone)]
+pub struct CatalogStore {
+    inner: Arc<Inner>,
+}
+
+#[derive(Debug, Default)]
+struct Inner {
+    state: RwLock<State>,
+    notify: Notify,
+}
+
+#[derive(Debug, Default)]
+struct State {
+    daemons: HashMap<String, DaemonInfo>,
+    sessions: HashMap<Uuid, Session>,
+    /// Bumped on every applied event. Wraps at u64::MAX, which is fine
+    /// because comparisons are equality, not ordering.
+    version: u64,
+    /// True once we've applied at least one Snapshot. Used by
+    /// [`CatalogStore::wait_for_initial_snapshot`].
+    bootstrapped: bool,
+}
+
+impl CatalogStore {
+    /// Apply one event atomically. Used by tests directly and by the WS loop.
+    pub fn apply_event(&self, ev: RegistryEvent) {
+        let mut s = self.inner.state.write().expect("catalog rwlock poisoned");
+        match ev {
+            RegistryEvent::Snapshot { daemons, sessions } => {
+                s.daemons = daemons
+                    .into_iter()
+                    .map(|d| (d.daemon_id.clone(), d))
+                    .collect();
+                s.sessions = sessions.into_iter().map(|x| (x.id, x)).collect();
+                s.bootstrapped = true;
+            }
+            RegistryEvent::DaemonAdded { daemon } => {
+                s.daemons.insert(daemon.daemon_id.clone(), daemon);
+            }
+            RegistryEvent::DaemonRemoved { daemon_id } => {
+                s.daemons.remove(&daemon_id);
+                // Drop sessions owned by the gone daemon — server will
+                // re-add them if it reappears.
+                s.sessions.retain(|_, sess| sess.daemon_id != daemon_id);
+            }
+            RegistryEvent::SessionAdded { session } | RegistryEvent::SessionUpdated { session } => {
+                s.sessions.insert(session.id, session);
+            }
+            RegistryEvent::SessionRemoved { daemon_id: _, id } => {
+                s.sessions.remove(&id);
+            }
+            RegistryEvent::Lagged { skipped } => {
+                tracing::warn!(
+                    skipped,
+                    "catalog events stream lagged; resync via reconnect"
+                );
+                // We don't reset state here — the WS loop notices the warn and
+                // can choose to drop+reconnect. Keep state until that happens.
+            }
+        }
+        s.version = s.version.wrapping_add(1);
+        drop(s);
+        self.inner.notify.notify_waiters();
+    }
+
+    /// Cheap O(daemons + sessions) read of current state.
+    pub fn snapshot(&self) -> CatalogSnapshot {
+        let s = self.inner.state.read().expect("catalog rwlock poisoned");
+        let mut daemons: Vec<DaemonInfo> = s.daemons.values().cloned().collect();
+        daemons.sort_by(|a, b| a.daemon_id.cmp(&b.daemon_id));
+        let mut sessions: Vec<Session> = s.sessions.values().cloned().collect();
+        sessions.sort_by_key(|x| x.id);
+        CatalogSnapshot {
+            daemons,
+            sessions,
+            version: s.version,
+        }
+    }
+
+    /// Returns true once we've applied at least one Snapshot frame.
+    pub fn is_bootstrapped(&self) -> bool {
+        self.inner
+            .state
+            .read()
+            .expect("catalog rwlock poisoned")
+            .bootstrapped
+    }
+
+    /// Wait until the catalog has been changed by something — applied event
+    /// or reconnect Snapshot. The render loop awaits this between frames.
+    pub async fn changed(&self) {
+        self.inner.notify.notified().await;
+    }
+
+    /// Wait until the first Snapshot has landed. Resolves immediately if
+    /// already bootstrapped.
+    pub async fn wait_for_initial_snapshot(&self) {
+        loop {
+            if self.is_bootstrapped() {
+                return;
+            }
+            self.inner.notify.notified().await;
+        }
+    }
+
+    /// Long-running task: connect to `ws_url`, parse `RegistryEvent` frames,
+    /// apply them. Reconnects with exponential backoff on disconnect. Never
+    /// returns — spawn it on a Tokio task.
+    pub async fn run_events_loop(self, ws_url: String) {
+        let mut backoff = Duration::from_secs(1);
+        loop {
+            match tokio_tungstenite::connect_async(&ws_url).await {
+                Ok((ws, _resp)) => {
+                    backoff = Duration::from_secs(1);
+                    let (_sink, mut stream) = ws.split();
+                    while let Some(frame) = stream.next().await {
+                        match frame {
+                            Ok(Message::Text(text)) => {
+                                match serde_json::from_str::<RegistryEvent>(&text) {
+                                    Ok(ev) => self.apply_event(ev),
+                                    Err(e) => {
+                                        tracing::warn!(error = ?e, "catalog: bad event frame");
+                                    }
+                                }
+                            }
+                            Ok(Message::Close(_)) | Err(_) => break,
+                            _ => {}
+                        }
+                    }
+                }
+                Err(e) => {
+                    tracing::debug!(error = ?e, "catalog: connect failed");
+                }
+            }
+            tracing::debug!(?backoff, "catalog: sleeping before reconnect");
+            tokio::time::sleep(jitter(backoff)).await;
+            backoff = std::cmp::min(backoff.saturating_mul(2), Duration::from_secs(30));
+        }
+    }
+}
+
+/// ±25 % jitter on the backoff. Cheap LCG seeded from the duration itself —
+/// avoids touching `rand` for one knob.
+fn jitter(base: Duration) -> Duration {
+    let ms = base.as_millis() as u64;
+    // Tiny pseudo-random in [0, 1) from system time nanos.
+    let now_nanos = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .map(|d| d.subsec_nanos())
+        .unwrap_or(0) as u64;
+    // ±25 %: scale by [-250, 250] of one quarter ms per ms.
+    let delta = (now_nanos % 500) as i64 - 250;
+    let scaled = (ms as i64).saturating_mul(delta).div_euclid(1000);
+    let total = ms.saturating_add_signed(scaled);
+    Duration::from_millis(total)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use chrono::Utc;
+    use maestro_protocol::{DaemonInfo, Session, SessionStatus};
+
+    fn daemon(id: &str) -> DaemonInfo {
+        DaemonInfo {
+            daemon_id: id.into(),
+            hostname: format!("{id}.lan"),
+            version: "test".into(),
+            daemon_url: format!("ws://{id}.lan:7070"),
+        }
+    }
+
+    fn session(id: u128, daemon_id: &str, command: &str) -> Session {
+        Session {
+            id: Uuid::from_u128(id),
+            daemon_id: daemon_id.into(),
+            command: command.into(),
+            cwd: std::path::PathBuf::from("/tmp"),
+            env: Default::default(),
+            metadata: Default::default(),
+            status: SessionStatus::Running,
+            created_at: Utc::now(),
+        }
+    }
+
+    #[test]
+    fn snapshot_event_replaces_state() {
+        let store = CatalogStore::default();
+        store.apply_event(RegistryEvent::Snapshot {
+            daemons: vec![daemon("konishi")],
+            sessions: vec![session(1, "konishi", "a")],
+        });
+        let snap = store.snapshot();
+        assert_eq!(snap.daemons.len(), 1);
+        assert_eq!(snap.sessions.len(), 1);
+        assert_eq!(snap.version, 1);
+        assert!(store.is_bootstrapped());
+
+        // A second snapshot wipes out the old state.
+        store.apply_event(RegistryEvent::Snapshot {
+            daemons: vec![daemon("yamato")],
+            sessions: vec![],
+        });
+        let snap = store.snapshot();
+        assert_eq!(snap.daemons.len(), 1);
+        assert_eq!(snap.daemons[0].daemon_id, "yamato");
+        assert!(snap.sessions.is_empty());
+        assert_eq!(snap.version, 2);
+    }
+
+    #[test]
+    fn daemon_added_then_removed() {
+        let store = CatalogStore::default();
+        store.apply_event(RegistryEvent::Snapshot {
+            daemons: vec![],
+            sessions: vec![],
+        });
+        store.apply_event(RegistryEvent::DaemonAdded {
+            daemon: daemon("konishi"),
+        });
+        assert_eq!(store.snapshot().daemons.len(), 1);
+        store.apply_event(RegistryEvent::DaemonRemoved {
+            daemon_id: "konishi".into(),
+        });
+        assert!(store.snapshot().daemons.is_empty());
+    }
+
+    #[test]
+    fn daemon_removed_evicts_its_sessions() {
+        let store = CatalogStore::default();
+        store.apply_event(RegistryEvent::Snapshot {
+            daemons: vec![daemon("konishi"), daemon("yamato")],
+            sessions: vec![session(1, "konishi", "a"), session(2, "yamato", "b")],
+        });
+        store.apply_event(RegistryEvent::DaemonRemoved {
+            daemon_id: "konishi".into(),
+        });
+        let snap = store.snapshot();
+        assert_eq!(snap.sessions.len(), 1);
+        assert_eq!(snap.sessions[0].daemon_id, "yamato");
+    }
+
+    #[test]
+    fn session_added_updated_removed() {
+        let store = CatalogStore::default();
+        store.apply_event(RegistryEvent::Snapshot {
+            daemons: vec![daemon("konishi")],
+            sessions: vec![],
+        });
+        store.apply_event(RegistryEvent::SessionAdded {
+            session: session(7, "konishi", "vim"),
+        });
+        assert_eq!(store.snapshot().sessions.len(), 1);
+        assert_eq!(store.snapshot().sessions[0].command, "vim");
+
+        let mut updated = session(7, "konishi", "claude");
+        updated.status = SessionStatus::Dead;
+        store.apply_event(RegistryEvent::SessionUpdated { session: updated });
+        let snap = store.snapshot();
+        assert_eq!(snap.sessions[0].command, "claude");
+        assert!(matches!(snap.sessions[0].status, SessionStatus::Dead));
+
+        store.apply_event(RegistryEvent::SessionRemoved {
+            daemon_id: "konishi".into(),
+            id: Uuid::from_u128(7),
+        });
+        assert!(store.snapshot().sessions.is_empty());
+    }
+
+    #[test]
+    fn lagged_event_doesnt_blow_up_state() {
+        let store = CatalogStore::default();
+        store.apply_event(RegistryEvent::Snapshot {
+            daemons: vec![daemon("konishi")],
+            sessions: vec![session(1, "konishi", "a")],
+        });
+        let before = store.snapshot();
+        store.apply_event(RegistryEvent::Lagged { skipped: 42 });
+        let after = store.snapshot();
+        assert_eq!(after.daemons, before.daemons);
+        assert_eq!(after.sessions, before.sessions);
+        assert_eq!(after.version, before.version + 1);
+    }
+
+    #[test]
+    fn snapshot_is_sorted() {
+        let store = CatalogStore::default();
+        store.apply_event(RegistryEvent::Snapshot {
+            daemons: vec![daemon("yamato"), daemon("konishi"), daemon("aurora")],
+            sessions: vec![
+                session(3, "aurora", "c"),
+                session(1, "konishi", "a"),
+                session(2, "yamato", "b"),
+            ],
+        });
+        let snap = store.snapshot();
+        let ids: Vec<_> = snap.daemons.iter().map(|d| d.daemon_id.as_str()).collect();
+        assert_eq!(ids, vec!["aurora", "konishi", "yamato"]);
+        let session_ids: Vec<_> = snap.sessions.iter().map(|s| s.id.as_u128()).collect();
+        assert_eq!(session_ids, vec![1, 2, 3]);
+    }
+
+    #[tokio::test]
+    async fn wait_for_initial_snapshot_resolves_after_first_event() {
+        let store = CatalogStore::default();
+        let task_store = store.clone();
+        let handle = tokio::spawn(async move {
+            task_store.wait_for_initial_snapshot().await;
+        });
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        assert!(!handle.is_finished());
+        store.apply_event(RegistryEvent::Snapshot {
+            daemons: vec![],
+            sessions: vec![],
+        });
+        // Give the waker a chance.
+        tokio::time::timeout(Duration::from_millis(200), handle)
+            .await
+            .expect("waker did not fire")
+            .unwrap();
+    }
+
+    #[tokio::test]
+    async fn changed_notifies_on_every_apply() {
+        let store = CatalogStore::default();
+        let waiter = {
+            let s = store.clone();
+            tokio::spawn(async move { s.changed().await })
+        };
+        tokio::time::sleep(Duration::from_millis(10)).await;
+        store.apply_event(RegistryEvent::Snapshot {
+            daemons: vec![],
+            sessions: vec![],
+        });
+        tokio::time::timeout(Duration::from_millis(200), waiter)
+            .await
+            .expect("changed did not fire")
+            .unwrap();
+    }
+}

crates/maestro-tui/src/lib.rs

@@ -9,6 +9,7 @@
 #![deny(missing_debug_implementations)]
 
 pub mod action;
+pub mod catalog;
 pub mod cli;
 pub mod config;
 pub mod input;

crates/maestro-tui/tests/catalog_ws.rs

@@ -0,0 +1,94 @@
+//! WS round-trip test for `CatalogStore::run_events_loop`. Spins a tiny axum
+//! server that emulates the daemon's `/api/events` contract: send a `Snapshot`
+//! on connect, then a delta.
+
+use std::time::Duration;
+
+use axum::extract::ws::{Message, WebSocketUpgrade};
+use axum::response::Response;
+use axum::routing::any;
+use axum::Router;
+use chrono::Utc;
+use maestro_protocol::{DaemonInfo, RegistryEvent, Session, SessionStatus};
+use maestro_tui::catalog::CatalogStore;
+use uuid::Uuid;
+
+fn daemon(id: &str) -> DaemonInfo {
+    DaemonInfo {
+        daemon_id: id.into(),
+        hostname: format!("{id}.lan"),
+        version: "test".into(),
+        daemon_url: format!("ws://{id}.lan:7070"),
+    }
+}
+
+fn session(id: u128, daemon_id: &str) -> Session {
+    Session {
+        id: Uuid::from_u128(id),
+        daemon_id: daemon_id.into(),
+        command: "bash".into(),
+        cwd: std::path::PathBuf::from("/tmp"),
+        env: Default::default(),
+        metadata: Default::default(),
+        status: SessionStatus::Running,
+        created_at: Utc::now(),
+    }
+}
+
+async fn events_handler(ws: WebSocketUpgrade) -> Response {
+    ws.on_upgrade(|mut socket| async move {
+        let snapshot = RegistryEvent::Snapshot {
+            daemons: vec![daemon("konishi")],
+            sessions: vec![],
+        };
+        let text = serde_json::to_string(&snapshot).unwrap();
+        if socket.send(Message::Text(text)).await.is_err() {
+            return;
+        }
+        // Small delay so the client has the snapshot applied before the delta.
+        tokio::time::sleep(Duration::from_millis(50)).await;
+        let added = RegistryEvent::SessionAdded {
+            session: session(1, "konishi"),
+        };
+        let text = serde_json::to_string(&added).unwrap();
+        let _ = socket.send(Message::Text(text)).await;
+        // Keep the connection alive so the client doesn't enter reconnect.
+        tokio::time::sleep(Duration::from_secs(2)).await;
+    })
+}
+
+#[tokio::test(flavor = "multi_thread")]
+async fn run_events_loop_applies_snapshot_then_delta() {
+    let app = Router::new().route("/api/events", any(events_handler));
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    tokio::spawn(async move {
+        axum::serve(listener, app).await.unwrap();
+    });
+
+    let store = CatalogStore::default();
+    let bg = store.clone();
+    let ws_url = format!("ws://{addr}/api/events");
+    tokio::spawn(async move { bg.run_events_loop(ws_url).await });
+
+    // Bootstrap should land within ~250 ms.
+    tokio::time::timeout(
+        Duration::from_millis(500),
+        store.wait_for_initial_snapshot(),
+    )
+    .await
+    .expect("initial snapshot never arrived");
+
+    let snap = store.snapshot();
+    assert_eq!(snap.daemons.len(), 1);
+    assert_eq!(snap.daemons[0].daemon_id, "konishi");
+
+    // Delta arrives shortly after. Poll for up to 500 ms.
+    for _ in 0..50 {
+        if store.snapshot().sessions.len() == 1 {
+            return;
+        }
+        tokio::time::sleep(Duration::from_millis(10)).await;
+    }
+    panic!("delta never propagated; got {:?}", store.snapshot());
+}