diff --git a/Cargo.lock b/Cargo.lock index 488f4e1cc..2e393e412 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6701,6 +6701,7 @@ dependencies = [ "tokio-rustls", "tokio-tungstenite 0.24.0", "tokio-util", + "tower-http", "tower-service", "uuid", ] @@ -6791,6 +6792,7 @@ name = "okena-transport" version = "0.1.0" dependencies = [ "async-channel", + "flate2", "futures", "log", "okena-core", diff --git a/crates/okena-app-core/src/remote_snapshot.rs b/crates/okena-app-core/src/remote_snapshot.rs index c3f1bec69..7e5c2f86d 100644 --- a/crates/okena-app-core/src/remote_snapshot.rs +++ b/crates/okena-app-core/src/remote_snapshot.rs @@ -219,5 +219,6 @@ pub fn build_state_response( claude_dialogs: Default::default(), claude_sessions: Vec::new(), claude_pr_links: Default::default(), + terminal_attention: Default::default(), } } diff --git a/crates/okena-cli/src/resolve.rs b/crates/okena-cli/src/resolve.rs index b330fd17f..440a06652 100644 --- a/crates/okena-cli/src/resolve.rs +++ b/crates/okena-cli/src/resolve.rs @@ -466,6 +466,7 @@ mod tests { claude_dialogs: Default::default(), claude_sessions: Vec::new(), claude_pr_links: Default::default(), + terminal_attention: Default::default(), } } diff --git a/crates/okena-core/src/api.rs b/crates/okena-core/src/api.rs index b8d1432d9..ddff9f435 100644 --- a/crates/okena-core/src/api.rs +++ b/crates/okena-core/src/api.rs @@ -69,6 +69,22 @@ pub struct StateResponse { #[serde(default, skip_serializing_if = "std::collections::HashMap::is_empty")] pub claude_pr_links: std::collections::HashMap>, + /// When each terminal last rang the bell or raised an `OSC 9` / `OSC 777` + /// notification, for terminals that ever did. Lets a client that does not + /// stream a terminal's output still flag it. Runtime-only; a client tells + /// new events by a timestamp that moved, never by comparing clocks. + #[serde(default, skip_serializing_if = "std::collections::HashMap::is_empty")] + pub terminal_attention: std::collections::HashMap, +} + +/// Unix-millis timestamps of a terminal's last bell and last desktop +/// notification (see [`StateResponse::terminal_attention`]). +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct ApiTerminalAttention { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bell_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub notification_at: Option, } /// OS window bounds in screen pixels. @@ -1818,6 +1834,7 @@ mod tests { claude_dialogs: Default::default(), claude_sessions: Vec::new(), claude_pr_links: Default::default(), + terminal_attention: Default::default(), }; let json = serde_json::to_string(&resp).unwrap(); let parsed: StateResponse = serde_json::from_str(&json).unwrap(); diff --git a/crates/okena-core/src/ws.rs b/crates/okena-core/src/ws.rs index 02d50c72b..2ea958f58 100644 --- a/crates/okena-core/src/ws.rs +++ b/crates/okena-core/src/ws.rs @@ -37,6 +37,12 @@ pub enum WsInbound { SetWindows { windows: Vec, }, + /// Whether this connection wants the periodic `system_stats_changed` + /// push (on by default). A client that never shows host metrics turns it + /// off; on a phone the 2 s cadence adds up to megabytes a day. + SetSystemStats { + enabled: bool, + }, SendText { terminal_id: String, text: String, diff --git a/crates/okena-daemon-core/src/command_loop.rs b/crates/okena-daemon-core/src/command_loop.rs index 11bbbc5a4..3a9433978 100644 --- a/crates/okena-daemon-core/src/command_loop.rs +++ b/crates/okena-daemon-core/src/command_loop.rs @@ -4457,6 +4457,22 @@ pub async fn daemon_command_loop( resp.claude_dialogs = claude_dialogs; resp.claude_sessions = claude_sessions; resp.claude_pr_links = claude_pr_links; + resp.terminal_attention = { + let registry = terminals.lock(); + registry + .iter() + .filter_map(|(id, term)| match term.attention() { + (None, None) => None, + (bell_at, notification_at) => Some(( + id.clone(), + okena_core::api::ApiTerminalAttention { + bell_at, + notification_at, + }, + )), + }) + .collect() + }; // `match` (not `.expect`) so the daemon-core crate stays clean // under `clippy::expect_used` had it been enabled — the serialize diff --git a/crates/okena-daemon-core/src/daemon.rs b/crates/okena-daemon-core/src/daemon.rs index 1eb078fae..b823323bc 100644 --- a/crates/okena-daemon-core/src/daemon.rs +++ b/crates/okena-daemon-core/src/daemon.rs @@ -636,7 +636,6 @@ impl DaemonCore { tokio::task::spawn_local(crate::git_poll::run_git_poll( reactor.workspace.clone(), git_status_tx.clone(), - reactor.state_version.clone(), remote_subscribed_terminals.clone(), remote_visible_projects.clone(), git_poll_trigger_rx, diff --git a/crates/okena-daemon-core/src/git_poll.rs b/crates/okena-daemon-core/src/git_poll.rs index dd055ad21..262035518 100644 --- a/crates/okena-daemon-core/src/git_poll.rs +++ b/crates/okena-daemon-core/src/git_poll.rs @@ -496,7 +496,6 @@ fn apply_github_result( ci_checks: &mut HashMap>, last: &mut HashMap, git_status_tx: &watch::Sender>, - state_version: &watch::Sender, ) { let GithubPollResult { head_generations, @@ -562,7 +561,7 @@ fn apply_github_result( status.pr_info = pr_infos.get(id).cloned().flatten(); status.ci_checks = ci_checks.get(id).cloned().flatten(); } - publish(last, &enriched, git_status_tx, state_version); + publish(last, &enriched, git_status_tx); } /// Change signals for the walk schedule besides HEAD triggers. @@ -583,12 +582,13 @@ pub struct GitChangeSignals { /// cycles stable even when targeted triggers wake the loop between cadence ticks. /// PR/CI lookups retain their existing visible-project adaptive cadence. /// -/// Bumps `state_version` on a real change so a snapshot/broadcast observer can -/// react; the *primary* output is the `git_status_tx` watch. +/// Publishes into the `git_status_tx` watch only. Remote clients get git status +/// as its own `git_status_changed` message and patch their copy of the state, so +/// a git change does not bump `state_version` (which would make every client +/// refetch the whole state). pub async fn run_git_poll( workspace: Arc>, git_status_tx: Arc>>, - state_version: watch::Sender, remote_subscribed_terminals: Arc>>>, remote_visible_projects: Arc>>>, mut trigger_rx: mpsc::UnboundedReceiver, @@ -783,7 +783,7 @@ pub async fn run_git_poll( // never block the branch/diff badge from appearing. // No receiver-count exit: receivers exist only while clients are connected, // so "none left" also means "none yet". The loop ends with the daemon's LocalSet. - publish(&mut last, &new_statuses, &git_status_tx, &state_version); + publish(&mut last, &new_statuses, &git_status_tx); // ── 4. Start GitHub PR/CI fan-out without blocking local git refreshes ─ // Only visible projects (plus anything explicitly asked for) and only @@ -886,7 +886,6 @@ pub async fn run_git_poll( &mut ci_checks, &mut last, &git_status_tx, - &state_version, ), GithubPassMessage::Finished(ids) => { for id in &ids { @@ -974,15 +973,14 @@ fn clear_github_cache_for_ids( } } -/// Broadcast the slimmed `ApiGitStatus` map into `git_status_tx` and bump -/// `state_version`, but only on a real change. `last` holds the previously +/// Broadcast the slimmed `ApiGitStatus` map into `git_status_tx`, but only on a +/// real change. `last` holds the previously /// published richer `GitStatus` map (the GUI's `commit_statuses` change check); /// no-ops when `new_statuses` equals it, so re-committing the same data is free. fn publish( last: &mut HashMap, new_statuses: &HashMap, git_status_tx: &watch::Sender>, - state_version: &watch::Sender, ) { if new_statuses == last { return; @@ -991,7 +989,6 @@ fn publish( let api_statuses: HashMap = last.iter().map(|(id, s)| (id.clone(), to_api(s))).collect(); git_status_tx.send_replace(api_statuses); - state_version.send_modify(|v| *v += 1); } #[cfg(test)] @@ -1006,7 +1003,6 @@ mod tests { let workspace = Arc::new(Mutex::new(Workspace::new(empty_workspace_data()))); let (tx, rx) = watch::channel(HashMap::::new()); drop(rx); - let (state_version, _svrx) = watch::channel(0u64); let subscribed = Arc::new(RwLock::new(HashMap::new())); let client_visible = Arc::new(RwLock::new(HashMap::new())); let (_trigger_tx, trigger_rx) = mpsc::unbounded_channel(); @@ -1016,7 +1012,6 @@ mod tests { _ = run_git_poll( workspace, Arc::new(tx), - state_version, subscribed, client_visible, trigger_rx, @@ -1471,13 +1466,11 @@ mod tests { let terminals: TerminalsRegistry = Arc::new(Mutex::new(HashMap::new())); terminals.lock().insert("t".into(), terminal.clone()); let (status_tx, status_rx) = watch::channel(HashMap::new()); - let (state_version, _) = watch::channel(0u64); let (trigger_tx, trigger_rx) = mpsc::unbounded_channel(); let (index_change_tx, index_change_rx) = mpsc::unbounded_channel(); tokio::task::spawn_local(run_git_poll( workspace, Arc::new(status_tx), - state_version, Arc::new(RwLock::new(HashMap::new())), Arc::new(RwLock::new(HashMap::new())), trigger_rx, @@ -1629,15 +1622,12 @@ mod tests { ci_checks: HashMap>, last: HashMap, git_status_tx: watch::Sender>, - state_version: watch::Sender, _rx: watch::Receiver>, - _state_rx: watch::Receiver, } impl ApplyHarness { fn new() -> Self { let (git_status_tx, _rx) = watch::channel(HashMap::new()); - let (state_version, _state_rx) = watch::channel(0); Self { schedule: GithubPollSchedule::default(), pr_infos: HashMap::new(), @@ -1650,9 +1640,7 @@ mod tests { }, )]), git_status_tx, - state_version, _rx, - _state_rx, } } @@ -1671,7 +1659,6 @@ mod tests { &mut self.ci_checks, &mut self.last, &self.git_status_tx, - &self.state_version, ); } } diff --git a/crates/okena-daemon-core/src/pty_loop.rs b/crates/okena-daemon-core/src/pty_loop.rs index 89961d2c3..f756e50dc 100644 --- a/crates/okena-daemon-core/src/pty_loop.rs +++ b/crates/okena-daemon-core/src/pty_loop.rs @@ -487,16 +487,23 @@ fn process_activity_edges( // Drain ALL THREE edges with `|=` (never `||`): each is a one-shot that must // be consumed to clear it, so short-circuiting would leak the notification // queue (and miss a bell that follows a command-finish in the same batch). + // + // Bells and notifications are also stamped on the terminal, so clients + // that do not stream it see them through the state. + let now_ms = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_millis() as u64); let active: Vec = { let reg = terminals.lock(); dirty_terminal_ids .iter() .filter(|tid| { reg.get(*tid).is_some_and(|t| { - let mut a = t.take_pending_command_finished(); - a |= t.take_pending_bell(); - a |= !t.take_pending_notifications().is_empty(); - a + let finished = t.take_pending_command_finished(); + let bell = t.take_pending_bell(); + let notification = !t.take_pending_notifications().is_empty(); + t.record_attention(bell, notification, now_ms); + finished || bell || notification }) }) .cloned() diff --git a/crates/okena-mobile-ffi/src/client/handler.rs b/crates/okena-mobile-ffi/src/client/handler.rs index 1bb2cdac0..b42f090bc 100644 --- a/crates/okena-mobile-ffi/src/client/handler.rs +++ b/crates/okena-mobile-ffi/src/client/handler.rs @@ -27,13 +27,17 @@ impl MobileConnectionHandler { &self.terminals } - /// Seconds since last WS activity (terminal output). + /// Seconds since anything last arrived on the WebSocket. pub fn seconds_since_activity(&self) -> f64 { self.last_activity.lock().elapsed().as_secs_f64() } } impl ConnectionHandler for MobileConnectionHandler { + fn on_inbound(&self) { + *self.last_activity.lock() = Instant::now(); + } + fn create_terminal( &self, _connection_id: &str, diff --git a/crates/okena-mobile-ffi/src/client/manager.rs b/crates/okena-mobile-ffi/src/client/manager.rs index ac7c7f248..e39384869 100644 --- a/crates/okena-mobile-ffi/src/client/manager.rs +++ b/crates/okena-mobile-ffi/src/client/manager.rs @@ -2,11 +2,12 @@ use crate::client::handler::MobileConnectionHandler; use crate::client::terminal_holder::{TerminalActivity, TerminalHolder}; use okena_core::api::{ - ActionRequest, ApiFullscreen, ApiLayoutNode, ApiUsage, ApiUsageProvider, StateResponse, + ActionRequest, ApiFullscreen, ApiLayoutNode, ApiTerminalAttention, ApiUsage, ApiUsageProvider, + StateResponse, }; use okena_transport::client::{ - ConnectionEvent, ConnectionStatus, RemoteClient, RemoteConnectionConfig, WsClientMessage, - make_prefixed_id, + ConnectionEvent, ConnectionStatus, RemoteClient, RemoteConnectionConfig, StateSyncPolicy, + TrafficStats, WsClientMessage, make_prefixed_id, }; use parking_lot::RwLock; use std::collections::{HashMap, HashSet}; @@ -18,8 +19,15 @@ static MANAGER: OnceLock = OnceLock::new(); pub struct ConnectionManager { runtime: Arc, connections: RwLock>, + /// The app is in the background: connections defer state fetches. + in_background: std::sync::atomic::AtomicBool, } +/// Least time between two state fetches on the phone. A busy workspace bumps +/// the state several times a second (agent status edges, finished commands, +/// git); each fetch downloads the whole snapshot. +const STATE_SYNC_MIN_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1); + struct MobileConnection { client: RwLock>, handler: Arc, @@ -274,6 +282,29 @@ fn apply_terminal_presentation( } } +/// Terminals whose bell / notification timestamp moved between two states: +/// `(terminal_id, bell, notification)`. A terminal absent from `previous` had +/// none yet, so any stamp it now has is new. Only movement counts, so the +/// phone never compares the server's clock with its own; the first state of a +/// connection is the baseline and raises nothing. +fn new_attention( + previous: &HashMap, + next: &HashMap, +) -> Vec<(String, bool, bool)> { + let moved = |before: Option, now: Option| now > before; + let mut raised: Vec<(String, bool, bool)> = next + .iter() + .filter_map(|(terminal_id, now)| { + let before = previous.get(terminal_id).copied().unwrap_or_default(); + let bell = moved(before.bell_at, now.bell_at); + let notification = moved(before.notification_at, now.notification_at); + (bell || notification).then(|| (terminal_id.clone(), bell, notification)) + }) + .collect(); + raised.sort(); + raised +} + fn merge_state_presentation(next: &mut StateResponse, previous: &StateResponse) { for project in &mut next.projects { let Some(previous_project) = previous.projects.iter().find(|old| old.id == project.id) @@ -363,6 +394,7 @@ impl ConnectionManager { ConnectionManager { runtime: Arc::new(runtime), connections: RwLock::new(HashMap::new()), + in_background: std::sync::atomic::AtomicBool::new(false), } }); } @@ -507,6 +539,19 @@ impl ConnectionManager { let (event_tx, event_rx) = async_channel::bounded::(256); let client = RemoteClient::new(config, self.runtime.clone(), handler.clone(), event_tx); + // The phone shows one terminal at a time: stream nothing until the UI + // names what is on screen (`set_streamed_terminals`), instead of + // downloading every terminal's output on connect. + client.set_streamed_terminals(Some(Vec::new())); + // Coalesce state fetches, hold them in the background, and skip the + // settings, which the app does not use. + client.set_state_sync(StateSyncPolicy { + min_interval: STATE_SYNC_MIN_INTERVAL, + deferred: self.in_background.load(Ordering::Relaxed), + fetch_settings: false, + }); + // The app shows no host metrics; skip the 2 s `system_stats` push. + client.set_system_stats_push(false); // Spawn event processor task let conn_id_clone = conn_id.clone(); @@ -613,6 +658,37 @@ impl ConnectionManager { } } + /// Stream PTY output only for these terminals (unprefixed ids): the ones + /// on screen. Survives reconnects; the rest keep their last content until + /// streamed again, when the server sends a fresh snapshot. + pub fn set_streamed_terminals(&self, conn_id: &str, terminal_ids: Vec) { + let connections = self.connections.read(); + if let Some(conn) = connections.get(conn_id) { + conn.client + .read() + .set_streamed_terminals(Some(terminal_ids)); + } + } + + /// The app went to (or left) the background. In the background no state is + /// fetched; a state that changed meanwhile is fetched once on return. + /// Unlike [`Self::on_app_background`] this also applies while the Android + /// connection service keeps the session alive. + pub fn set_app_in_background(&self, background: bool) { + self.in_background.store(background, Ordering::Relaxed); + for conn in self.connections.read().values() { + conn.client.read().set_state_sync_deferred(background); + } + } + + /// Payload bytes a connection has received since it was created. + pub fn traffic_stats(&self, conn_id: &str) -> Option { + let connections = self.connections.read(); + connections + .get(conn_id) + .map(|conn| conn.client.read().traffic_stats()) + } + /// Pair with the remote server using a pairing code. pub fn pair(&self, conn_id: &str, code: &str) { let connections = self.connections.read(); @@ -966,6 +1042,29 @@ impl ConnectionManager { ConnectionEvent::StateReceived { mut state, .. } => { let mut cache = conn.state_cache.write(); if let Some(previous) = cache.as_ref() { + // Bells and notifications of terminals the phone does + // not stream reach it only through the state. One a + // terminal already showed in its own output since the + // last state is the same event, maybe already seen. + let raised = + new_attention(&previous.terminal_attention, &state.terminal_attention); + let terminals = conn.handler.terminals().read(); + let mut local: HashMap<&str, (bool, bool)> = terminals + .iter() + .map(|(id, holder)| (id.as_str(), holder.take_local_attention())) + .collect(); + for (terminal_id, bell, notification) in raised { + let prefixed = make_prefixed_id(&conn_id, &terminal_id); + if let Some(holder) = terminals.get(&prefixed) { + let (bell_here, notification_here) = + local.remove(prefixed.as_str()).unwrap_or_default(); + holder.raise_attention( + bell && !bell_here, + notification && !notification_here, + ); + } + } + drop(terminals); merge_state_presentation(&mut state, previous); } else { state.fullscreen_terminal = None; @@ -1010,6 +1109,37 @@ mod tests { use okena_core::shell::ShellType; use okena_core::types::SplitDirection; + fn attention(bell_at: Option, notification_at: Option) -> ApiTerminalAttention { + ApiTerminalAttention { + bell_at, + notification_at, + } + } + + #[test] + fn only_moved_attention_stamps_raise() { + let previous = HashMap::from([ + ("t1".to_string(), attention(Some(10), None)), + ("t2".to_string(), attention(Some(10), Some(20))), + ]); + let next = HashMap::from([ + // Bell rang again. + ("t1".to_string(), attention(Some(15), None)), + // Unchanged. + ("t2".to_string(), attention(Some(10), Some(20))), + // New terminal with a notification. + ("t3".to_string(), attention(None, Some(5))), + ]); + assert_eq!( + new_attention(&previous, &next), + vec![ + ("t1".to_string(), true, false), + ("t3".to_string(), false, true), + ] + ); + assert!(new_attention(&next, &next).is_empty()); + } + fn terminal(id: &str, minimized: bool, shell_type: ShellType) -> ApiLayoutNode { ApiLayoutNode::Terminal { terminal_id: Some(id.to_string()), @@ -1060,6 +1190,7 @@ mod tests { claude_dialogs: Default::default(), claude_sessions: Vec::new(), claude_pr_links: Default::default(), + terminal_attention: Default::default(), } } diff --git a/crates/okena-mobile-ffi/src/client/terminal_holder.rs b/crates/okena-mobile-ffi/src/client/terminal_holder.rs index abab5fe34..96834dccf 100644 --- a/crates/okena-mobile-ffi/src/client/terminal_holder.rs +++ b/crates/okena-mobile-ffi/src/client/terminal_holder.rs @@ -55,12 +55,15 @@ pub(crate) const FREEZE_MAX_DURATION: Duration = Duration::from_secs(15); /// cause duplicates. pub(crate) struct MobileEventListener { bell: Arc, + /// Same edge, consumed by [`TerminalHolder::take_local_attention`]. + bell_seen_locally: Arc, } impl EventListener for MobileEventListener { fn send_event(&self, event: TermEvent) { if let TermEvent::Bell = event { self.bell.store(true, Ordering::Relaxed); + self.bell_seen_locally.store(true, Ordering::Relaxed); } } } @@ -134,6 +137,11 @@ pub struct TerminalHolder { bell: Arc, /// Unseen desktop notification (OSC 9 / 777). notification: AtomicBool, + /// A bell / notification arrived in this terminal's own output since the + /// last state was applied. The state's attention stamps for that event + /// must not raise it again: the user may already have seen it here. + bell_seen_locally: Arc, + notification_seen_locally: AtomicBool, /// Parser for [`NotificationScanner`], kept across chunks so a sequence /// split between two frames is still recognized. notify_parser: Mutex, @@ -214,10 +222,14 @@ impl TerminalHolder { .saturating_sub(1) .clamp(1, MAX_OUTPUT_STEP); let bell = Arc::new(AtomicBool::new(false)); + let bell_seen_locally = Arc::new(AtomicBool::new(false)); let term = Term::new( config, &term_size, - MobileEventListener { bell: bell.clone() }, + MobileEventListener { + bell: bell.clone(), + bell_seen_locally: bell_seen_locally.clone(), + }, ); Self { @@ -231,6 +243,8 @@ impl TerminalHolder { output_step, bell, notification: AtomicBool::new(false), + bell_seen_locally, + notification_seen_locally: AtomicBool::new(false), notify_parser: Mutex::new(Parser::new()), last_output_at: AtomicU64::new(0), freeze: Mutex::new(None), @@ -262,6 +276,8 @@ impl TerminalHolder { self.notify_parser.lock().advance(&mut scanner, data); if scanner.seen { self.notification.store(true, Ordering::Relaxed); + self.notification_seen_locally + .store(true, Ordering::Relaxed); } self.last_output_at.store(monotonic_ms(), Ordering::Relaxed); let mut freeze = self.freeze.lock(); @@ -288,8 +304,11 @@ impl TerminalHolder { let mut freeze = self.freeze.lock(); self.replay_frozen(&mut freeze); let bell = self.bell.load(Ordering::Relaxed); + let bell_seen_locally = self.bell_seen_locally.load(Ordering::Relaxed); self.feed(data); self.bell.store(bell, Ordering::Relaxed); + self.bell_seen_locally + .store(bell_seen_locally, Ordering::Relaxed); self.end_freeze_if_unselected(&mut freeze); } @@ -353,6 +372,29 @@ impl TerminalHolder { } } + /// Consume "a bell / notification arrived in this terminal's own output + /// since the last state": `(bell, notification)`. Called once per applied + /// state, for every terminal. + pub fn take_local_attention(&self) -> (bool, bool) { + ( + self.bell_seen_locally.swap(false, Ordering::Relaxed), + self.notification_seen_locally + .swap(false, Ordering::Relaxed), + ) + } + + /// A bell / notification the server reported for this terminal. One the + /// terminal already showed in its own output since the last state (see + /// [`Self::take_local_attention`]) is skipped by the caller. + pub fn raise_attention(&self, bell: bool, notification: bool) { + if bell { + self.bell.store(true, Ordering::Relaxed); + } + if notification { + self.notification.store(true, Ordering::Relaxed); + } + } + /// The user viewed the terminal: clear the unseen bell and notification. pub fn mark_seen(&self) { self.bell.store(false, Ordering::Relaxed); @@ -788,6 +830,20 @@ mod tests { use super::*; use okena_core::theme::DARK_THEME; + #[test] + fn local_attention_is_reported_once_per_state() { + let holder = TerminalHolder::new(20, 5); + assert_eq!(holder.take_local_attention(), (false, false)); + holder.process_output(b"\x07\x1b]9;build done\x07"); + assert_eq!(holder.take_local_attention(), (true, true)); + // Consumed: the next state finds nothing local. + assert_eq!(holder.take_local_attention(), (false, false)); + // Seeing it does not hide that it happened here. + holder.process_output(b"\x07"); + holder.mark_seen(); + assert_eq!(holder.take_local_attention(), (true, false)); + } + #[test] fn bell_sets_unseen_until_marked_seen() { let holder = TerminalHolder::new(20, 5); diff --git a/crates/okena-mobile-ffi/src/lib.rs b/crates/okena-mobile-ffi/src/lib.rs index c22126d6e..022487730 100644 --- a/crates/okena-mobile-ffi/src/lib.rs +++ b/crates/okena-mobile-ffi/src/lib.rs @@ -46,7 +46,7 @@ use crate::client::manager::ConnectionManager; pub use types::{ CellData, ConnectionStatus, CursorShape, CursorState, FolderInfo, FullscreenInfo, ProjectInfo, - ScrollInfo, SelectionBounds, ServiceInfo, TerminalActivity, + ScrollInfo, SelectionBounds, ServiceInfo, TerminalActivity, TrafficInfo, }; pub use usage::{ ExtraUsageInfo, PoolAccountInfo, PoolAmountInfo, PoolInfo, PoolNonstopInfo, PoolPaceInfo, @@ -80,6 +80,31 @@ pub fn on_app_background() { ConnectionManager::get().on_app_background(); } +/// Stream PTY output only for `terminal_ids`, the terminals on screen; an +/// empty list streams nothing (dashboard, background). A connection starts +/// with nothing streamed. A terminal that comes back on screen gets a fresh +/// snapshot from the server. +#[uniffi::export] +pub fn set_streamed_terminals(conn_id: String, terminal_ids: Vec) { + ConnectionManager::get().set_streamed_terminals(&conn_id, terminal_ids); +} + +/// The app went to (`true`) or left (`false`) the background. Call on every +/// change, also while the Android connection service runs: in the background +/// no workspace state is fetched, and a changed state is fetched once on return. +#[uniffi::export] +pub fn set_app_in_background(background: bool) { + ConnectionManager::get().set_app_in_background(background); +} + +/// Payload bytes a connection has received since it was created, by kind. +#[uniffi::export] +pub fn get_traffic_stats(conn_id: String) -> Option { + ConnectionManager::get() + .traffic_stats(&conn_id) + .map(Into::into) +} + /// Retry one connection now ("Retry now" in the UI): skips the backoff and /// resets the attempt budget, keeping terminals and cached state. #[uniffi::export] diff --git a/crates/okena-mobile-ffi/src/types.rs b/crates/okena-mobile-ffi/src/types.rs index a3159c5c4..7ecd1478b 100644 --- a/crates/okena-mobile-ffi/src/types.rs +++ b/crates/okena-mobile-ffi/src/types.rs @@ -303,3 +303,34 @@ impl TerminalActivity { } } } + +/// Payload bytes a connection has received since the app connected it (TLS, +/// HTTP and WebSocket framing excluded). +#[derive(Debug, Clone, uniffi::Record)] +pub struct TrafficInfo { + /// Live PTY output. + pub pty_bytes: u64, + /// Screen snapshots sent when a terminal starts streaming. + pub snapshot_bytes: u64, + /// Workspace state bodies as received. + pub state_bytes: u64, + /// Number of workspace state fetches. + pub state_fetches: u64, + /// Settings bodies. + pub settings_bytes: u64, + /// Every other WebSocket message. + pub event_bytes: u64, +} + +impl From for TrafficInfo { + fn from(s: okena_transport::client::TrafficStats) -> Self { + Self { + pty_bytes: s.pty_bytes, + snapshot_bytes: s.snapshot_bytes, + state_bytes: s.state_bytes, + state_fetches: s.state_fetches, + settings_bytes: s.settings_bytes, + event_bytes: s.event_bytes, + } + } +} diff --git a/crates/okena-remote-server/Cargo.toml b/crates/okena-remote-server/Cargo.toml index f96f860c8..a4f587e6e 100644 --- a/crates/okena-remote-server/Cargo.toml +++ b/crates/okena-remote-server/Cargo.toml @@ -33,6 +33,9 @@ axum = { version = "0.8", features = ["ws"] } hyper = { version = "1", features = ["server", "http1", "http2"] } hyper-util = { version = "0.1", features = ["server-auto", "tokio"] } tower-service = "0.3" +# gzip for the JSON endpoints: a full /v1/state is hundreds of KB and remote +# (mobile) clients refetch it on every state change. +tower-http = { version = "0.6", default-features = false, features = ["compression-gzip"] } # TLS (self-signed cert + dual-stack http/https on one port) rcgen = "0.13" diff --git a/crates/okena-remote-server/src/CLAUDE.md b/crates/okena-remote-server/src/CLAUDE.md index 55d2148d4..f4bcfc599 100644 --- a/crates/okena-remote-server/src/CLAUDE.md +++ b/crates/okena-remote-server/src/CLAUDE.md @@ -42,6 +42,10 @@ logic lives in `crates/okena-remote-client/` (desktop) and `okena-transport` - **Thread boundary**: All mutable state access happens on the daemon reactor's `LocalSet`. The tokio server only serializes/deserializes and forwards via channels. - **Discovery file**: `remote.json` (in config dir) contains the port and auth info so clients can auto-discover the running instance. +- **Push cadence**: `state_changed` makes clients refetch the whole `/v1/state`, so bump + `state_version` only for changes that live in the state. Git status has its own + `git_status_changed` push and does not bump it. `system_stats_changed` goes out every 2 s + unless the client sent `set_system_stats` with `enabled: false`. - **PTY fan-out**: `PtyBroadcaster` uses tokio's `broadcast` channel so multiple WebSocket clients can subscribe to the same terminal's output independently. - **Window model**: `GET /v1/state` returns `windows` (`ApiWindow[]`). The daemon serves a **single synthetic main window** (`command_loop.rs`, `GetState` arm) with `visible_project_ids` populated and the presentation fields (`focused_project_id`, `focused_terminal_id`, `fullscreen`, `bounds`, `folder_filter`, `sidebar_open`) left `None` — those are client-owned by design, not missing. Once a desktop declares its windows (`WsInbound::SetWindows`: id, kind, active, visible projects; kept per connection in `AppState::remote_windows`, dropped on close), `routes/state.rs` lists those instead, in connection order — so `okena ls` and `--window` see the desktop's real windows. The flat `focused_project_id`/`fullscreen_terminal` stay for backward compatibility. - **Per-window action targeting**: `FocusTerminal`, `SetProjectShowInOverview`, `SetFullscreen`, `SetProjectLayout`, `SetStatusLanes` and `SetOverviewStatusLanes` accept an optional `window` field (`"main"` | extra UUID), parsed by `WindowId::from_wire` (`okena-state`). On the daemon, `None` and `"main"` target its synthetic main window; for an extra UUID it only validates `SetProjectShowInOverview` / `SetFullscreen` / `SetStatusLanes` (project, terminal) and leaves applying them to the desktop the server pushes them to. `POST /v1/actions` rejects an extra UUID no connected desktop declared via `SetWindows` (also `CreateWorktree`'s `window`) with "window not found", before the daemon runs the action. diff --git a/crates/okena-remote-server/src/routes/mod.rs b/crates/okena-remote-server/src/routes/mod.rs index b0706f55b..dbc100aae 100644 --- a/crates/okena-remote-server/src/routes/mod.rs +++ b/crates/okena-remote-server/src/routes/mod.rs @@ -193,10 +193,13 @@ pub fn build_router( // Routes that require auth let protected = Router::new() - .route("/v1/state", axum::routing::get(state::get_state)) + .route( + "/v1/state", + axum::routing::get(state::get_state).layer(gzip()), + ) .route( "/v1/claude/terminals/{terminal_id}", - axum::routing::get(claude::get_session), + axum::routing::get(claude::get_session).layer(gzip()), ) .route( "/v1/claude/sessions/{session_id}", @@ -204,7 +207,7 @@ pub fn build_router( ) .route( "/v1/claude/terminals/{terminal_id}/transcript", - axum::routing::get(claude::get_transcript), + axum::routing::get(claude::get_transcript).layer(gzip()), ) .route( "/v1/claude/terminals/{terminal_id}/subagents", @@ -218,7 +221,10 @@ pub fn build_router( "/v1/claude/terminals/{terminal_id}/dialogs/{dialog_id}", axum::routing::post(claude::post_dialog_decision), ) - .route("/v1/actions", axum::routing::post(actions::post_actions)) + .route( + "/v1/actions", + axum::routing::post(actions::post_actions).layer(gzip()), + ) .route( "/v1/files/download", axum::routing::post(download::post_download), @@ -239,14 +245,17 @@ pub fn build_router( .layer(DefaultBodyLimit::max(paste_image::FILE_UPLOAD_LIMIT)), ) .route("/v1/refresh", axum::routing::post(refresh::post_refresh)) - .route("/v1/usage", axum::routing::get(usage::get_usage)) + .route( + "/v1/usage", + axum::routing::get(usage::get_usage).layer(gzip()), + ) .route( "/v1/usage/refresh", axum::routing::post(usage::post_refresh), ) .route( "/v1/usage/pool/{account_id}", - axum::routing::get(usage::get_pool_account_history), + axum::routing::get(usage::get_pool_account_history).layer(gzip()), ) .route("/v1/tokens", axum::routing::get(tokens::list_tokens)) .route( @@ -339,6 +348,15 @@ pub fn build_router( .with_state(state) } +/// gzip for the JSON routes, applied per route rather than router-wide so file +/// downloads and the WebSocket upgrade keep their raw bodies (and a download +/// keeps its `Content-Length`). A full `/v1/state` runs to hundreds of KB and a +/// remote client refetches it on every `state_changed`; JSON compresses ~7x. +/// Only responses to requests that send `Accept-Encoding: gzip` are compressed. +fn gzip() -> tower_http::compression::CompressionLayer { + tower_http::compression::CompressionLayer::new() +} + /// Serve embedded web client assets (SPA with index.html fallback for client-side routing). async fn serve_web_asset(uri: axum::http::Uri) -> axum::response::Response { use axum::response::IntoResponse; @@ -624,6 +642,44 @@ mod tests { ); } + #[tokio::test] + async fn json_routes_are_gzipped_only_when_the_client_accepts_it() { + let store = Arc::new(AuthStore::with_secret(vec![7u8; 32])); + let token = paired_token(&store); + let mut router = test_router(store); + let encoding = |accept: Option<&'static str>| { + let mut builder = axum::http::Request::builder() + .uri("/v1/usage") + .header("authorization", format!("Bearer {token}")); + if let Some(accept) = accept { + builder = builder.header("accept-encoding", accept); + } + let mut req = builder.body(Body::empty()).expect("build request"); + req.extensions_mut().insert(OFF_HOST); + req + }; + + let response = router + .call(encoding(Some("gzip"))) + .await + .expect("the router is infallible"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response + .headers() + .get("content-encoding") + .map(|v| v.to_str().unwrap()), + Some("gzip") + ); + + let response = router + .call(encoding(None)) + .await + .expect("the router is infallible"); + assert_eq!(response.status(), StatusCode::OK); + assert!(response.headers().get("content-encoding").is_none()); + } + #[tokio::test] async fn pool_account_history_is_served_while_the_pool_lists_it() { let store = Arc::new(AuthStore::with_secret(vec![7u8; 32])); @@ -1036,6 +1092,56 @@ mod tests { ); } + /// A client that opts out of host metrics gets no more `system_stats_changed` + /// pushes (they come every 2 s otherwise). + #[tokio::test] + async fn a_client_can_opt_out_of_system_stats() { + use futures::StreamExt as _; + + let store = Arc::new(AuthStore::with_secret(vec![7u8; 32])); + let token = paired_token(&store); + let addr = spawn_stream_server(store).await; + let mut socket = connect_stream(addr).await; + send_frame( + &mut socket, + serde_json::json!({ "type": "auth", "token": token }), + ) + .await; + assert_eq!( + next_frame_type(&mut socket).await.as_deref(), + Some("auth_ok") + ); + + send_frame( + &mut socket, + serde_json::json!({ "type": "set_system_stats", "enabled": false }), + ) + .await; + // Messages are handled in order: once the ping is answered, the + // opt-out has been applied. + send_frame(&mut socket, serde_json::json!({ "type": "ping" })).await; + loop { + match next_frame_type(&mut socket).await.as_deref() { + Some("pong") => break, + Some(_) => continue, + None => panic!("stream ended before the pong"), + } + } + + let quiet = tokio::time::timeout(std::time::Duration::from_millis(2500), async { + while let Some(Ok(frame)) = socket.next().await { + if let TestFrame::Text(text) = frame { + assert!( + !text.contains("system_stats_changed"), + "stats pushed after the opt-out: {text}" + ); + } + } + }) + .await; + assert!(quiet.is_err(), "the stream must stay open"); + } + /// Revocation must end the established stream, not merely refuse the next /// handshake — the connection already has terminal read/write. #[tokio::test] diff --git a/crates/okena-remote-server/src/routes/stream.rs b/crates/okena-remote-server/src/routes/stream.rs index 18d68b405..5b75bbd2b 100644 --- a/crates/okena-remote-server/src/routes/stream.rs +++ b/crates/okena-remote-server/src/routes/stream.rs @@ -191,6 +191,7 @@ async fn handle_ws( let mut auth_deadline = session_deadline(session.as_ref()); let mut system_stats = SystemStatsCache::new(); let mut system_stats_interval = tokio::time::interval(SYSTEM_STATS_REFRESH_INTERVAL); + let mut system_stats_enabled = true; system_stats_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); // Pin the writer handle for use in select! @@ -292,6 +293,9 @@ async fn handle_ws( } } } + Ok(WsInbound::SetSystemStats { enabled }) => { + system_stats_enabled = enabled; + } Ok(WsInbound::SetVisibleProjects { project_ids }) => { // Full replacement set, kept even when empty: the git // poller trusts a declared viewport over subscriptions. @@ -648,7 +652,7 @@ async fn handle_ws( // Periodic host metrics for remote status UI. Kept separate from // `state_version` so status refreshes do not force workspace resync. - _ = system_stats_interval.tick() => { + _ = system_stats_interval.tick(), if system_stats_enabled => { system_stats.refresh(); let resp = serde_json::to_string(&WsOutbound::SystemStatsChanged { stats: system_stats.stats(), diff --git a/crates/okena-terminal/src/terminal/meta.rs b/crates/okena-terminal/src/terminal/meta.rs index 267413a18..554ef071b 100644 --- a/crates/okena-terminal/src/terminal/meta.rs +++ b/crates/okena-terminal/src/terminal/meta.rs @@ -217,6 +217,28 @@ impl Terminal { .swap(false, std::sync::atomic::Ordering::Relaxed) } + /// Stamp the drained bell / notification edges with `now_ms` (Unix millis). + /// Called by the daemon's PTY loop right after `take_pending_bell` / + /// `take_pending_notifications`. + pub fn record_attention(&self, bell: bool, notification: bool, now_ms: u64) { + if bell { + self.last_bell_at + .store(now_ms, std::sync::atomic::Ordering::Relaxed); + } + if notification { + self.last_notification_at + .store(now_ms, std::sync::atomic::Ordering::Relaxed); + } + } + + /// Unix millis of the last recorded bell and notification, if any. + pub fn attention(&self) -> (Option, Option) { + let load = |at: &std::sync::atomic::AtomicU64| { + Some(at.load(std::sync::atomic::Ordering::Relaxed)).filter(|&ms| ms != 0) + }; + (load(&self.last_bell_at), load(&self.last_notification_at)) + } + /// Consume the one-shot "a command finished (OSC 133 ;D) since last drain" /// edge. Returns true if a command completed since the previous call, then /// resets it. The PTY event loop uses this to bump the owning project's diff --git a/crates/okena-terminal/src/terminal/mod.rs b/crates/okena-terminal/src/terminal/mod.rs index 5c56a207d..c87ff5c89 100644 --- a/crates/okena-terminal/src/terminal/mod.rs +++ b/crates/okena-terminal/src/terminal/mod.rs @@ -174,6 +174,12 @@ pub struct Terminal { /// exactly once instead of on every batch while `has_bell` stays set. pub(super) bell_pending: Arc, + /// Unix millis of the last bell / desktop notification the daemon's PTY + /// loop drained, 0 when none yet. Published in the state so clients that + /// do not stream this terminal can still flag it. + pub(super) last_bell_at: AtomicU64, + pub(super) last_notification_at: AtomicU64, + /// "The user marked this pane unread by hand" flag. Holds `has_bell` /// against the render path's clear-on-focus, so marking the pane you are /// looking at actually sticks; released when focus leaves, so the next @@ -519,6 +525,8 @@ impl Terminal { title, has_bell, bell_pending, + last_bell_at: AtomicU64::new(0), + last_notification_at: AtomicU64::new(0), manual_unread: AtomicBool::new(false), has_notification: AtomicBool::new(false), pending_clipboard, diff --git a/crates/okena-terminal/src/terminal/tests/unread.rs b/crates/okena-terminal/src/terminal/tests/unread.rs index 73c908be1..d29b84773 100644 --- a/crates/okena-terminal/src/terminal/tests/unread.rs +++ b/crates/okena-terminal/src/terminal/tests/unread.rs @@ -71,3 +71,17 @@ fn toggle_flips_both_ways_and_clears_a_shell_bell() { assert!(!terminal.toggle_unread(), "shell bell -> read"); assert!(!terminal.has_bell()); } + +#[test] +fn recorded_attention_keeps_the_latest_bell_and_notification() { + let terminal = terminal(); + assert_eq!(terminal.attention(), (None, None)); + + terminal.record_attention(true, false, 1_000); + assert_eq!(terminal.attention(), (Some(1_000), None)); + + // A batch with neither edge leaves both stamps alone. + terminal.record_attention(false, false, 2_000); + terminal.record_attention(false, true, 3_000); + assert_eq!(terminal.attention(), (Some(1_000), Some(3_000))); +} diff --git a/crates/okena-transport/Cargo.toml b/crates/okena-transport/Cargo.toml index d7c1e6f8f..6c9bcd6df 100644 --- a/crates/okena-transport/Cargo.toml +++ b/crates/okena-transport/Cargo.toml @@ -7,7 +7,7 @@ license = "MIT" [features] default = [] # Async client engine: WS connection manager, TLS certificate pinning (TOFU). -client = ["dep:tokio", "dep:reqwest", "dep:tokio-tungstenite", "dep:async-channel", "dep:futures", "dep:rustls", "dep:rustls-pki-types", "dep:sha2"] +client = ["dep:tokio", "dep:reqwest", "dep:tokio-tungstenite", "dep:async-channel", "dep:futures", "dep:rustls", "dep:rustls-pki-types", "dep:sha2", "dep:flate2"] # Blocking HTTP helper + remote_action over it (used by extensions and the # file/git read paths). TLS deps are shared with the async client so blocking # remote actions enforce the same pinned-certificate policy. @@ -29,7 +29,7 @@ log = "0.4" # Client feature deps (async networking) tokio = { version = "1", features = ["rt-multi-thread", "net", "sync", "macros", "time"], optional = true } -reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json"], optional = true } +reqwest = { version = "0.12", default-features = false, features = ["rustls-tls", "json", "gzip"], optional = true } tokio-tungstenite = { version = "0.24", features = ["rustls-tls-native-roots"], optional = true } async-channel = { version = "2.3", optional = true } futures = { version = "0.3", optional = true } @@ -41,6 +41,8 @@ futures = { version = "0.3", optional = true } rustls = { version = "0.23", default-features = false, features = ["ring", "tls12", "logging", "std"], optional = true } rustls-pki-types = { version = "1", optional = true } sha2 = { version = "0.10", optional = true } +# Decodes gzipped /v1/state by hand so the session can count wire bytes. +flate2 = { version = "1", optional = true } [dev-dependencies] tempfile = "3" diff --git a/crates/okena-transport/src/client/connection.rs b/crates/okena-transport/src/client/connection.rs index e8fb5d926..136f9255f 100644 --- a/crates/okena-transport/src/client/connection.rs +++ b/crates/okena-transport/src/client/connection.rs @@ -10,7 +10,7 @@ use okena_core::api::{ActionRequest, ApiSystemStats, ApiUsage, ApiWindow, StateR use okena_core::ws::ClientPresentationRequest; use futures::{Sink, Stream}; -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; @@ -80,6 +80,8 @@ impl Sink for AnyWsStream { fn unix_http_client(path: &str) -> Result { reqwest::Client::builder() .unix_socket(path) + // Same-host socket: compressing would only spend CPU on both ends. + .no_gzip() .build() .map_err(|error| format!("Cannot initialise Unix socket HTTP client: {error}")) } @@ -371,6 +373,192 @@ struct SessionControl { /// While true, the reconnect loop dials nothing (the app is in the /// background, where Android drops packets and attempts only burn budget). paused: Arc>, + /// Which terminals' PTY output to stream (unprefixed ids). `None` streams + /// every terminal in the state (desktop); `Some` streams only the listed + /// ones that exist, so a client showing one terminal does not download the + /// output of all of them. + streams: Arc>>>, + /// How eagerly the session refetches state after `state_changed`. + state_sync: Arc>, + /// Bytes received, by kind. Lives as long as the client, across sessions. + traffic: Arc, + /// Whether the server should push host metrics (`system_stats_changed`). + /// Only a `false` is ever sent; servers default to on. + system_stats: Arc>, +} + +/// How a session follows `state_changed`, which makes it refetch the whole +/// `/v1/state` snapshot (and, by default, the settings). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct StateSyncPolicy { + /// Least time between two state fetches. Changes inside the window are + /// coalesced into one fetch at its end; the first change after a quiet + /// window fetches at once. Zero fetches on every change. + pub min_interval: Duration, + /// Only note that the state is stale and fetch once this turns false + /// (an app in the background, where nobody sees the state). + pub deferred: bool, + /// Fetch the settings on connect and with every state fetch. A client + /// that ignores `SettingsChanged` turns it off. + pub fetch_settings: bool, +} + +impl Default for StateSyncPolicy { + fn default() -> Self { + Self { + min_interval: Duration::ZERO, + deferred: false, + fetch_settings: true, + } + } +} + +/// Payload bytes a client has received since it was created. Excludes TLS, +/// HTTP and WebSocket framing overhead. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct TrafficStats { + /// Live PTY output. + pub pty_bytes: u64, + /// Screen snapshots sent on (re)subscribe. + pub snapshot_bytes: u64, + /// `/v1/state` bodies as received (compressed when the server gzips). + pub state_bytes: u64, + /// Number of `/v1/state` fetches. + pub state_fetches: u64, + /// Settings bodies. + pub settings_bytes: u64, + /// Every other WebSocket message (state, git, usage and stats events). + pub event_bytes: u64, +} + +#[derive(Default)] +struct TrafficCounters { + pty: std::sync::atomic::AtomicU64, + snapshot: std::sync::atomic::AtomicU64, + state: std::sync::atomic::AtomicU64, + state_fetches: std::sync::atomic::AtomicU64, + settings: std::sync::atomic::AtomicU64, + events: std::sync::atomic::AtomicU64, +} + +impl TrafficCounters { + fn add(counter: &std::sync::atomic::AtomicU64, bytes: usize) { + counter.fetch_add(bytes as u64, std::sync::atomic::Ordering::Relaxed); + } + + fn record_state(&self, bytes: usize) { + Self::add(&self.state, bytes); + Self::add(&self.state_fetches, 1); + } + + fn snapshot(&self) -> TrafficStats { + let load = |c: &std::sync::atomic::AtomicU64| c.load(std::sync::atomic::Ordering::Relaxed); + TrafficStats { + pty_bytes: load(&self.pty), + snapshot_bytes: load(&self.snapshot), + state_bytes: load(&self.state), + state_fetches: load(&self.state_fetches), + settings_bytes: load(&self.settings), + event_bytes: load(&self.events), + } + } +} + +/// When the session should next fetch the state. +struct StateRefresh { + stale: bool, + last_fetch: Instant, +} + +impl StateRefresh { + fn fetched_at(now: Instant) -> Self { + Self { + stale: false, + last_fetch: now, + } + } + + fn on_changed(&mut self) { + self.stale = true; + } + + fn on_fetched(&mut self, now: Instant) { + self.stale = false; + self.last_fetch = now; + } + + /// The instant to fetch at, or `None` while there is nothing to fetch or + /// fetching is deferred. May be in the past (fetch now). + fn due(&self, policy: &StateSyncPolicy) -> Option { + (self.stale && !policy.deferred).then(|| self.last_fetch + policy.min_interval) + } +} + +/// Why a `/v1/state` fetch failed. +enum StateFetchError { + Unauthorized(reqwest::StatusCode), + Status(reqwest::StatusCode), + Other(String), +} + +impl std::fmt::Display for StateFetchError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Unauthorized(status) | Self::Status(status) => write!(f, "HTTP {status}"), + Self::Other(error) => f.write_str(error), + } + } +} + +/// Fetch `/v1/state`, returning it with its body size as received. The +/// session client has transparent decompression off, so a gzipped body is +/// decoded here and the count is what crossed the network. +async fn fetch_state( + client: &reqwest::Client, + base_url: &str, + token: &str, + accept_gzip: bool, +) -> Result<(StateResponse, usize), StateFetchError> { + let mut request = client + .get(format!("{base_url}/v1/state")) + .header("Authorization", format!("Bearer {token}")) + .timeout(Duration::from_secs(10)); + if accept_gzip { + request = request.header(reqwest::header::ACCEPT_ENCODING, "gzip"); + } + let response = request + .send() + .await + .map_err(|e| StateFetchError::Other(e.to_string()))?; + let status = response.status(); + if status == reqwest::StatusCode::UNAUTHORIZED { + return Err(StateFetchError::Unauthorized(status)); + } + if !status.is_success() { + return Err(StateFetchError::Status(status)); + } + let gzipped = response + .headers() + .get(reqwest::header::CONTENT_ENCODING) + .is_some_and(|value| value.as_bytes().eq_ignore_ascii_case(b"gzip")); + let body = response + .bytes() + .await + .map_err(|e| StateFetchError::Other(e.to_string()))?; + let state = parse_state_body(&body, gzipped).map_err(StateFetchError::Other)?; + Ok((state, body.len())) +} + +fn parse_state_body(body: &[u8], gzipped: bool) -> Result { + if gzipped { + let mut json = Vec::new(); + std::io::Read::read_to_end(&mut flate2::read::GzDecoder::new(body), &mut json) + .map_err(|e| format!("cannot decompress state: {e}"))?; + serde_json::from_slice(&json) + } else { + serde_json::from_slice(body) + } + .map_err(|e| format!("cannot parse state: {e}")) } impl SessionControl { @@ -379,6 +567,10 @@ impl SessionControl { kick: Arc::new(tokio::sync::watch::channel(0).0), probe: Arc::new(tokio::sync::watch::channel(0).0), paused: Arc::new(tokio::sync::watch::channel(false).0), + streams: Arc::new(tokio::sync::watch::channel(None).0), + state_sync: Arc::new(tokio::sync::watch::channel(StateSyncPolicy::default()).0), + traffic: Arc::new(TrafficCounters::default()), + system_stats: Arc::new(tokio::sync::watch::channel(true).0), } } @@ -393,6 +585,140 @@ impl SessionControl { fn set_paused(&self, paused: bool) { self.paused.send_replace(paused); } + + fn set_streams(&self, streams: Option>) { + self.streams.send_if_modified(|current| { + if *current == streams { + return false; + } + *current = streams; + true + }); + } +} + +/// The terminals of `state` whose output should stream under `filter`, in +/// state order. `None` wants all of them. +fn wanted_streams(state: &StateResponse, filter: Option<&HashSet>) -> Vec { + let mut ids = collect_state_terminal_ids(state); + if let Some(filter) = filter { + ids.retain(|id| filter.contains(id)); + } + ids +} + +/// What to send to move the session's subscriptions from `subscribed` to +/// `wanted`: `(subscribe, unsubscribe)`, each in a stable order. +fn subscription_changes( + subscribed: &HashSet, + wanted: &[String], +) -> (Vec, Vec) { + let subscribe: Vec = wanted + .iter() + .filter(|id| !subscribed.contains(*id)) + .cloned() + .collect(); + let wanted: HashSet<&String> = wanted.iter().collect(); + let mut unsubscribe: Vec = subscribed + .iter() + .filter(|id| !wanted.contains(id)) + .cloned() + .collect(); + unsubscribe.sort(); + (subscribe, unsubscribe) +} + +/// The session's view of its terminal subscriptions. +#[derive(Default)] +struct StreamSubscriptions { + /// Terminals the session wants streamed (sent Subscribe, no Unsubscribe). + subscribed: HashSet, + /// Subscribes sent per terminal whose `subscribed` reply has not arrived. + /// The server hands out a fresh stream id per Subscribe, so only the reply + /// to the latest one carries a live id. + pending: HashMap, +} + +impl StreamSubscriptions { + fn sent_subscribe(&mut self, ids: &[String]) { + for id in ids { + self.subscribed.insert(id.clone()); + *self.pending.entry(id.clone()).or_default() += 1; + } + } + + fn sent_unsubscribe(&mut self, ids: &[String]) { + for id in ids { + self.subscribed.remove(id); + } + } + + /// Keep the mappings of a `subscribed` reply that are still live: the + /// terminal is wanted and this answers its latest Subscribe. A reply that + /// crossed an Unsubscribe, or one superseded by a later Subscribe, names a + /// stream the server has already dropped. + fn accept_reply(&mut self, mut mappings: HashMap) -> HashMap { + mappings.retain(|id, _| { + let outstanding = match self.pending.get_mut(id) { + Some(count) => { + *count = count.saturating_sub(1); + *count + } + None => 0, + }; + if outstanding == 0 { + self.pending.remove(id); + } + outstanding == 0 && self.subscribed.contains(id) + }); + mappings + } +} + +/// Move the session's subscriptions to `wanted`. Sends block rather than +/// `try_send`: a dropped Subscribe leaves a terminal silently never streaming, +/// a dropped Unsubscribe keeps downloading output nobody shows. +async fn reconcile_streams( + subs: &mut StreamSubscriptions, + wanted: &[String], + ws_tx: &async_channel::Sender, + stream_map: &std::sync::RwLock>, + reverse_stream_map: &mut HashMap, +) { + let (subscribe, unsubscribe) = subscription_changes(&subs.subscribed, wanted); + if !unsubscribe.is_empty() { + // Forget the stream ids now: input to these terminals falls back to the + // JSON path, and frames still in flight are not routed. + subs.sent_unsubscribe(&unsubscribe); + if let Ok(mut map) = stream_map.write() { + for id in &unsubscribe { + if let Some(stream_id) = map.remove(id) { + reverse_stream_map.remove(&stream_id); + } + } + } + let count = unsubscribe.len(); + if let Err(e) = ws_tx + .send(WsClientMessage::Unsubscribe { + terminal_ids: unsubscribe, + }) + .await + { + log::warn!("failed to send Unsubscribe for {count} terminals: {e}"); + } + } + if !subscribe.is_empty() { + subs.sent_subscribe(&subscribe); + let count = subscribe.len(); + if let Err(e) = ws_tx + .send(WsClientMessage::Subscribe { + terminal_ids: subscribe, + }) + .await + { + log::warn!("failed to send Subscribe for {count} terminals: {e}"); + } + } } /// Wait until the control is not paused. Returns immediately when it isn't. @@ -563,6 +889,10 @@ pub trait ConnectionHandler: Send + Sync + 'static { fn on_terminal_snapshot(&self, prefixed_id: &str, data: &[u8]) { self.on_terminal_output(prefixed_id, data); } + /// Any frame arrived on the live session (PTY data, state events, pong, + /// periodic stats). For a liveness indicator that must not depend on + /// terminal output, which a client streaming few terminals rarely gets. + fn on_inbound(&self) {} /// Terminal removed — clean up platform terminal object. fn remove_terminal(&self, prefixed_id: &str); /// Resize a terminal's grid to match the server's dimensions. @@ -664,6 +994,51 @@ impl RemoteClient { self.control.set_paused(paused); } + /// Limit PTY streaming to these terminals (unprefixed ids); `None` streams + /// every terminal (the default). Takes effect on the live session at once + /// and on every reconnect. A terminal that is not streamed keeps its last + /// content until it is streamed again, when the server sends a fresh + /// snapshot. + pub fn set_streamed_terminals(&self, terminal_ids: Option>) { + self.control + .set_streams(terminal_ids.map(|ids| ids.into_iter().collect())); + } + + /// Set how the session follows `state_changed` (see [`StateSyncPolicy`]). + /// Applies to the live session at once and to every reconnect. + pub fn set_state_sync(&self, policy: StateSyncPolicy) { + self.control.state_sync.send_if_modified(|current| { + let changed = *current != policy; + *current = policy; + changed + }); + } + + /// Defer (or resume) state fetches, keeping the rest of the policy. On + /// resume a stale state is fetched at once. + pub fn set_state_sync_deferred(&self, deferred: bool) { + self.control.state_sync.send_if_modified(|policy| { + let changed = policy.deferred != deferred; + policy.deferred = deferred; + changed + }); + } + + /// Ask the server to push host metrics (`system_stats_changed`, every 2 s) + /// or not. On by default; applies to the live session and every reconnect. + pub fn set_system_stats_push(&self, enabled: bool) { + self.control.system_stats.send_if_modified(|current| { + let changed = *current != enabled; + *current = enabled; + changed + }); + } + + /// Payload bytes received since this client was created. + pub fn traffic_stats(&self) -> TrafficStats { + self.control.traffic.snapshot() + } + /// Declare which of this connection's projects the client currently renders. /// /// Full replacement set of unprefixed ids. Stored for reconnects and pushed @@ -1526,42 +1901,41 @@ impl RemoteClient { // Step 3: Fetch state via HTTP let base_url = config.http_origin(); + // Off-host the state is fetched gzipped and decoded by hand (see + // `fetch_state`); the local socket is not worth compressing. + let accept_gzip = local_unix_path(config).is_none(); let client = if let Some(path) = local_unix_path(config) { unix_http_client(path) } else { - crate::client::tls::build_reqwest_client( + crate::client::tls::reqwest_client_builder( config.tls, config.pinned_cert_sha256.clone(), observed.clone(), ) + .no_gzip() + .build() + .map_err(|error| format!("Cannot initialise HTTP client: {error}")) } .map_err(SessionError::Transient)?; + let traffic = control.traffic.clone(); + let mut sync_rx = control.state_sync.subscribe(); + let fetch_settings = sync_rx.borrow_and_update().fetch_settings; log::info!("State fetch starting at +{}ms", elapsed_ms()); - let state_resp = client - .get(format!("{}/v1/state", base_url)) - .header("Authorization", format!("Bearer {}", token)) - .timeout(std::time::Duration::from_secs(10)) - .send() - .await - .map_err(|e| SessionError::Transient(format!("Failed to fetch state: {}", e)))?; - - if state_resp.status() == reqwest::StatusCode::UNAUTHORIZED { - return Err(SessionError::Auth(format!( - "State fetch failed: HTTP {}", - state_resp.status() - ))); - } - if !state_resp.status().is_success() { - return Err(SessionError::Transient(format!( - "State fetch failed: HTTP {}", - state_resp.status() - ))); - } - - let state: StateResponse = state_resp - .json() - .await - .map_err(|e| SessionError::Transient(format!("Failed to parse state: {}", e)))?; + let state = match fetch_state(&client, &base_url, token, accept_gzip).await { + Ok((state, bytes)) => { + traffic.record_state(bytes); + state + } + Err(error @ StateFetchError::Unauthorized(_)) => { + return Err(SessionError::Auth(format!("State fetch failed: {error}"))); + } + Err(error) => { + return Err(SessionError::Transient(format!( + "State fetch failed: {error}" + ))); + } + }; + let mut refresh = StateRefresh::fetched_at(Instant::now()); log::info!("State fetched at +{}ms", elapsed_ms()); // Step 4: Sync terminal objects. @@ -1587,23 +1961,31 @@ impl RemoteClient { state: state.clone(), }) .await; - match fetch_remote_settings(&client, &base_url, token).await { - Ok(settings) => { - let _ = event_tx - .send(ConnectionEvent::SettingsChanged { - connection_id: config.id.clone(), - settings, - }) - .await; + if fetch_settings { + match fetch_remote_settings(&client, &base_url, token).await { + Ok(settings) => { + TrafficCounters::add(&traffic.settings, settings.to_string().len()); + let _ = event_tx + .send(ConnectionEvent::SettingsChanged { + connection_id: config.id.clone(), + settings, + }) + .await; + } + Err(error) => log::warn!("{error}"), } - Err(error) => log::warn!("{error}"), } - // Step 5: Subscribe to all terminal streams - if !terminal_ids.is_empty() { + // Step 5: Subscribe to the terminal streams the client wants (all of + // them unless it narrowed the set with `set_streamed_terminals`). + let mut streams_rx = control.streams.subscribe(); + let initial_streams = wanted_streams(&state, streams_rx.borrow_and_update().as_ref()); + let mut subscribed = StreamSubscriptions::default(); + if !initial_streams.is_empty() { + subscribed.sent_subscribe(&initial_streams); let subscribe_msg = serde_json::json!({ "type": "subscribe", - "terminal_ids": terminal_ids, + "terminal_ids": initial_streams, }); futures::SinkExt::send( &mut ws_write, @@ -1651,6 +2033,19 @@ impl RemoteClient { .map_err(|e| SessionError::Transient(format!("Failed to send windows: {}", e)))?; } + // Host metrics are on by default server-side; only an opt-out is sent, + // so servers that predate the message never see it from clients that + // keep the default. + let mut system_stats_rx = control.system_stats.subscribe(); + if !*system_stats_rx.borrow_and_update() { + let msg = serde_json::json!({ "type": "set_system_stats", "enabled": false }); + futures::SinkExt::send(&mut ws_write, tungstenite::Message::Text(msg.to_string())) + .await + .map_err(|e| { + SessionError::Transient(format!("Failed to send system stats opt-out: {e}")) + })?; + } + // Notify connected *reached_connected = true; log::info!( @@ -1753,6 +2148,12 @@ impl RemoteClient { "windows": windows, }) } + WsClientMessage::SetSystemStats { enabled } => { + serde_json::json!({ + "type": "set_system_stats", + "enabled": enabled, + }) + } WsClientMessage::Ping => serde_json::json!({ "type": "ping" }), }; if let Err(e) = futures::SinkExt::send( @@ -1780,8 +2181,90 @@ impl RemoteClient { ping_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); let mut probe_rx = control.probe.subscribe(); loop { + let state_due = refresh.due(&sync_rx.borrow_and_update()); let frame = tokio::select! { frame = futures::StreamExt::next(&mut ws_read) => frame, + Ok(()) = sync_rx.changed() => continue, + Ok(()) = system_stats_rx.changed() => { + let enabled = *system_stats_rx.borrow_and_update(); + let _ = ws_tx_clone.try_send(WsClientMessage::SetSystemStats { enabled }); + continue; + } + _ = tokio::time::sleep_until(tokio::time::Instant::from_std( + state_due.unwrap_or_else(Instant::now), + )), if state_due.is_some() => { + // Reuse the session HTTP client (built once at Step 3) so + // connection pooling / keep-alive is preserved across + // refetches instead of rebuilding a client per event. + match fetch_state(&client, &base_url, token, accept_gzip).await { + Ok((new_state, bytes)) => { + traffic.record_state(bytes); + let diff = diff_states(&cached_state, &new_state); + let new_size_map = collect_terminal_sizes(&new_state); + + // Add new terminals via handler + for tid in &diff.added_terminals { + let prefixed = make_prefixed_id(&config_id, tid); + let (cols, rows) = + new_size_map.get(tid).copied().unwrap_or((0, 0)); + handler_clone.create_terminal( + &config_id, + tid, + &prefixed, + ws_tx_clone.clone(), + cols, + rows, + ); + } + + // Remove old terminals via handler + for tid in &diff.removed_terminals { + let prefixed = make_prefixed_id(&config_id, tid); + handler_clone.remove_terminal(&prefixed); + } + + // Stream what is wanted now: new terminals that + // pass the filter start, removed ones stop. + let wanted = wanted_streams(&new_state, streams_rx.borrow().as_ref()); + reconcile_streams( + &mut subscribed, + &wanted, + &ws_tx_clone, + &stream_map, + &mut reverse_stream_map, + ) + .await; + + cached_state = new_state.clone(); + + let _ = event_tx_clone + .send(ConnectionEvent::StateReceived { + connection_id: config_id.clone(), + state: new_state, + }) + .await; + } + Err(error) => log::warn!("State re-fetch failed: {error}"), + } + if sync_rx.borrow().fetch_settings { + match fetch_remote_settings(&client, &base_url, token).await { + Ok(settings) => { + TrafficCounters::add(&traffic.settings, settings.to_string().len()); + let _ = event_tx_clone + .send(ConnectionEvent::SettingsChanged { + connection_id: config_id.clone(), + settings, + }) + .await; + } + Err(error) => log::warn!("{error}"), + } + } + // A failed fetch is not retried until the next change, + // as before: retrying at once could spin on a dead route. + refresh.on_fetched(Instant::now()); + continue; + } _ = tokio::time::sleep_until(tokio::time::Instant::from_std(liveness.deadline())) => { match liveness.on_deadline(Instant::now()) { LivenessAction::SendProbe => { @@ -1812,6 +2295,19 @@ impl RemoteClient { let _ = ws_tx_clone.try_send(WsClientMessage::Ping); continue; } + Ok(()) = streams_rx.changed() => { + let wanted = + wanted_streams(&cached_state, streams_rx.borrow_and_update().as_ref()); + reconcile_streams( + &mut subscribed, + &wanted, + &ws_tx_clone, + &stream_map, + &mut reverse_stream_map, + ) + .await; + continue; + } Ok(()) = probe_rx.changed() => { if liveness.on_probe_request(Instant::now(), WS_PROBE_MIN_QUIET) { log::info!( @@ -1832,13 +2328,19 @@ impl RemoteClient { }; if matches!(frame, Some(Ok(_))) { liveness.on_inbound(Instant::now()); + handler_clone.on_inbound(); } match frame { Some(Ok(tungstenite::Message::Binary(data))) => { // Generic binary frame: [proto:1][type:1][stream_id:4 BE][payload...] - if let Some((frame_type, stream_id, payload)) = - okena_core::ws::parse_binary_frame(&data) - { + let parsed = okena_core::ws::parse_binary_frame(&data); + let counter = match parsed { + Some((okena_core::ws::FRAME_TYPE_PTY, _, _)) => &traffic.pty, + Some((okena_core::ws::FRAME_TYPE_SNAPSHOT, _, _)) => &traffic.snapshot, + _ => &traffic.events, + }; + TrafficCounters::add(counter, data.len()); + if let Some((frame_type, stream_id, payload)) = parsed { match frame_type { okena_core::ws::FRAME_TYPE_PTY | okena_core::ws::FRAME_TYPE_SNAPSHOT => { @@ -1859,6 +2361,7 @@ impl RemoteClient { } } Some(Ok(tungstenite::Message::Text(text))) => { + TrafficCounters::add(&traffic.events, text.len()); // JSON message match serde_json::from_str::(&text) { Ok(value) => { @@ -1871,6 +2374,9 @@ impl RemoteClient { mappings.clone(), ) { + // Only live ids: never route to or send input on + // a stream the server has already dropped. + let map = subscribed.accept_reply(map); log::info!("Subscribed to {} terminal streams", map.len()); for (terminal_id, stream_id) in &map { reverse_stream_map @@ -1915,121 +2421,9 @@ impl RemoteClient { } } "state_changed" => { - log::info!("State changed on remote server"); - // Reuse the session HTTP client (built once at - // Step 3) so connection pooling / keep-alive is - // preserved across state_changed events instead - // of rebuilding a client per event. - match client - .get(format!("{}/v1/state", base_url)) - .header("Authorization", format!("Bearer {}", token)) - .timeout(std::time::Duration::from_secs(10)) - .send() - .await - { - Ok(resp) if resp.status().is_success() => { - if let Ok(new_state) = - resp.json::().await - { - let diff = diff_states(&cached_state, &new_state); - let new_size_map = - collect_terminal_sizes(&new_state); - - // Add new terminals via handler - for tid in &diff.added_terminals { - let prefixed = - make_prefixed_id(&config_id, tid); - let (cols, rows) = new_size_map - .get(tid) - .copied() - .unwrap_or((0, 0)); - handler_clone.create_terminal( - &config_id, - tid, - &prefixed, - ws_tx_clone.clone(), - cols, - rows, - ); - } - - // Remove old terminals via handler - for tid in &diff.removed_terminals { - let prefixed = - make_prefixed_id(&config_id, tid); - handler_clone.remove_terminal(&prefixed); - } - - // Subscribe to new terminals. Use a blocking - // send (not try_send): dropping this on a full - // channel would leave the new terminals silently - // never streaming output. - if !diff.added_terminals.is_empty() - && let Err(e) = ws_tx_clone - .send(WsClientMessage::Subscribe { - terminal_ids: diff - .added_terminals - .clone(), - }) - .await - { - log::warn!( - "failed to send Subscribe for {} terminals: {}", - diff.added_terminals.len(), - e - ); - } - - // Unsubscribe from removed terminals. Likewise - // blocking — a dropped Unsubscribe leaks a stream - // for an already-gone terminal. - if !diff.removed_terminals.is_empty() - && let Err(e) = ws_tx_clone - .send(WsClientMessage::Unsubscribe { - terminal_ids: diff - .removed_terminals - .clone(), - }) - .await - { - log::warn!( - "failed to send Unsubscribe for {} terminals: {}", - diff.removed_terminals.len(), - e - ); - } - - cached_state = new_state.clone(); - - let _ = event_tx_clone - .send(ConnectionEvent::StateReceived { - connection_id: config_id.clone(), - state: new_state, - }) - .await; - } - } - Ok(resp) => { - log::warn!( - "State re-fetch failed: HTTP {}", - resp.status() - ); - } - Err(e) => { - log::warn!("State re-fetch failed: {}", e); - } - } - match fetch_remote_settings(&client, &base_url, token).await { - Ok(settings) => { - let _ = event_tx_clone - .send(ConnectionEvent::SettingsChanged { - connection_id: config_id.clone(), - settings, - }) - .await; - } - Err(error) => log::warn!("{error}"), - } + // Fetched by the refresh branch above, at + // once or coalesced per the sync policy. + refresh.on_changed(); } "pong" => { // Keep-alive response, ignore @@ -2418,6 +2812,199 @@ mod tests { use super::*; use std::sync::atomic::{AtomicUsize, Ordering}; + fn state_with_terminals(projects: &[(&str, &str)]) -> StateResponse { + let projects: Vec = projects + .iter() + .map(|(project_id, terminal_id)| { + serde_json::json!({ + "id": project_id, + "name": project_id, + "path": "/tmp", + "show_in_overview": true, + "layout": { + "type": "terminal", + "terminal_id": terminal_id, + "minimized": false, + "detached": false, + }, + "terminal_names": {}, + "folder_color": "default", + }) + }) + .collect(); + serde_json::from_value(serde_json::json!({ + "state_version": 1, + "projects": projects, + "focused_project_id": null, + "fullscreen_terminal": null, + })) + .expect("test state deserializes") + } + + #[test] + fn state_refresh_fetches_at_once_without_an_interval() { + let t0 = Instant::now(); + let mut refresh = StateRefresh::fetched_at(t0); + let policy = StateSyncPolicy::default(); + assert_eq!(refresh.due(&policy), None); + refresh.on_changed(); + assert_eq!(refresh.due(&policy), Some(t0)); + refresh.on_fetched(t0); + assert_eq!(refresh.due(&policy), None); + } + + #[test] + fn state_refresh_coalesces_changes_inside_the_interval() { + let t0 = Instant::now(); + let mut refresh = StateRefresh::fetched_at(t0); + let policy = StateSyncPolicy { + min_interval: Duration::from_secs(1), + ..StateSyncPolicy::default() + }; + // Several changes right after a fetch: one fetch when the window ends. + refresh.on_changed(); + refresh.on_changed(); + assert_eq!(refresh.due(&policy), Some(t0 + Duration::from_secs(1))); + // A change that comes after a quiet window is already due when it + // arrives: the window ended a second after the last fetch. + refresh.on_fetched(t0); + refresh.on_changed(); + let arrives = t0 + Duration::from_secs(5); + assert!(refresh.due(&policy).unwrap() < arrives); + } + + #[test] + fn deferred_state_refresh_waits_and_then_fetches_once() { + let t0 = Instant::now(); + let mut refresh = StateRefresh::fetched_at(t0); + let mut policy = StateSyncPolicy { + deferred: true, + ..StateSyncPolicy::default() + }; + refresh.on_changed(); + refresh.on_changed(); + assert_eq!(refresh.due(&policy), None); + policy.deferred = false; + assert_eq!(refresh.due(&policy), Some(t0)); + } + + #[test] + fn state_bodies_parse_plain_and_gzipped() { + let state = state_with_terminals(&[("p1", "t1")]); + let json = serde_json::to_vec(&state).unwrap(); + let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default()); + std::io::Write::write_all(&mut encoder, &json).unwrap(); + let gzipped = encoder.finish().unwrap(); + + for (body, gzip) in [(&json, false), (&gzipped, true)] { + let parsed = parse_state_body(body, gzip).expect("state parses"); + assert_eq!(wanted_streams(&parsed, None), vec!["t1"]); + } + assert!(parse_state_body(&json, true).is_err()); + } + + #[test] + fn without_a_filter_every_terminal_streams() { + let state = state_with_terminals(&[("p1", "t1"), ("p2", "t2")]); + assert_eq!(wanted_streams(&state, None), vec!["t1", "t2"]); + } + + #[test] + fn a_filter_streams_only_listed_terminals_that_exist() { + let state = state_with_terminals(&[("p1", "t1"), ("p2", "t2")]); + let filter: HashSet = ["t2", "gone"].map(String::from).into(); + assert_eq!(wanted_streams(&state, Some(&filter)), vec!["t2"]); + assert!(wanted_streams(&state, Some(&HashSet::new())).is_empty()); + } + + #[test] + fn subscription_changes_start_new_streams_and_stop_unwanted_ones() { + let subscribed: HashSet = ["t1", "t2"].map(String::from).into(); + let wanted = vec!["t2".to_string(), "t3".to_string()]; + assert_eq!( + subscription_changes(&subscribed, &wanted), + (vec!["t3".to_string()], vec!["t1".to_string()]) + ); + assert_eq!( + subscription_changes(&subscribed, &["t1".to_string(), "t2".to_string()]), + (vec![], vec![]) + ); + } + + #[test] + fn only_the_reply_to_the_latest_subscribe_maps_a_stream() { + let mut subs = StreamSubscriptions::default(); + let t1 = || vec!["t1".to_string()]; + // Subscribe, unsubscribe and subscribe again before any reply. + subs.sent_subscribe(&t1()); + subs.sent_unsubscribe(&t1()); + subs.sent_subscribe(&t1()); + + // The first reply names stream 1, which the server already dropped. + assert!( + subs.accept_reply(HashMap::from([("t1".to_string(), 1)])) + .is_empty() + ); + // The reply to the second Subscribe is live. + assert_eq!( + subs.accept_reply(HashMap::from([("t1".to_string(), 2)])), + HashMap::from([("t1".to_string(), 2)]) + ); + assert!(subs.pending.is_empty()); + } + + #[test] + fn a_reply_after_an_unsubscribe_maps_nothing() { + let mut subs = StreamSubscriptions::default(); + subs.sent_subscribe(&["t1".to_string()]); + subs.sent_unsubscribe(&["t1".to_string()]); + assert!( + subs.accept_reply(HashMap::from([("t1".to_string(), 1)])) + .is_empty() + ); + assert!(subs.pending.is_empty()); + } + + #[tokio::test] + async fn reconcile_forgets_stream_ids_of_stopped_terminals() { + let (tx, rx) = async_channel::unbounded(); + let mut subscribed = StreamSubscriptions { + subscribed: ["t1", "t2"].map(String::from).into(), + pending: HashMap::new(), + }; + let stream_map = std::sync::RwLock::new(HashMap::from([ + ("t1".to_string(), 1), + ("t2".to_string(), 2), + ])); + let mut reverse = HashMap::from([(1, "t1".to_string()), (2, "t2".to_string())]); + + reconcile_streams( + &mut subscribed, + &["t2".to_string(), "t3".to_string()], + &tx, + &stream_map, + &mut reverse, + ) + .await; + + assert_eq!(subscribed.subscribed, ["t2", "t3"].map(String::from).into()); + assert_eq!(subscribed.pending, HashMap::from([("t3".to_string(), 1)])); + assert_eq!( + *stream_map.read().unwrap(), + HashMap::from([("t2".to_string(), 2)]) + ); + assert_eq!(reverse, HashMap::from([(2, "t2".to_string())])); + assert!(matches!( + rx.try_recv(), + Ok(WsClientMessage::Unsubscribe { terminal_ids }) if terminal_ids == ["t1"] + )); + assert!(matches!( + rx.try_recv(), + Ok(WsClientMessage::Subscribe { terminal_ids }) if terminal_ids == ["t3"] + )); + assert!(rx.try_recv().is_err()); + } + #[derive(Default)] struct TestHandler { remove_all_calls: AtomicUsize, @@ -2908,6 +3495,7 @@ mod tests { claude_dialogs: Default::default(), claude_sessions: Vec::new(), claude_pr_links: Default::default(), + terminal_attention: Default::default(), })); client.reconnect(); diff --git a/crates/okena-transport/src/client/mod.rs b/crates/okena-transport/src/client/mod.rs index 480b3bf6e..077d0577e 100644 --- a/crates/okena-transport/src/client/mod.rs +++ b/crates/okena-transport/src/client/mod.rs @@ -7,7 +7,7 @@ pub mod tls; pub mod types; pub use crate::{LOCAL_DAEMON_CONNECTION_ID, LocalEndpoint, RemoteConnectionConfig}; -pub use connection::{ConnectionHandler, RemoteClient}; +pub use connection::{ConnectionHandler, RemoteClient, StateSyncPolicy, TrafficStats}; pub use id::{is_remote_terminal, make_prefixed_id, strip_prefix}; pub use state::{ StateDiff, collect_all_terminal_ids, collect_layout_terminal_ids, collect_state_terminal_ids, diff --git a/crates/okena-transport/src/client/state.rs b/crates/okena-transport/src/client/state.rs index a20bdc556..d55c109f5 100644 --- a/crates/okena-transport/src/client/state.rs +++ b/crates/okena-transport/src/client/state.rs @@ -180,6 +180,7 @@ mod tests { claude_dialogs: Default::default(), claude_sessions: Vec::new(), claude_pr_links: Default::default(), + terminal_attention: Default::default(), } } diff --git a/crates/okena-transport/src/client/types.rs b/crates/okena-transport/src/client/types.rs index b13cd4bc1..854d370d5 100644 --- a/crates/okena-transport/src/client/types.rs +++ b/crates/okena-transport/src/client/types.rs @@ -42,6 +42,8 @@ pub enum WsClientMessage { /// Declare this client's windows, as a full replacement list. Server-side /// project ids (unprefixed) — `GET /v1/state` lists them. SetWindows { windows: Vec }, + /// Whether the server should push `system_stats_changed` to this client. + SetSystemStats { enabled: bool }, /// Application-level keepalive (`{"type":"ping"}`); the server answers /// `pong`. Sent by the session itself to detect dead connections. Ping, diff --git a/crates/okena-transport/src/remote_http.rs b/crates/okena-transport/src/remote_http.rs index dbfaaf9d2..7fe55e700 100644 --- a/crates/okena-transport/src/remote_http.rs +++ b/crates/okena-transport/src/remote_http.rs @@ -13,6 +13,8 @@ pub fn async_client_and_url( if let Some(crate::LocalEndpoint::UnixSocket { path: socket_path }) = &config.local_endpoint { let client = reqwest::Client::builder() .unix_socket(socket_path.as_str()) + // Same-host socket: compressing would only spend CPU on both ends. + .no_gzip() .build() .map_err(|error| format!("Cannot initialise Unix socket HTTP client: {error}"))?; return Ok((client, config.http_url(path))); @@ -38,6 +40,8 @@ pub fn blocking_client_and_url( if let Some(crate::LocalEndpoint::UnixSocket { path: socket_path }) = &config.local_endpoint { let client = reqwest::blocking::Client::builder() .unix_socket(socket_path.as_str()) + // Same-host socket: compressing would only spend CPU on both ends. + .no_gzip() .timeout(timeout) .connect_timeout(std::time::Duration::from_secs(5)) .build() diff --git a/crates/okena-transport/src/tls.rs b/crates/okena-transport/src/tls.rs index 0d6566427..8d62b5b79 100644 --- a/crates/okena-transport/src/tls.rs +++ b/crates/okena-transport/src/tls.rs @@ -184,14 +184,25 @@ pub fn build_reqwest_client( pinned: Option, observed: ObservedFingerprint, ) -> Result { - if !tls { - return Ok(reqwest::Client::new()); - } - let config = pinned_client_config(pinned, observed); - reqwest::Client::builder() - .use_preconfigured_tls(config) + reqwest_client_builder(tls, pinned, observed) .build() - .map_err(|error| format!("Cannot initialise pinned TLS client: {error}")) + .map_err(|error| format!("Cannot initialise HTTP client: {error}")) +} + +/// The builder behind [`build_reqwest_client`], for callers that need to +/// adjust it (e.g. turn off transparent decompression to measure the wire). +#[cfg(any(feature = "client", feature = "cancellable-http"))] +pub fn reqwest_client_builder( + tls: bool, + pinned: Option, + observed: ObservedFingerprint, +) -> reqwest::ClientBuilder { + let builder = reqwest::Client::builder(); + if tls { + builder.use_preconfigured_tls(pinned_client_config(pinned, observed)) + } else { + builder + } } /// Build a blocking reqwest client for a connection. This applies the same diff --git a/crates/okena-workspace/src/remote_apply.rs b/crates/okena-workspace/src/remote_apply.rs index c5db36107..228c41f7b 100644 --- a/crates/okena-workspace/src/remote_apply.rs +++ b/crates/okena-workspace/src/remote_apply.rs @@ -661,6 +661,7 @@ mod tests { claude_dialogs: Default::default(), claude_sessions: Vec::new(), claude_pr_links: Default::default(), + terminal_attention: Default::default(), } } diff --git a/crates/okena-workspace/src/state.rs b/crates/okena-workspace/src/state.rs index de18ab04b..1db106e10 100644 --- a/crates/okena-workspace/src/state.rs +++ b/crates/okena-workspace/src/state.rs @@ -2883,6 +2883,7 @@ mod workspace_tests { claude_dialogs: Default::default(), claude_sessions: Vec::new(), claude_pr_links: Default::default(), + terminal_attention: Default::default(), }), } } @@ -4578,6 +4579,7 @@ mod gpui_tests { claude_dialogs: Default::default(), claude_sessions: Vec::new(), claude_pr_links: Default::default(), + terminal_attention: Default::default(), }), } } diff --git a/docs/reference/mobile.md b/docs/reference/mobile.md index 39977d9e8..95265cd8a 100644 --- a/docs/reference/mobile.md +++ b/docs/reference/mobile.md @@ -103,6 +103,59 @@ Soft keyboard / KeyToolbar tap → sendText() / sendSpecialKey() (JSI) → ConnectionManager.send_ws_message() → WsClientMessage::SendText → WS → server PTY stdin ``` +### What streams + +The phone streams PTY output only for the terminal on screen. A connection starts with nothing +streamed (`ConnectionManager` sets an empty filter before connecting), and the workspace store +calls `setStreamedTerminals` with the selected terminal, or with `[]` while the dashboard is open, +the app is in the background or polling stopped. The filter lives in the transport +(`RemoteClient::set_streamed_terminals`), so it survives reconnects; the session reconciles its +subscriptions on every filter change and `state_changed`. A terminal that streams again gets a +fresh snapshot from the server, so switching terminals costs one snapshot. + +Consequence: the output-based "running" signal comes only from the streamed terminal. Bells and +OSC 9/777 notifications of the others come with the state: the daemon stamps each terminal's last +bell and notification (`StateResponse.terminal_attention`, Unix millis), and the phone raises a +terminal's unseen flag when its stamp moved between two states (never by comparing clocks; the +first state of a connection is the baseline). Agent status (`terminalAgentStatus`) and Claude +dialogs also come with the state. A bell the terminal already showed in its own output since the +last state is not raised again from the stamp. The stamps need a daemon that sends +`terminal_attention`; against an older one, bells and notifications of terminals off screen are +not flagged. The connection dot's liveness counts any inbound frame (pong, state events), not +terminal output; on a quiet connection only the pong to the 15 s ping arrives, so the dot stays +green up to 20 s and turns red from 45 s. + +The daemon's git poller puts projects with a streamed terminal on its responsive tier, and the +phone declares no visible projects. With only the phone connected, projects other than the one on +screen get a full git walk only after a change signal (terminal output, index, HEAD) or every +5 minutes, so their line counts on the dashboard can lag behind edits that print nothing. This is +deliberate: every git change is pushed to every client as the full `git_status_changed` map, so a +faster tier for all projects would cost data. + +The phone opts out of the 2 s `system_stats_changed` push (`set_system_stats` with +`enabled: false`); it shows no host metrics. Git status arrives as `git_status_changed` and is +patched into the cached state; it no longer bumps `state_version`, so a git change does not make +any client refetch the whole state. + +The JSON routes (`/v1/state`, `/v1/actions`, usage, Claude session/transcript) are gzipped for +clients that send `Accept-Encoding: gzip`; the transport's reqwest client does, except over the +local Unix socket. + +### State fetches + +Every `state_changed` makes the transport refetch the whole `/v1/state`. The phone sets a +`StateSyncPolicy` on each connection: at most one fetch per second (changes inside the window +are coalesced into one fetch at its end, the first change after a quiet window fetches at once), +none in the background (the store calls `setAppInBackground` on every lifecycle change, also while +the Android connection service keeps the session alive; a state that changed meanwhile is fetched +once on return), and no settings, which the app does not use. The desktop keeps the default +policy: fetch on every change, settings included. + +The transport counts payload bytes per connection (`traffic_stats`: PTY, snapshots, state with +the number of fetches, settings, other events). The state is fetched with transparent +decompression off and decoded by hand, so the count is the size on the wire. Settings shows it +under Data while connected. + ### State sync (project list, layouts) ``` diff --git a/mobile/rn/src/App.tsx b/mobile/rn/src/App.tsx index b397ab01e..65eff3351 100644 --- a/mobile/rn/src/App.tsx +++ b/mobile/rn/src/App.tsx @@ -97,9 +97,11 @@ export const App: React.FC = () => { // A failed restore just leaves the user on the server list. }); // The store keeps background reconnects enabled while the Android service runs. - const appState = AppState.addEventListener('change', (next) => - connectionStore().getState().onAppStateChange(next), - ); + const appState = AppState.addEventListener('change', (next) => { + connectionStore().getState().onAppStateChange(next); + // Background streams nothing, foreground the terminal on screen again. + workspaceStore().getState().syncStreams(); + }); return () => { appState.remove(); unbindNav(); diff --git a/mobile/rn/src/components/ConnectionDot.tsx b/mobile/rn/src/components/ConnectionDot.tsx index cc650453d..270570806 100644 --- a/mobile/rn/src/components/ConnectionDot.tsx +++ b/mobile/rn/src/components/ConnectionDot.tsx @@ -1,7 +1,7 @@ /** * ConnectionDot.tsx — the connection-quality dot of the app bars: green when - * the server sent data in the last 3 s, amber up to 10 s or while - * reconnecting, red after that. + * the server sent anything in the last {@link ACTIVITY_LIVE_S} s, amber up to + * {@link ACTIVITY_STALE_S} s or while reconnecting, red after that. */ import React from 'react'; @@ -12,6 +12,7 @@ import { useConnectionStore, useWorkspaceStore, } from '../state'; +import { ACTIVITY_LIVE_S, ACTIVITY_STALE_S } from '../state/workspaceStore'; import { OkenaColors } from '../theme'; export const ConnectionDot: React.FC = () => { @@ -19,9 +20,9 @@ export const ConnectionDot: React.FC = () => { const secondsSinceActivity = useWorkspaceStore((s) => s.secondsSinceActivity); const color = reconnecting ? OkenaColors.warning - : secondsSinceActivity < 3 + : secondsSinceActivity < ACTIVITY_LIVE_S ? OkenaColors.success - : secondsSinceActivity < 10 + : secondsSinceActivity < ACTIVITY_STALE_S ? OkenaColors.warning : OkenaColors.error; return ; diff --git a/mobile/rn/src/components/DataUsage.test.ts b/mobile/rn/src/components/DataUsage.test.ts new file mode 100644 index 000000000..21c30c1e9 --- /dev/null +++ b/mobile/rn/src/components/DataUsage.test.ts @@ -0,0 +1,32 @@ +/// + +import { formatBytes, trafficRows } from './DataUsage'; + +describe('data usage', () => { + it('formats byte counts', () => { + expect(formatBytes(512)).toBe('512 B'); + expect(formatBytes(1536)).toBe('1.5 KB'); + expect(formatBytes(234 * 1024)).toBe('234 KB'); + expect(formatBytes(12.3 * 1024 * 1024)).toBe('12 MB'); + expect(formatBytes(3 * 1024 ** 3)).toBe('3.0 GB'); + }); + + it('lists each kind and a total', () => { + const rows = trafficRows({ + ptyBytes: 1024, + snapshotBytes: 1024, + stateBytes: 2048, + stateFetches: 1, + settingsBytes: 512, + eventBytes: 512, + }); + expect(rows.map((r) => r.value)).toEqual([ + '1.0 KB', + '1.0 KB', + '2.0 KB · 1 fetch', + '1.0 KB', + '5.0 KB', + ]); + expect(rows.at(-1)?.label).toBe('Total'); + }); +}); diff --git a/mobile/rn/src/components/DataUsage.tsx b/mobile/rn/src/components/DataUsage.tsx new file mode 100644 index 000000000..efe992c14 --- /dev/null +++ b/mobile/rn/src/components/DataUsage.tsx @@ -0,0 +1,122 @@ +/** + * DataUsage.tsx — what the current connection has downloaded, by kind. + * + * Reads `getTrafficStats` (payload bytes counted in the Rust transport since + * the connection was created) every couple of seconds while mounted. TLS, + * HTTP and WebSocket framing are not included, so the system's per-app + * counter reads somewhat higher. + */ + +import React, { useEffect, useState } from 'react'; +import { StyleSheet, Text, View } from 'react-native'; + +import type { ConnId, OkenaNative, TrafficStats } from '../native/okena'; +import { OkenaColors } from '../theme'; + +export const DATA_USAGE_POLL_MS = 2000; + +export type TrafficNative = Pick; + +export function formatBytes(bytes: number): string { + if (bytes < 1024) return `${bytes} B`; + const units = ['KB', 'MB', 'GB']; + let value = bytes / 1024; + let unit = 0; + while (value >= 1024 && unit < units.length - 1) { + value /= 1024; + unit += 1; + } + return `${value < 10 ? value.toFixed(1) : Math.round(value)} ${units[unit]}`; +} + +export interface TrafficRow { + label: string; + value: string; +} + +/** The rows shown for `stats`, total last. */ +export function trafficRows(stats: TrafficStats): TrafficRow[] { + const total = + stats.ptyBytes + + stats.snapshotBytes + + stats.stateBytes + + stats.settingsBytes + + stats.eventBytes; + const fetches = stats.stateFetches === 1 ? '1 fetch' : `${stats.stateFetches} fetches`; + return [ + { label: 'Terminal output', value: formatBytes(stats.ptyBytes) }, + { label: 'Terminal snapshots', value: formatBytes(stats.snapshotBytes) }, + { + label: 'Workspace state', + value: `${formatBytes(stats.stateBytes)} · ${fetches}`, + }, + { + label: 'Events and settings', + value: formatBytes(stats.eventBytes + stats.settingsBytes), + }, + { label: 'Total', value: formatBytes(total) }, + ]; +} + +export const DataUsageGroup: React.FC<{ native: TrafficNative; connId: ConnId }> = ({ + native, + connId, +}) => { + const [stats, setStats] = useState(() => + native.getTrafficStats(connId), + ); + useEffect(() => { + setStats(native.getTrafficStats(connId)); + const timer = setInterval( + () => setStats(native.getTrafficStats(connId)), + DATA_USAGE_POLL_MS, + ); + return () => clearInterval(timer); + }, [native, connId]); + + if (!stats) return null; + const rows = trafficRows(stats); + return ( + + {rows.map((row, i) => ( + + + {row.label} + + + {row.value} + + + ))} + + ); +}; + +const styles = StyleSheet.create({ + group: { + marginHorizontal: 16, + borderRadius: 12, + backgroundColor: OkenaColors.surfaceElevated, + overflow: 'hidden', + }, + row: { + flexDirection: 'row', + alignItems: 'center', + justifyContent: 'space-between', + minHeight: 44, + paddingHorizontal: 16, + paddingVertical: 8, + }, + rowDivider: { + borderBottomWidth: StyleSheet.hairlineWidth, + borderBottomColor: OkenaColors.border, + }, + label: { color: OkenaColors.textPrimary, fontSize: 15 }, + value: { color: OkenaColors.textSecondary, fontSize: 15, fontVariant: ['tabular-nums'] }, + total: { fontWeight: '600', color: OkenaColors.textPrimary }, +}); diff --git a/mobile/rn/src/native/okena.ts b/mobile/rn/src/native/okena.ts index 1308a1121..aed1d73b1 100644 --- a/mobile/rn/src/native/okena.ts +++ b/mobile/rn/src/native/okena.ts @@ -465,6 +465,16 @@ export interface OkenaNative { */ onAppBackground(): void; + /** + * The app went to (`true`) or left (`false`) the background. Call on every + * change, also while the Android connection service runs: in the background + * no workspace state is fetched; a changed state is fetched once on return. + */ + setAppInBackground(background: boolean): void; + + /** Payload bytes the connection has received since it connected, by kind. */ + getTrafficStats(connId: ConnId): TrafficStats | undefined; + /** * Retry one connection now: skip the backoff and reset the attempt budget, * keeping terminals and cached state. Only meaningful when @@ -561,14 +571,23 @@ export interface OkenaNative { getProjectOrder(connId: ConnId): ProjectId[]; /** - * Activity signals of every terminal the connection mirrors (the client - * subscribes to all of them). Cheap: one call per state poll. + * Activity signals of every terminal the connection mirrors. Output time + * only moves for streamed terminals (see {@link setStreamedTerminals}); + * bells and notifications of the rest arrive with the state. Cheap: one + * call per state poll. */ getTerminalActivity(connId: ConnId): TerminalActivity[]; /** The user viewed a terminal: clear its unseen bell / notification. */ markTerminalSeen(connId: ConnId, terminalId: TerminalId): void; + /** + * Stream PTY output only for these terminals (the ones on screen); `[]` + * streams nothing. A connection starts with nothing streamed. A terminal + * that is streamed again gets a fresh snapshot from the server. + */ + setStreamedTerminals(connId: ConnId, terminalIds: TerminalId[]): void; + /** The fullscreen terminal, if one is active. */ getFullscreenTerminal(connId: ConnId): FullscreenInfo | undefined; @@ -887,6 +906,29 @@ function toProjectInfo(p: GenProjectInfo): ProjectInfo { }; } +/** Payload bytes received since connecting (framing overhead excluded). */ +export interface TrafficStats { + ptyBytes: number; + snapshotBytes: number; + stateBytes: number; + stateFetches: number; + settingsBytes: number; + eventBytes: number; +} + +type GenTrafficStats = { [K in keyof TrafficStats]: bigint | number }; + +function toTrafficStats(s: GenTrafficStats): TrafficStats { + return { + ptyBytes: Number(s.ptyBytes), + snapshotBytes: Number(s.snapshotBytes), + stateBytes: Number(s.stateBytes), + stateFetches: Number(s.stateFetches), + settingsBytes: Number(s.settingsBytes), + eventBytes: Number(s.eventBytes), + }; +} + type GenTerminalActivity = Omit & { lastOutputMsAgo?: bigint | number; }; @@ -938,6 +980,12 @@ export function getOkenaNative(): OkenaNative { markTerminalSeen: (connId: ConnId, terminalId: TerminalId) => { if (typeof gen.markTerminalSeen === 'function') gen.markTerminalSeen(connId, terminalId); }, + // Bindings generated before `set_streamed_terminals` existed stream + // every terminal, as before. + setStreamedTerminals: (connId: ConnId, terminalIds: TerminalId[]) => { + if (typeof gen.setStreamedTerminals === 'function') + gen.setStreamedTerminals(connId, terminalIds); + }, // Bindings generated before `scroll_wheel` / `take_dirty` existed: keep // scrolling the local scrollback and keep the always-repaint loop. scrollWheel: @@ -973,6 +1021,14 @@ export function getOkenaNative(): OkenaNative { onAppBackground: () => { if (typeof gen.onAppBackground === 'function') gen.onAppBackground(); }, + setAppInBackground: (background: boolean) => { + if (typeof gen.setAppInBackground === 'function') gen.setAppInBackground(background); + }, + getTrafficStats: (connId: ConnId) => { + if (typeof gen.getTrafficStats !== 'function') return undefined; + const stats = gen.getTrafficStats(connId) as GenTrafficStats | undefined; + return stats === undefined ? undefined : toTrafficStats(stats); + }, reconnectNow: (connId: ConnId) => { if (typeof gen.reconnectNow === 'function') gen.reconnectNow(connId); }, diff --git a/mobile/rn/src/screens/SettingsSheet.tsx b/mobile/rn/src/screens/SettingsSheet.tsx index b929aa50f..4fdb87159 100644 --- a/mobile/rn/src/screens/SettingsSheet.tsx +++ b/mobile/rn/src/screens/SettingsSheet.tsx @@ -15,6 +15,8 @@ * Projects: the project order of the drawer and the switcher (Activity, the * default, or Manual). * + * Data: what the current connection has downloaded (only while connected). + * * Advanced: the terminal renderer toggle (native view, the default, or the * Skia fallback) for the native renderer rollout. */ @@ -41,6 +43,9 @@ import { type ProjectSortMode, } from '../state/settingsStore'; import { OkenaColors, OkenaTypography, TerminalTheme } from '../theme'; +import { DataUsageGroup, type TrafficNative } from '../components/DataUsage'; +import { getOkenaNative } from '../native/okena'; +import { useConnectionStore } from '../state/connectionStore'; /** JetBrains Mono advance width as a fraction of the font size (600/1000 em). */ const MONO_ADVANCE_EM = 0.6; @@ -55,10 +60,23 @@ export function formatFontSize(size: number): string { return Number.isInteger(size) ? `${size} pt` : `${size.toFixed(1)} pt`; } -export const SettingsSheet: React.FC<{ visible: boolean; onClose: () => void }> = ({ - visible, - onClose, -}) => { +function defaultTrafficNative(): TrafficNative | undefined { + try { + return getOkenaNative(); + } catch { + // Off-device without generated bindings: no data section. + return undefined; + } +} + +export const SettingsSheet: React.FC<{ + visible: boolean; + onClose: () => void; + /** Injected for tests; defaults to the native module. */ + native?: TrafficNative; +}> = ({ visible, onClose, native }) => { + const connId = useConnectionStore((s) => (s.status.kind === 'connected' ? s.connId : null)); + const trafficNative = React.useMemo(() => native ?? defaultTrafficNative(), [native]); const insets = useSafeAreaInsets(); const { width } = useWindowDimensions(); const fontSize = useSettingsStore((s) => s.terminalFontSize); @@ -216,6 +234,17 @@ export const SettingsSheet: React.FC<{ visible: boolean; onClose: () => void }> : "The desktop sidebar's manual order, with folders."} + {connId && trafficNative ? ( + <> + DATA + + + Downloaded from this server since connecting. Only the terminal on screen streams + its output; TLS and protocol overhead are not counted. + + + ) : null} + ADVANCED diff --git a/mobile/rn/src/state/connectionStore.test.ts b/mobile/rn/src/state/connectionStore.test.ts index c73ee2f87..c0d6f5324 100644 --- a/mobile/rn/src/state/connectionStore.test.ts +++ b/mobile/rn/src/state/connectionStore.test.ts @@ -41,6 +41,7 @@ function stubNative( secondsSinceActivity: () => 0, onAppForeground: () => {}, onAppBackground: () => {}, + setAppInBackground: () => {}, reconnectNow: () => {}, canReconnectNow: () => true, ...overrides, @@ -444,6 +445,31 @@ describe('reconnect handling', () => { expect(onAppForeground).toHaveBeenCalledTimes(1); }); + it('defers state fetches in the background, also while the service runs', () => { + const setAppInBackground = jest.fn(); + setup(stubNative({ setAppInBackground })); + configureConnectionStore({ + service: { setActive: jest.fn(), isRunning: () => true }, + }); + const store = useConnectionStore.getState(); + + store.onAppStateChange('background'); + expect(setAppInBackground).toHaveBeenLastCalledWith(true); + store.onAppStateChange('active'); + expect(setAppInBackground).toHaveBeenLastCalledWith(false); + }); + + it('resumes state fetches when iOS returns through inactive', () => { + const setAppInBackground = jest.fn(); + setup(stubNative({ setAppInBackground })); + const store = useConnectionStore.getState(); + + store.onAppStateChange('background'); + store.onAppStateChange('inactive'); + store.onAppStateChange('active'); + expect(setAppInBackground).toHaveBeenLastCalledWith(false); + }); + it('keeps reconnects running in the background only while the service is running', () => { const onAppBackground = jest.fn(); setup(stubNative({ onAppBackground })); diff --git a/mobile/rn/src/state/connectionStore.ts b/mobile/rn/src/state/connectionStore.ts index 08942c1e3..a6639e70c 100644 --- a/mobile/rn/src/state/connectionStore.ts +++ b/mobile/rn/src/state/connectionStore.ts @@ -68,6 +68,7 @@ export type ConnectionNative = Pick< | 'secondsSinceActivity' | 'onAppForeground' | 'onAppBackground' + | 'setAppInBackground' | 'reconnectNow' | 'canReconnectNow' >; @@ -493,9 +494,15 @@ export const useConnectionStore: UseBoundStore> = lastAppState = next; const native = deps().native; if (next === 'background') { + // State fetches wait for the foreground even while the service keeps + // the session alive; reconnects pause only without the service. + native.setAppInBackground(true); if (!deps().service.isRunning()) native.onAppBackground(); return; } + // Resume state fetches on any return to active: iOS may go background → + // inactive → active, and a deferral left on would freeze the state. + if (next === 'active') native.setAppInBackground(false); if (next !== 'active' || previous !== 'background') return; native.onAppForeground(); const { connId } = get(); diff --git a/mobile/rn/src/state/sessionRestore.test.ts b/mobile/rn/src/state/sessionRestore.test.ts index d901e7fc8..e597f004b 100644 --- a/mobile/rn/src/state/sessionRestore.test.ts +++ b/mobile/rn/src/state/sessionRestore.test.ts @@ -62,6 +62,7 @@ function stubNative(connect: jest.Mock): ConnectionNative { secondsSinceActivity: () => 0, onAppForeground: () => {}, onAppBackground: () => {}, + setAppInBackground: () => {}, reconnectNow: () => {}, canReconnectNow: () => true, }; diff --git a/mobile/rn/src/state/workspaceStore.test.ts b/mobile/rn/src/state/workspaceStore.test.ts index 4ed714274..b3a3c4b37 100644 --- a/mobile/rn/src/state/workspaceStore.test.ts +++ b/mobile/rn/src/state/workspaceStore.test.ts @@ -64,6 +64,7 @@ function stubNative(layouts: string[]): WorkspaceNative { secondsSinceActivity: () => 0, getTerminalActivity: () => [], markTerminalSeen: () => {}, + setStreamedTerminals: () => {}, focusTerminal: () => Promise.resolve(), getUsage: () => undefined, getUsageRevision: () => 0, @@ -780,3 +781,90 @@ describe('workspace agent usage', () => { expect(useWorkspaceStore.getState().usageRefreshing).toBe(false); }); }); + +describe('terminal streaming', () => { + let streamed: string[][]; + let foreground: boolean; + + beforeEach(() => { + jest.useFakeTimers(); + streamed = []; + foreground = true; + configureWorkspaceStore({ + native: { + ...stubNative([TABS_FIRST_ACTIVE]), + setStreamedTerminals: (_c, ids) => { + streamed.push(ids); + }, + }, + isForeground: () => foreground, + isInBackground: () => !foreground, + }); + }); + + afterEach(() => { + useWorkspaceStore.getState().stop(); + configureWorkspaceStore({ isForeground: undefined, isInBackground: undefined }); + jest.useRealTimers(); + }); + + it('streams nothing on the dashboard and only the terminal on screen', () => { + useWorkspaceStore.getState().start('conn-1'); + // The connection opens on the dashboard. + expect(streamed).toEqual([[]]); + + useWorkspaceStore.getState().showWorkspace(); + expect(streamed.at(-1)).toEqual(['t1']); + + useWorkspaceStore.getState().selectTerminal('t2'); + expect(streamed.at(-1)).toEqual(['t2']); + + useWorkspaceStore.getState().showDashboard(); + expect(streamed.at(-1)).toEqual([]); + }); + + it('sends a set only when it changes', () => { + useWorkspaceStore.getState().start('conn-1'); + useWorkspaceStore.getState().showWorkspace(); + const sent = streamed.length; + + jest.advanceTimersByTime(WORKSPACE_POLL_MS * 3); + expect(streamed.length).toBe(sent); + }); + + it('stops streaming in the background and resumes in the foreground', () => { + useWorkspaceStore.getState().start('conn-1'); + useWorkspaceStore.getState().showWorkspace(); + + foreground = false; + useWorkspaceStore.getState().syncStreams(); + expect(streamed.at(-1)).toEqual([]); + + foreground = true; + useWorkspaceStore.getState().syncStreams(); + expect(streamed.at(-1)).toEqual(['t1']); + }); + + it('keeps streaming while iOS is inactive', () => { + let background = false; + configureWorkspaceStore({ + isForeground: () => false, + isInBackground: () => background, + }); + useWorkspaceStore.getState().start('conn-1'); + useWorkspaceStore.getState().showWorkspace(); + expect(streamed.at(-1)).toEqual(['t1']); + + background = true; + useWorkspaceStore.getState().syncStreams(); + expect(streamed.at(-1)).toEqual([]); + }); + + it('stops streaming when polling stops', () => { + useWorkspaceStore.getState().start('conn-1'); + useWorkspaceStore.getState().showWorkspace(); + + useWorkspaceStore.getState().stop(); + expect(streamed.at(-1)).toEqual([]); + }); +}); diff --git a/mobile/rn/src/state/workspaceStore.ts b/mobile/rn/src/state/workspaceStore.ts index 1fa8ddab9..fd765d981 100644 --- a/mobile/rn/src/state/workspaceStore.ts +++ b/mobile/rn/src/state/workspaceStore.ts @@ -16,6 +16,12 @@ * - the server's agent usage (`getUsage`), fetched only when * `getUsageRevision` moved, so an unchanged snapshot costs one call. * + * Only the terminal on screen streams its PTY output (`setStreamedTerminals`): + * none while the dashboard is open or the app is in the background. The + * output-based "running" signal therefore only comes from the terminal on + * screen; bells, notifications, agent status and Claude dialogs of every + * terminal come with the state. + * * The terminal on screen counts as seen: each poll (and each selection) * clears its unseen bell / notification and acknowledges a `done` agent. Its * output does not make its project "running". While the dashboard is open no @@ -69,6 +75,15 @@ import { } from '../models/activityOrder'; import type { TerminalActivity } from '../native/okena'; +/** + * Connection dot thresholds (s since anything last arrived). The phone streams + * at most one terminal and skips host metrics, so on a quiet connection the + * pong to the transport's 15 s ping is the only frame: green up to one missed + * beat, red from 45 s, where the transport itself starts probing. + */ +export const ACTIVITY_LIVE_S = 20; +export const ACTIVITY_STALE_S = 45; + /** Poll interval (ms) for the workspace state — Dart used 1000ms. */ export const WORKSPACE_POLL_MS = 1000; @@ -99,6 +114,7 @@ export type WorkspaceNative = Pick< | 'secondsSinceActivity' | 'getTerminalActivity' | 'markTerminalSeen' + | 'setStreamedTerminals' | 'focusTerminal' | 'getUsage' | 'getUsageRevision' @@ -109,10 +125,16 @@ export type WorkspaceNative = Pick< export interface WorkspaceDeps { native: WorkspaceNative; /** - * Whether the app is in the foreground. The terminal on screen only counts - * as seen while it is. + * Whether the app is in the foreground (`active`). The terminal on screen + * only counts as seen while it is. */ isForeground: () => boolean; + /** + * Whether the app is in the background. The terminal on screen streams + * until it is: `inactive` (Control Center, a system alert, Face ID) is a + * transient overlay, and resubscribing would download a fresh snapshot. + */ + isInBackground: () => boolean; } /** @@ -207,6 +229,12 @@ export interface WorkspaceState { * {@link selectProject}) or on the current selection. */ showWorkspace(projectId?: ProjectId, terminalId?: TerminalId): void; + /** + * Stream only the terminal on screen (none on the dashboard or in the + * background). Runs after every poll and selection; call it when the app + * changes foreground state, since polling may be paused in the background. + */ + syncStreams(): void; /** Acknowledge {@link restoreProjectMissing}. */ clearRestoreProjectMissing(): void; /** @@ -226,6 +254,9 @@ function deps(): WorkspaceDeps { native: injectedDeps.native ?? getOkenaNative(), isForeground: injectedDeps.isForeground ?? (() => AppState.currentState === 'active'), + isInBackground: + injectedDeps.isInBackground ?? + (() => AppState.currentState === 'background'), }; } return resolvedDeps; @@ -250,6 +281,8 @@ let activeConnId: ConnId | null = null; let previousTerminalIds: Set | null = null; /** Selection to reopen once the first non-empty project list arrives. */ let pendingRestore: SelectionMemory | null = null; +/** Terminals last passed to `setStreamedTerminals`, joined; null = not yet sent. */ +let streamedKey: string | null = null; /** Latest per-terminal activity, reused when a selection re-derives signals. */ let lastActivity: TerminalActivity[] = []; /** @@ -283,6 +316,10 @@ function stopPolling(): void { clearInterval(pollTimer); pollTimer = null; } + // Nothing is on screen any more; a connection that stays up stops streaming. + if (activeConnId !== null && streamedKey !== '') + deps().native.setStreamedTerminals(activeConnId, []); + streamedKey = null; activeConnId = null; previousTerminalIds = null; pendingRestore = null; @@ -466,6 +503,24 @@ function resolveSelectedProject( } /** One poll tick — refresh remote state + run auto-selection. Mirrors `_pollState`. */ +/** + * Tell the native side which terminals to stream: the one on screen, or none + * on the dashboard and in the background. Sent only when the set changes. + */ +function syncStreams(get: StoreApi['getState']): void { + const connId = activeConnId; + if (!connId) return; + const { dashboardOpen, selectedTerminalId } = get(); + const ids = + !dashboardOpen && selectedTerminalId !== null && !deps().isInBackground() + ? [selectedTerminalId] + : []; + const key = ids.join('\n'); + if (key === streamedKey) return; + streamedKey = key; + deps().native.setStreamedTerminals(connId, ids); +} + function pollState( get: StoreApi['getState'], set: StoreApi['setState'], @@ -622,12 +677,12 @@ function pollState( } } - // Connection health (drives the staleness indicator's 3s / 10s thresholds). + // Connection health (drives the connection dot, see ACTIVITY_LIVE_S). const newActivity = native.secondsSinceActivity(connId); const oldActivity = prev.secondsSinceActivity; if ( - oldActivity < 3 !== newActivity < 3 || - oldActivity < 10 !== newActivity < 10 + oldActivity < ACTIVITY_LIVE_S !== newActivity < ACTIVITY_LIVE_S || + oldActivity < ACTIVITY_STALE_S !== newActivity < ACTIVITY_STALE_S ) { changed = true; } @@ -639,6 +694,7 @@ function pollState( // threshold crossed (cheap, avoids a stale value). set({ secondsSinceActivity: newActivity }); } + syncStreams(get); } /** @@ -781,6 +837,7 @@ export const useWorkspaceStore: UseBoundStore> = : null, ...(connId ? signalsPatch(get, projects, terminalId, connId) : {}), }); + syncStreams(get); // Opening a project is activity: the daemon stamps `lastActivityAt` // (`RecordProjectActivity`), which the activity order sorts by. if (connId && terminalId !== null && selectedProjectId !== projectId) { @@ -804,6 +861,7 @@ export const useWorkspaceStore: UseBoundStore> = ? signalsPatch(get, get().projects, terminalId, connId) : {}), }); + syncStreams(get); }, setTerminalView(terminalId, view) { @@ -816,6 +874,7 @@ export const useWorkspaceStore: UseBoundStore> = showDashboard() { if (get().dashboardOpen) return; set({ dashboardOpen: true }); + syncStreams(get); }, showWorkspace(projectId, terminalId) { @@ -829,6 +888,11 @@ export const useWorkspaceStore: UseBoundStore> = set( signalsPatch(get, get().projects, get().selectedTerminalId, connId), ); + syncStreams(get); + }, + + syncStreams() { + syncStreams(get); }, clearRestoreProjectMissing() { diff --git a/web/src/App.tsx b/web/src/App.tsx index 246e26922..5ffa5e56f 100644 --- a/web/src/App.tsx +++ b/web/src/App.tsx @@ -42,6 +42,9 @@ export function App() { case "state_changed": fetchState(); break; + case "git_status_changed": + dispatch({ type: "set_git_statuses", statuses: msg.projects }); + break; case "auth_failed": clearToken(); setAuthed(false); diff --git a/web/src/state/store.ts b/web/src/state/store.ts index 1964cbd31..6c1fa12e3 100644 --- a/web/src/state/store.ts +++ b/web/src/state/store.ts @@ -1,5 +1,5 @@ import { createContext, useContext } from "react"; -import type { StateResponse, WsOutbound } from "../api/types"; +import type { ApiGitStatus, StateResponse, WsOutbound } from "../api/types"; import { WsManager, type WsStatus } from "../api/websocket"; // ── Terminal Registry ─────────────────────────────────────────────────────── @@ -57,10 +57,37 @@ export interface AppState { streamMappings: Record; /** Terminal whose Claude chat panel is open, if any. */ claudePanelTerminalId: string | null; + /** + * The last `git_status_changed` map of this WS connection. Git changes do + * not trigger a state refetch, so a `/v1/state` response that was in flight + * when the push arrived must not overwrite it. Cleared when the connection + * status changes (no push is replayed on reconnect). + */ + gitStatuses: Record | null; +} + +/** + * Apply a pushed git status map. The push is the daemon's complete map, so a + * project it lacks has no status, except in a state response that is newer + * than the push (`keepMissing`): a project added since keeps its own status. + */ +function withGitStatuses( + workspace: StateResponse, + statuses: Record, + keepMissing: boolean, +): StateResponse { + return { + ...workspace, + projects: workspace.projects.map((p) => ({ + ...p, + git_status: statuses[p.id] ?? (keepMissing ? p.git_status : null), + })), + }; } export type AppAction = | { type: "set_workspace"; workspace: StateResponse } + | { type: "set_git_statuses"; statuses: Record } | { type: "select_project"; projectId: string } | { type: "select_terminal"; terminalId: string | null } | { type: "set_sidebar_open"; open: boolean } @@ -73,7 +100,21 @@ export type AppAction = export function appReducer(state: AppState, action: AppAction): AppState { switch (action.type) { case "set_workspace": - return { ...state, workspace: action.workspace }; + return { + ...state, + workspace: state.gitStatuses + ? withGitStatuses(action.workspace, state.gitStatuses, true) + : action.workspace, + }; + case "set_git_statuses": + // Git status changes arrive as their own message, not as a state change. + return { + ...state, + gitStatuses: action.statuses, + workspace: state.workspace + ? withGitStatuses(state.workspace, action.statuses, false) + : null, + }; case "select_project": return { ...state, selectedProjectId: action.projectId }; case "select_terminal": @@ -81,7 +122,11 @@ export function appReducer(state: AppState, action: AppAction): AppState { case "set_sidebar_open": return { ...state, sidebarOpen: action.open }; case "set_ws_status": - return { ...state, wsStatus: action.status }; + return { + ...state, + wsStatus: action.status, + gitStatuses: action.status === state.wsStatus ? state.gitStatuses : null, + }; case "set_stream_mappings": return { ...state, streamMappings: { ...state.streamMappings, ...action.mappings } }; case "clear_stream_mappings": @@ -101,6 +146,7 @@ export const initialState: AppState = { wsStatus: "disconnected", streamMappings: {}, claudePanelTerminalId: null, + gitStatuses: null, }; // ── Context ─────────────────────────────────────────────────────────────────