chore: consolidate all in-progress parallel-session WIP

Wholesale commit of every uncommitted change across the tree, at the user's explicit request — host refactor-campaign W1 (native.rs facade + native/ dir, library/ + mgmt/ splits), Android, core. These streams were mid-flight and not individually built/tested together; this supersedes the per-session HOLD markers. Consolidating so everything lands on main in one pass. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-16 20:08:29 +02:00
parent 07e2836601
commit ecfa71212d
67 changed files with 9456 additions and 8062 deletions
@@ -0,0 +1,449 @@
+//! Host lifecycle event bus (scripting-and-hooks RFC §4, phase 0).
+//!
+//! A process-wide broadcast bus + bounded catch-up ring for **lifecycle events**: client
+//! connect/disconnect, session and stream start/end, pairing decisions, virtual-display
+//! create/release, library mutations, host start/stop. Fire sites on BOTH planes call
+//! [`emit`]; consumers ([`EventBus::subscribe`]) get a ring-backed catch-up plus a live
+//! tail — the shape `GET /api/v1/events` (SSE, phase 1) and the hook runner (phase 2)
+//! consume. Until those land, the bus is host-internal.
+//!
+//! Design notes (mirrors [`crate::log_capture`], the shipped ring precedent):
+//! - Events carry a monotonically increasing `seq` (1-based) and a wall-clock `ts_ms`.
+//!   A consumer resumes with `since = last seen seq`; one that fell off the ring gets
+//!   `dropped = true` and should resync via the REST snapshots.
+//! - The wire shape is **versioned and additive-only** within a major ([`SCHEMA_VERSION`]):
+//!   fields and kinds may be added, never removed or renamed. The JSON snapshot tests below
+//!   are the review gate — a failing snapshot IS a schema change.
+//! - **Payload hygiene: events never carry secrets** — no PINs, no tokens, no key material.
+//!   Client names and fingerprints are fine (already exposed via the management API).
+//! - Emission is fire-and-forget and cheap (a mutex push + a non-blocking broadcast send);
+//!   nothing here sits in a streaming hot path. Slow consumers lag (`RecvError::Lagged`)
+//!   rather than buffering unboundedly; the SSE layer disconnects them.
+
+use serde::{Deserialize, Serialize};
+use std::collections::VecDeque;
+use std::sync::{Mutex, OnceLock};
+use std::time::{SystemTime, UNIX_EPOCH};
+use tokio::sync::broadcast;
+use utoipa::ToSchema;
+
+/// Wire-shape major version, carried on every event. Additive-only within a major; removing
+/// or renaming a field is a new major (and, at the API layer, a new endpoint negotiation).
+pub const SCHEMA_VERSION: u32 = 1;
+
+/// Catch-up ring capacity. Events are small (a few hundred bytes) and low-rate (lifecycle,
+/// not per-frame), so 1024 spans hours of ordinary host activity.
+const RING_CAPACITY: usize = 1024;
+/// Live-tail channel depth per subscriber before a slow consumer starts lagging.
+const BROADCAST_CAPACITY: usize = 256;
+
+/// One host lifecycle event, as it will appear on the wire (`data:` of one SSE frame).
+#[derive(Serialize, Deserialize, ToSchema, Clone, Debug)]
+pub struct HostEvent {
+    /// Monotonic sequence number (1-based) — a consumer resumes with `since = last seen`.
+    pub seq: u64,
+    /// Unix timestamp in milliseconds (the [`crate::log_capture::LogEntry`] convention).
+    pub ts_ms: u64,
+    /// Wire-shape version ([`SCHEMA_VERSION`]).
+    pub schema: u32,
+    /// The event kind + payload, flattened: `"kind": "stream.started", …payload…`.
+    #[serde(flatten)]
+    pub kind: EventKind,
+}
+
+/// Which protocol plane an event originated from. Hooks and scripts filter on it — a hook
+/// that fires for native clients but not Moonlight clients is a bug, not a v2 feature.
+#[derive(Serialize, Deserialize, ToSchema, Clone, Copy, Debug, PartialEq, Eq)]
+#[serde(rename_all = "lowercase")]
+pub enum Plane {
+    /// The native punktfunk/1 plane (QUIC).
+    Native,
+    /// The GameStream/Moonlight compat plane (`--gamestream`).
+    Gamestream,
+}
+
+/// Why a client went away. `Quit` is a deliberate user "stop" (the typed close code);
+/// `Timeout` is a transport idle timeout (the client vanished); `Error` is everything else.
+#[derive(Serialize, Deserialize, ToSchema, Clone, Copy, Debug, PartialEq, Eq)]
+#[serde(rename_all = "lowercase")]
+pub enum DisconnectReason {
+    Quit,
+    Timeout,
+    Error,
+}
+
+/// The connecting/disconnecting client's identity.
+#[derive(Serialize, Deserialize, ToSchema, Clone, Debug)]
+pub struct ClientRef {
+    /// Client-supplied device name; may be empty (an anonymous or compat-plane client).
+    pub name: String,
+    /// Hex SHA-256 certificate fingerprint, when the client presented one.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub fingerprint: Option<String>,
+    pub plane: Plane,
+}
+
+/// A live A/V session (the plane-neutral notion the Dashboard shows).
+#[derive(Serialize, Deserialize, ToSchema, Clone, Debug)]
+pub struct SessionRef {
+    /// Host-local session id (unique within this host process).
+    pub id: u64,
+    /// Short client label (cert-fingerprint prefix, or peer IP for an anonymous client).
+    pub client: String,
+    /// Negotiated mode, `WxH@Hz` (e.g. `"3840x2160@120"`).
+    pub mode: String,
+    pub hdr: bool,
+}
+
+/// A live video stream (what the stream marker file reflects).
+#[derive(Serialize, Deserialize, ToSchema, Clone, Debug)]
+pub struct StreamRef {
+    /// Negotiated mode, `WxH@Hz`.
+    pub mode: String,
+    pub hdr: bool,
+    /// Client-supplied device name; may be empty.
+    pub client: String,
+    /// The launched app/title for this stream, when one was requested (store-qualified id on
+    /// the native plane, app title on the GameStream plane).
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub app: Option<String>,
+    pub plane: Plane,
+}
+
+/// A device in the pairing flow.
+#[derive(Serialize, Deserialize, ToSchema, Clone, Debug)]
+pub struct DeviceRef {
+    /// Sanitized device name (the pairing store's copy).
+    pub name: String,
+    /// Hex certificate fingerprint.
+    pub fingerprint: String,
+    pub plane: Plane,
+}
+
+/// The event catalog (RFC §4). Serialized internally tagged as `"kind": "<domain>.<verb>"`,
+/// flattened into [`HostEvent`]. **Additive-only** within [`SCHEMA_VERSION`].
+#[derive(Serialize, Deserialize, ToSchema, Clone, Debug)]
+#[serde(tag = "kind")]
+pub enum EventKind {
+    #[serde(rename = "client.connected")]
+    ClientConnected { client: ClientRef },
+    #[serde(rename = "client.disconnected")]
+    ClientDisconnected {
+        client: ClientRef,
+        reason: DisconnectReason,
+    },
+    #[serde(rename = "session.started")]
+    SessionStarted { session: SessionRef },
+    #[serde(rename = "session.ended")]
+    SessionEnded { session: SessionRef },
+    #[serde(rename = "stream.started")]
+    StreamStarted { stream: StreamRef },
+    #[serde(rename = "stream.stopped")]
+    StreamStopped { stream: StreamRef },
+    #[serde(rename = "pairing.pending")]
+    PairingPending { device: DeviceRef },
+    #[serde(rename = "pairing.completed")]
+    PairingCompleted { device: DeviceRef },
+    #[serde(rename = "pairing.denied")]
+    PairingDenied { device: DeviceRef },
+    #[serde(rename = "display.created")]
+    DisplayCreated {
+        /// The virtual-display backend that minted it (`VirtualDisplay::name`).
+        backend: String,
+        /// `WxH@Hz`.
+        mode: String,
+    },
+    #[serde(rename = "display.released")]
+    DisplayReleased {
+        /// How many kept displays this release retired.
+        count: u32,
+    },
+    #[serde(rename = "library.changed")]
+    LibraryChanged {
+        /// What mutated the library: `"manual"` today; a provider id once the provider
+        /// API (RFC §8) lands.
+        source: String,
+    },
+    #[serde(rename = "host.started")]
+    HostStarted {
+        version: String,
+        /// Whether the GameStream/Moonlight compat plane is enabled.
+        gamestream: bool,
+    },
+    #[serde(rename = "host.stopping")]
+    HostStopping,
+}
+
+impl EventKind {
+    /// The wire kind string (`"stream.started"`, …) — for filters and log lines.
+    pub fn name(&self) -> &'static str {
+        match self {
+            EventKind::ClientConnected { .. } => "client.connected",
+            EventKind::ClientDisconnected { .. } => "client.disconnected",
+            EventKind::SessionStarted { .. } => "session.started",
+            EventKind::SessionEnded { .. } => "session.ended",
+            EventKind::StreamStarted { .. } => "stream.started",
+            EventKind::StreamStopped { .. } => "stream.stopped",
+            EventKind::PairingPending { .. } => "pairing.pending",
+            EventKind::PairingCompleted { .. } => "pairing.completed",
+            EventKind::PairingDenied { .. } => "pairing.denied",
+            EventKind::DisplayCreated { .. } => "display.created",
+            EventKind::DisplayReleased { .. } => "display.released",
+            EventKind::LibraryChanged { .. } => "library.changed",
+            EventKind::HostStarted { .. } => "host.started",
+            EventKind::HostStopping => "host.stopping",
+        }
+    }
+}
+
+/// Formats a mode as the wire's `WxH@Hz` string.
+pub fn mode_str(width: u32, height: u32, hz: u32) -> String {
+    format!("{width}x{height}@{hz}")
+}
+
+/// One consumer's view: the ring-backed catch-up plus a live-tail receiver, taken atomically
+/// (no event can fall between `catch_up` and the first `rx.recv()`, and none is in both).
+pub struct Subscription {
+    /// Events with `seq > since`, oldest first.
+    pub catch_up: Vec<HostEvent>,
+    /// True when events between `since` and the first caught-up one were already evicted —
+    /// the consumer should resync via the REST snapshots (the `LogPage.dropped` contract).
+    pub dropped: bool,
+    /// The live tail. A consumer that can't keep up sees `RecvError::Lagged`.
+    pub rx: broadcast::Receiver<HostEvent>,
+}
+
+/// The process-wide event bus: a bounded seq-numbered ring (catch-up) + a broadcast channel
+/// (live tail).
+pub struct EventBus {
+    inner: Mutex<Ring>,
+    tx: broadcast::Sender<HostEvent>,
+}
+
+struct Ring {
+    events: VecDeque<HostEvent>,
+    next_seq: u64,
+}
+
+impl EventBus {
+    fn new() -> Self {
+        let (tx, _) = broadcast::channel(BROADCAST_CAPACITY);
+        Self {
+            inner: Mutex::new(Ring {
+                events: VecDeque::with_capacity(RING_CAPACITY),
+                next_seq: 1,
+            }),
+            tx,
+        }
+    }
+
+    /// Stamp, ring-buffer, and broadcast one event. Fire-and-forget: no receivers is fine
+    /// (the ring still records it for a later subscriber's catch-up).
+    pub fn emit(&self, kind: EventKind) {
+        let ts_ms = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map(|d| d.as_millis() as u64)
+            .unwrap_or(0);
+        let mut ring = self.inner.lock().unwrap_or_else(|e| e.into_inner());
+        let ev = HostEvent {
+            seq: ring.next_seq,
+            ts_ms,
+            schema: SCHEMA_VERSION,
+            kind,
+        };
+        ring.next_seq += 1;
+        if ring.events.len() == RING_CAPACITY {
+            ring.events.pop_front();
+        }
+        ring.events.push_back(ev.clone());
+        // Send while still holding the ring lock: it serializes with `subscribe` (which also
+        // takes the lock), so an event lands either in a subscriber's catch-up or on its live
+        // tail — never both, never neither. `send` is non-blocking, the hold is trivial.
+        let _ = self.tx.send(ev);
+    }
+
+    /// Subscribe with a resume cursor: events with `seq > since` come back as catch-up, the
+    /// returned receiver carries everything after. `since = 0` means "from the ring start".
+    pub fn subscribe(&self, since: u64) -> Subscription {
+        let ring = self.inner.lock().unwrap_or_else(|e| e.into_inner());
+        let rx = self.tx.subscribe();
+        let first_seq = ring.events.front().map_or(ring.next_seq, |e| e.seq);
+        let dropped = since != 0 && since + 1 < first_seq;
+        let catch_up = ring
+            .events
+            .iter()
+            .filter(|e| e.seq > since)
+            .cloned()
+            .collect();
+        Subscription {
+            catch_up,
+            dropped,
+            rx,
+        }
+    }
+}
+
+/// The process-wide bus — a `OnceLock` singleton (the [`crate::log_capture::ring`] shape) so
+/// fire sites across both planes and the API layer share it without threading an `Arc`.
+pub fn bus() -> &'static EventBus {
+    static BUS: OnceLock<EventBus> = OnceLock::new();
+    BUS.get_or_init(EventBus::new)
+}
+
+/// Emit one lifecycle event on the process-wide bus. Cheap and non-blocking — safe from any
+/// thread, including RAII `Drop` paths.
+pub fn emit(kind: EventKind) {
+    bus().emit(kind);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn ev(name: &str) -> EventKind {
+        EventKind::LibraryChanged {
+            source: name.to_string(),
+        }
+    }
+
+    #[test]
+    fn seq_is_monotonic_and_catch_up_resumes() {
+        let bus = EventBus::new();
+        for i in 0..5 {
+            bus.emit(ev(&format!("m{i}")));
+        }
+        let sub = bus.subscribe(0);
+        assert_eq!(
+            sub.catch_up.iter().map(|e| e.seq).collect::<Vec<_>>(),
+            vec![1, 2, 3, 4, 5]
+        );
+        assert!(!sub.dropped);
+        assert!(sub.catch_up.iter().all(|e| e.schema == SCHEMA_VERSION));
+
+        // Resume from a cursor mid-ring.
+        let sub = bus.subscribe(3);
+        assert_eq!(
+            sub.catch_up.iter().map(|e| e.seq).collect::<Vec<_>>(),
+            vec![4, 5]
+        );
+        assert!(!sub.dropped);
+
+        // Cursor at the tip: empty catch-up, not a gap.
+        let sub = bus.subscribe(5);
+        assert!(sub.catch_up.is_empty());
+        assert!(!sub.dropped);
+    }
+
+    #[test]
+    fn eviction_reports_dropped() {
+        let bus = EventBus::new();
+        for i in 0..(RING_CAPACITY + 50) {
+            bus.emit(ev(&format!("m{i}")));
+        }
+        // Seqs 1..=50 were evicted; a cursor inside the gap must flag it.
+        let sub = bus.subscribe(10);
+        assert!(sub.dropped);
+        assert_eq!(sub.catch_up.first().map(|e| e.seq), Some(51));
+        // A fresh consumer (since = 0) is a backfill, not a gap.
+        let sub = bus.subscribe(0);
+        assert!(!sub.dropped);
+        assert_eq!(sub.catch_up.len(), RING_CAPACITY);
+    }
+
+    #[tokio::test]
+    async fn live_tail_continues_exactly_after_catch_up() {
+        let bus = EventBus::new();
+        bus.emit(ev("before-1"));
+        bus.emit(ev("before-2"));
+        let mut sub = bus.subscribe(0);
+        assert_eq!(sub.catch_up.len(), 2);
+        // Emitted after subscribe → on the live tail only, starting at exactly seq 3.
+        bus.emit(ev("after"));
+        let live = sub.rx.recv().await.expect("live event");
+        assert_eq!(live.seq, 3);
+        assert_eq!(live.kind.name(), "library.changed");
+        // Nothing duplicated: the tail holds only what wasn't in the catch-up.
+        assert!(sub.rx.try_recv().is_err());
+    }
+
+    /// The wire shape IS the contract (additive-only, RFC §4): these snapshots are the review
+    /// gate — if one fails, the change renames/removes a field and needs a schema-version bump,
+    /// not a test update.
+    #[test]
+    fn wire_shape_snapshots() {
+        let ev = HostEvent {
+            seq: 4182,
+            ts_ms: 1_700_000_000_000,
+            schema: 1,
+            kind: EventKind::StreamStarted {
+                stream: StreamRef {
+                    mode: mode_str(3840, 2160, 120),
+                    hdr: true,
+                    client: "Living Room TV".into(),
+                    app: Some("steam:570".into()),
+                    plane: Plane::Native,
+                },
+            },
+        };
+        assert_eq!(
+            serde_json::to_string(&ev).unwrap(),
+            r#"{"seq":4182,"ts_ms":1700000000000,"schema":1,"kind":"stream.started","stream":{"mode":"3840x2160@120","hdr":true,"client":"Living Room TV","app":"steam:570","plane":"native"}}"#
+        );
+
+        let ev = HostEvent {
+            seq: 1,
+            ts_ms: 1_700_000_000_000,
+            schema: 1,
+            kind: EventKind::ClientDisconnected {
+                client: ClientRef {
+                    name: "Deck".into(),
+                    fingerprint: Some("b1c2".into()),
+                    plane: Plane::Gamestream,
+                },
+                reason: DisconnectReason::Timeout,
+            },
+        };
+        assert_eq!(
+            serde_json::to_string(&ev).unwrap(),
+            r#"{"seq":1,"ts_ms":1700000000000,"schema":1,"kind":"client.disconnected","client":{"name":"Deck","fingerprint":"b1c2","plane":"gamestream"},"reason":"timeout"}"#
+        );
+
+        let ev = HostEvent {
+            seq: 2,
+            ts_ms: 1_700_000_000_000,
+            schema: 1,
+            kind: EventKind::HostStopping,
+        };
+        assert_eq!(
+            serde_json::to_string(&ev).unwrap(),
+            r#"{"seq":2,"ts_ms":1700000000000,"schema":1,"kind":"host.stopping"}"#
+        );
+    }
+
+    #[test]
+    fn wire_shape_roundtrips() {
+        let ev = HostEvent {
+            seq: 7,
+            ts_ms: 3,
+            schema: 1,
+            kind: EventKind::PairingPending {
+                device: DeviceRef {
+                    name: "iPad Pro".into(),
+                    fingerprint: "ab12".into(),
+                    plane: Plane::Native,
+                },
+            },
+        };
+        let json = serde_json::to_string(&ev).unwrap();
+        let back: HostEvent = serde_json::from_str(&json).unwrap();
+        assert_eq!(back.seq, 7);
+        assert_eq!(back.kind.name(), "pairing.pending");
+        match back.kind {
+            EventKind::PairingPending { device } => {
+                assert_eq!(device.name, "iPad Pro");
+                assert_eq!(device.plane, Plane::Native);
+            }
+            other => panic!("wrong kind: {other:?}"),
+        }
+    }
+}