refactor(maestro): collapse the daemon's double event fan-out

A background task, a second broadcast channel and a whole module existed
to remap SessionEvent onto CatalogEvent one variant at a time.
SessionManager now emits CatalogEvent directly.

Deletes local_events.rs, the publisher task, the LocalEventBus channel,
enum SessionEvent, LocalApiState::events, and the duplicate capacity
constant — events_capacity was declared twice for one logical stream.

Lag semantics improve. There were two lag points: the publisher falling
behind the manager, which broadcast one Lagged to every client, and each
WS subscriber falling behind the bus. Only the per-subscriber one is left,
which pump_events already handled — a slow client now makes only itself
resync.

Adds the first daemon-level tests for WS /api/events: the Snapshot-then-
deltas contract, and that a late subscriber gets a Snapshot of current
state. Both were described only in a comment.

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

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:?}"),
+        }
+    }
+}