push terminal session changes and alert evaluator health

wakey-control-plane/src/api/alerts.rs

@@ -8,6 +8,7 @@ use axum::extract::{Query, State};
 use axum::http::StatusCode;
 use axum::response::IntoResponse;
 use serde::Deserialize;
+use tokio::select;
 use tokio::sync::{RwLock, broadcast};
 use tokio::time::Duration;
 use tracing::warn;
@@ -105,15 +106,16 @@ pub struct AlertsSnapshot {
 #[derive(Clone)]
 pub struct AlertsHub {
     tx: broadcast::Sender<String>,
-    latest_payload: Arc<RwLock<Option<String>>>,
-    latest_alerts: Arc<RwLock<Vec<AlertState>>>,
-    evaluator_status: Arc<RwLock<EvaluatorStatus>>,
+    status: Arc<RwLock<HubStatus>>,
 }
 
 #[derive(Default)]
-struct EvaluatorStatus {
+struct HubStatus {
+    latest_payload: Option<String>,
+    latest_alerts: Vec<AlertState>,
     last_success_unix: Option<u64>,
     failure: Option<String>,
+    last_health_available: Option<bool>,
 }
 
 const ALERT_SNAPSHOT_MAX_AGE_SECS: u64 = 15;
@@ -129,9 +131,7 @@ impl AlertsHub {
         let (tx, _) = broadcast::channel(16);
         Self {
             tx,
-            latest_payload: Arc::new(RwLock::new(None)),
-            latest_alerts: Arc::new(RwLock::new(Vec::new())),
-            evaluator_status: Arc::new(RwLock::new(EvaluatorStatus::default())),
+            status: Arc::new(RwLock::new(HubStatus::default())),
         }
     }
 
@@ -140,27 +140,49 @@ impl AlertsHub {
     }
 
     pub async fn latest_alerts(&self) -> Vec<AlertState> {
-        self.latest_alerts.read().await.clone()
-    }
-
-    pub async fn latest_payload(&self) -> Option<String> {
-        self.latest_payload.read().await.clone()
+        self.status.read().await.latest_alerts.clone()
     }
 
     async fn mark_unavailable(&self, reason: impl Into<String>) {
-        self.evaluator_status.write().await.failure = Some(reason.into());
+        let mut status = self.status.write().await;
+        if status.failure.is_some() {
+            return;
+        }
+        status.failure = Some(reason.into());
+        let reason = status.failure.as_deref().unwrap_or_default().to_owned();
+        status.last_health_available = Some(false);
+        self.broadcast_health(false, Some(reason));
     }
 
     async fn unavailable_reason(&self) -> Option<String> {
-        let status = self.evaluator_status.read().await;
-        if let Some(failure) = &status.failure {
-            return Some(failure.clone());
+        let status = self.status.read().await;
+        evaluator_unavailable_reason(&status, now_unix())
+    }
+
+    async fn resync_payloads(&self) -> (String, Option<String>) {
+        let status = self.status.read().await;
+        (
+            encode_health(evaluator_unavailable_reason(&status, now_unix())),
+            status.latest_payload.clone(),
+        )
+    }
+
+    async fn broadcast_health_if_changed(&self) {
+        let mut status = self.status.write().await;
+        if status.last_success_unix.is_none() || status.failure.is_some() {
+            return;
         }
-        match status.last_success_unix {
-            Some(ts) if now_unix().saturating_sub(ts) <= ALERT_SNAPSHOT_MAX_AGE_SECS => None,
-            Some(_) => Some("alert evaluator snapshot is stale".into()),
-            None => Some("alert evaluator has not produced a snapshot yet".into()),
+        let reason = evaluator_unavailable_reason(&status, now_unix());
+        let available = reason.is_none();
+        if status.last_health_available == Some(available) {
+            return;
         }
+        status.last_health_available = Some(available);
+        self.broadcast_health(available, reason);
+    }
+
+    fn broadcast_health(&self, available: bool, reason: Option<String>) {
+        let _ = self.tx.send(encode_health_message(available, reason));
     }
 
     pub async fn publish(&self, snapshot: AlertsSnapshot) {
@@ -174,16 +196,49 @@ impl AlertsHub {
             warn!("failed to encode alerts snapshot payload");
             return;
         };
-        *self.latest_alerts.write().await = snapshot.alerts;
-        *self.latest_payload.write().await = Some(encoded.clone());
-        let mut status = self.evaluator_status.write().await;
+        let mut status = self.status.write().await;
+        let was_unavailable = evaluator_unavailable_reason(&status, now_unix()).is_some();
+        status.latest_alerts = snapshot.alerts;
+        status.latest_payload = Some(encoded.clone());
         status.last_success_unix = Some(snapshot.ts_unix);
         status.failure = None;
-        drop(status);
+        let available = evaluator_unavailable_reason(&status, now_unix()).is_none();
+        if available {
+            status.last_health_available = Some(true);
+        }
+        if was_unavailable && available {
+            self.broadcast_health(true, None);
+        }
         let _ = self.tx.send(encoded);
     }
 }
 
+fn evaluator_unavailable_reason(status: &HubStatus, now: u64) -> Option<String> {
+    if let Some(failure) = &status.failure {
+        return Some(failure.clone());
+    }
+    match status.last_success_unix {
+        Some(ts) if now.saturating_sub(ts) <= ALERT_SNAPSHOT_MAX_AGE_SECS => None,
+        Some(_) => Some("alert evaluator snapshot is stale".into()),
+        None => Some("alert evaluator has not produced a snapshot yet".into()),
+    }
+}
+
+fn encode_health(reason: Option<String>) -> String {
+    encode_health_message(reason.is_none(), reason)
+}
+
+fn encode_health_message(available: bool, reason: Option<String>) -> String {
+    let mut message = serde_json::json!({
+        "type": "alerts_health",
+        "available": available,
+    });
+    if let Some(reason) = reason {
+        message["reason"] = serde_json::Value::String(reason);
+    }
+    message.to_string()
+}
+
 /// Background task that owns alert evaluation: on a fixed ticker it recomputes
 /// the snapshot, persists transitions, and publishes to subscribers so neither
 /// the GET endpoint nor individual WS clients evaluate alerts themselves.
@@ -271,22 +326,49 @@ pub async fn alerts_stream(
 
 async fn alerts_stream_socket(state: AppState, mut socket: WebSocket) {
     let mut rx = state.alerts.subscribe();
+    let mut freshness_tick = tokio::time::interval(Duration::from_secs(1));
+    freshness_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
-    if let Some(encoded) = state.alerts.latest_payload().await
+    let (health, snapshot) = state.alerts.resync_payloads().await;
+    if socket.send(Message::Text(health.into())).await.is_err() {
+        return;
+    }
+    if let Some(encoded) = snapshot
         && socket.send(Message::Text(encoded.into())).await.is_err()
     {
         return;
     }
 
     loop {
-        match rx.recv().await {
-            Ok(encoded) => {
-                if socket.send(Message::Text(encoded.into())).await.is_err() {
-                    break;
+        select! {
+            incoming = socket.recv() => {
+                match incoming {
+                    Some(Ok(Message::Close(_))) | None | Some(Err(_)) => break,
+                    Some(Ok(_)) => {}
                 }
             }
-            Err(broadcast::error::RecvError::Lagged(_)) => continue,
-            Err(broadcast::error::RecvError::Closed) => break,
+            update = rx.recv() => {
+                match update {
+                    Ok(encoded) => {
+                        if socket.send(Message::Text(encoded.into())).await.is_err() {
+                            break;
+                        }
+                    }
+                    Err(broadcast::error::RecvError::Lagged(_)) => {
+                        let (health, snapshot) = state.alerts.resync_payloads().await;
+                        if socket.send(Message::Text(health.into())).await.is_err() {
+                            break;
+                        }
+                        if let Some(encoded) = snapshot
+                            && socket.send(Message::Text(encoded.into())).await.is_err()
+                        {
+                            break;
+                        }
+                    }
+                    Err(broadcast::error::RecvError::Closed) => break,
+                }
+            }
+            _ = freshness_tick.tick() => state.alerts.broadcast_health_if_changed().await,
         }
     }
 }
@@ -648,12 +730,24 @@ mod tests {
     async fn alert_hub_reports_initial_failure_and_recovers_on_snapshot() {
         let hub = AlertsHub::new();
         assert!(hub.unavailable_reason().await.is_some());
+        let mut updates = hub.subscribe();
 
         hub.mark_unavailable("store query failed").await;
         assert_eq!(
             hub.unavailable_reason().await.as_deref(),
             Some("store query failed")
         );
+        let failure: serde_json::Value =
+            serde_json::from_str(&updates.try_recv().unwrap()).unwrap();
+        assert_eq!(failure["type"], "alerts_health");
+        assert_eq!(failure["available"], false);
+        assert_eq!(failure["reason"], "store query failed");
+
+        hub.mark_unavailable("same outage, later attempt").await;
+        assert!(
+            updates.try_recv().is_err(),
+            "failure notifications deduplicate"
+        );
 
         hub.publish(AlertsSnapshot {
             ts_unix: super::now_unix(),
@@ -662,6 +756,14 @@ mod tests {
         })
         .await;
         assert_eq!(hub.unavailable_reason().await, None);
+        let recovery: serde_json::Value =
+            serde_json::from_str(&updates.try_recv().unwrap()).unwrap();
+        assert_eq!(recovery["type"], "alerts_health");
+        assert_eq!(recovery["available"], true);
+        assert_eq!(recovery.get("reason"), None);
+        let snapshot: serde_json::Value =
+            serde_json::from_str(&updates.try_recv().unwrap()).unwrap();
+        assert_eq!(snapshot["type"], "alerts_snapshot");
     }
 
     #[tokio::test]
@@ -678,4 +780,49 @@ mod tests {
             Some("alert evaluator snapshot is stale")
         );
     }
+
+    #[tokio::test]
+    async fn alert_hub_initial_resync_marks_cached_snapshot_stale() {
+        let hub = AlertsHub::new();
+        let (initial_health, initial_snapshot) = hub.resync_payloads().await;
+        let initial_health: serde_json::Value = serde_json::from_str(&initial_health).unwrap();
+        assert_eq!(initial_health["available"], false);
+        assert_eq!(
+            initial_health["reason"],
+            "alert evaluator has not produced a snapshot yet"
+        );
+        assert!(initial_snapshot.is_none());
+
+        hub.publish(AlertsSnapshot {
+            ts_unix: 0,
+            alerts: Vec::new(),
+            recent_transitions: Vec::new(),
+        })
+        .await;
+        let (stale_health, cached_snapshot) = hub.resync_payloads().await;
+        let stale_health: serde_json::Value = serde_json::from_str(&stale_health).unwrap();
+        assert_eq!(stale_health["available"], false);
+        assert_eq!(stale_health["reason"], "alert evaluator snapshot is stale");
+        assert!(cached_snapshot.is_some());
+    }
+
+    #[tokio::test]
+    async fn alert_hub_broadcasts_stale_transition_once() {
+        let hub = AlertsHub::new();
+        let mut updates = hub.subscribe();
+        hub.publish(AlertsSnapshot {
+            ts_unix: 0,
+            alerts: Vec::new(),
+            recent_transitions: Vec::new(),
+        })
+        .await;
+        let _snapshot = updates.try_recv().unwrap();
+
+        hub.broadcast_health_if_changed().await;
+        let stale: serde_json::Value = serde_json::from_str(&updates.try_recv().unwrap()).unwrap();
+        assert_eq!(stale["available"], false);
+        assert_eq!(stale["reason"], "alert evaluator snapshot is stale");
+        hub.broadcast_health_if_changed().await;
+        assert!(updates.try_recv().is_err());
+    }
 }

wakey-control-plane/src/api/mod.rs

@@ -19,7 +19,7 @@ pub use control::{
 };
 pub use terminals::{
     agent_terminal_ws, attach_terminal, close_terminal, create_terminal, get_terminal,
-    list_terminals, operator_terminal_ws,
+    list_terminals, operator_terminal_ws, terminal_events_ws,
 };
 
 use axum::response::{IntoResponse, Response};

wakey-control-plane/src/api/terminals.rs

@@ -6,6 +6,7 @@ use axum::response::Response;
 use futures_util::{SinkExt, StreamExt};
 use serde::{Deserialize, Serialize};
 use std::time::Duration;
+use tokio::sync::broadcast;
 use tracing::{info, warn};
 use wakey_agent::protocol::{
     AgentCapability, DEFAULT_TERMINAL_MAX_SESSIONS, DEFAULT_TERMINAL_SESSION_TTL_SECONDS,
@@ -244,6 +245,49 @@ pub async fn operator_terminal_ws(
         .on_upgrade(move |socket| handle_operator_terminal_socket(state, terminal_id, socket))
 }
 
+pub async fn terminal_events_ws(ws: WebSocketUpgrade, State(state): State<AppState>) -> Response {
+    ws.on_upgrade(move |socket| handle_terminal_events_socket(state, socket))
+}
+
+async fn handle_terminal_events_socket(state: AppState, socket: WebSocket) {
+    let mut events = state.terminals.subscribe();
+    let (mut write, mut read) = socket.split();
+    if write
+        .send(Message::Text(r#"{"type":"sessions_changed"}"#.into()))
+        .await
+        .is_err()
+    {
+        return;
+    }
+
+    loop {
+        tokio::select! {
+            event = events.recv() => {
+                match event {
+                    Ok(()) | Err(broadcast::error::RecvError::Lagged(_)) => {
+                        if write.send(Message::Text(r#"{"type":"sessions_changed"}"#.into())).await.is_err() {
+                            break;
+                        }
+                    }
+                    Err(broadcast::error::RecvError::Closed) => break,
+                }
+            }
+            incoming = read.next() => {
+                match incoming {
+                    Some(Ok(Message::Ping(payload))) => {
+                        if write.send(Message::Pong(payload)).await.is_err() {
+                            break;
+                        }
+                    }
+                    Some(Ok(Message::Pong(_))) => {}
+                    Some(Ok(Message::Close(_))) | None | Some(Err(_)) => break,
+                    Some(Ok(Message::Text(_))) | Some(Ok(Message::Binary(_))) => {}
+                }
+            }
+        }
+    }
+}
+
 async fn handle_agent_terminal_socket(state: AppState, terminal_id: String, mut socket: WebSocket) {
     let auth = match receive_text_handshake(&mut socket).await.and_then(|text| {
         serde_json::from_str::<TerminalAgentHandshake>(&text).map_err(|_| "invalid handshake")

wakey-control-plane/src/runtime/mod.rs

@@ -148,6 +148,10 @@ fn control_api_routes() -> Router<AppState> {
             "/api/v1/control/terminals",
             get(api::list_terminals).post(api::create_terminal),
         )
+        .route(
+            "/api/v1/control/terminals/events/ws",
+            get(api::terminal_events_ws),
+        )
         .route(
             "/api/v1/control/terminals/{terminal_id}",
             get(api::get_terminal).delete(api::close_terminal),

wakey-control-plane/src/runtime/terminals.rs

@@ -2,7 +2,7 @@ use std::collections::{HashMap, VecDeque};
 use std::sync::Arc;
 use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
-use tokio::sync::{Mutex, mpsc};
+use tokio::sync::{Mutex, broadcast, mpsc};
 use uuid::Uuid;
 use wakey_agent::protocol::{AgentTerminalSession, TerminalId};
 
@@ -12,6 +12,7 @@ pub const TERMINAL_PENDING_AGENT_BYTES: usize = 256 * 1024;
 pub const TERMINAL_ATTACH_TIMEOUT: Duration = Duration::from_secs(10);
 const TERMINAL_TOMBSTONE_TTL: Duration = Duration::from_secs(5 * 60);
 const TERMINAL_MAX_TOMBSTONES: usize = 1024;
+const TERMINAL_EVENTS_CAPACITY: usize = 32;
 
 #[derive(Clone, Debug)]
 pub enum TerminalRelayFrame {
@@ -24,6 +25,7 @@ pub enum TerminalRelayFrame {
 pub struct TerminalRegistry {
     inner: Arc<Mutex<HashMap<String, TerminalSession>>>,
     closed: Arc<Mutex<HashMap<String, Instant>>>,
+    events: broadcast::Sender<()>,
 }
 
 struct TerminalSession {
@@ -69,12 +71,22 @@ impl Default for TerminalRegistry {
 
 impl TerminalRegistry {
     pub fn new() -> Self {
+        let (events, _) = broadcast::channel(TERMINAL_EVENTS_CAPACITY);
         Self {
             inner: Arc::new(Mutex::new(HashMap::new())),
             closed: Arc::new(Mutex::new(HashMap::new())),
+            events,
         }
     }
 
+    pub fn subscribe(&self) -> broadcast::Receiver<()> {
+        self.events.subscribe()
+    }
+
+    fn notify_changed(&self) {
+        let _ = self.events.send(());
+    }
+
     #[cfg(test)]
     async fn create(&self, agent_id: String) -> Result<CreatedTerminal, &'static str> {
         self.create_with_limits(
@@ -95,13 +107,17 @@ impl TerminalRegistry {
         session_ttl_seconds: u64,
     ) -> Result<CreatedTerminal, &'static str> {
         let mut sessions = self.inner.lock().await;
-        prune_expired_sessions(&mut sessions);
+        let pruned = prune_expired_sessions(&mut sessions) > 0;
         if sessions
             .values()
             .filter(|session| session.agent_id == agent_id)
             .count()
             >= max_sessions.max(1)
         {
+            drop(sessions);
+            if pruned {
+                self.notify_changed();
+            }
             return Err("agent_terminal_limit_reached");
         }
 
@@ -112,8 +128,13 @@ impl TerminalRegistry {
             .duration_since(UNIX_EPOCH)
             .unwrap_or_default()
             .as_secs();
-        let expiry = expiry_for_created_at(created_at_unix, session_ttl_seconds)
-            .ok_or("terminal_ttl_invalid")?;
+        let Some(expiry) = expiry_for_created_at(created_at_unix, session_ttl_seconds) else {
+            drop(sessions);
+            if pruned {
+                self.notify_changed();
+            }
+            return Err("terminal_ttl_invalid");
+        };
         sessions.insert(
             terminal_id.clone(),
             TerminalSession {
@@ -133,6 +154,8 @@ impl TerminalRegistry {
                 operator_detached_at: None,
             },
         );
+        drop(sessions);
+        self.notify_changed();
 
         Ok(CreatedTerminal {
             terminal_id,
@@ -145,6 +168,7 @@ impl TerminalRegistry {
 
     pub async fn remove(&self, terminal_id: &str) -> Option<String> {
         let session = self.inner.lock().await.remove(terminal_id)?;
+        self.notify_changed();
         self.remember_closed(terminal_id).await;
         if let Some(tx) = session.agent_tx {
             let _ =
@@ -196,10 +220,10 @@ impl TerminalRegistry {
             })
             .map(|session| session.terminal_id.as_str())
             .collect::<std::collections::HashSet<_>>();
-        let stale_ids = {
+        let (stale_ids, pruned) = {
             let mut sessions = self.inner.lock().await;
-            prune_expired_sessions(&mut sessions);
-            sessions
+            let pruned = prune_expired_sessions(&mut sessions) > 0;
+            let stale_ids = sessions
                 .iter()
                 .filter(|(terminal_id, session)| {
                     session.agent_id == agent_id
@@ -207,13 +231,18 @@ impl TerminalRegistry {
                         && !reported_ids.contains(terminal_id.as_str())
                 })
                 .map(|(terminal_id, _)| terminal_id.clone())
-                .collect::<Vec<_>>()
+                .collect::<Vec<_>>();
+            (stale_ids, pruned)
         };
+        if pruned {
+            self.notify_changed();
+        }
         for terminal_id in stale_ids {
             self.remove(&terminal_id).await;
         }
 
         let mut credentials = Vec::new();
+        let mut changed = false;
         let mut sessions = self.inner.lock().await;
         for reported_session in reported {
             let Some(expiry) = expiry_for_created_at(
@@ -223,9 +252,9 @@ impl TerminalRegistry {
                 continue;
             };
             let terminal_id = reported_session.terminal_id.as_str().to_string();
-            let session = sessions
-                .entry(terminal_id)
-                .or_insert_with(|| TerminalSession {
+            let session = sessions.entry(terminal_id).or_insert_with(|| {
+                changed = true;
+                TerminalSession {
                     agent_id: agent_id.to_string(),
                     created_at_unix: reported_session.created_at_unix,
                     expiry,
@@ -240,7 +269,8 @@ impl TerminalRegistry {
                     operator_id: None,
                     operator_connection_id: None,
                     operator_detached_at: Some(Instant::now()),
-                });
+                }
+            });
             if session.agent_id != agent_id || session.agent_tx.is_some() {
                 continue;
             }
@@ -249,14 +279,24 @@ impl TerminalRegistry {
             session.relay_token = Some(token.clone());
             credentials.push((reported_session.terminal_id.clone(), token));
         }
+        drop(sessions);
+        if changed {
+            self.notify_changed();
+        }
         credentials
     }
 
     pub async fn detach_agent(&self, terminal_id: &str) -> Option<String> {
         let mut sessions = self.inner.lock().await;
         let session = sessions.get_mut(terminal_id)?;
+        let changed = session.agent_tx.is_some();
         session.agent_tx = None;
-        Some(session.agent_id.clone())
+        let agent_id = session.agent_id.clone();
+        drop(sessions);
+        if changed {
+            self.notify_changed();
+        }
+        Some(agent_id)
     }
 
     /// Issues a token while atomically handing this operator's attachment to
@@ -269,17 +309,21 @@ impl TerminalRegistry {
     ) -> Result<String, &'static str> {
         validate_operator_id(operator_id)?;
         let mut sessions = self.inner.lock().await;
-        let target = active_session(&mut sessions, terminal_id)?;
+        let target = active_session(&mut sessions, terminal_id, &self.events)?;
         if target.operator_tx.is_some() && target.operator_id.as_deref() != Some(operator_id) {
             return Err("terminal_operator_already_attached");
         }
 
-        detach_operator_sessions(&mut sessions, operator_id);
+        let detached = detach_operator_sessions(&mut sessions, operator_id);
 
-        let session = active_session(&mut sessions, terminal_id)?;
+        let session = active_session(&mut sessions, terminal_id, &self.events)?;
         let token = new_token();
         session.attachment_token = Some(token.clone());
         session.attachment_operator_id = Some(operator_id.to_string());
+        drop(sessions);
+        if detached {
+            self.notify_changed();
+        }
         Ok(token)
     }
 
@@ -290,7 +334,7 @@ impl TerminalRegistry {
         relay_token: &str,
     ) -> Result<(mpsc::Receiver<TerminalRelayFrame>, Vec<TerminalRelayFrame>), &'static str> {
         let mut sessions = self.inner.lock().await;
-        let session = active_session(&mut sessions, terminal_id)?;
+        let session = active_session(&mut sessions, terminal_id, &self.events)?;
         if session.agent_id != agent_id {
             return Err("terminal_agent_mismatch");
         }
@@ -306,6 +350,8 @@ impl TerminalRegistry {
         session.agent_tx = Some(tx);
         let pending = session.pending_agent.drain(..).collect();
         session.pending_agent_bytes = 0;
+        drop(sessions);
+        self.notify_changed();
         Ok((rx, pending))
     }
 
@@ -317,7 +363,7 @@ impl TerminalRegistry {
     ) -> Result<mpsc::Receiver<TerminalRelayFrame>, &'static str> {
         validate_operator_id(operator_id)?;
         let mut sessions = self.inner.lock().await;
-        let session = active_session(&mut sessions, terminal_id)?;
+        let session = active_session(&mut sessions, terminal_id, &self.events)?;
         if session.operator_tx.is_some() {
             return Err("terminal_operator_already_attached");
         }
@@ -336,7 +382,7 @@ impl TerminalRegistry {
         // WebSocket presents that token, it still atomically releases any
         // older session held by the same browser tab.
         detach_operator_sessions(&mut sessions, operator_id);
-        let session = active_session(&mut sessions, terminal_id)?;
+        let session = active_session(&mut sessions, terminal_id, &self.events)?;
         session.attachment_token = None;
         session.attachment_operator_id = None;
         session.operator_detached_at = None;
@@ -344,6 +390,8 @@ impl TerminalRegistry {
         session.operator_tx = Some(tx);
         session.operator_id = Some(operator_id.to_string());
         session.operator_connection_id = Some(attachment_token.to_string());
+        drop(sessions);
+        self.notify_changed();
         Ok(rx)
     }
 
@@ -357,7 +405,7 @@ impl TerminalRegistry {
         // leave input stranded in the pre-attachment queue.
         let tx = {
             let mut sessions = self.inner.lock().await;
-            let session = active_session(&mut sessions, terminal_id)?;
+            let session = active_session(&mut sessions, terminal_id, &self.events)?;
             if let Some(tx) = session.agent_tx.clone() {
                 tx
             } else {
@@ -385,7 +433,7 @@ impl TerminalRegistry {
     ) -> Result<(), &'static str> {
         let operator_tx = {
             let mut sessions = self.inner.lock().await;
-            let session = active_session(&mut sessions, terminal_id)?;
+            let session = active_session(&mut sessions, terminal_id, &self.events)?;
             session.operator_tx.clone()
         };
 
@@ -397,13 +445,19 @@ impl TerminalRegistry {
                     // Detach its stale sender; the agent's parsed screen remains
                     // authoritative and will reconstruct the next attachment.
                     let mut sessions = self.inner.lock().await;
-                    let session = active_session(&mut sessions, terminal_id)?;
+                    let session = active_session(&mut sessions, terminal_id, &self.events)?;
+                    let mut changed = false;
                     if session
                         .operator_tx
                         .as_ref()
                         .is_some_and(|current| current.same_channel(&tx))
                     {
                         clear_operator_attachment(session);
+                        changed = true;
+                    }
+                    drop(sessions);
+                    if changed {
+                        self.notify_changed();
                     }
                     return Ok(());
                 }
@@ -445,25 +499,36 @@ impl TerminalRegistry {
         }
         let detached_at = Instant::now();
         clear_operator_attachment(session);
+        drop(sessions);
+        self.notify_changed();
         Some(detached_at)
     }
 
     pub async fn summary(&self, terminal_id: &str) -> Option<TerminalSummary> {
         let mut sessions = self.inner.lock().await;
-        prune_expired_sessions(&mut sessions);
-        sessions
+        let pruned = prune_expired_sessions(&mut sessions) > 0;
+        let summary = sessions
             .get(terminal_id)
-            .map(|session| terminal_summary(terminal_id, session))
+            .map(|session| terminal_summary(terminal_id, session));
+        drop(sessions);
+        if pruned {
+            self.notify_changed();
+        }
+        summary
     }
 
     pub async fn summaries(&self) -> Vec<TerminalSummary> {
         let mut sessions = self.inner.lock().await;
-        prune_expired_sessions(&mut sessions);
+        let removed = prune_expired_sessions(&mut sessions);
         let mut summaries = sessions
             .iter()
             .map(|(terminal_id, session)| terminal_summary(terminal_id, session))
             .collect::<Vec<_>>();
         summaries.sort_by_key(|session| std::cmp::Reverse(session.created_at_unix));
+        drop(sessions);
+        if removed > 0 {
+            self.notify_changed();
+        }
         summaries
     }
 }
@@ -482,12 +547,14 @@ fn terminal_summary(terminal_id: &str, session: &TerminalSession) -> TerminalSum
 fn active_session<'a>(
     sessions: &'a mut HashMap<String, TerminalSession>,
     terminal_id: &str,
+    events: &broadcast::Sender<()>,
 ) -> Result<&'a mut TerminalSession, &'static str> {
     if sessions
         .get(terminal_id)
         .is_some_and(|session| session.expiry.is_expired(Instant::now()))
     {
         sessions.remove(terminal_id);
+        let _ = events.send(());
         return Err("terminal_expired");
     }
     sessions.get_mut(terminal_id).ok_or("terminal_not_found")
@@ -500,12 +567,18 @@ fn validate_operator_id(operator_id: &str) -> Result<(), &'static str> {
     Ok(())
 }
 
-fn detach_operator_sessions(sessions: &mut HashMap<String, TerminalSession>, operator_id: &str) {
+fn detach_operator_sessions(
+    sessions: &mut HashMap<String, TerminalSession>,
+    operator_id: &str,
+) -> bool {
+    let mut detached = false;
     for session in sessions.values_mut() {
         if session.operator_id.as_deref() == Some(operator_id) {
             clear_operator_attachment(session);
+            detached = true;
         }
     }
+    detached
 }
 
 fn clear_operator_attachment(session: &mut TerminalSession) {
@@ -523,9 +596,11 @@ fn prune_tombstones(closed: &mut HashMap<String, Instant>) {
     closed.retain(|_, closed_at| closed_at.elapsed() < TERMINAL_TOMBSTONE_TTL);
 }
 
-fn prune_expired_sessions(sessions: &mut HashMap<String, TerminalSession>) {
+fn prune_expired_sessions(sessions: &mut HashMap<String, TerminalSession>) -> usize {
     let now = Instant::now();
+    let old_len = sessions.len();
     sessions.retain(|_, session| !session.expiry.is_expired(now));
+    old_len - sessions.len()
 }
 
 #[derive(Clone, Copy, Debug)]
@@ -990,4 +1065,127 @@ mod tests {
 
         assert!(registry.summary(&created.terminal_id).await.is_some());
     }
+
+    #[tokio::test]
+    async fn session_events_invalidate_mutations_but_not_reads_or_terminal_bytes() {
+        let registry = TerminalRegistry::new();
+        let mut events = registry.subscribe();
+        let created = registry.create("router".into()).await.expect("create");
+        assert!(events.try_recv().is_ok());
+
+        registry.summaries().await;
+        registry.summary(&created.terminal_id).await;
+        assert!(events.try_recv().is_err(), "ordinary reads emit no event");
+
+        let mut operator = registry
+            .attach_operator(&created.terminal_id, &created.attachment_token, OPERATOR_A)
+            .await
+            .expect("attach operator");
+        assert!(events.try_recv().is_ok());
+        registry
+            .relay_from_agent(
+                &created.terminal_id,
+                TerminalRelayFrame::Binary(b"terminal output".to_vec()),
+            )
+            .await
+            .expect("relay terminal bytes");
+        operator.recv().await.expect("terminal bytes delivered");
+        assert!(events.try_recv().is_err(), "terminal bytes emit no event");
+
+        registry
+            .detach_operator(&created.terminal_id, &created.attachment_token)
+            .await
+            .expect("detach operator");
+        assert!(events.try_recv().is_ok());
+        registry.remove(&created.terminal_id).await.expect("remove");
+        assert!(events.try_recv().is_ok());
+    }
+
+    #[tokio::test]
+    async fn lagged_session_receiver_recovers_with_latest_invalidation() {
+        let registry = TerminalRegistry::new();
+        let mut events = registry.subscribe();
+        for _ in 0..(TERMINAL_EVENTS_CAPACITY + 3) {
+            registry
+                .create_with_limits("router".into(), TERMINAL_EVENTS_CAPACITY + 3, 0)
+                .await
+                .expect("create");
+        }
+
+        assert!(matches!(
+            events.try_recv(),
+            Err(broadcast::error::TryRecvError::Lagged(_))
+        ));
+        assert!(events.try_recv().is_ok());
+        assert_eq!(
+            registry.summaries().await.len(),
+            TERMINAL_EVENTS_CAPACITY + 3
+        );
+    }
+
+    #[tokio::test]
+    async fn agent_attach_detach_and_reconciliation_invalidate_catalog_changes() {
+        let registry = TerminalRegistry::new();
+        let mut events = registry.subscribe();
+        let terminal_id = TerminalId::new("reconciled-session").expect("terminal id");
+        let reported = AgentTerminalSession {
+            terminal_id: terminal_id.clone(),
+            created_at_unix: 0,
+            session_ttl_seconds: 0,
+        };
+
+        let credentials = registry
+            .reconcile_agent_sessions("router", std::slice::from_ref(&reported))
+            .await;
+        assert_eq!(credentials.len(), 1);
+        assert!(events.try_recv().is_ok(), "adoption changes the catalog");
+
+        let (_, relay_token) = &credentials[0];
+        registry
+            .attach_agent(terminal_id.as_str(), "router", relay_token)
+            .await
+            .expect("attach agent");
+        assert!(events.try_recv().is_ok());
+        registry
+            .detach_agent(terminal_id.as_str())
+            .await
+            .expect("detach agent");
+        assert!(events.try_recv().is_ok());
+
+        registry.reconcile_agent_sessions("router", &[]).await;
+        assert!(
+            events.try_recv().is_ok(),
+            "reconciliation removes stale session"
+        );
+        assert!(registry.summary(terminal_id.as_str()).await.is_none());
+        assert!(events.try_recv().is_err());
+    }
+
+    #[tokio::test]
+    async fn pruning_sibling_keeps_requested_summary_and_notifies_once() {
+        let registry = TerminalRegistry::new();
+        let mut events = registry.subscribe();
+        let expired = registry.create("router".into()).await.expect("expired");
+        let live = registry.create("router".into()).await.expect("live");
+        events.try_recv().expect("expired create event");
+        events.try_recv().expect("live create event");
+        registry
+            .inner
+            .lock()
+            .await
+            .get_mut(&expired.terminal_id)
+            .expect("expired session")
+            .expiry = TerminalExpiry::Finite {
+            deadline: Instant::now(),
+            unix: expired.created_at_unix,
+        };
+
+        let summary = registry
+            .summary(&live.terminal_id)
+            .await
+            .expect("live sibling remains available");
+        assert_eq!(summary.terminal_id, live.terminal_id);
+        assert!(matches!(events.try_recv(), Ok(())));
+        assert!(events.try_recv().is_err());
+    }
 }