feat(host/events): GET /api/v1/events — SSE lifecycle event stream (M1)

Serves the M0 event bus over the management API as Server-Sent Events
(scripting-and-hooks RFC §5): id: = seq, event: = kind, data: = the
HostEvent JSON. Standard Last-Event-ID (or ?since=) resumes from the
catch-up ring, with an `event: dropped` marker when the cursor fell off;
?kinds= filters server-side (exact kinds or `domain.*` prefixes).

Bounds per RFC §9.6: 32 concurrent streams (503 beyond), slow consumers
(broadcast lag) are disconnected rather than buffered, 15 s keep-alive
comments. Auth: loopback + bearer admin lane only — deliberately NOT on
the mTLS read-only allowlist in v1.

Note: api/openapi.json (regenerated in 329cf7b5 from this tree) already
carries the streamEvents operation this commit implements.

Verified live on Linux: catch-up + mid-stream library.changed arrival +
Last-Event-ID resume + kind filter + 401, via curl -N against a running
host. 335 host tests green (incl. the spec drift test), clippy clean.

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

crates/punktfunk-host/Cargo.toml

@@ -56,6 +56,9 @@ tokio-rustls = "0.26"
 hyper = { version = "1", features = ["server", "http1", "http2"] }
 hyper-util = { version = "0.1", features = ["server", "server-auto", "tokio", "service"] }
 tower = { version = "0.5", features = ["util"] }
+# Stream combinators for the mgmt API's SSE event feed (`GET /api/v1/events`) — already in the
+# tree transitively (axum/hyper) and as a Linux target dep; control plane only.
+futures-util = "0.3"
 rusty_enet = "0.4"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"

crates/punktfunk-host/src/mgmt.rs

@@ -32,6 +32,7 @@ use utoipa_scalar::{Scalar, Servable};
 mod auth;
 mod clients;
 mod display;
+mod events;
 mod gpu;
 mod host;
 mod library;
@@ -215,7 +216,8 @@ fn api_router_parts() -> (Router<Arc<MgmtState>>, utoipa::openapi::OpenApi) {
                     stats::stats_recording_get,
                     stats::stats_recording_delete
                 ))
-                .routes(routes!(stats::logs_get)),
+                .routes(routes!(stats::logs_get))
+                .routes(routes!(events::stream_events)),
         )
         .split_for_parts()
 }
@@ -251,6 +253,7 @@ pub fn openapi_json() -> String {
         (name = "library", description = "Game library: installed-store titles (Steam) plus user-curated custom entries"),
         (name = "stats", description = "Streaming performance-stats capture: arm/stop a recording, read the live + saved time-series for graphing"),
         (name = "logs", description = "Host log stream: the newest in-memory log entries, cursor-paged for live following"),
+        (name = "events", description = "Host lifecycle events: an SSE stream (client/session/stream lifecycle, pairing, displays, library, host) with Last-Event-ID resume and server-side kind filters"),
     )
 )]
 struct ApiDoc;

crates/punktfunk-host/src/mgmt/events.rs

