agentmux_srv/server/
http_viewer_terminal.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! `GET /agentmux/viewer/terminals/:block_id/feed`: one terminal's output,
5//! live and read-only, for a paired device
6//! (`agentmux-mobile/docs/reports/REPORT_PANE_LIST_TERMINALS_AND_FROZEN_FEED_2026_10_12.md`
7//! §2.4, §2.5 phase B). Served beside the other viewer routes, on the viewer
8//! listener only, for the viewer token only.
9//!
10//! Off unless the user turns on "Share terminals with paired devices"
11//! (`viewer:shareterminals`): a terminal's scrollback routinely holds
12//! secrets, and terminals have no per-pane hide flag. The shell drawer of an
13//! agent hidden from paired devices is hidden with it. Nothing is ever sent
14//! to the terminal: no input, no resize. The device renders at the size the
15//! desktop gave the PTY (`size` events).
16//!
17//! Wire (SSE): `snapshot { offset, next_offset, data64, cols?, rows? }`, the
18//! last 256 KB of the terminal's output from a line start; `append { offset,
19//! data64 }` per write; `size { cols, rows }` when the PTY is resized;
20//! `reset { reason }` before a fresh snapshot, or before the stream ends
21//! (`behind`, `hidden`, `not_shared`); `: hb` every 15 s. Ids are the byte
22//! offset after the event; a reconnect whose `Last-Event-ID` is still within
23//! 256 KB of the end (a snapshot's worth) gets the bytes it missed as one
24//! `append`.
25
26use std::convert::Infallible;
27use std::time::Duration;
28
29use axum::body::Bytes;
30use axum::extract::Path;
31use base64::Engine as _;
32use futures_util::Stream;
33
34use super::http_viewer::{agent_hidden, ViewerCaller, FEED_HEARTBEAT, HEARTBEAT, RETRY_PREAMBLE, STATUS_POLL};
35use super::*;
36use crate::backend::viewer::feed::{self, HubEvent, TERM_FILE};
37use crate::backend::viewer::terminal::{self, TermStep};
38use crate::backend::viewer::{FeedLease, FeedRefused};
39
40/// Settings key: let paired devices watch this computer's terminals. Off
41/// unless set.
42pub(crate) const SETTING_SHARE_TERMINALS: &str = "viewer:shareterminals";
43
44/// Whether the user lets paired devices watch terminals.
45pub(crate) fn terminals_shared(state: &AppState) -> bool {
46    state
47        .config_watcher
48        .get_settings()
49        .extra
50        .get(SETTING_SHARE_TERMINALS)
51        .and_then(|v| v.as_bool())
52        .unwrap_or(false)
53}
54
55/// Why a block's terminal can't be watched.
56#[derive(Debug, Clone, Copy, PartialEq, Eq)]
57pub(crate) enum Unwatchable {
58    /// Gone, or not a terminal.
59    NotATerminal,
60    /// The shell drawer of an agent hidden from paired devices.
61    Hidden,
62}
63
64/// Whether `block_id` is a terminal a paired device may watch (sharing aside).
65pub(crate) fn terminal_watchable(state: &AppState, block_id: &str) -> Result<(), Unwatchable> {
66    use crate::backend::obj::{meta_get_string, Block};
67    let Ok(Some(block)) = state.mstore.get::<Block>(block_id) else {
68        return Err(Unwatchable::NotATerminal);
69    };
70    if meta_get_string(&block.meta, "view", "") != "term" {
71        return Err(Unwatchable::NotATerminal);
72    }
73    // An agent CLI can run in a terminal pane: that pane is the agent's, and
74    // hidden with it.
75    if agent_hidden(state, block_id) {
76        return Err(Unwatchable::Hidden);
77    }
78    let key = crate::backend::blockcontroller::META_KEY_SHELL_SUBBLOCK_ID;
79    let parent = state
80        .mstore
81        .get_all::<Block>()
82        .unwrap_or_default()
83        .into_iter()
84        .find(|b| meta_get_string(&b.meta, key, "") == block_id);
85    match parent {
86        Some(agent) if agent_hidden(state, &agent.oid) => Err(Unwatchable::Hidden),
87        _ => Ok(()),
88    }
89}
90
91fn refused(status: StatusCode, error: &str) -> Response {
92    (status, Json(json!({ "error": error }))).into_response()
93}
94
95pub(super) async fn handle_viewer_terminal_feed(
96    State(state): State<AppState>,
97    Extension(caller): Extension<ViewerCaller>,
98    Path(block_id): Path<String>,
99    headers: axum::http::HeaderMap,
100) -> Response {
101    if !terminals_shared(&state) {
102        return refused(StatusCode::FORBIDDEN, "this computer doesn't share its terminals");
103    }
104    let st = state.clone();
105    let id = block_id.clone();
106    let watchable = tokio::task::spawn_blocking(move || terminal_watchable(&st, &id)).await;
107    if !matches!(watchable, Ok(Ok(()))) {
108        return refused(StatusCode::NOT_FOUND, "no such terminal");
109    }
110    let lease = match state.viewer.open_feed(&caller.device_id) {
111        Ok(lease) => lease,
112        Err(FeedRefused::PerDevice) => {
113            return refused(StatusCode::SERVICE_UNAVAILABLE, "too many open feeds for this device")
114        }
115        Err(FeedRefused::Total) => return refused(StatusCode::SERVICE_UNAVAILABLE, "too many open feeds"),
116    };
117    let last_event_id = headers
118        .get("last-event-id")
119        .and_then(|v| v.to_str().ok())
120        .and_then(|v| v.trim().parse::<u64>().ok());
121    tracing::info!(device = %caller.device_id, block = %block_id, resume = last_event_id.is_some(), "viewer terminal feed: opened");
122    let stream = terminal_stream(
123        TermSource::of(&state, &block_id),
124        last_event_id,
125        lease,
126        TermTiming::default(),
127        TermProbe::of(&state, &block_id),
128    );
129    (
130        [(header::CONTENT_TYPE, "text/event-stream"), (header::CACHE_CONTROL, "no-cache")],
131        Body::from_stream(stream),
132    )
133        .into_response()
134}
135
136/// A terminal feed's clocks.
137#[derive(Debug, Clone, Copy)]
138pub(crate) struct TermTiming {
139    pub heartbeat: Duration,
140    pub size_poll: Duration,
141}
142
143impl Default for TermTiming {
144    fn default() -> Self {
145        Self { heartbeat: FEED_HEARTBEAT, size_poll: STATUS_POLL }
146    }
147}
148
149/// What a terminal feed asks while it runs: the PTY's size, and whether it
150/// may still be watched (`Some(reason)` ends the stream).
151pub(crate) struct TermProbe {
152    pub size: Box<dyn Fn() -> Option<(u16, u16)> + Send>,
153    pub closed: Box<dyn Fn() -> Option<&'static str> + Send>,
154}
155
156impl TermProbe {
157    pub(crate) fn of(state: &AppState, block_id: &str) -> Self {
158        let size_block = block_id.to_string();
159        let (st, block) = (state.clone(), block_id.to_string());
160        Self {
161            size: Box::new(move || crate::backend::blockcontroller::term_size::get(&size_block)),
162            closed: Box::new(move || {
163                if !terminals_shared(&st) {
164                    return Some("not_shared");
165                }
166                match terminal_watchable(&st, &block) {
167                    Ok(()) => None,
168                    Err(Unwatchable::Hidden) => Some("hidden"),
169                    Err(Unwatchable::NotATerminal) => Some("gone"),
170                }
171            }),
172        }
173    }
174}
175
176/// Where a terminal feed reads the terminal's output from: its `term` file.
177pub(crate) struct TermSource {
178    filestore: Arc<crate::backend::storage::filestore::FileStore>,
179    hub: Arc<feed::FeedHub>,
180    block_id: String,
181}
182
183/// A snapshot as read: its bytes start at `offset`; the file ends at `next`.
184struct TermSnapshot {
185    offset: u64,
186    next: u64,
187    data: Vec<u8>,
188}
189
190impl TermSource {
191    pub(crate) fn of(state: &AppState, block_id: &str) -> Self {
192        Self { filestore: state.filestore.clone(), hub: state.viewer.hub.clone(), block_id: block_id.to_string() }
193    }
194
195    /// The file's size and its bytes from `from` (`None`: from
196    /// [`terminal::SNAPSHOT_TAIL_BYTES`] before the end), on the blocking pool.
197    async fn read_from(&self, from: Option<u64>) -> Option<(u64, u64, Vec<u8>)> {
198        let (fs, block) = (self.filestore.clone(), self.block_id.clone());
199        tokio::task::spawn_blocking(move || {
200            let size = match fs.stat(&block, TERM_FILE) {
201                Ok(Some(file)) => file.size.max(0) as u64,
202                Ok(None) => return Some((0, 0, Vec::new())),
203                Err(_) => return None,
204            };
205            let start = from.unwrap_or_else(|| size.saturating_sub(terminal::SNAPSHOT_TAIL_BYTES));
206            // Nothing to read, or more than one event may carry (a resume far
207            // behind on a terminal whose file has no cap): not read at all.
208            if start >= size || size - start > terminal::SNAPSHOT_TAIL_BYTES {
209                return Some((start, size, Vec::new()));
210            }
211            let (at, data) = fs.read_at(&block, TERM_FILE, start as i64, (size - start) as i64).ok()?;
212            Some((at.max(0) as u64, size, data))
213        })
214        .await
215        .ok()
216        .flatten()
217    }
218
219    /// The tail, from a line start.
220    async fn snapshot(&self) -> Option<TermSnapshot> {
221        let (offset, next, mut data) = self.read_from(None).await?;
222        let start = terminal::tail_start(offset, &data);
223        data.drain(..start);
224        Some(TermSnapshot { offset: offset + start as u64, next, data })
225    }
226
227    /// The bytes from `from` on, for a reconnect: `None` when `from` is past
228    /// the end (the file was truncated) or more than a snapshot holds was
229    /// written since (a snapshot then). One event never carries more than a
230    /// snapshot would: a device's event-stream parser bounds a line's length.
231    async fn resume(&self, from: u64) -> Option<(u64, Vec<u8>)> {
232        let (at, size, data) = self.read_from(Some(from)).await?;
233        if from > size || at != from || size - from > terminal::SNAPSHOT_TAIL_BYTES {
234            return None;
235        }
236        Some((at, data))
237    }
238}
239
240fn b64(data: &[u8]) -> String {
241    base64::engine::general_purpose::STANDARD.encode(data)
242}
243
244fn snapshot_event(s: &TermSnapshot, size: Option<(u16, u16)>) -> Bytes {
245    let mut data = json!({ "offset": s.offset, "next_offset": s.next, "data64": b64(&s.data) });
246    if let Some((rows, cols)) = size {
247        data["rows"] = json!(rows);
248        data["cols"] = json!(cols);
249    }
250    feed::sse_event("snapshot", Some(&s.next.to_string()), &data)
251}
252
253fn append_event(offset: u64, data: &[u8]) -> Bytes {
254    let end = offset + data.len() as u64;
255    feed::sse_event("append", Some(&end.to_string()), &json!({ "offset": offset, "data64": b64(data) }))
256}
257
258fn size_event((rows, cols): (u16, u16)) -> Bytes {
259    feed::sse_event("size", None, &json!({ "rows": rows, "cols": cols }))
260}
261
262fn reset_event(reason: &str) -> Bytes {
263    feed::sse_event("reset", None, &json!({ "reason": reason }))
264}
265
266/// One terminal feed: `retry`, a `snapshot` (or, for a reconnect that still
267/// holds, the bytes it missed as one `append`), then an `append` per write,
268/// `size` on a resize, `reset` + `snapshot` when the file is truncated or the
269/// feed loses its place, and a heartbeat on a quiet stream. Ends with `reset`
270/// when the device falls more than 1 MB behind or the terminal may no longer
271/// be watched, and silently when the device is revoked or goes away.
272pub(crate) fn terminal_stream(
273    source: TermSource,
274    last_event_id: Option<u64>,
275    lease: FeedLease,
276    timing: TermTiming,
277    probe: TermProbe,
278) -> impl Stream<Item = Result<Bytes, Infallible>> + Send + 'static {
279    async_stream::stream! {
280        let lease = lease;
281        // Subscribed before anything is read, so nothing written between the
282        // read and the first event is missed.
283        let mut sub = source.hub.subscribe_file(&source.block_id, TERM_FILE);
284        yield Ok(Bytes::from_static(RETRY_PREAMBLE));
285        let mut shown_size = (probe.size)();
286
287        let resumed = match last_event_id {
288            Some(from) => source.resume(from).await,
289            None => None,
290        };
291        let mut next = match resumed {
292            Some((at, data)) => {
293                if !data.is_empty() {
294                    yield Ok(append_event(at, &data));
295                }
296                if let Some(size) = shown_size {
297                    yield Ok(size_event(size));
298                }
299                Some(at + data.len() as u64)
300            }
301            None => match source.snapshot().await {
302                Some(snapshot) => {
303                    yield Ok(snapshot_event(&snapshot, shown_size));
304                    Some(snapshot.next)
305                }
306                None => None,
307            },
308        };
309        if next.is_none() {
310            yield Ok(reset_event("unavailable"));
311        }
312
313        let mut tick = tokio::time::interval_at(tokio::time::Instant::now() + timing.heartbeat, timing.heartbeat);
314        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
315        let mut size_tick = tokio::time::interval_at(tokio::time::Instant::now() + timing.size_poll, timing.size_poll);
316        size_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
317        while let Some(cursor) = next.as_mut() {
318            let mut resync: Option<String> = None;
319            tokio::select! {
320                biased;
321                _ = lease.token.cancelled() => break,
322                event = sub.recv() => match event {
323                    None => break,
324                    Some(HubEvent::Behind) => {
325                        yield Ok(reset_event("behind"));
326                        break;
327                    }
328                    Some(HubEvent::Rewritten { fileop }) => resync = Some(fileop),
329                    Some(HubEvent::Append { offset, records, .. }) => match terminal::apply_term_append(cursor, offset, &records) {
330                        TermStep::Send { offset, data } => yield Ok(append_event(offset, &data)),
331                        TermStep::Skip => {}
332                        TermStep::Resync(reason) => resync = Some(reason.to_string()),
333                    },
334                },
335                _ = size_tick.tick() => {
336                    let size = (probe.size)();
337                    if size.is_some() && size != shown_size {
338                        shown_size = size;
339                        if let Some(size) = size {
340                            yield Ok(size_event(size));
341                        }
342                    }
343                }
344                _ = tick.tick() => {
345                    if let Some(reason) = (probe.closed)() {
346                        yield Ok(reset_event(reason));
347                        break;
348                    }
349                    yield Ok(Bytes::from_static(HEARTBEAT));
350                }
351            }
352            if let Some(reason) = resync {
353                tracing::info!(block = %source.block_id, %reason, "viewer terminal feed: reset; sending a fresh snapshot");
354                yield Ok(reset_event(&reason));
355                match source.snapshot().await {
356                    Some(snapshot) => {
357                        next = Some(snapshot.next);
358                        yield Ok(snapshot_event(&snapshot, shown_size));
359                    }
360                    None => {
361                        yield Ok(reset_event("unavailable"));
362                        break;
363                    }
364                }
365            }
366        }
367    }
368}
369
370#[cfg(test)]
371#[path = "tests/viewer/viewer_terminal_tests.rs"]
372mod tests;
agentmux_srv/server/http_viewer_terminal.rs

agentmux_srv/server/
http_viewer_terminal.rs
