agentmux_srv/server/
http_viewer_sysinfo.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! This computer's CPU and memory for a paired device's Sysinfo, read-only
5//! (agentmux-mobile's PLAN_CHANNEL_AS_ONE_LIST_2026_10_13 §3a):
6//!
7//! - `GET /agentmux/viewer/sysinfo`: the latest sample, a `TimeSeriesData`
8//!   (`{ts, values, uptime_secs}`, the metric names the desktop's Sysinfo
9//!   reads: `cpu`, `cpu:N`, `mem:*`, `net:*`, `disk:*`). 204 before the
10//!   first sample.
11//! - `GET /agentmux/viewer/sysinfo/feed?since=<ms>`: the same as SSE, one
12//!   `sample` event per sample, its `id` its `ts`: first the kept history
13//!   after `since` (or after `Last-Event-ID`; all of it with neither), then
14//!   each new sample, with a heartbeat on a quiet stream.
15//!
16//! Both read the samples the sysinfo loop already keeps
17//! (`Broker::read_event_history`), so a phone watching costs no extra
18//! sampling. Built on REPORT_SYSINFO_REWRITE_AND_MODERN_MONITOR_2026_10_13's
19//! data layer as agreed with its author: when that layer's shared scheduler
20//! and `since seq N` read exist, the feed holds a lease while open and
21//! resumes by sequence, and `since=<ms>` keeps working.
22
23use std::convert::Infallible;
24use std::sync::Arc;
25use std::time::Duration;
26
27use axum::body::Bytes;
28use axum::extract::Query;
29use futures_util::Stream;
30use serde::Deserialize;
31
32use super::http_viewer::{ViewerCaller, FEED_HEARTBEAT, HEARTBEAT, RETRY_PREAMBLE};
33use super::*;
34use crate::backend::mps::{Broker, EVENT_SYS_INFO};
35use crate::backend::viewer::{FeedLease, FeedRefused};
36
37/// How often the feed looks for new samples: the sysinfo loop's own cadence.
38const SAMPLE_POLL: Duration = Duration::from_secs(1);
39/// The most samples one poll reads back. More than a poll can miss even
40/// when the loop runs at its fastest (0.2 s) through a long stall.
41const POLL_READ: usize = 64;
42/// The most samples kept, and so the most a feed sends first.
43const HISTORY_READ: usize = 1024;
44
45/// The samples kept for the whole computer (the sysinfo loop also files them
46/// under its connection's scope; the unscoped ring holds them all).
47const GLOBAL_SCOPE: &str = "";
48
49#[derive(Deserialize, Default)]
50pub(crate) struct SysinfoFeedQuery {
51    since: Option<i64>,
52}
53
54/// One kept sample, as `(ts, the TimeSeriesData as sent)`.
55fn samples(broker: &Broker, max: usize) -> Vec<(i64, serde_json::Value)> {
56    broker
57        .read_event_history(EVENT_SYS_INFO, GLOBAL_SCOPE, max)
58        .into_iter()
59        .filter_map(|e| {
60            let data = e.data?;
61            let ts = data.get("ts")?.as_i64()?;
62            Some((ts, data))
63        })
64        .collect()
65}
66
67pub(super) async fn handle_viewer_sysinfo(State(state): State<AppState>) -> Response {
68    match samples(&state.broker, 1).pop() {
69        Some((_, data)) => Json(data).into_response(),
70        None => StatusCode::NO_CONTENT.into_response(),
71    }
72}
73
74pub(super) async fn handle_viewer_sysinfo_feed(
75    State(state): State<AppState>,
76    Extension(caller): Extension<ViewerCaller>,
77    Query(q): Query<SysinfoFeedQuery>,
78    headers: axum::http::HeaderMap,
79) -> Response {
80    let lease = match state.viewer.open_feed(&caller.device_id) {
81        Ok(lease) => lease,
82        Err(FeedRefused::PerDevice) => {
83            return (StatusCode::SERVICE_UNAVAILABLE, Json(json!({ "error": "too many open feeds for this device" })))
84                .into_response()
85        }
86        Err(FeedRefused::Total) => {
87            return (StatusCode::SERVICE_UNAVAILABLE, Json(json!({ "error": "too many open feeds" }))).into_response()
88        }
89    };
90    // A reconnect resumes after the last sample it got.
91    let since = headers
92        .get("last-event-id")
93        .and_then(|v| v.to_str().ok())
94        .and_then(|v| v.trim().parse::<i64>().ok())
95        .or(q.since);
96    tracing::info!(device = %caller.device_id, resume = since.is_some(), "viewer sysinfo feed: opened");
97    let stream = sysinfo_stream(state.broker.clone(), since, lease, SysinfoTiming::default());
98    (
99        [(header::CONTENT_TYPE, "text/event-stream"), (header::CACHE_CONTROL, "no-cache")],
100        Body::from_stream(stream),
101    )
102        .into_response()
103}
104
105/// A Sysinfo feed's clocks.
106#[derive(Debug, Clone, Copy)]
107pub(crate) struct SysinfoTiming {
108    pub heartbeat: Duration,
109    pub poll: Duration,
110}
111
112impl Default for SysinfoTiming {
113    fn default() -> Self {
114        Self { heartbeat: FEED_HEARTBEAT, poll: SAMPLE_POLL }
115    }
116}
117
118fn sample_event(ts: i64, data: &serde_json::Value) -> Bytes {
119    Bytes::from(format!("event: sample\nid: {ts}\ndata: {data}\n\n"))
120}
121
122/// The kept samples a resume asks for: those after `since`, by their time.
123/// All of them with no `since`, or when every kept sample is older than it
124/// (the computer's clock went back since the device's last one).
125fn history_after(kept: Vec<(i64, serde_json::Value)>, since: Option<i64>) -> Vec<(i64, serde_json::Value)> {
126    let Some(since) = since else { return kept };
127    if kept.iter().all(|(ts, _)| *ts < since) {
128        return kept;
129    }
130    kept.into_iter().filter(|(ts, _)| *ts > since).collect()
131}
132
133/// The samples kept after `cursor` (the last one sent), by their place in
134/// the ring, not their time: a clock that goes back doesn't hide new ones.
135/// The newest `POLL_READ` usually hold it; when they don't (a stalled
136/// stream), the whole ring is read, so nothing kept is skipped. A cursor
137/// no longer kept means everything kept is newer.
138fn samples_after(broker: &Broker, cursor: Option<&serde_json::Value>) -> Vec<(i64, serde_json::Value)> {
139    let recent = samples(broker, POLL_READ);
140    let Some(cursor) = cursor else { return recent };
141    if let Some(at) = recent.iter().rposition(|(_, d)| d == cursor) {
142        return recent[at + 1..].to_vec();
143    }
144    let kept = samples(broker, HISTORY_READ);
145    match kept.iter().rposition(|(_, d)| d == cursor) {
146        Some(at) => kept[at + 1..].to_vec(),
147        None => kept,
148    }
149}
150
151/// The kept samples after `since` (all of them with none), then each new
152/// one; a heartbeat when nothing new came for `timing.heartbeat`. Ends when
153/// the device is revoked or goes away.
154pub(crate) fn sysinfo_stream(
155    broker: Arc<Broker>,
156    since: Option<i64>,
157    lease: FeedLease,
158    timing: SysinfoTiming,
159) -> impl Stream<Item = Result<Bytes, Infallible>> + Send + 'static {
160    async_stream::stream! {
161        let lease = lease;
162        yield Ok(Bytes::from_static(RETRY_PREAMBLE));
163        let kept = samples(&broker, HISTORY_READ);
164        // The newest kept sample is where the live part starts, whether or
165        // not the resume sent it.
166        let mut cursor: Option<serde_json::Value> = kept.last().map(|(_, d)| d.clone());
167        for (ts, data) in history_after(kept, since) {
168            yield Ok(sample_event(ts, &data));
169        }
170        let mut poll = tokio::time::interval_at(tokio::time::Instant::now() + timing.poll, timing.poll);
171        poll.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
172        let mut quiet_since = tokio::time::Instant::now();
173        loop {
174            tokio::select! {
175                biased;
176                _ = lease.token.cancelled() => break,
177                _ = poll.tick() => {
178                    let mut sent = false;
179                    for (ts, data) in samples_after(&broker, cursor.as_ref()) {
180                        sent = true;
181                        yield Ok(sample_event(ts, &data));
182                        cursor = Some(data);
183                    }
184                    if sent {
185                        quiet_since = tokio::time::Instant::now();
186                    } else if quiet_since.elapsed() >= timing.heartbeat {
187                        quiet_since = tokio::time::Instant::now();
188                        yield Ok(Bytes::from_static(HEARTBEAT));
189                    }
190                }
191            }
192        }
193    }
194}
195
196#[cfg(test)]
197#[path = "tests/viewer/viewer_sysinfo_tests.rs"]
198mod tests;
agentmux_srv/server/http_viewer_sysinfo.rs

agentmux_srv/server/
http_viewer_sysinfo.rs
