diff --git a/README.md b/README.md index cb0f60a..38ded81 100644 --- a/README.md +++ b/README.md @@ -207,10 +207,23 @@ Control-plane admin API includes token management endpoints: - `GET /api/v1/control/audit/events?agent_id=&event_type=&limit=` - `GET /api/v1/control/alerts?lookback_seconds=900` - `GET /api/v1/control/alerts/history?since_unix=&limit=` -- `GET /api/v1/control/alerts/ws` (websocket snapshots + recent transitions) +- `GET /api/v1/control/alerts/ws` (websocket snapshots, recent transitions, and evaluator health) +- `GET /api/v1/control/terminals/events/ws` (websocket session-list invalidations) - `DELETE /api/v1/control/agents/{agent_id}` - `PATCH /api/v1/control/agents/{agent_id}/nickname` +Alert evaluation runs once per control plane every five seconds. The alerts GET +returns `503 alerts_unavailable` until a successful evaluation, after evaluation +failure, or when its snapshot is stale. The existing alerts stream also sends +`alerts_health` messages with `available` and an optional `reason`; the UI keeps +the last results visible with a stale warning until recovery. Custom alert-rule +query overrides are rejected with `400 unsupported_alert_override`. + +Terminal session events carry only `sessions_changed` invalidations. Clients +refetch the session list after changes or reconnection and retain a 15-second +fallback refresh. Events do not grant session ownership; attachment still uses +the existing server-side locking and scoped credentials. + If commands still appear silent, verify both processes are running with `-v` and that `RUST_LOG` is not overriding to a stricter level. diff --git a/docs/adr/0002-remote-terminal-sessions.md b/docs/adr/0002-remote-terminal-sessions.md index 913c42f..b9dfbb8 100644 --- a/docs/adr/0002-remote-terminal-sessions.md +++ b/docs/adr/0002-remote-terminal-sessions.md @@ -72,6 +72,8 @@ When the operator socket disconnects, the session becomes detached. Navigation, The operator UI lists live sessions as tabs and stores only the active terminal ID in browser-tab-scoped `sessionStorage`. On reload it prefers that ID and briefly waits for the previous WebSocket to detach before requesting a fresh attachment credential. The stored ID is a navigation hint, not an ownership credential; another attached operator still locks the session at the control plane. +The protected `/api/v1/control/terminals/events/ws` stream sends `sessions_changed` invalidations when the volatile session catalog or attachment ownership changes. These events contain no terminal bytes or attachment credentials: browsers refetch the ordinary session list, coalescing bursts and repeating a fetch if another event arrives during it. Subscribing and reconnecting also invalidate the list. The 15-second refresh remains a recovery fallback; the stream is a freshness hint, never an authorization or locking mechanism. + After operator attachment, the control plane sends a `snapshot` request. The agent serializes up to 1,000 retained physical rows followed by its parsed current screen and input modes as terminal escape bytes, sends them before subsequent live output, and then signals the PTY's current foreground process group with `SIGWINCH`. Historical rows retain their terminal attributes and are replayed as ordinary terminal output so xterm builds its native scrollback; the final formatted state restores the exact live screen. The signal lets full-screen applications redraw without assuming that the original shell remains in the foreground. Signal failure is logged but never terminates the session. The connected browser independently retains up to 10,000 rows because that memory belongs to the operator device rather than the agent host. Wakey carries a small, read-only extension to `vt100` because its public API exposes only the active grid. A snapshot streams bounded history from the primary grid, redraws that grid with its saved cursor and attributes, and then reconstructs the alternate grid when a full-screen application is active. The parser is never resized or switched while taking a snapshot. Consequently, reconnecting during `btop` restores the application while preserving the hidden shell history and state that reappear when it exits. diff --git a/ui/src/App.tsx b/ui/src/App.tsx index 1d8c73d..5d690e2 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -1,4 +1,4 @@ -import { lazy, Suspense, useEffect, useState } from "react"; +import { lazy, Suspense, useEffect, useRef, useState } from "react"; import { Navigate, Route, Routes } from "react-router-dom"; import { toast } from "sonner"; @@ -40,22 +40,48 @@ export function App() { const [history, setHistory] = useState([]); const [audit, setAudit] = useState([]); const [selectedAgentId, setSelectedAgentId] = useState(""); + const [alertsHealth, setAlertsHealth] = useState<{ + available: boolean; + reason?: string; + } | null>(null); + const [alertsStreamConnected, setAlertsStreamConnected] = useState(false); + const streamHealthReceived = useRef(false); + const alertsRevision = useRef(0); const [state, setState] = useState("idle"); + async function loadAlertsWithoutBlocking() { + try { + const nextAlerts = await fetchAlerts(); + if (!streamHealthReceived.current) { + setAlertsHealth({ available: true }); + } + return nextAlerts; + } catch (err) { + if (!streamHealthReceived.current) { + setAlertsHealth({ available: false, reason: String(err) }); + } + return null; + } + } + async function loadAll() { + const revision = alertsRevision.current; setState("loading"); try { const [nextAgents, nextAlerts, nextHistory, nextAudit] = await Promise.all([ fetchAgents(), - fetchAlerts(), + loadAlertsWithoutBlocking(), fetchAlertHistory(20), fetchAudit(30), ]); setAgents(nextAgents); - setAlerts(nextAlerts); - setHistory(nextHistory); + // Streamed state wins over HTTP requests started before its arrival. + if (revision === alertsRevision.current) { + if (nextAlerts) setAlerts(nextAlerts); + setHistory(nextHistory); + } setAudit(nextAudit); const firstConnectedAgentId = nextAgents.find((agent) => agent.connected)?.agent_id ?? ""; @@ -100,12 +126,15 @@ export function App() { } async function refreshAlertsAndHistory() { + const revision = alertsRevision.current; const [nextAlerts, nextHistory] = await Promise.all([ - fetchAlerts(), + loadAlertsWithoutBlocking(), fetchAlertHistory(20), ]); - setAlerts(nextAlerts); - setHistory(nextHistory); + if (revision === alertsRevision.current) { + if (nextAlerts) setAlerts(nextAlerts); + setHistory(nextHistory); + } } useEffect(() => { @@ -117,15 +146,14 @@ export function App() { let disposed = false; let attempt = 0; let openedAt = 0; - let wsGeneration = 0; let reconnectTimer: number | undefined; const resync = () => { - const generation = wsGeneration; + const generation = alertsRevision.current; void Promise.all([fetchAlerts(), fetchAlertHistory(20)]) .then(([nextAlerts, nextHistory]) => { // A WS snapshot delivered while this request was in flight is newer. - if (generation !== wsGeneration) return; + if (disposed || generation !== alertsRevision.current) return; setAlerts(nextAlerts); setHistory(nextHistory); }) @@ -138,18 +166,32 @@ export function App() { socket = new WebSocket(wsUrl); socket.onopen = () => { + if (disposed) return; openedAt = Date.now(); + setAlertsStreamConnected(true); // Pull anything missed while the stream was down. resync(); }; socket.onmessage = (evt) => { - wsGeneration += 1; + if (disposed) return; + alertsRevision.current += 1; try { const payload = JSON.parse(String(evt.data)) as { + type?: string; + available?: boolean; + reason?: string; alerts?: Alert[]; recent_transitions?: AlertTransition[]; }; + if (payload.type === "alerts_health") { + streamHealthReceived.current = true; + setAlertsHealth({ + available: payload.available === true, + reason: payload.reason, + }); + return; + } if (payload.alerts) setAlerts(payload.alerts); if (payload.recent_transitions) setHistory(payload.recent_transitions); @@ -161,6 +203,7 @@ export function App() { socket.onclose = () => { socket = null; if (disposed) return; + setAlertsStreamConnected(false); // Only a connection that stayed up proves the stream is healthy; an // accept-then-close server must not pin us to the first backoff step. if (openedAt && Date.now() - openedAt > 10_000) attempt = 0; @@ -217,6 +260,8 @@ export function App() { transitions={history} audit={audit} loading={state === "loading"} + alertsHealth={alertsHealth} + alertsStreamConnected={alertsStreamConnected} onRefresh={loadAll} /> } @@ -259,6 +304,8 @@ export function App() { } diff --git a/ui/src/pages/AlertsPage.tsx b/ui/src/pages/AlertsPage.tsx index 6f67efd..d4b3835 100644 --- a/ui/src/pages/AlertsPage.tsx +++ b/ui/src/pages/AlertsPage.tsx @@ -16,10 +16,18 @@ import { Textarea } from "@/components/ui/textarea"; type Props = { alerts: Alert[]; transitions: AlertTransition[]; + alertsHealth: { available: boolean; reason?: string } | null; + alertsStreamConnected: boolean; onRefresh: () => Promise; }; -export function AlertsPage({ alerts, transitions, onRefresh }: Props) { +export function AlertsPage({ + alerts, + transitions, + alertsHealth, + alertsStreamConnected, + onRefresh, +}: Props) { const [severity, setSeverity] = useState("all"); const [status, setStatus] = useState("all"); const [kind, setKind] = useState("all"); @@ -66,9 +74,27 @@ export function AlertsPage({ alerts, transitions, onRefresh }: Props) { Active Alerts - +
+
+ {!alertsStreamConnected &&
Alert stream disconnected
} + {alertsHealth?.available === false && ( +
+ Alert results may be stale or unavailable + {alertsHealth.reason ? `: ${alertsHealth.reason}` : ""} +
+ )} +
+ +
diff --git a/ui/src/pages/DashboardPage.tsx b/ui/src/pages/DashboardPage.tsx index 0aea20b..07b38d9 100644 --- a/ui/src/pages/DashboardPage.tsx +++ b/ui/src/pages/DashboardPage.tsx @@ -27,6 +27,8 @@ type Props = { transitions: AlertTransition[]; audit: AuditEvent[]; loading: boolean; + alertsHealth: { available: boolean; reason?: string } | null; + alertsStreamConnected: boolean; onRefresh: () => void; }; @@ -66,6 +68,8 @@ export function DashboardPage({ transitions, audit, loading, + alertsHealth, + alertsStreamConnected, onRefresh, }: Props) { const connected = agents.filter((a) => a.connected).length; @@ -165,6 +169,20 @@ export function DashboardPage({ + {(!alertsStreamConnected || alertsHealth?.available === false) && ( +
+ {!alertsStreamConnected &&
Alert stream disconnected
} + {alertsHealth?.available === false && ( +
+ Alert results may be stale or unavailable + {alertsHealth.reason ? `: ${alertsHealth.reason}` : ""} +
+ )} +
+ )}
{alerts.length}
{Object.keys(severityBuckets).length > 0 ? (
diff --git a/ui/src/pages/TerminalPage.tsx b/ui/src/pages/TerminalPage.tsx index 0ee4427..fd48385 100644 --- a/ui/src/pages/TerminalPage.tsx +++ b/ui/src/pages/TerminalPage.tsx @@ -513,9 +513,38 @@ export function TerminalPage({ terminalRef.current = terminal; fitRef.current = fit; + let cancelled = false; + let refreshInFlight: Promise | null = null; + let refreshRequested = false; + const refreshSessions = (): Promise => { + if (refreshInFlight) { + refreshRequested = true; + return refreshInFlight; + } + refreshInFlight = (async () => { + let listed: TerminalSession[] = []; + do { + refreshRequested = false; + listed = await listTerminals(); + if (!cancelled) { + setSessions((current) => + reconcileTerminalSessions(current, listed), + ); + } + } while (refreshRequested && !cancelled); + return listed; + })().finally(() => { + refreshInFlight = null; + if (refreshRequested && !cancelled) { + void refreshSessions().catch(() => undefined); + } + }); + return refreshInFlight; + }; + async function restoreTerminalSession() { try { - const listed = await listTerminals(); + const listed = await refreshSessions(); if (cancelled) return; setSessions(orderTerminalSessions(listed)); const rememberedId = window.sessionStorage.getItem( @@ -557,7 +586,6 @@ export function TerminalPage({ } } - let cancelled = false; const input = terminal.onData((data) => writeTerminalData(data, true)); let resizeFrame = 0; let lastHostWidth = 0; @@ -575,30 +603,61 @@ export function TerminalPage({ }); }); resizeObserver.observe(hostRef.current); - void restoreTerminalSession(); - let refreshInFlight = false; - const refreshSessions = () => { - if (refreshInFlight) return; - refreshInFlight = true; - void listTerminals() - .then((listed) => { - if (!cancelled) { - setSessions((current) => - reconcileTerminalSessions(current, listed), - ); + let eventsSocket: WebSocket | null = null; + let reconnectTimer = 0; + let healthyTimer = 0; + let reconnectAttempt = 0; + const connectEvents = () => { + if (cancelled) return; + const socket = new WebSocket( + websocketUrl("/api/v1/control/terminals/events/ws"), + ); + eventsSocket = socket; + socket.onmessage = (event) => { + if (cancelled || eventsSocket !== socket) return; + if (typeof event.data !== "string") return; + try { + const message = JSON.parse(event.data) as { type?: string }; + if (message.type === "sessions_changed") { + void refreshSessions().catch(() => undefined); } - }) - .catch(() => { - // The attached terminal transport remains authoritative while a - // background list refresh is temporarily unavailable. - }) - .finally(() => { - refreshInFlight = false; - }); + } catch { + // Ignore malformed event frames; the fallback refresh remains active. + } + }; + socket.onopen = () => { + window.clearTimeout(healthyTimer); + healthyTimer = window.setTimeout(() => { + if (eventsSocket === socket && socket.readyState === WebSocket.OPEN) { + reconnectAttempt = 0; + } + }, 10_000); + }; + socket.onclose = () => { + if (cancelled || eventsSocket !== socket) return; + window.clearTimeout(healthyTimer); + const ceiling = Math.min(500 * 2 ** reconnectAttempt, 12_000); + const delay = Math.min( + Math.round(ceiling * (0.8 + Math.random() * 0.4)), + 15_000, + ); + reconnectAttempt = Math.min(reconnectAttempt + 1, 5); + reconnectTimer = window.setTimeout(connectEvents, delay); + }; + socket.onerror = () => socket.close(); }; - refreshSessionsRef.current = refreshSessions; - const refreshTimer = window.setInterval(refreshSessions, 15_000); - window.addEventListener("focus", refreshSessions); + connectEvents(); + void restoreTerminalSession(); + refreshSessionsRef.current = () => { + void refreshSessions().catch(() => undefined); + }; + const refreshTimer = window.setInterval(() => { + void refreshSessions().catch(() => undefined); + }, 15_000); + const refreshOnFocus = () => { + void refreshSessions().catch(() => undefined); + }; + window.addEventListener("focus", refreshOnFocus); return () => { cancelled = true; @@ -606,7 +665,11 @@ export function TerminalPage({ titleChange.dispose(); resizeObserver.disconnect(); window.clearInterval(refreshTimer); - window.removeEventListener("focus", refreshSessions); + window.removeEventListener("focus", refreshOnFocus); + window.clearTimeout(reconnectTimer); + window.clearTimeout(healthyTimer); + eventsSocket?.close(); + eventsSocket = null; refreshSessionsRef.current = () => {}; window.cancelAnimationFrame(resizeFrame); detachTransport(); diff --git a/wakey-control-plane/src/api/alerts.rs b/wakey-control-plane/src/api/alerts.rs index 4a2c6ac..b262662 100644 --- a/wakey-control-plane/src/api/alerts.rs +++ b/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, - latest_payload: Arc>>, - latest_alerts: Arc>>, - evaluator_status: Arc>, + status: Arc>, } #[derive(Default)] -struct EvaluatorStatus { +struct HubStatus { + latest_payload: Option, + latest_alerts: Vec, last_success_unix: Option, failure: Option, + last_health_available: Option, } 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 { - self.latest_alerts.read().await.clone() - } - - pub async fn latest_payload(&self) -> Option { - self.latest_payload.read().await.clone() + self.status.read().await.latest_alerts.clone() } async fn mark_unavailable(&self, reason: impl Into) { - 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 { - 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) { + 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) { + 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 { + 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 { + encode_health_message(reason.is_none(), reason) +} + +fn encode_health_message(available: bool, reason: Option) -> 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()); + } } diff --git a/wakey-control-plane/src/api/mod.rs b/wakey-control-plane/src/api/mod.rs index ebfa7e9..997adda 100644 --- a/wakey-control-plane/src/api/mod.rs +++ b/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}; diff --git a/wakey-control-plane/src/api/terminals.rs b/wakey-control-plane/src/api/terminals.rs index 26b6944..85a982e 100644 --- a/wakey-control-plane/src/api/terminals.rs +++ b/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) -> 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::(&text).map_err(|_| "invalid handshake") diff --git a/wakey-control-plane/src/runtime/mod.rs b/wakey-control-plane/src/runtime/mod.rs index 41b854b..44a749e 100644 --- a/wakey-control-plane/src/runtime/mod.rs +++ b/wakey-control-plane/src/runtime/mod.rs @@ -148,6 +148,10 @@ fn control_api_routes() -> Router { "/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), diff --git a/wakey-control-plane/src/runtime/terminals.rs b/wakey-control-plane/src/runtime/terminals.rs index 2f21bf9..3024a8b 100644 --- a/wakey-control-plane/src/runtime/terminals.rs +++ b/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>>, closed: Arc>>, + 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 { self.create_with_limits( @@ -95,13 +107,17 @@ impl TerminalRegistry { session_ttl_seconds: u64, ) -> Result { 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 { 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::>(); - 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::>() + .collect::>(); + (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 { 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 { 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, Vec), &'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, &'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 { 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 { 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::>(); 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, 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, operator_id: &str) { +fn detach_operator_sessions( + sessions: &mut HashMap, + 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) { closed.retain(|_, closed_at| closed_at.elapsed() < TERMINAL_TOMBSTONE_TTL); } -fn prune_expired_sessions(sessions: &mut HashMap) { +fn prune_expired_sessions(sessions: &mut HashMap) -> 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()); + } }