@@ -0,0 +1,236 @@
+//! `GET /api/v1/events` — the host lifecycle event stream (scripting-and-hooks RFC §5, M1).
+//!
+//! Server-Sent Events over the existing HTTPS serve loop: each event goes out as one SSE frame
+//! with `id:` = the event's `seq`, `event:` = its kind (`stream.started`, …), and `data:` = the
+//! [`crate::events::HostEvent`] JSON. Consumers resume with the standard `Last-Event-ID` header
+//! (or its `?since=` query twin); a consumer that fell off the catch-up ring gets a synthetic
+//! `event: dropped` frame first and should resync via the REST snapshots (`/status`, `/clients`,
+//! …). `?kinds=` filters server-side (exact kinds or `domain.*` prefixes, comma-separated).
+//!
+//! Bounds (RFC §9.6): at most [`MAX_EVENT_STREAMS`] concurrent streams (503 beyond — the
+//! consumer retries; SSE clients reconnect by themselves), and a consumer too slow for the
+//! live tail (broadcast lag) is **disconnected**, never buffered unboundedly — its reconnect
+//! resumes from the ring via `Last-Event-ID`.
+//!
+//! Auth: deliberately NOT on the mTLS read-only allowlist (`auth::cert_may_access`) — the
+//! stream is part of the loopback + bearer admin lane in v1 (RFC §5; revisit when paired
+//! clients want an activity feed).
+
+use super::shared::*;
+use axum::http::HeaderMap;
+use axum::response::sse::{Event, KeepAlive, Sse};
+use std::collections::VecDeque;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::time::Duration;
+use tokio::sync::broadcast;
+
+/// Concurrent SSE connection cap. Operators run a handful of consumers (console feed, a couple
+/// of scripts/plugins); 32 is generous headroom while bounding a runaway reconnect loop.
+const MAX_EVENT_STREAMS: usize = 32;
+
+/// SSE keep-alive comment interval — detects a dead peer and keeps middleboxes from idling
+/// the connection out between (low-rate) lifecycle events.
+const KEEP_ALIVE: Duration = Duration::from_secs(15);
+
+static LIVE_STREAMS: AtomicUsize = AtomicUsize::new(0);
+
+/// RAII slot in the connection cap: taken before the stream starts, released when the SSE body
+/// is dropped (client disconnect, slow-consumer cut, server shutdown). `pub(crate)` only for
+/// the [`test_support`] saturation helper's return type.
+pub(crate) struct StreamSlot;
+
+fn try_acquire_slot() -> Option<StreamSlot> {
+    LIVE_STREAMS
+        .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| {
+            (n < MAX_EVENT_STREAMS).then_some(n + 1)
+        })
+        .ok()
+        .map(|_| StreamSlot)
+}
+
+impl Drop for StreamSlot {
+    fn drop(&mut self) {
+        LIVE_STREAMS.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+
+/// `?kinds=stream.*,pairing.pending` — exact kind names or `domain.*` prefixes. `None` = all.
+struct KindFilter(Option<Vec<String>>);
+
+impl KindFilter {
+    fn parse(kinds: Option<&str>) -> Self {
+        let pats: Option<Vec<String>> = kinds.map(|s| {
+            s.split(',')
+                .map(str::trim)
+                .filter(|p| !p.is_empty())
+                .map(str::to_string)
+                .collect()
+        });
+        KindFilter(pats.filter(|p| !p.is_empty()))
+    }
+
+    fn matches(&self, kind: &str) -> bool {
+        match &self.0 {
+            None => true,
+            Some(pats) => pats.iter().any(|p| match p.strip_suffix(".*") {
+                Some(prefix) => kind
+                    .strip_prefix(prefix)
+                    .is_some_and(|rest| rest.starts_with('.')),
+                None => p == kind,
+            }),
+        }
+    }
+}
+
+#[derive(Deserialize)]
+pub(crate) struct EventsQuery {
+    /// Resume cursor: stream events with `seq > since`. The `Last-Event-ID` header (what an SSE
+    /// client sends on auto-reconnect) takes precedence when both are present.
+    since: Option<u64>,
+    /// Comma-separated kind filter (`stream.*,pairing.pending`).
+    kinds: Option<String>,
+}
+
+/// One [`crate::events::HostEvent`] as an SSE frame: `id:` = seq (drives `Last-Event-ID`),
+/// `event:` = kind, `data:` = the full event JSON (kind included — the frame is self-contained
+/// for consumers that read `data` only).
+fn sse_event(ev: &crate::events::HostEvent) -> Event {
+    Event::default()
+        .id(ev.seq.to_string())
+        .event(ev.kind.name())
+        .data(serde_json::to_string(ev).unwrap_or_else(|_| "{}".to_string()))
+}
+
+/// Everything the poll loop owns; dropping it (the response body going away) releases the
+/// connection-cap slot and the broadcast receiver.
+struct StreamState {
+    /// The dropped marker (when applicable) + the filtered catch-up, served before the live tail.
+    pending: VecDeque<Event>,
+    rx: broadcast::Receiver<crate::events::HostEvent>,
+    filter: KindFilter,
+    _slot: StreamSlot,
+}
+
+/// Stream host lifecycle events (SSE)
+///
+/// Server-Sent Events stream of the host's lifecycle events: client connect/disconnect, session
+/// and stream start/end, pairing decisions, display create/release, library changes, host
+/// start/stop — both protocol planes. Frames carry `id:` = the event's monotonic `seq`,
+/// `event:` = its kind, and `data:` = the event JSON (schema-versioned, additive-only).
+///
+/// Resume: standard `Last-Event-ID` (or `?since=`) replays from the in-memory ring; a consumer
+/// that fell off the ring receives an `event: dropped` frame first and should resync via the
+/// REST snapshots. Keep-alive comments are sent every 15 s.
+#[utoipa::path(
+    get,
+    path = "/events",
+    tag = "events",
+    operation_id = "streamEvents",
+    params(
+        ("since" = Option<u64>, Query, description = "Resume cursor: only events with `seq` greater than this are sent (the ring keeps the newest ~1024). `Last-Event-ID` takes precedence."),
+        ("kinds" = Option<String>, Query, description = "Comma-separated server-side kind filter: exact kinds (`pairing.pending`) or `domain.*` prefixes (`stream.*`)."),
+        ("Last-Event-ID" = Option<u64>, Header, description = "SSE auto-reconnect cursor — the `id:` of the last received frame."),
+    ),
+    responses(
+        (status = OK, description = "SSE stream; each frame's `data:` is one HostEvent", body = crate::events::HostEvent, content_type = "text/event-stream"),
+        (status = UNAUTHORIZED, description = "Missing or invalid bearer token", body = ApiError),
+        (status = SERVICE_UNAVAILABLE, description = "Concurrent event-stream cap reached — retry shortly", body = ApiError),
+    )
+)]
+pub(crate) async fn stream_events(Query(q): Query<EventsQuery>, headers: HeaderMap) -> Response {
+    let Some(slot) = try_acquire_slot() else {
+        return api_error(
+            StatusCode::SERVICE_UNAVAILABLE,
+            "event-stream connection cap reached — close an existing stream or retry",
+        );
+    };
+    // The header is what an SSE client re-sends on auto-reconnect, so it is always the newer
+    // cursor when both it and the original URL's `?since=` are present.
+    let since = headers
+        .get("last-event-id")
+        .and_then(|v| v.to_str().ok())
+        .and_then(|v| v.trim().parse::<u64>().ok())
+        .or(q.since)
+        .unwrap_or(0);
+    let filter = KindFilter::parse(q.kinds.as_deref());
+    let sub = crate::events::bus().subscribe(since);
+
+    let mut pending = VecDeque::new();
+    if sub.dropped {
+        // The consumer's cursor precedes the ring: tell it so (the `LogPage.dropped` contract)
+        // — its move is a REST resync, not trust in a complete replay.
+        pending.push_back(
+            Event::default()
+                .event("dropped")
+                .data(r#"{"dropped":true}"#),
+        );
+    }
+    pending.extend(
+        sub.catch_up
+            .iter()
+            .filter(|ev| filter.matches(ev.kind.name()))
+            .map(sse_event),
+    );
+
+    let state = StreamState {
+        pending,
+        rx: sub.rx,
+        filter,
+        _slot: slot,
+    };
+    let stream = futures_util::stream::unfold(state, |mut st| async move {
+        loop {
+            if let Some(ev) = st.pending.pop_front() {
+                return Some((Ok::<_, std::convert::Infallible>(ev), st));
+            }
+            match st.rx.recv().await {
+                Ok(ev) => {
+                    if st.filter.matches(ev.kind.name()) {
+                        return Some((Ok(sse_event(&ev)), st));
+                    }
+                }
+                // Lagged = this consumer is too slow for the live tail: cut it loose (bounded
+                // memory, RFC §9.6) — its auto-reconnect resumes from the ring, which flags
+                // `dropped` if it also fell off that. Closed = host shutdown.
+                Err(broadcast::error::RecvError::Lagged(_))
+                | Err(broadcast::error::RecvError::Closed) => return None,
+            }
+        }
+    });
+    Sse::new(stream)
+        .keep_alive(KeepAlive::new().interval(KEEP_ALIVE))
+        .into_response()
+}
+
+#[cfg(test)]
+pub(crate) mod test_support {
+    /// Fill the connection cap and hand the slots to a test ([`Drop`] frees them). The events
+    /// tests serialize on a lock so the full cap can't 503 an unrelated concurrent test.
+    pub(crate) fn saturate_slots() -> Vec<super::StreamSlot> {
+        std::iter::from_fn(super::try_acquire_slot).collect()
+    }
+}
+
+#[cfg(test)]
+mod unit_tests {
+    use super::*;
+
+    #[test]
+    fn kind_filter_semantics() {
+        let all = KindFilter::parse(None);
+        assert!(all.matches("stream.started"));
+
+        let f = KindFilter::parse(Some("stream.*, pairing.pending"));
+        assert!(f.matches("stream.started"));
+        assert!(f.matches("stream.stopped"));
+        assert!(f.matches("pairing.pending"));
+        assert!(!f.matches("pairing.completed"));
+        assert!(!f.matches("client.connected"));
+        // A prefix pattern must match on the dot boundary, not raw text.
+        assert!(!f.matches("streamx.started"));
+
+        // Empty/blank filter strings mean "no filter", not "nothing matches".
+        assert!(KindFilter::parse(Some("")).matches("host.started"));
+        assert!(KindFilter::parse(Some(" , ")).matches("host.started"));
+    }
+}

crates/punktfunk-host/src/mgmt/tests.rs

@@ -946,3 +946,212 @@ async fn logs_endpoint_pages_by_cursor() {
     assert!(json["entries"].as_array().unwrap().is_empty());
     assert_eq!(json["next"].as_u64().unwrap(), after);
 }
+
+// ------------------------------------------------------------------ events (SSE)
+
+/// Serializes the events-route tests: they share the process-global event bus and the
+/// connection-cap counter, so the cap test must never 503 a concurrently running stream test.
+static EVENTS_TEST_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+/// `get_req` + the default test bearer, pre-attached (these tests read streaming bodies
+/// directly instead of going through `send`).
+fn events_req(path: &str) -> axum::http::Request<Body> {
+    let mut req = get_req(path);
+    req.headers_mut().insert(
+        axum::http::header::AUTHORIZATION,
+        axum::http::HeaderValue::from_static("Bearer test-secret"),
+    );
+    req
+}
+
+/// The next SSE frame as text, or `None` when the stream ended / nothing arrived in time.
+async fn next_sse_chunk(body: &mut Body) -> Option<String> {
+    match tokio::time::timeout(std::time::Duration::from_secs(5), body.frame()).await {
+        Ok(Some(Ok(frame))) => frame
+            .into_data()
+            .ok()
+            .map(|b| String::from_utf8_lossy(&b).into_owned()),
+        _ => None,
+    }
+}
+
+/// Every `data:` payload in accumulated SSE text, parsed as JSON.
+fn sse_data_events(text: &str) -> Vec<serde_json::Value> {
+    text.lines()
+        .filter_map(|l| l.strip_prefix("data: "))
+        .filter_map(|d| serde_json::from_str(d).ok())
+        .collect()
+}
+
+#[tokio::test]
+async fn events_stream_requires_bearer() {
+    let app = test_app(test_state(), None);
+    let mut req = get_req("/api/v1/events");
+    req.headers_mut().insert(
+        axum::http::header::AUTHORIZATION,
+        axum::http::HeaderValue::from_static("Bearer wrong"),
+    );
+    let resp = app.clone().oneshot(req).await.expect("infallible");
+    assert_eq!(resp.status(), StatusCode::UNAUTHORIZED);
+}
+
+/// The full consumer contract on one route: ring catch-up, the server-side kind filter, the
+/// live tail on the same connection, `?since=`/`Last-Event-ID` resume, and the `dropped`
+/// marker for a cursor that fell off the ring.
+#[tokio::test]
+async fn events_stream_catch_up_filter_resume_tail_and_dropped() {
+    use crate::events::EventKind;
+    let _l = EVENTS_TEST_LOCK.lock().await;
+    let app = test_app(test_state(), None);
+    let uniq = format!("evt-{}-{:p}", std::process::id(), &0u8 as *const u8);
+    let m1 = format!("{uniq}-one");
+
+    // Noise of a different kind (must be filtered out), then our marker.
+    crate::events::emit(EventKind::DisplayReleased { count: 424_242 });
+    crate::events::emit(EventKind::LibraryChanged { source: m1.clone() });
+
+    let resp = app
+        .clone()
+        .oneshot(events_req("/api/v1/events?kinds=library.changed"))
+        .await
+        .expect("infallible");
+    assert_eq!(resp.status(), StatusCode::OK);
+    let ctype = resp
+        .headers()
+        .get(axum::http::header::CONTENT_TYPE)
+        .and_then(|v| v.to_str().ok())
+        .unwrap_or_default()
+        .to_string();
+    assert!(
+        ctype.starts_with("text/event-stream"),
+        "content-type: {ctype}"
+    );
+
+    // Catch-up must deliver m1 (other tests' library.changed events may interleave — scan).
+    let mut body = resp.into_body();
+    let mut seen = String::new();
+    while !seen.contains(&m1) {
+        let chunk = next_sse_chunk(&mut body)
+            .await
+            .expect("catch-up delivers the marker event");
+        seen.push_str(&chunk);
+    }
+    assert!(
+        !seen.contains("event: display.released"),
+        "kind filter must drop other kinds: {seen}"
+    );
+    assert!(
+        seen.contains("event: library.changed"),
+        "frame kind: {seen}"
+    );
+    let m1_seq = sse_data_events(&seen)
+        .iter()
+        .find(|e| e["source"] == m1.as_str())
+        .and_then(|e| e["seq"].as_u64())
+        .expect("marker frame carries the full event JSON with its seq");
+
+    // Live tail on the SAME connection. If a concurrent test floods the broadcast channel the
+    // slow-consumer cut ends this stream — then the documented client move (reconnect with the
+    // last seen id) must deliver m2 instead, so follow it rather than flaking.
+    let m2 = format!("{uniq}-two");
+    crate::events::emit(EventKind::LibraryChanged { source: m2.clone() });
+    let mut tail = String::new();
+    loop {
+        match next_sse_chunk(&mut body).await {
+            Some(chunk) => {
+                tail.push_str(&chunk);
+                if tail.contains(&m2) {
+                    break;
+                }
+            }
+            None => {
+                let resp = app
+                    .clone()
+                    .oneshot(events_req(&format!(
+                        "/api/v1/events?since={m1_seq}&kinds=library.changed"
+                    )))
+                    .await
+                    .expect("infallible");
+                body = resp.into_body();
+            }
+        }
+    }
+    drop(body);
+
+    // Resume from m1's seq: m2 is caught up, m1 is not.
+    let resp = app
+        .clone()
+        .oneshot(events_req(&format!(
+            "/api/v1/events?since={m1_seq}&kinds=library.changed"
+        )))
+        .await
+        .expect("infallible");
+    let mut body = resp.into_body();
+    let mut resumed = String::new();
+    while !resumed.contains(&m2) {
+        let chunk = next_sse_chunk(&mut body)
+            .await
+            .expect("resume catch-up delivers m2");
+        resumed.push_str(&chunk);
+    }
+    assert!(!resumed.contains(&m1), "since-cursor must exclude m1");
+    drop(body);
+
+    // Last-Event-ID beats ?since (it is the newer cursor on an SSE auto-reconnect).
+    let mut req = events_req("/api/v1/events?since=0&kinds=library.changed");
+    req.headers_mut().insert(
+        "last-event-id",
+        axum::http::HeaderValue::from_str(&m1_seq.to_string()).unwrap(),
+    );
+    let resp = app.clone().oneshot(req).await.expect("infallible");
+    let mut body = resp.into_body();
+    let mut resumed = String::new();
+    while !resumed.contains(&m2) {
+        let chunk = next_sse_chunk(&mut body)
+            .await
+            .expect("header-resume catch-up delivers m2");
+        resumed.push_str(&chunk);
+    }
+    assert!(!resumed.contains(&m1), "Last-Event-ID must exclude m1");
+    drop(body);
+
+    // A cursor that fell off the ring gets the dropped marker first. Flood the ring past
+    // capacity, then resume from seq 1.
+    for _ in 0..1100 {
+        crate::events::emit(EventKind::DisplayReleased { count: 1 });
+    }
+    let resp = app
+        .clone()
+        .oneshot(events_req("/api/v1/events?since=1"))
+        .await
+        .expect("infallible");
+    let mut body = resp.into_body();
+    let first = next_sse_chunk(&mut body).await.expect("dropped marker");
+    assert!(first.contains("event: dropped"), "first frame: {first}");
+    assert!(
+        first.contains(r#"{"dropped":true}"#),
+        "marker data: {first}"
+    );
+}
+
+#[tokio::test]
+async fn events_stream_connection_cap() {
+    let _l = EVENTS_TEST_LOCK.lock().await;
+    let app = test_app(test_state(), None);
+
+    let slots = super::events::test_support::saturate_slots();
+    let resp = app
+        .clone()
+        .oneshot(events_req("/api/v1/events"))
+        .await
+        .expect("infallible");
+    assert_eq!(resp.status(), StatusCode::SERVICE_UNAVAILABLE);
+    drop(slots);
+
+    let resp = app
+        .clone()
+        .oneshot(events_req("/api/v1/events"))
+        .await
+        .expect("infallible");
+    assert_eq!(resp.status(), StatusCode::OK, "cap frees with the slots");
+}