feat(qbzd): live event stream — GET /api/events (SSE) + qbzd watch

Add a Server-Sent Events feed so a client (a plasmoid, a bar applet, a script)
reacts to playback/queue/library changes without polling — the last CONSOLE
surface.

Server (api/sse.rs): `GET /api/events` streams one SSE frame per CoreEvent
(`event: <type>` + the tagged `data:` JSON). Because the control-plane serve
loop is single-threaded, an open stream would starve it, so the route is moved
onto its OWN thread (Request is Send) after the same origin/token gate every
route runs; the bus is drained with `blocking_recv()` there. Bulky/internal
events (search payloads, loading/download/navigation hints, diagnostics) are
filtered out; everything else — playback, queue, volume, auth, favorites,
playlists, errors, device changes — is emitted. ApiState gains the bus sender.

Client (`qbzd watch`): a bespoke reqwest client with NO read timeout (only the
connect is bounded); default output is newline-delimited JSON (one data payload
per line, `qbzd watch | jq`), `--raw` passes the SSE frames through verbatim.
P1_ROUTES → 26 with the route-table test updated (caller: `qbzd watch`).

283 tests green.

cc

crates/qbzd/src/api/mod.rs

@@ -26,6 +26,7 @@ pub mod radio;
 pub mod reco;
 pub mod search;
 pub mod settings;
+pub mod sse;
 pub mod status;
 
 use std::io::Cursor;
@@ -33,13 +34,15 @@ use std::net::SocketAddr;
 use std::sync::{Arc, Mutex};
 use std::time::Instant;
 
-use tiny_http::{Header, Request, Response};
+use tiny_http::{Header, Method, Request, Response};
+use tokio::sync::broadcast;
 
 use crate::adapter::DaemonAdapter;
 use crate::paths::ProfileRoots;
 use crate::state::DaemonShared;
 use qbz_app::shell::AppRuntime;
 use qbz_audio::settings::AudioSettingsStore;
+use qbz_models::CoreEvent;
 
 /// The counted P0 route table (02 §3.2) — 17, FINAL, this test now guards the
 /// budget forever. A route exists only iff a shipped client calls it (§3.1.4).
@@ -101,6 +104,7 @@ pub const P1_ROUTES: &[(&str, &str)] = &[
     ("POST", "/api/playlist/delete"),
     ("POST", "/api/playlist/tracks/add"),
     ("POST", "/api/playlist/tracks/remove"),
+    ("GET", "/api/events"),
 ];
 
 /// A socket bound at boot step 5, not yet serving. Wraps the tiny_http server
