diff --git a/crates/maestro-daemon/src/cli.rs b/crates/maestro-daemon/src/cli.rs index fb7fa7b..542440b 100644 --- a/crates/maestro-daemon/src/cli.rs +++ b/crates/maestro-daemon/src/cli.rs @@ -12,8 +12,7 @@ use crate::sidecar::SidecarStore; use crate::tmux::Tmux; use crate::worktree::WorktreeManager; use crate::{ - default_config_path, default_state_dir, load_config, local_api, spawn_session_publisher, - EnvOverrides, LocalEventBus, SessionManager, + default_config_path, default_state_dir, load_config, local_api, EnvOverrides, SessionManager, }; #[derive(Debug, Subcommand)] @@ -72,15 +71,11 @@ async fn run(config_override: Option) -> Result<(), Box> { tracing::info!(recovered, "adopted existing tmux sessions"); manager.start_liveness_watcher(DEFAULT_LIVENESS_INTERVAL); - let events = LocalEventBus::new(); - spawn_session_publisher(manager.clone(), events.clone()); - let router = local_api::router(local_api::state_for( manager, config.hostname.clone(), config.home_dir.clone(), config.spawn_presets.clone(), - events, )); let listener = TcpListener::bind(config.bind_addr).await?; diff --git a/crates/maestro-daemon/src/lib.rs b/crates/maestro-daemon/src/lib.rs index 7dd2b81..f40ca02 100644 --- a/crates/maestro-daemon/src/lib.rs +++ b/crates/maestro-daemon/src/lib.rs @@ -14,7 +14,6 @@ pub mod config; pub mod dirlist; pub mod gitroot; pub mod local_api; -pub mod local_events; pub mod panes; pub mod proc; pub mod pty; @@ -25,5 +24,4 @@ pub mod tmux; pub mod worktree; pub use config::{default_config_path, default_state_dir, load_config, DaemonConfig, EnvOverrides}; -pub use local_events::{spawn_session_publisher, LocalEventBus}; -pub use session::{SessionError, SessionEvent, SessionManager}; +pub use session::{SessionError, SessionManager}; diff --git a/crates/maestro-daemon/src/local_api.rs b/crates/maestro-daemon/src/local_api.rs index 8c4b672..926e048 100644 --- a/crates/maestro-daemon/src/local_api.rs +++ b/crates/maestro-daemon/src/local_api.rs @@ -25,7 +25,6 @@ use maestro_protocol::{ use serde::Deserialize; use uuid::Uuid; -use crate::local_events::LocalEventBus; use crate::session::SessionManager; #[derive(Debug, Clone)] @@ -37,9 +36,6 @@ pub struct LocalApiState { pub home_dir: Arc, /// Resolved spawn presets (#22/#26) advertised in `GET /api/info`. pub presets: Arc>, - /// Fan-out of session deltas (NAP-248). The `/api/events` WS subscribes - /// here; everything else just reads REST. - pub events: LocalEventBus, } pub fn router(state: LocalApiState) -> Router { @@ -161,14 +157,12 @@ pub fn state_for( hostname: impl Into, home_dir: impl Into, presets: Vec, - events: LocalEventBus, ) -> LocalApiState { LocalApiState { manager, hostname: Arc::new(hostname.into()), home_dir: Arc::new(home_dir.into()), presets: Arc::new(presets), - events, } } @@ -221,7 +215,7 @@ async fn pump_events( // 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 mut rx = s.manager.events(); let snapshot = CatalogEvent::Snapshot { sessions: s.manager.snapshot().await, diff --git a/crates/maestro-daemon/src/local_events.rs b/crates/maestro-daemon/src/local_events.rs deleted file mode 100644 index bac1aca..0000000 --- a/crates/maestro-daemon/src/local_events.rs +++ /dev/null @@ -1,109 +0,0 @@ -//! Fan-out of session deltas for the `WS /api/events` endpoint (NAP-248). -//! -//! [`spawn_session_publisher`] translates the `SessionManager::events()` stream -//! from `SessionEvent` into the matching [`CatalogEvent`] variant and publishes -//! it here. The `WS /api/events` handler sends a one-shot `Snapshot` on connect, -//! then forwards live deltas until the client disconnects. `Lagged` tells -//! clients to resync via REST. - -use std::sync::Arc; - -use maestro_protocol::CatalogEvent; -use tokio::sync::broadcast; - -use crate::session::{SessionEvent, SessionManager}; - -/// Capacity of the per-bus broadcast channel. Sourced from the shared -/// [`maestro_config::BroadcastTuning`] (NAP-273). -const BUS_CAPACITY: usize = maestro_config::BroadcastTuning::DEFAULT.events_capacity; - -/// Cheap-to-clone handle around a `broadcast::Sender`. -#[derive(Debug, Clone)] -pub struct LocalEventBus { - sender: Arc>, -} - -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: CatalogEvent) { - let _ = self.sender.send(event); - } - - pub fn subscribe(&self) -> broadcast::Receiver { - self.sender.subscribe() - } -} - -impl Default for LocalEventBus { - fn default() -> Self { - Self::new() - } -} - -/// Spawn a task that converts each `SessionEvent` from `manager` into the -/// matching `CatalogEvent` and publishes it on `bus`. Exits when the -/// `SessionManager` is dropped (the events channel closes). -pub fn spawn_session_publisher(manager: SessionManager, bus: LocalEventBus) { - tokio::spawn(async move { - let mut rx = manager.events(); - loop { - match rx.recv().await { - Ok(SessionEvent::Added(session)) => { - bus.publish(CatalogEvent::SessionAdded { session }); - } - Ok(SessionEvent::Updated(session)) => { - bus.publish(CatalogEvent::SessionUpdated { session }); - } - Ok(SessionEvent::Removed(id)) => { - bus.publish(CatalogEvent::SessionRemoved { id }); - } - Err(broadcast::error::RecvError::Lagged(skipped)) => { - bus.publish(CatalogEvent::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(CatalogEvent::Lagged { skipped: 7 }); - match rx.recv().await { - Ok(CatalogEvent::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(CatalogEvent::Lagged { skipped: 1 }); - } - - #[tokio::test] - async fn clones_share_state() { - let a = LocalEventBus::new(); - let b = a.clone(); - let mut rx = a.subscribe(); - b.publish(CatalogEvent::Lagged { skipped: 3 }); - match rx.recv().await { - Ok(CatalogEvent::Lagged { skipped }) => assert_eq!(skipped, 3), - other => panic!("unexpected: {other:?}"), - } - } -} diff --git a/crates/maestro-daemon/src/session.rs b/crates/maestro-daemon/src/session.rs index 75718b1..b07fc72 100644 --- a/crates/maestro-daemon/src/session.rs +++ b/crates/maestro-daemon/src/session.rs @@ -12,7 +12,8 @@ use std::time::Duration; use chrono::Utc; use maestro_protocol::{ - ActivityState, CreateSessionRequest, ErrorInfo, Session, SessionStatus, WorktreeSpec, + ActivityState, CatalogEvent, CreateSessionRequest, ErrorInfo, Session, SessionStatus, + WorktreeSpec, }; use tokio::sync::{broadcast, Mutex}; use uuid::Uuid; @@ -34,13 +35,6 @@ pub const EVENTS_CAPACITY: usize = maestro_config::BroadcastTuning::DEFAULT.even /// How often the liveness watcher polls tmux for sessions that disappeared. pub const DEFAULT_LIVENESS_INTERVAL: Duration = Duration::from_secs(2); -#[derive(Debug, Clone)] -pub enum SessionEvent { - Added(Session), - Removed(Uuid), - Updated(Session), -} - /// Per-session daemon-local bookkeeping that complements the public `Session`. #[derive(Debug)] pub struct SessionRecord { @@ -106,7 +100,7 @@ struct Inner { /// but disabled (`0`-day) history no-ops internally. spawn_history: Option, sessions: Mutex>, - events: broadcast::Sender, + events: broadcast::Sender, } impl SessionManager { @@ -173,7 +167,9 @@ impl SessionManager { .unwrap_or_default() } - pub fn events(&self) -> broadcast::Receiver { + /// Subscribe to session deltas. Each subscriber gets its own lag domain — + /// a slow client makes only itself resync. + pub fn events(&self) -> broadcast::Receiver { self.inner.events.subscribe() } @@ -306,7 +302,9 @@ impl SessionManager { }, ); } - let _ = self.inner.events.send(SessionEvent::Added(session.clone())); + let _ = self.inner.events.send(CatalogEvent::SessionAdded { + session: session.clone(), + }); // Best-effort: log the spawn for the recents picker (#24). A log-write // failure must never fail an otherwise-successful spawn, so the error @@ -437,7 +435,10 @@ impl SessionManager { bridge, }, ); - let _ = self.inner.events.send(SessionEvent::Added(session)); + let _ = self + .inner + .events + .send(CatalogEvent::SessionAdded { session }); adopted += 1; } Ok(adopted) @@ -574,7 +575,7 @@ async fn run_activity_pass(inner: &Arc) { } }; if let Some(session) = updated { - let _ = inner.events.send(SessionEvent::Updated(session)); + let _ = inner.events.send(CatalogEvent::SessionUpdated { session }); } } } @@ -617,5 +618,5 @@ async fn cleanup_record(inner: &Arc, id: Uuid, record: SessionRecord, kil let _ = inner.worktree_mgr.remove(base, path); } } - let _ = inner.events.send(SessionEvent::Removed(id)); + let _ = inner.events.send(CatalogEvent::SessionRemoved { id }); } diff --git a/crates/maestro-daemon/tests/attach.rs b/crates/maestro-daemon/tests/attach.rs index dd4469a..bcfec11 100644 --- a/crates/maestro-daemon/tests/attach.rs +++ b/crates/maestro-daemon/tests/attach.rs @@ -11,7 +11,7 @@ use std::time::Duration; use common::DaemonFixture; use futures_util::{SinkExt, StreamExt}; -use maestro_daemon::{local_api, LocalEventBus, SessionManager}; +use maestro_daemon::{local_api, SessionManager}; use maestro_protocol::CreateSessionRequest; use tempfile::TempDir; use tokio::net::TcpListener; @@ -37,7 +37,6 @@ async fn spawn_daemon_ws() -> Harness { "test-host", "/home/test", Vec::new(), - LocalEventBus::new(), )); let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) .await diff --git a/crates/maestro-daemon/tests/local_api.rs b/crates/maestro-daemon/tests/local_api.rs index 7bdd2c0..c1be910 100644 --- a/crates/maestro-daemon/tests/local_api.rs +++ b/crates/maestro-daemon/tests/local_api.rs @@ -8,7 +8,7 @@ use std::time::Duration; use common::DaemonFixture; use futures_util::{SinkExt, StreamExt}; -use maestro_daemon::{local_api, LocalEventBus}; +use maestro_daemon::local_api; use maestro_protocol::{CreateSessionRequest, Session}; use tokio::net::TcpListener; use tokio::sync::oneshot; @@ -28,7 +28,6 @@ struct Harness { async fn spawn(hostname: &str) -> Harness { let fx = DaemonFixture::new(); let manager = fx.manager.clone(); - let events = LocalEventBus::new(); // Built-in presets (claude/opencode/zsh + custom) so `/api/info` advertises // a known set (#26). let presets = maestro_config::SpawnSection::default().resolve().unwrap(); @@ -37,7 +36,6 @@ async fn spawn(hostname: &str) -> Harness { hostname, "/home/test", presets, - events, )); let listener = TcpListener::bind(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))) .await @@ -63,6 +61,10 @@ fn url(addr: SocketAddr, path: &str) -> String { format!("http://{addr}{path}") } +fn ws_url(addr: SocketAddr, path: &str) -> String { + format!("ws://{addr}{path}") +} + #[tokio::test] async fn info_returns_daemon_identity() { let h = spawn("naps-laptop").await; @@ -260,3 +262,103 @@ async fn attach_ws_for_unknown_session_returns_404() { "{msg}" ); } + +/// `WS /api/events` had no daemon-level test: the Snapshot-then-deltas contract +/// and the subscribe-before-snapshot ordering were only described in a comment. +#[tokio::test] +async fn events_ws_sends_a_snapshot_then_streams_deltas() { + let h = spawn("konishi").await; + let cwd = tempfile::TempDir::new().unwrap(); + + let (mut ws, _) = connect_async(ws_url(h.addr, "/api/events")) + .await + .expect("events ws"); + + match next_event(&mut ws).await { + maestro_protocol::CatalogEvent::Snapshot { sessions } => assert!(sessions.is_empty()), + other => panic!("first frame must be a Snapshot, got {other:?}"), + } + + let created = h + .manager + .create(CreateSessionRequest { + cwd: cwd.path().to_path_buf(), + command: "sleep 30".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }) + .await + .expect("create"); + + match next_event(&mut ws).await { + maestro_protocol::CatalogEvent::SessionAdded { session } => { + assert_eq!(session.id, created.id); + } + other => panic!("expected SessionAdded, got {other:?}"), + } + + h.manager.kill(created.id).await.expect("kill"); + + // Activity ticks may interleave; the removal is what we are waiting for. + let deadline = tokio::time::Instant::now() + Duration::from_secs(5); + loop { + assert!( + tokio::time::Instant::now() < deadline, + "timed out waiting for SessionRemoved" + ); + if let maestro_protocol::CatalogEvent::SessionRemoved { id } = next_event(&mut ws).await { + assert_eq!(id, created.id); + break; + } + } +} + +/// Each subscriber has its own lag domain now that the publisher task is gone: +/// two clients connected at different times both get their own Snapshot. +#[tokio::test] +async fn every_events_subscriber_gets_its_own_snapshot() { + let h = spawn("konishi").await; + let cwd = tempfile::TempDir::new().unwrap(); + + let (mut first, _) = connect_async(ws_url(h.addr, "/api/events")).await.unwrap(); + let _ = next_event(&mut first).await; + + h.manager + .create(CreateSessionRequest { + cwd: cwd.path().to_path_buf(), + command: "sleep 30".into(), + env: HashMap::new(), + metadata: HashMap::new(), + worktree: None, + }) + .await + .expect("create"); + + let (mut second, _) = connect_async(ws_url(h.addr, "/api/events")).await.unwrap(); + match next_event(&mut second).await { + maestro_protocol::CatalogEvent::Snapshot { sessions } => { + assert_eq!( + sessions.len(), + 1, + "a late subscriber must see current state" + ); + } + other => panic!("expected Snapshot, got {other:?}"), + } +} + +async fn next_event(ws: &mut WsClient) -> maestro_protocol::CatalogEvent { + loop { + match tokio::time::timeout(Duration::from_secs(5), ws.next()) + .await + .expect("events ws timed out") + { + Some(Ok(WsMessage::Text(t))) => { + return serde_json::from_str(&t).expect("decode catalog event") + } + Some(Ok(_)) => continue, + other => panic!("unexpected events frame: {other:?}"), + } + } +} diff --git a/crates/maestro-daemon/tests/session_decisions.rs b/crates/maestro-daemon/tests/session_decisions.rs index 244fb37..af4acc5 100644 --- a/crates/maestro-daemon/tests/session_decisions.rs +++ b/crates/maestro-daemon/tests/session_decisions.rs @@ -10,11 +10,11 @@ use std::sync::Arc; use std::time::Duration; use maestro_daemon::panes::{Call, FakePanes, Op}; -use maestro_daemon::session::{SessionError, SessionEvent, SessionManager}; +use maestro_daemon::session::{SessionError, SessionManager}; use maestro_daemon::sidecar::SidecarStore; use maestro_daemon::tmux::PaneInfo; use maestro_daemon::worktree::WorktreeManager; -use maestro_protocol::{ActivityState, CreateSessionRequest}; +use maestro_protocol::{ActivityState, CatalogEvent, CreateSessionRequest}; use tempfile::TempDir; use tokio::sync::broadcast::error::TryRecvError; use uuid::Uuid; @@ -171,7 +171,7 @@ async fn liveness_reaps_only_the_session_tmux_lost() { let mut removed = Vec::new(); while let Ok(ev) = events.try_recv() { - if let SessionEvent::Removed(id) = ev { + if let CatalogEvent::SessionRemoved { id } = ev { removed.push(id); } } @@ -244,12 +244,12 @@ async fn activity_pass_emits_updated_only_when_something_changed() { } async fn wait_for_updated( - events: &mut tokio::sync::broadcast::Receiver, + events: &mut tokio::sync::broadcast::Receiver, ) -> maestro_protocol::Session { let deadline = tokio::time::Instant::now() + Duration::from_secs(3); while tokio::time::Instant::now() < deadline { match events.try_recv() { - Ok(SessionEvent::Updated(s)) => return s, + Ok(CatalogEvent::SessionUpdated { session }) => return session, Ok(_) => continue, Err(_) => tokio::time::sleep(Duration::from_millis(10)).await, } diff --git a/crates/maestro-daemon/tests/session_manager.rs b/crates/maestro-daemon/tests/session_manager.rs index 359487c..1411b08 100644 --- a/crates/maestro-daemon/tests/session_manager.rs +++ b/crates/maestro-daemon/tests/session_manager.rs @@ -8,7 +8,7 @@ use std::sync::Arc; use std::time::Duration; use common::{wait_for_async, DaemonFixture}; -use maestro_daemon::SessionEvent; +use maestro_protocol::CatalogEvent; use maestro_protocol::CreateSessionRequest; use tempfile::TempDir; @@ -345,7 +345,7 @@ async fn liveness_watcher_reaps_dead_sessions_and_emits_removed() { let deadline = std::time::Instant::now() + Duration::from_secs(1); while std::time::Instant::now() < deadline && !saw_removed { match tokio::time::timeout(Duration::from_millis(50), events.recv()).await { - Ok(Ok(SessionEvent::Removed(id))) if id == s.id => saw_removed = true, + Ok(Ok(CatalogEvent::SessionRemoved { id })) if id == s.id => saw_removed = true, Ok(Ok(_)) => continue, _ => break, }