feat(maestro): daemon WS /api/events for live catalog updates (NAP-248)

New module `local_events::LocalEventBus` (broadcast<RegistryEvent>) with
two publishers:

* `spawn_session_publisher` — translates SessionManager's own session
  events into the matching `RegistryEvent` variants (with daemon_id
  filled in for the SessionRemoved case).
* `CatalogMirror::apply` — re-publishes peer events after applying them
  to the local mirror, so subscribers see the post-apply state.

New `WS /api/events` route on the daemon: subscribes to the bus, sends
an initial `Snapshot` built from manager.snapshot() ∪ catalog state,
then forwards live deltas. `Lagged` follows the registry's convention
(client resyncs via REST).

Revert the REST-seed hack in web/src/events.ts — the WS endpoint now
exists, so `CatalogStore::start()` can go back to just `connect()`.

E2E: subscribe to a daemon's events WS, expect Snapshot then a
SessionAdded after POSTing /api/sessions.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

crates/maestro-daemon/src/catalog_mirror.rs

@@ -27,6 +27,10 @@ struct State {
 #[derive(Debug, Clone, Default)]
 pub struct CatalogMirror {
     inner: Arc<Mutex<State>>,
+    /// Optional fan-out (NAP-248). When attached, every applied event is
+    /// re-published here so `WS /api/events` subscribers see peer state
+    /// changes the same way the registry's clients do.
+    bus: Option<crate::local_events::LocalEventBus>,
 }
 
 impl CatalogMirror {
@@ -34,40 +38,59 @@ impl CatalogMirror {
         Self::default()
     }
 
+    /// Attach a [`LocalEventBus`](crate::local_events::LocalEventBus) so peer
+    /// catalog deltas reach the daemon's `/api/events` subscribers.
+    pub fn with_bus(mut self, bus: crate::local_events::LocalEventBus) -> Self {
+        self.bus = Some(bus);
+        self
+    }
+
     /// Apply one delta. Snapshot events replace the mirror wholesale;
     /// `Lagged` resets it on the assumption that the next full snapshot will
     /// arrive later (today the registry doesn't re-send one — see follow-up
     /// notes in `handler.rs`).
     pub async fn apply(&self, event: RegistryEvent) {
-        let mut s = self.inner.lock().await;
-        match event {
-            RegistryEvent::Snapshot { daemons, sessions } => {
-                s.daemons.clear();
-                s.sessions.clear();
-                for d in daemons {
-                    s.daemons.insert(d.daemon_id.clone(), d);
+        {
+            let mut s = self.inner.lock().await;
+            match &event {
+                RegistryEvent::Snapshot { daemons, sessions } => {
+                    s.daemons.clear();
+                    s.sessions.clear();
+                    for d in daemons {
+                        s.daemons.insert(d.daemon_id.clone(), d.clone());
+                    }
+                    for sess in sessions {
+                        s.sessions
+                            .insert((sess.daemon_id.clone(), sess.id), sess.clone());
+                    }
                 }
-                for sess in sessions {
-                    s.sessions.insert((sess.daemon_id.clone(), sess.id), sess);
+                RegistryEvent::DaemonAdded { daemon } => {
+                    s.daemons.insert(daemon.daemon_id.clone(), daemon.clone());
+                }
+                RegistryEvent::DaemonRemoved { daemon_id } => {
+                    s.daemons.remove(daemon_id);
+                    s.sessions.retain(|(d, _), _| d != daemon_id);
+                }
+                RegistryEvent::SessionAdded { session }
+                | RegistryEvent::SessionUpdated { session } => {
+                    s.sessions
+                        .insert((session.daemon_id.clone(), session.id), session.clone());
+                }
+                RegistryEvent::SessionRemoved { daemon_id, id } => {
+                    s.sessions.remove(&(daemon_id.clone(), *id));
+                }
+                RegistryEvent::Lagged { skipped } => {
+                    tracing::warn!(
+                        skipped = *skipped,
+                        "catalog mirror lagged — local view may be stale"
+                    );
                 }
             }
-            RegistryEvent::DaemonAdded { daemon } => {
-                s.daemons.insert(daemon.daemon_id.clone(), daemon);
-            }
-            RegistryEvent::DaemonRemoved { daemon_id } => {
-                s.daemons.remove(&daemon_id);
-                s.sessions.retain(|(d, _), _| d != &daemon_id);
-            }
-            RegistryEvent::SessionAdded { session } | RegistryEvent::SessionUpdated { session } => {
-                s.sessions
-                    .insert((session.daemon_id.clone(), session.id), session);
-            }
-            RegistryEvent::SessionRemoved { daemon_id, id } => {
-                s.sessions.remove(&(daemon_id, id));
-            }
-            RegistryEvent::Lagged { skipped } => {
-                tracing::warn!(skipped, "catalog mirror lagged — local view may be stale");
-            }
+        }
+        // Re-publish AFTER releasing the lock so subscribers reading `daemons()`
+        // / `sessions()` in response observe the new state.
+        if let Some(bus) = &self.bus {
+            bus.publish(event);
         }
     }
 

crates/maestro-daemon/src/lib.rs

@@ -13,6 +13,7 @@ pub mod config;
 pub mod configure;
 pub mod jwt;
 pub mod local_api;
+pub mod local_events;
 pub mod pty;
 pub mod registry_client;
 pub mod session;
@@ -25,5 +26,6 @@ pub use catalog_mirror::CatalogMirror;
 pub use config::{default_config_path, default_state_dir, load_config, DaemonConfig, EnvOverrides};
 pub use configure::{run as run_configure, ConfigureError, ConfigureInputs};
 pub use jwt::{issue as issue_jwt, JwtError, JwtVerifier, PairAttachVerifier};
+pub use local_events::{spawn_session_publisher, LocalEventBus};
 pub use registry_client::{ClientError, RegistryClient, HEARTBEAT_INTERVAL};
 pub use session::{SessionError, SessionEvent, SessionManager};

crates/maestro-daemon/src/local_api.rs

@@ -33,6 +33,7 @@ use serde::Serialize;
 use uuid::Uuid;
 
 use crate::catalog_mirror::CatalogMirror;
+use crate::local_events::LocalEventBus;
 use crate::session::SessionManager;
 
 /// Daemon identity returned by `GET /api/web/info`. Browsers render this in the
@@ -54,6 +55,9 @@ pub struct LocalApiState {
     /// Peer-only mirror of the global catalog. Empty until the daemon's
     /// registry client connects and receives the snapshot.
     pub catalog: CatalogMirror,
+    /// Daemon-local fan-out of own + peer catalog deltas (NAP-248). The
+    /// `/api/events` WS subscribes here; everything else just reads REST.
+    pub events: LocalEventBus,
 }
 
 pub fn router(state: LocalApiState) -> Router {
@@ -62,6 +66,7 @@ pub fn router(state: LocalApiState) -> Router {
         .route("/api/sessions/:id", delete(kill_session))
         .route("/api/daemons", get(list_daemons))
         .route("/api/web/info", get(web_info))
+        .route("/api/events", any(events_ws))
         .with_state(state)
 }
 
@@ -166,15 +171,90 @@ pub fn state_for(
     hostname: impl Into<String>,
     daemon_url: impl Into<String>,
     catalog: CatalogMirror,
+    events: LocalEventBus,
 ) -> LocalApiState {
     LocalApiState {
         manager,
         hostname: Arc::new(hostname.into()),
         daemon_url: Arc::new(daemon_url.into()),
         catalog,
+        events,
     }
 }
 
+// ---------------------------------------------------------------------------
+// WS /api/events — live catalog deltas for browser/CLI subscribers (NAP-248)
+// ---------------------------------------------------------------------------
+
+async fn events_ws(
+    State(s): State<LocalApiState>,
+    ws: axum::extract::WebSocketUpgrade,
+) -> Response {
+    ws.on_upgrade(move |socket| async move {
+        let _ = pump_events(socket, s).await;
+    })
+}
+
+async fn pump_events(
+    socket: axum::extract::ws::WebSocket,
+    s: LocalApiState,
+) -> Result<(), axum::Error> {
+    use axum::extract::ws::Message;
+    use futures_util::{SinkExt, StreamExt};
+    use maestro_protocol::RegistryEvent;
+    use tokio::sync::broadcast::error::RecvError;
+
+    let (mut sink, mut stream) = socket.split();
+
+    // Subscribe BEFORE building the snapshot so deltas applied while we're
+    // marshaling can't slip between the two and disappear. The UI is
+    // idempotent for repeated *Added events, so any overlap is harmless.
+    let mut rx = s.events.subscribe();
+
+    let self_info = DaemonInfo {
+        daemon_id: s.manager.daemon_id().to_string(),
+        hostname: (*s.hostname).clone(),
+        version: PROTOCOL_VERSION.into(),
+        daemon_url: (*s.daemon_url).clone(),
+    };
+    let mut daemons = vec![self_info];
+    daemons.extend(s.catalog.daemons().await);
+    let mut sessions = s.manager.snapshot().await;
+    sessions.extend(s.catalog.sessions().await);
+    let snapshot = RegistryEvent::Snapshot { daemons, sessions };
+    let text = serde_json::to_string(&snapshot).expect("snapshot encode");
+    if sink.send(Message::Text(text)).await.is_err() {
+        return Ok(());
+    }
+
+    loop {
+        tokio::select! {
+            ev = rx.recv() => {
+                let event = match ev {
+                    Ok(e) => e,
+                    Err(RecvError::Lagged(n)) => RegistryEvent::Lagged { skipped: n },
+                    Err(RecvError::Closed) => break,
+                };
+                let text = serde_json::to_string(&event).expect("event encode");
+                if sink.send(Message::Text(text)).await.is_err() {
+                    break;
+                }
+            }
+            frame = stream.next() => {
+                // Server-push only. Anything from the client either closes or
+                // is ignored.
+                match frame {
+                    None => break,
+                    Some(Err(_)) => break,
+                    Some(Ok(Message::Close(_))) => break,
+                    Some(Ok(_)) => {}
+                }
+            }
+        }
+    }
+    Ok(())
+}
+
 // ---------------------------------------------------------------------------
 // Broker WS — `WS /api/attach/:session_id`
 // ---------------------------------------------------------------------------

crates/maestro-daemon/src/local_events.rs

@@ -0,0 +1,124 @@
+//! Daemon-local fan-out of `RegistryEvent`s for the `WS /api/events`
+//! endpoint (NAP-248).
+//!
+//! Two sources feed this bus:
+//!
+//! 1. **Own-session events** — the daemon's `SessionManager::events()` stream,
+//!    translated from `SessionEvent` into the matching `RegistryEvent` variant
+//!    by [`spawn_session_publisher`].
+//! 2. **Peer-catalog events** — applied to the daemon's [`CatalogMirror`] when
+//!    the registry fans them out (NAP-246). `CatalogMirror::apply` publishes
+//!    each event here after mutating its local view.
+//!
+//! Subscribers get the merged stream. The `WS /api/events` handler builds a
+//! one-shot `Snapshot` of own ∪ peer state on connect, then forwards live
+//! deltas until the client disconnects. `Lagged` follows the registry's
+//! existing convention — clients resync via REST.
+
+use std::sync::Arc;
+
+use maestro_protocol::RegistryEvent;
+use tokio::sync::broadcast;
+
+use crate::session::{SessionEvent, SessionManager};
+
+/// Capacity of the per-bus broadcast channel. Matches the registry's
+/// `EVENT_CHANNEL_CAPACITY` so a slow client behaves the same on either side.
+const BUS_CAPACITY: usize = 256;
+
+/// Cheap-to-clone handle around a `broadcast::Sender<RegistryEvent>`.
+#[derive(Debug, Clone)]
+pub struct LocalEventBus {
+    sender: Arc<broadcast::Sender<RegistryEvent>>,
+}
+
+impl LocalEventBus {
+    pub fn new() -> Self {
+        let (sender, _) = broadcast::channel(BUS_CAPACITY);
+        Self {
+            sender: Arc::new(sender),
+        }
+    }
+
+    /// Push an event onto the bus. No-op if there are no subscribers.
+    pub fn publish(&self, event: RegistryEvent) {
+        let _ = self.sender.send(event);
+    }
+
+    pub fn subscribe(&self) -> broadcast::Receiver<RegistryEvent> {
+        self.sender.subscribe()
+    }
+}
+
+impl Default for LocalEventBus {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Spawn a task that converts each `SessionEvent` from `manager` into the
+/// matching `RegistryEvent` and publishes it on `bus`. Exits when the
+/// `SessionManager` is dropped (the events channel closes).
+pub fn spawn_session_publisher(manager: SessionManager, bus: LocalEventBus) {
+    let daemon_id = manager.daemon_id().to_string();
+    tokio::spawn(async move {
+        let mut rx = manager.events();
+        loop {
+            match rx.recv().await {
+                Ok(SessionEvent::Added(session)) => {
+                    bus.publish(RegistryEvent::SessionAdded { session });
+                }
+                Ok(SessionEvent::Updated(session)) => {
+                    bus.publish(RegistryEvent::SessionUpdated { session });
+                }
+                Ok(SessionEvent::Removed(id)) => {
+                    bus.publish(RegistryEvent::SessionRemoved {
+                        daemon_id: daemon_id.clone(),
+                        id,
+                    });
+                }
+                Err(broadcast::error::RecvError::Lagged(skipped)) => {
+                    bus.publish(RegistryEvent::Lagged { skipped });
+                }
+                Err(broadcast::error::RecvError::Closed) => return,
+            }
+        }
+    });
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn publish_visible_to_subscriber() {
+        let bus = LocalEventBus::new();
+        let mut rx = bus.subscribe();
+        bus.publish(RegistryEvent::Lagged { skipped: 7 });
+        match rx.recv().await {
+            Ok(RegistryEvent::Lagged { skipped }) => assert_eq!(skipped, 7),
+            other => panic!("unexpected: {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn no_subscribers_swallows_event() {
+        let bus = LocalEventBus::new();
+        // No one's listening — `publish` must not panic and must not block.
+        bus.publish(RegistryEvent::DaemonRemoved {
+            daemon_id: "ghost".into(),
+        });
+    }
+
+    #[tokio::test]
+    async fn clones_share_state() {
+        let a = LocalEventBus::new();
+        let b = a.clone();
+        let mut rx = a.subscribe();
+        b.publish(RegistryEvent::Lagged { skipped: 3 });
+        match rx.recv().await {
+            Ok(RegistryEvent::Lagged { skipped }) => assert_eq!(skipped, 3),
+            other => panic!("unexpected: {other:?}"),
+        }
+    }
+}

crates/maestro-daemon/src/main.rs

@@ -15,7 +15,8 @@ use maestro_daemon::tmux::Tmux;
 use maestro_daemon::worktree::WorktreeManager;
 use maestro_daemon::{
     attach, default_config_path, default_state_dir, load_config, local_api, run_configure,
-    ConfigureInputs, EnvOverrides, JwtVerifier, PairAttachVerifier, RegistryClient, SessionManager,
+    spawn_session_publisher, ConfigureInputs, EnvOverrides, JwtVerifier, LocalEventBus,
+    PairAttachVerifier, RegistryClient, SessionManager,
 };
 use tokio::net::TcpListener;
 use tracing_subscriber::EnvFilter;
@@ -139,15 +140,22 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
         config.allow_inbound.iter().cloned(),
     );
 
+    // Daemon-local event bus (NAP-248): own session events + peer catalog
+    // deltas funnel through here for `WS /api/events` subscribers.
+    let events = LocalEventBus::new();
+    spawn_session_publisher(manager.clone(), events.clone());
+
     // Build the registry client first so we can hand its catalog mirror to
-    // the local-API state — both halves share the same Arc<Mutex<…>>.
-    let client = RegistryClient::new(
+    // the local-API state — both halves share the same Arc<Mutex<…>>. Hand
+    // the event bus to the catalog mirror so peer deltas fan out too.
+    let client = RegistryClient::new_with_bus(
         config.registry_url.clone(),
         config.daemon_id.clone(),
         config.hostname.clone(),
         config.daemon_url.clone(),
         config.token.clone(),
         manager.clone(),
+        Some(events.clone()),
     )
     .with_perimeter(config.perimeter_id.clone(), config.private_addr.clone());
     let catalog = client.catalog();
@@ -157,6 +165,7 @@ async fn run(config_override: Option<PathBuf>) -> Result<(), Box<dyn Error>> {
         config.hostname.clone(),
         config.daemon_url.clone(),
         catalog.clone(),
+        events,
     );
     let broker_state = local_api::BrokerState {
         manager: manager.clone(),

crates/maestro-daemon/src/registry_client.rs

@@ -74,8 +74,34 @@ impl RegistryClient {
         daemon_url: impl Into<String>,
         token: impl Into<String>,
         manager: SessionManager,
+    ) -> Self {
+        Self::new_with_bus(
+            registry_url,
+            daemon_id,
+            hostname,
+            daemon_url,
+            token,
+            manager,
+            None,
+        )
+    }
+
+    /// Variant that wires a [`LocalEventBus`](crate::LocalEventBus) into the
+    /// catalog mirror so peer deltas reach `WS /api/events` (NAP-248).
+    pub fn new_with_bus(
+        registry_url: impl Into<String>,
+        daemon_id: impl Into<String>,
+        hostname: impl Into<String>,
+        daemon_url: impl Into<String>,
+        token: impl Into<String>,
+        manager: SessionManager,
+        bus: Option<crate::LocalEventBus>,
     ) -> Self {
         let (broker, broker_rx) = BrokerHandle::new();
+        let catalog = match &bus {
+            Some(b) => CatalogMirror::new().with_bus(b.clone()),
+            None => CatalogMirror::new(),
+        };
         Self {
             registry_url: registry_url.into(),
             daemon_id: daemon_id.into(),
@@ -85,7 +111,7 @@ impl RegistryClient {
             perimeter_id: None,
             private_addr: None,
             manager,
-            catalog: CatalogMirror::new(),
+            catalog,
             broker,
             broker_rx,
         }

crates/maestro-daemon/tests/local_api.rs

@@ -11,7 +11,7 @@ use std::time::Duration;
 
 use common::DaemonFixture;
 use futures_util::{SinkExt, StreamExt};
-use maestro_daemon::{local_api, BrokerHandle, CatalogMirror};
+use maestro_daemon::{local_api, BrokerHandle, CatalogMirror, LocalEventBus};
 use maestro_protocol::{CreateSessionRequest, DaemonInfo, RegistryEvent, Session};
 use tokio::net::TcpListener;
 use tokio::sync::oneshot;
@@ -33,7 +33,14 @@ async fn spawn(daemon_id: &str, hostname: &str, daemon_url: &str) -> Harness {
     let fx = DaemonFixture::new(daemon_id);
     let manager = fx.manager.clone();
     let catalog = CatalogMirror::new();
-    let state = local_api::state_for(manager.clone(), hostname, daemon_url, catalog.clone());
+    let events = LocalEventBus::new();
+    let state = local_api::state_for(
+        manager.clone(),
+        hostname,
+        daemon_url,
+        catalog.clone(),
+        events,
+    );
     // NAP-247: tests don't exercise the cross-host broker path; hand the
     // router an isolated BrokerHandle whose receiver is dropped, so any
     // resolve_attach attempt fails fast with NoControlLink.

crates/maestro-e2e/tests/e2e.rs

@@ -642,3 +642,69 @@ async fn broker_rejects_when_target_doesnt_allow_inbound() {
     drop(beta);
     drop(registry);
 }
+
+#[tokio::test(flavor = "multi_thread")]
+async fn daemon_events_ws_streams_snapshot_then_session_added() {
+    // NAP-248: subscribe to a daemon's WS /api/events, expect an initial
+    // Snapshot, then a SessionAdded after we spawn a session via REST.
+    let (_reg, daemons, client) = bring_up(&["alpha"]).await;
+    let cwd = TempDir::new().unwrap();
+
+    let alpha_addr = daemons[0].bind_addr;
+    let ws_url = format!("ws://{alpha_addr}/api/events");
+    let (mut ws, _) = tokio_tungstenite::connect_async(ws_url)
+        .await
+        .expect("events WS handshake");
+
+    // First frame must be a Snapshot (the WS sends it eagerly on connect).
+    let snapshot = next_event(&mut ws).await;
+    match &snapshot {
+        maestro_protocol::RegistryEvent::Snapshot { daemons, sessions } => {
+            assert!(daemons.iter().any(|d| d.daemon_id == "alpha"));
+            assert!(sessions.is_empty());
+        }
+        other => panic!("expected initial Snapshot, got {other:?}"),
+    }
+
+    let session = client
+        .spawn_session(&create_req("alpha", cwd.path(), "sleep 30"))
+        .await
+        .expect("spawn alpha");
+
+    // SessionAdded must show up on the WS within a reasonable window.
+    let saw_added = tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            match next_event(&mut ws).await {
+                maestro_protocol::RegistryEvent::SessionAdded { session: s }
+                    if s.id == session.id =>
+                {
+                    return true;
+                }
+                _ => continue,
+            }
+        }
+    })
+    .await
+    .unwrap_or(false);
+    assert!(
+        saw_added,
+        "events WS never delivered SessionAdded for own session"
+    );
+
+    client.kill_session(session.id).await.expect("kill");
+}
+
+async fn next_event<S>(ws: &mut S) -> maestro_protocol::RegistryEvent
+where
+    S: StreamExt<Item = Result<WsMessage, tokio_tungstenite::tungstenite::Error>> + Unpin,
+{
+    loop {
+        match ws.next().await {
+            Some(Ok(WsMessage::Text(t))) => {
+                return serde_json::from_str(&t).expect("RegistryEvent")
+            }
+            Some(Ok(_)) => continue,
+            other => panic!("WS closed waiting for event: {other:?}"),
+        }
+    }
+}