@@ -118,6 +122,9 @@ pub struct BoundServer {
 pub struct ApiState {
     pub runtime: Arc<AppRuntime<DaemonAdapter>>,
     pub shared: Arc<Mutex<DaemonShared>>,
+    /// The CoreEvent bus (DaemonAdapter sender). `/api/events` subscribes a
+    /// receiver per SSE connection; no other route touches it.
+    pub bus: broadcast::Sender<CoreEvent>,
     pub roots: ProfileRoots,
     pub token: Option<String>,
     /// The bound address, echoed verbatim by `/api/info`.
@@ -249,6 +256,32 @@ pub fn serve(server: BoundServer, state: ApiState) -> ApiHandle {
         .name("qbzd-api".into())
         .spawn(move || {
             for mut req in srv.incoming_requests() {
+                // `/api/events` is a long-lived SSE stream: it would block this
+                // single serving thread forever. Move it onto its OWN thread
+                // (Request is Send) so the control plane keeps answering. The
+                // origin/token gate is applied first, identically to `route`.
+                let is_events = *req.method() == Method::Get
+                    && req.url().split('?').next() == Some("/api/events");
+                if is_events {
+                    let has_origin = req.headers().iter().any(|h| h.field.equiv("Origin"));
+                    let auth = req
+                        .headers()
+                        .iter()
+                        .find(|h| h.field.equiv("Authorization"))
+                        .map(|h| h.value.as_str().to_owned());
+                    if let Some(reject) =
+                        access_gate(has_origin, "GET", "/api/events", auth.as_deref(), state.token.as_deref())
+                    {
+                        let _ = req.respond(reject.response());
+                        continue;
+                    }
+                    let rx = state.bus.subscribe();
+                    std::thread::Builder::new()
+                        .name("qbzd-sse".into())
+                        .spawn(move || sse::stream(req, rx))
+                        .ok();
+                    continue;
+                }
                 let resp = route(&state, &mut req);
                 let _ = req.respond(resp);
             }
@@ -539,7 +572,8 @@ mod tests {
         // §3.1.4 HARD RULE, applied to the content-verb door). Row 19:
         // GET /api/search — caller: `qbzd search`. Count is pinned so a route
         // with no caller cannot creep in; P1 must never overlap P0.
-        assert_eq!(P1_ROUTES.len(), 25);
+        assert_eq!(P1_ROUTES.len(), 26);
+        assert!(P1_ROUTES.contains(&("GET", "/api/events"))); // caller: `qbzd watch`
         assert!(P1_ROUTES.contains(&("GET", "/api/discover")));
         assert!(P1_ROUTES.contains(&("GET", "/api/lyrics")));
         assert!(P1_ROUTES.contains(&("POST", "/api/reco/playlist")));

crates/qbzd/src/api/sse.rs

@@ -0,0 +1,156 @@
+// crates/qbzd/src/api/sse.rs — the `GET /api/events` Server-Sent Events stream
+// (CONSOLE ext). A live push feed of CoreEvents so a client (a plasmoid, a bar
+// applet, `qbzd watch`) reacts to playback/queue/library changes without
+// polling.
+//
+// Concurrency: the control-plane serve loop is single-threaded, so an open SSE
+// stream would starve every other request. `serve()` therefore moves this onto
+// its OWN thread (Request is Send); `stream` blocks there until the client
+// disconnects (respond() errors) or the bus closes. The rusqlite-free bus is a
+// tokio broadcast, drained here with `blocking_recv()` from the plain thread.
+//
+// Wire format: one SSE frame per emitted event —
+//   event: <CoreEvent type>\n
+//   data: {"type":"…","data":{…}}\n\n
+// The `data` line is the CoreEvent's own `#[serde(tag="type",content="data")]`
+// JSON (single-line). A leading `: …` comment primes the stream; a dropped-lag
+// notice is sent as a comment rather than silently losing ordering.
+use std::io::{Cursor, Read};
+
+use tiny_http::{Header, Request, Response, StatusCode};
+use tokio::sync::broadcast;
+
+use qbz_models::CoreEvent;
+
+/// Stream CoreEvents to one client until it disconnects. Runs on a dedicated
+/// thread; `req.respond` blocks here, writing chunked as `SseReader` yields.
+pub fn stream(req: Request, rx: broadcast::Receiver<CoreEvent>) {
+    let headers = vec![
+        header("Content-Type", "text/event-stream"),
+        header("Cache-Control", "no-cache"),
+        header("Connection", "keep-alive"),
+        // Defeat proxy/response buffering so frames flush immediately.
+        header("X-Accel-Buffering", "no"),
+    ];
+    let response = Response::new(StatusCode(200), headers, SseReader::new(rx), None, None);
+    let _ = req.respond(response);
+}
+
+fn header(name: &str, value: &str) -> Header {
+    // Both are static ASCII — construction cannot fail.
+    Header::from_bytes(name.as_bytes(), value.as_bytes()).expect("static header")
+}
+
+/// A blocking `Read` over the CoreEvent bus: each `read` serves the current
+/// frame's bytes, and when a frame is exhausted it blocks for the next event.
+/// `Ok(0)` (EOF) only when the bus closes (daemon shutting down).
+struct SseReader {
+    rx: broadcast::Receiver<CoreEvent>,
+    frame: Cursor<Vec<u8>>,
+    primed: bool,
+}
+
+impl SseReader {
+    fn new(rx: broadcast::Receiver<CoreEvent>) -> Self {
+        SseReader { rx, frame: Cursor::new(Vec::new()), primed: false }
+    }
+
+    /// The next frame to send, blocking for a bus event. `None` = bus closed.
+    fn next_frame(&mut self) -> Option<Vec<u8>> {
+        if !self.primed {
+            self.primed = true;
+            return Some(b": qbzd event stream\n\n".to_vec());
+        }
+        loop {
+            match self.rx.blocking_recv() {
+                Ok(ev) => {
+                    if let Some(frame) = format_event(&ev) {
+                        return Some(frame.into_bytes());
+                    }
+                    // Not an emitted event — keep waiting.
+                }
+                Err(broadcast::error::RecvError::Lagged(n)) => {
+                    return Some(format!(": lagged {n} event(s)\n\n").into_bytes());
+                }
+                Err(broadcast::error::RecvError::Closed) => return None,
+            }
+        }
+    }
+}
+
+impl Read for SseReader {
+    fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
+        loop {
+            let n = self.frame.read(out)?;
+            if n > 0 {
+                return Ok(n);
+            }
+            match self.next_frame() {
+                Some(bytes) => self.frame = Cursor::new(bytes),
+                None => return Ok(0), // bus closed → end the stream
+            }
+        }
+    }
+}
+
+/// Render one CoreEvent as an SSE frame, or `None` for events not worth pushing
+/// to a UI client (bulky search payloads, internal loading/download/navigation
+/// hints, diagnostics). Everything else — playback, queue, volume, auth,
+/// favorites, playlists, errors, device changes — is emitted.
+fn format_event(ev: &CoreEvent) -> Option<String> {
+    if !emit(ev) {
+        return None;
+    }
+    let value = serde_json::to_value(ev).ok()?;
+    let typ = value.get("type").and_then(|v| v.as_str()).unwrap_or("event").to_string();
+    let data = serde_json::to_string(&value).ok()?;
+    Some(format!("event: {typ}\ndata: {data}\n\n"))
+}
+
+fn emit(ev: &CoreEvent) -> bool {
+    use CoreEvent::*;
+    !matches!(
+        ev,
+        SearchResultsReceived { .. }
+            | LoadingStarted { .. }
+            | LoadingCompleted { .. }
+            | DownloadProgress { .. }
+            | DownloadCompleted { .. }
+            | NavigateToAlbum { .. }
+            | NavigateToArtist { .. }
+            | NavigateToPlaylist { .. }
+            | AudioDiagnostic { .. }
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use qbz_models::PlaybackState;
+
+    #[test]
+    fn playback_event_becomes_a_typed_sse_frame() {
+        let frame = format_event(&CoreEvent::PlaybackStateChanged {
+            state: PlaybackState::Playing,
+        })
+        .expect("playback event is emitted");
+        assert!(frame.starts_with("event: PlaybackStateChanged\n"));
+        assert!(frame.contains("data: {"));
+        assert!(frame.ends_with("\n\n"));
+        // The data line carries the tagged CoreEvent JSON.
+        assert!(frame.contains("\"type\":\"PlaybackStateChanged\""));
+    }
+
+    #[test]
+    fn bulky_and_internal_events_are_not_emitted() {
+        assert!(format_event(&CoreEvent::LoadingStarted { operation: "x".into() }).is_none());
+        assert!(format_event(&CoreEvent::DownloadCompleted { track_id: 1 }).is_none());
+        assert!(format_event(&CoreEvent::NavigateToArtist { artist_id: 1 }).is_none());
+    }
+
+    #[test]
+    fn volume_and_queue_events_are_emitted() {
+        assert!(format_event(&CoreEvent::VolumeChanged { volume: 0.5 }).is_some());
+        assert!(format_event(&CoreEvent::ShuffleChanged { enabled: true }).is_some());
+    }
+}

crates/qbzd/src/cli/client.rs

@@ -101,7 +101,7 @@ fn normalize_hostport(h: &str) -> String {
 
 /// §1.5 token discovery: `QBZD_TOKEN` (remote or local), else — only when
 /// targeting the local daemon — the local `qbzd.toml` `[server] token`.
-fn resolve_token(target: &Target, roots: &ProfileRoots) -> Option<String> {
+pub(crate) fn resolve_token(target: &Target, roots: &ProfileRoots) -> Option<String> {
     if let Ok(t) = std::env::var("QBZD_TOKEN") {
         if !t.is_empty() {
             return Some(t);

crates/qbzd/src/cli/mod.rs

@@ -22,3 +22,4 @@ pub mod search;
 pub mod settings;
 pub mod status;
 pub mod transport;
+pub mod watch;

crates/qbzd/src/cli/watch.rs

@@ -0,0 +1,99 @@
+// crates/qbzd/src/cli/watch.rs — `qbzd watch`, the CLI caller for the
+// `GET /api/events` SSE stream (CONSOLE ext). Opens a long-lived connection and
+// prints events as they arrive; the shipped verb behind the P1 route (§3.1.4).
+//
+// Default output is newline-delimited JSON — one CoreEvent `data` payload per
+// line, pipe-friendly (`qbzd watch | jq`). `--raw` passes the SSE frames
+// through verbatim (event:/data:/comment lines). Unlike `ApiClient` this uses a
+// bespoke reqwest client with NO read timeout (the stream is meant to stay
+// open); only the connect attempt is bounded.
+use std::io::Write;
+use std::time::Duration;
+
+use crate::cli::client::{resolve_host, resolve_token, CliError};
+use crate::paths::ProfileRoots;
+
+pub async fn watch(host: Option<String>, raw: bool, roots: &ProfileRoots) -> i32 {
+    let target = resolve_host(host);
+    let token = resolve_token(&target, roots);
+    let base = format!("http://{}", target.addr);
+
+    let client = match reqwest::Client::builder()
+        .connect_timeout(Duration::from_secs(2))
+        .build()
+    {
+        Ok(c) => c,
+        Err(e) => {
+            eprintln!("error: {e}");
+            return 1;
+        }
+    };
+
+    let mut builder = client
+        .get(format!("{base}/api/events"))
+        .header(reqwest::header::ACCEPT, "text/event-stream");
+    if let Some(t) = &token {
+        builder = builder.bearer_auth(t);
+    }
+
+    let resp = match builder.send().await {
+        Ok(r) => r,
+        Err(e) => {
+            let err = if e.is_connect() || e.is_timeout() {
+                CliError::Unreachable(target.addr.clone())
+            } else {
+                CliError::Runtime(format!("request failed: {e}"))
+            };
+            eprintln!("{err}");
+            return err.exit_code();
+        }
+    };
+
+    if !resp.status().is_success() {
+        let code = resp.status().as_u16();
+        let body = resp.text().await.unwrap_or_default();
+        let hint = if code == 401 || code == 403 {
+            " — check QBZD_TOKEN / the daemon [server] token"
+        } else {
+            ""
+        };
+        eprintln!("error: daemon returned {code}{hint}: {}", body.trim());
+        return 1;
+    }
+
+    let mut resp = resp;
+    let stdout = std::io::stdout();
+    let mut pending = String::new();
+    loop {
+        match resp.chunk().await {
+            Ok(Some(bytes)) => {
+                if raw {
+                    let mut lock = stdout.lock();
+                    let _ = lock.write_all(&bytes);
+                    let _ = lock.flush();
+                    continue;
+                }
+                pending.push_str(&String::from_utf8_lossy(&bytes));
+                // Emit each COMPLETE line's `data:` payload; keep the tail.
+                while let Some(nl) = pending.find('\n') {
+                    let line: String = pending.drain(..=nl).collect();
+                    if let Some(data) = line.trim_end().strip_prefix("data:") {
+                        let mut lock = stdout.lock();
+                        let _ = writeln!(lock, "{}", data.trim());
+                        let _ = lock.flush();
+                    }
+                    // event: / comment (`:`) / blank lines are dropped in parsed mode
+                }
+            }
+            // The daemon closed the stream (shutting down): treat as unreachable.
+            Ok(None) => {
+                eprintln!("{}", CliError::Unreachable(target.addr.clone()));
+                return CliError::Unreachable(target.addr).exit_code();
+            }
+            Err(e) => {
+                eprintln!("error: event stream read failed: {e}");
+                return 1;
+            }
+        }
+    }
+}

crates/qbzd/src/daemon.rs

@@ -153,6 +153,7 @@ pub async fn run(roots: ProfileRoots, cfg: QbzdConfig, warns: Vec<String>) -> Re
         crate::api::ApiState {
             runtime: booted.runtime.clone(),
             shared: booted.shared.clone(),
+            bus: booted.bus.clone(),
             roots: roots.clone(),
             token: cfg.server.token.filter(|t| !t.trim().is_empty()),
             bind: bind_addr.to_string(),

crates/qbzd/src/main.rs

@@ -47,6 +47,8 @@ enum Cmd {
     Ping   { #[arg(long)] json: bool },
     /// One-line now-playing
     Now    { #[arg(long)] json: bool },
+    /// Stream live daemon events (SSE); default = newline-delimited JSON
+    Watch  { #[arg(long)] raw: bool },
     /// Search Qobuz — top hits with ids (--ids pipes into `queue add -`)
     Search {
         query: String,
@@ -339,6 +341,10 @@ async fn main() {
             let roots = paths::ProfileRoots::resolve(None, None);
             cli::transport::now(cli.host, json, &roots).await
         }
+        Cmd::Watch { raw } => {
+            let roots = paths::ProfileRoots::resolve(None, None);
+            cli::watch::watch(cli.host, raw, &roots).await
+        }
         Cmd::Search { query, kind, limit, offset, ids, json } => {
             let roots = paths::ProfileRoots::resolve(None, None);
             cli::search::search(cli.host, query, kind, limit, offset, ids, json, &roots).await