From 2c2cb748802dbf646de7b62afa7531226ed7299b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Fri, 13 Nov 2026 17:33:21 +0100 Subject: [PATCH 01/12] perf(remote): gzip the JSON api routes A full /v1/state runs to hundreds of KB and remote clients refetch it on every state change; JSON compresses about 7x. Compression is per route so downloads and the WebSocket upgrade keep their raw bodies. The transport client accepts gzip except over the local Unix socket. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + crates/okena-remote-server/Cargo.toml | 3 + crates/okena-remote-server/src/routes/mod.rs | 68 +++++++++++++++++-- crates/okena-transport/Cargo.toml | 2 +- .../okena-transport/src/client/connection.rs | 2 + crates/okena-transport/src/remote_http.rs | 4 ++ 6 files changed, 73 insertions(+), 7 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 488f4e1cc..a99b629a7 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", ] 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/routes/mod.rs b/crates/okena-remote-server/src/routes/mod.rs index b0706f55b..8749f7260 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])); diff --git a/crates/okena-transport/Cargo.toml b/crates/okena-transport/Cargo.toml index d7c1e6f8f..c97a3cdc7 100644 --- a/crates/okena-transport/Cargo.toml +++ b/crates/okena-transport/Cargo.toml @@ -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 } diff --git a/crates/okena-transport/src/client/connection.rs b/crates/okena-transport/src/client/connection.rs index e8fb5d926..f0b1d02d4 100644 --- a/crates/okena-transport/src/client/connection.rs +++ b/crates/okena-transport/src/client/connection.rs @@ -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}")) } 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() From 6247ba7c7daa8188e6c3ec8f158e9aa22c29d903 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Fri, 13 Nov 2026 17:33:21 +0100 Subject: [PATCH 02/12] perf(mobile): stream only the terminal on screen The phone subscribed to the PTY output of every terminal, also on the dashboard and in the background, while it shows at most one. The transport gains a stream filter that survives reconnects; the mobile app starts with nothing streamed and streams the selected terminal only. Connection liveness now counts any inbound frame instead of terminal output. Co-Authored-By: Claude Opus 5.5 --- crates/okena-mobile-ffi/src/client/handler.rs | 6 +- crates/okena-mobile-ffi/src/client/manager.rs | 16 + crates/okena-mobile-ffi/src/lib.rs | 9 + .../okena-transport/src/client/connection.rs | 288 +++++++++++++++--- docs/reference/mobile.md | 19 ++ mobile/rn/src/App.tsx | 8 +- mobile/rn/src/native/okena.ts | 18 +- mobile/rn/src/state/workspaceStore.test.ts | 72 +++++ mobile/rn/src/state/workspaceStore.ts | 45 +++ 9 files changed, 432 insertions(+), 49 deletions(-) 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..52b79a669 100644 --- a/crates/okena-mobile-ffi/src/client/manager.rs +++ b/crates/okena-mobile-ffi/src/client/manager.rs @@ -507,6 +507,10 @@ 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())); // Spawn event processor task let conn_id_clone = conn_id.clone(); @@ -613,6 +617,18 @@ 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)); + } + } + /// Pair with the remote server using a pairing code. pub fn pair(&self, conn_id: &str, code: &str) { let connections = self.connections.read(); diff --git a/crates/okena-mobile-ffi/src/lib.rs b/crates/okena-mobile-ffi/src/lib.rs index c22126d6e..67c7bce0e 100644 --- a/crates/okena-mobile-ffi/src/lib.rs +++ b/crates/okena-mobile-ffi/src/lib.rs @@ -80,6 +80,15 @@ 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); +} + /// 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-transport/src/client/connection.rs b/crates/okena-transport/src/client/connection.rs index f0b1d02d4..2b8b61acd 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}; @@ -373,6 +373,11 @@ 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>>>, } impl SessionControl { @@ -381,6 +386,7 @@ 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), } } @@ -395,6 +401,95 @@ 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) +} + +/// 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( + subscribed: &mut HashSet, + wanted: &[String], + ws_tx: &async_channel::Sender, + stream_map: &std::sync::RwLock>, + reverse_stream_map: &mut HashMap, +) { + let (subscribe, unsubscribe) = subscription_changes(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. + for id in &unsubscribe { + subscribed.remove(id); + } + 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() { + subscribed.extend(subscribe.iter().cloned()); + 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. @@ -565,6 +660,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. @@ -666,6 +765,16 @@ 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())); + } + /// Declare which of this connection's projects the client currently renders. /// /// Full replacement set of unprefixed ids. Stored for reconnects and pushed @@ -1601,11 +1710,15 @@ impl RemoteClient { 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: HashSet = initial_streams.iter().cloned().collect(); + if !initial_streams.is_empty() { let subscribe_msg = serde_json::json!({ "type": "subscribe", - "terminal_ids": terminal_ids, + "terminal_ids": initial_streams, }); futures::SinkExt::send( &mut ws_write, @@ -1814,6 +1927,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!( @@ -1834,6 +1960,7 @@ impl RemoteClient { }; if matches!(frame, Some(Ok(_))) { liveness.on_inbound(Instant::now()); + handler_clone.on_inbound(); } match frame { Some(Ok(tungstenite::Message::Binary(data))) => { @@ -1868,11 +1995,17 @@ impl RemoteClient { match msg_type { "subscribed" => { if let Some(mappings) = value.get("mappings") - && let Ok(map) = + && let Ok(mut map) = serde_json::from_value::>( mappings.clone(), ) { + // A reply that crossed an Unsubscribe: the server has + // dropped those streams already, so don't route to + // or send input on their ids. + map.retain(|terminal_id, _| { + subscribed.contains(terminal_id) + }); log::info!("Subscribed to {} terminal streams", map.len()); for (terminal_id, stream_id) in &map { reverse_stream_map @@ -1962,44 +2095,20 @@ impl RemoteClient { 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 - ); - } + // 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(); @@ -2420,6 +2529,99 @@ 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 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![]) + ); + } + + #[tokio::test] + async fn reconcile_forgets_stream_ids_of_stopped_terminals() { + let (tx, rx) = async_channel::unbounded(); + let mut subscribed: HashSet = ["t1", "t2"].map(String::from).into(); + 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, ["t2", "t3"].map(String::from).into()); + 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, diff --git a/docs/reference/mobile.md b/docs/reference/mobile.md index 39977d9e8..15b66b554 100644 --- a/docs/reference/mobile.md +++ b/docs/reference/mobile.md @@ -103,6 +103,25 @@ 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: bell, OSC 9/777 notification and the output-based "running" signal come only from +the streamed terminal. Agent status (`terminalAgentStatus`) and Claude dialogs come with the state +and cover every terminal. The connection dot's liveness counts any inbound frame (pong, the 2 s +system stats), not terminal output. + +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 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/native/okena.ts b/mobile/rn/src/native/okena.ts index 1308a1121..44ca9481f 100644 --- a/mobile/rn/src/native/okena.ts +++ b/mobile/rn/src/native/okena.ts @@ -561,14 +561,22 @@ 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. Bell, + * notification and output time only move for streamed terminals (see + * {@link setStreamedTerminals}). 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; @@ -938,6 +946,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: diff --git a/mobile/rn/src/state/workspaceStore.test.ts b/mobile/rn/src/state/workspaceStore.test.ts index 4ed714274..d08c3a0ad 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,74 @@ 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, + }); + }); + + afterEach(() => { + useWorkspaceStore.getState().stop(); + configureWorkspaceStore({ isForeground: 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('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..63778b136 100644 --- a/mobile/rn/src/state/workspaceStore.ts +++ b/mobile/rn/src/state/workspaceStore.ts @@ -16,6 +16,11 @@ * - 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. Bell, + * notification and output-based "running" signals therefore only come from + * the terminal on screen; agent status and Claude dialogs 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 @@ -99,6 +104,7 @@ export type WorkspaceNative = Pick< | 'secondsSinceActivity' | 'getTerminalActivity' | 'markTerminalSeen' + | 'setStreamedTerminals' | 'focusTerminal' | 'getUsage' | 'getUsageRevision' @@ -207,6 +213,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; /** @@ -250,6 +262,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 +297,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 +484,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().isForeground() + ? [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'], @@ -639,6 +675,7 @@ function pollState( // threshold crossed (cheap, avoids a stale value). set({ secondsSinceActivity: newActivity }); } + syncStreams(get); } /** @@ -781,6 +818,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 +842,7 @@ export const useWorkspaceStore: UseBoundStore> = ? signalsPatch(get, get().projects, terminalId, connId) : {}), }); + syncStreams(get); }, setTerminalView(terminalId, view) { @@ -816,6 +855,7 @@ export const useWorkspaceStore: UseBoundStore> = showDashboard() { if (get().dashboardOpen) return; set({ dashboardOpen: true }); + syncStreams(get); }, showWorkspace(projectId, terminalId) { @@ -829,6 +869,11 @@ export const useWorkspaceStore: UseBoundStore> = set( signalsPatch(get, get().projects, get().selectedTerminalId, connId), ); + syncStreams(get); + }, + + syncStreams() { + syncStreams(get); }, clearRestoreProjectMissing() { From e42fd5fe555dffdd938e628e350e076827bb43e7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 12:11:05 +0100 Subject: [PATCH 03/12] feat(transport): state sync policy and traffic counters A session refetches the whole /v1/state on every state_changed. Clients can now set a minimum interval between fetches (changes inside it are coalesced), defer fetches entirely (an app in the background) and skip the settings fetch. The default keeps the old behaviour. The client also counts received payload bytes by kind. The state is fetched with transparent decompression off and decoded by hand, so its count is the size on the wire. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 1 + crates/okena-transport/Cargo.toml | 4 +- .../okena-transport/src/client/connection.rs | 522 +++++++++++++----- crates/okena-transport/src/client/mod.rs | 2 +- crates/okena-transport/src/tls.rs | 25 +- 5 files changed, 416 insertions(+), 138 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a99b629a7..2e393e412 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6792,6 +6792,7 @@ name = "okena-transport" version = "0.1.0" dependencies = [ "async-channel", + "flate2", "futures", "log", "okena-core", diff --git a/crates/okena-transport/Cargo.toml b/crates/okena-transport/Cargo.toml index c97a3cdc7..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. @@ -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 2b8b61acd..fca95f5a2 100644 --- a/crates/okena-transport/src/client/connection.rs +++ b/crates/okena-transport/src/client/connection.rs @@ -378,6 +378,184 @@ struct SessionControl { /// 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, +} + +/// 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 { @@ -387,6 +565,8 @@ impl SessionControl { 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()), } } @@ -775,6 +955,31 @@ impl RemoteClient { .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 + }); + } + + /// 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 @@ -1637,42 +1842,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. @@ -1698,16 +1902,19 @@ 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 the terminal streams the client wants (all of @@ -1895,8 +2102,85 @@ 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, + _ = 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 => { @@ -1965,9 +2249,14 @@ impl RemoteClient { 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 => { @@ -1988,6 +2277,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) => { @@ -2050,97 +2340,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); - } - - // 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; - } - } - 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 @@ -2558,6 +2760,68 @@ mod tests { .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")]); 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/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 From 7c5223ed426455b3518be5db7c4bf54f21f7b1d0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 12:11:05 +0100 Subject: [PATCH 04/12] perf(mobile): coalesce state fetches and show data usage The phone fetches the workspace state at most once a second, not at all in the background (also while the Android connection service keeps the session open) and no longer fetches settings it never used. Settings gains a Data section with what the connection has downloaded, by kind. Co-Authored-By: Claude Opus 5.5 --- crates/okena-mobile-ffi/src/client/manager.rs | 38 +++++- crates/okena-mobile-ffi/src/lib.rs | 18 ++- crates/okena-mobile-ffi/src/types.rs | 31 +++++ docs/reference/mobile.md | 15 +++ mobile/rn/src/components/DataUsage.test.ts | 32 +++++ mobile/rn/src/components/DataUsage.tsx | 122 ++++++++++++++++++ mobile/rn/src/native/okena.ts | 41 ++++++ mobile/rn/src/screens/SettingsSheet.tsx | 37 +++++- mobile/rn/src/state/connectionStore.test.ts | 16 +++ mobile/rn/src/state/connectionStore.ts | 5 + mobile/rn/src/state/sessionRestore.test.ts | 1 + 11 files changed, 349 insertions(+), 7 deletions(-) create mode 100644 mobile/rn/src/components/DataUsage.test.ts create mode 100644 mobile/rn/src/components/DataUsage.tsx diff --git a/crates/okena-mobile-ffi/src/client/manager.rs b/crates/okena-mobile-ffi/src/client/manager.rs index 52b79a669..0617b7e58 100644 --- a/crates/okena-mobile-ffi/src/client/manager.rs +++ b/crates/okena-mobile-ffi/src/client/manager.rs @@ -5,8 +5,8 @@ use okena_core::api::{ ActionRequest, ApiFullscreen, ApiLayoutNode, 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 +18,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, @@ -363,6 +370,7 @@ impl ConnectionManager { ConnectionManager { runtime: Arc::new(runtime), connections: RwLock::new(HashMap::new()), + in_background: std::sync::atomic::AtomicBool::new(false), } }); } @@ -511,6 +519,13 @@ impl ConnectionManager { // 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, + }); // Spawn event processor task let conn_id_clone = conn_id.clone(); @@ -629,6 +644,25 @@ impl ConnectionManager { } } + /// 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(); diff --git a/crates/okena-mobile-ffi/src/lib.rs b/crates/okena-mobile-ffi/src/lib.rs index 67c7bce0e..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, @@ -89,6 +89,22 @@ 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/docs/reference/mobile.md b/docs/reference/mobile.md index 15b66b554..3ae547111 100644 --- a/docs/reference/mobile.md +++ b/docs/reference/mobile.md @@ -122,6 +122,21 @@ The JSON routes (`/v1/state`, `/v1/actions`, usage, Claude session/transcript) a 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/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 44ca9481f..37c214ff9 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 @@ -895,6 +905,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; }; @@ -987,6 +1020,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..e3197c984 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,21 @@ 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); + expect(setAppInBackground).toHaveBeenCalledTimes(2); + }); + 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..a0023c2f4 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,10 +494,14 @@ 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; } if (next !== 'active' || previous !== 'background') return; + native.setAppInBackground(false); native.onAppForeground(); const { connId } = get(); if (!connId) return; 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, }; From da803106a98f1d946eb7e2a523a70bc8c4d8cee7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 12:48:44 +0100 Subject: [PATCH 05/12] perf(daemon): stop bumping state_version on git status changes Git status already reaches clients as its own git_status_changed push, which the desktop and mobile clients patch into their copy of the state. The extra state_version bump made every client refetch the whole state on each git change. The web client now applies git_status_changed too. Co-Authored-By: Claude Opus 5.5 --- crates/okena-daemon-core/src/daemon.rs | 1 - crates/okena-daemon-core/src/git_poll.rs | 29 +++++++----------------- web/src/App.tsx | 3 +++ web/src/state/store.ts | 16 ++++++++++++- 4 files changed, 26 insertions(+), 23 deletions(-) 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/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..3ec14bd2f 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 ─────────────────────────────────────────────────────── @@ -61,6 +61,7 @@ export interface AppState { 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 } @@ -74,6 +75,19 @@ export function appReducer(state: AppState, action: AppAction): AppState { switch (action.type) { case "set_workspace": return { ...state, workspace: action.workspace }; + case "set_git_statuses": + // Git status changes arrive as their own message, not as a state change. + if (!state.workspace) return state; + return { + ...state, + workspace: { + ...state.workspace, + projects: state.workspace.projects.map((p) => ({ + ...p, + git_status: action.statuses[p.id] ?? null, + })), + }, + }; case "select_project": return { ...state, selectedProjectId: action.projectId }; case "select_terminal": From a29ade05c06a923cec4d3010b39a8d0d4dcc25d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 12:49:03 +0100 Subject: [PATCH 06/12] perf(remote): let clients opt out of system stats pushes The server pushes system_stats_changed every 2 s to every client. A client that shows no host metrics can now send set_system_stats with enabled: false; the mobile app does. Clients keeping the default never send the message, so older servers are unaffected. Co-Authored-By: Claude Opus 5.5 --- crates/okena-core/src/ws.rs | 6 +++ crates/okena-mobile-ffi/src/client/manager.rs | 2 + crates/okena-remote-server/src/CLAUDE.md | 4 ++ crates/okena-remote-server/src/routes/mod.rs | 50 +++++++++++++++++++ .../okena-remote-server/src/routes/stream.rs | 6 ++- .../okena-transport/src/client/connection.rs | 38 ++++++++++++++ crates/okena-transport/src/client/types.rs | 2 + 7 files changed, 107 insertions(+), 1 deletion(-) 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-mobile-ffi/src/client/manager.rs b/crates/okena-mobile-ffi/src/client/manager.rs index 0617b7e58..1ac14a707 100644 --- a/crates/okena-mobile-ffi/src/client/manager.rs +++ b/crates/okena-mobile-ffi/src/client/manager.rs @@ -526,6 +526,8 @@ impl ConnectionManager { 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(); 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 8749f7260..dbc100aae 100644 --- a/crates/okena-remote-server/src/routes/mod.rs +++ b/crates/okena-remote-server/src/routes/mod.rs @@ -1092,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-transport/src/client/connection.rs b/crates/okena-transport/src/client/connection.rs index fca95f5a2..9c7912b88 100644 --- a/crates/okena-transport/src/client/connection.rs +++ b/crates/okena-transport/src/client/connection.rs @@ -382,6 +382,9 @@ struct SessionControl { 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 @@ -567,6 +570,7 @@ impl SessionControl { 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), } } @@ -975,6 +979,16 @@ impl RemoteClient { }); } + /// 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() @@ -1973,6 +1987,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!( @@ -2075,6 +2102,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( @@ -2106,6 +2139,11 @@ impl RemoteClient { 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() => { 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, From 13ede1269a8e0b884efb3455bbcc22d8ebbb216b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 12:53:01 +0100 Subject: [PATCH 07/12] feat(remote): publish each terminal's last bell and notification Clients that stream only some terminals (the mobile app streams the one on screen) could no longer see bells and OSC 9/777 notifications of the rest. The daemon now stamps both edges on the terminal and sends them in StateResponse.terminal_attention. The mobile app raises a terminal's unseen flag when its stamp moved between two states, so it never compares the server's clock with its own. Co-Authored-By: Claude Opus 5.5 # Conflicts: # crates/okena-app-core/src/remote_snapshot.rs # crates/okena-cli/src/resolve.rs # crates/okena-core/src/api.rs # crates/okena-daemon-core/src/command_loop.rs # crates/okena-mobile-ffi/src/client/manager.rs # crates/okena-transport/src/client/connection.rs # crates/okena-transport/src/client/state.rs # crates/okena-workspace/src/remote_apply.rs # crates/okena-workspace/src/state.rs --- crates/okena-app-core/src/remote_snapshot.rs | 1 + crates/okena-cli/src/resolve.rs | 1 + crates/okena-core/src/api.rs | 17 +++++ crates/okena-daemon-core/src/command_loop.rs | 16 +++++ crates/okena-daemon-core/src/pty_loop.rs | 15 ++-- crates/okena-mobile-ffi/src/client/manager.rs | 72 ++++++++++++++++++- .../src/client/terminal_holder.rs | 11 +++ crates/okena-terminal/src/terminal/meta.rs | 22 ++++++ crates/okena-terminal/src/terminal/mod.rs | 8 +++ .../src/terminal/tests/unread.rs | 14 ++++ .../okena-transport/src/client/connection.rs | 1 + crates/okena-transport/src/client/state.rs | 1 + crates/okena-workspace/src/remote_apply.rs | 1 + crates/okena-workspace/src/state.rs | 2 + docs/reference/mobile.md | 16 +++-- mobile/rn/src/native/okena.ts | 7 +- mobile/rn/src/state/workspaceStore.ts | 7 +- 17 files changed, 197 insertions(+), 15 deletions(-) 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-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/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/manager.rs b/crates/okena-mobile-ffi/src/client/manager.rs index 1ac14a707..5cba13dfa 100644 --- a/crates/okena-mobile-ffi/src/client/manager.rs +++ b/crates/okena-mobile-ffi/src/client/manager.rs @@ -2,7 +2,8 @@ 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, StateSyncPolicy, @@ -281,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) @@ -1018,6 +1042,20 @@ 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. + let raised = + new_attention(&previous.terminal_attention, &state.terminal_attention); + if !raised.is_empty() { + let terminals = conn.handler.terminals().read(); + for (terminal_id, bell, notification) in raised { + if let Some(holder) = + terminals.get(&make_prefixed_id(&conn_id, &terminal_id)) + { + holder.raise_attention(bell, notification); + } + } + } merge_state_presentation(&mut state, previous); } else { state.fullscreen_terminal = None; @@ -1062,6 +1100,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()), @@ -1112,6 +1181,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..8472be7e6 100644 --- a/crates/okena-mobile-ffi/src/client/terminal_holder.rs +++ b/crates/okena-mobile-ffi/src/client/terminal_holder.rs @@ -353,6 +353,17 @@ impl TerminalHolder { } } + /// A bell / notification the server reported for this terminal (it is not + /// streamed, so the phone did not see the bytes itself). + 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); 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/src/client/connection.rs b/crates/okena-transport/src/client/connection.rs index 9c7912b88..fb697a650 100644 --- a/crates/okena-transport/src/client/connection.rs +++ b/crates/okena-transport/src/client/connection.rs @@ -3414,6 +3414,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/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-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 3ae547111..eef36abda 100644 --- a/docs/reference/mobile.md +++ b/docs/reference/mobile.md @@ -113,10 +113,18 @@ the app is in the background or polling stopped. The filter lives in the transpo 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: bell, OSC 9/777 notification and the output-based "running" signal come only from -the streamed terminal. Agent status (`terminalAgentStatus`) and Claude dialogs come with the state -and cover every terminal. The connection dot's liveness counts any inbound frame (pong, the 2 s -system stats), not terminal output. +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. The connection dot's liveness counts any inbound frame (pong, +state events), not terminal output. + +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 diff --git a/mobile/rn/src/native/okena.ts b/mobile/rn/src/native/okena.ts index 37c214ff9..aed1d73b1 100644 --- a/mobile/rn/src/native/okena.ts +++ b/mobile/rn/src/native/okena.ts @@ -571,9 +571,10 @@ export interface OkenaNative { getProjectOrder(connId: ConnId): ProjectId[]; /** - * Activity signals of every terminal the connection mirrors. Bell, - * notification and output time only move for streamed terminals (see - * {@link setStreamedTerminals}). 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[]; diff --git a/mobile/rn/src/state/workspaceStore.ts b/mobile/rn/src/state/workspaceStore.ts index 63778b136..8b0d145dc 100644 --- a/mobile/rn/src/state/workspaceStore.ts +++ b/mobile/rn/src/state/workspaceStore.ts @@ -17,9 +17,10 @@ * `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. Bell, - * notification and output-based "running" signals therefore only come from - * the terminal on screen; agent status and Claude dialogs come with the state. + * 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 From e2ef53924adf9128a8d1edf19002b23a4b609909 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 14:18:57 +0100 Subject: [PATCH 08/12] fix(transport): map only the reply to a terminal's latest subscribe The server hands out a fresh stream id per Subscribe. When a terminal was subscribed, unsubscribed and subscribed again before the first reply arrived, that reply installed the dropped id and input sent on it was lost until the second reply. The session now counts outstanding Subscribes per terminal and applies only the reply to the latest one. Co-Authored-By: Claude Opus 5.5 --- .../okena-transport/src/client/connection.rs | 113 +++++++++++++++--- 1 file changed, 97 insertions(+), 16 deletions(-) diff --git a/crates/okena-transport/src/client/connection.rs b/crates/okena-transport/src/client/connection.rs index fb697a650..136f9255f 100644 --- a/crates/okena-transport/src/client/connection.rs +++ b/crates/okena-transport/src/client/connection.rs @@ -628,23 +628,68 @@ fn subscription_changes( (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( - subscribed: &mut HashSet, + 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(subscribed, wanted); + 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. - for id in &unsubscribe { - subscribed.remove(id); - } + subs.sent_unsubscribe(&unsubscribe); if let Ok(mut map) = stream_map.write() { for id in &unsubscribe { if let Some(stream_id) = map.remove(id) { @@ -663,7 +708,7 @@ async fn reconcile_streams( } } if !subscribe.is_empty() { - subscribed.extend(subscribe.iter().cloned()); + subs.sent_subscribe(&subscribe); let count = subscribe.len(); if let Err(e) = ws_tx .send(WsClientMessage::Subscribe { @@ -1935,8 +1980,9 @@ impl RemoteClient { // 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: HashSet = initial_streams.iter().cloned().collect(); + 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": initial_streams, @@ -2323,17 +2369,14 @@ impl RemoteClient { match msg_type { "subscribed" => { if let Some(mappings) = value.get("mappings") - && let Ok(mut map) = + && let Ok(map) = serde_json::from_value::>( mappings.clone(), ) { - // A reply that crossed an Unsubscribe: the server has - // dropped those streams already, so don't route to - // or send input on their ids. - map.retain(|terminal_id, _| { - subscribed.contains(terminal_id) - }); + // 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 @@ -2888,10 +2931,47 @@ mod tests { ); } + #[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: HashSet = ["t1", "t2"].map(String::from).into(); + 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), @@ -2907,7 +2987,8 @@ mod tests { ) .await; - assert_eq!(subscribed, ["t2", "t3"].map(String::from).into()); + 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)]) From 1406a5fc92c61ac073542da628716bae9c75a507 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 14:18:57 +0100 Subject: [PATCH 09/12] fix(web): keep the pushed git status over an older state response Git changes no longer trigger a state refetch, so a /v1/state response that was in flight when git_status_changed arrived could overwrite the newer git status until the next git change. The web client now keeps the last pushed map and applies it over every state response of the connection. Also document why the phone's dashboard git counts can lag. Co-Authored-By: Claude Opus 5.5 --- docs/reference/mobile.md | 7 ++++++ web/src/state/store.ts | 46 +++++++++++++++++++++++++++++++--------- 2 files changed, 43 insertions(+), 10 deletions(-) diff --git a/docs/reference/mobile.md b/docs/reference/mobile.md index eef36abda..018ef87c6 100644 --- a/docs/reference/mobile.md +++ b/docs/reference/mobile.md @@ -121,6 +121,13 @@ first state of a connection is the baseline). Agent status (`terminalAgentStatus dialogs also come with the state. The connection dot's liveness counts any inbound frame (pong, state events), not terminal output. +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 diff --git a/web/src/state/store.ts b/web/src/state/store.ts index 3ec14bd2f..1c182c76f 100644 --- a/web/src/state/store.ts +++ b/web/src/state/store.ts @@ -57,6 +57,26 @@ 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; +} + +function withGitStatuses( + workspace: StateResponse, + statuses: Record, +): StateResponse { + return { + ...workspace, + projects: workspace.projects.map((p) => ({ + ...p, + git_status: statuses[p.id] ?? null, + })), + }; } export type AppAction = @@ -74,19 +94,20 @@ 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) + : action.workspace, + }; case "set_git_statuses": // Git status changes arrive as their own message, not as a state change. - if (!state.workspace) return state; return { ...state, - workspace: { - ...state.workspace, - projects: state.workspace.projects.map((p) => ({ - ...p, - git_status: action.statuses[p.id] ?? null, - })), - }, + gitStatuses: action.statuses, + workspace: state.workspace + ? withGitStatuses(state.workspace, action.statuses) + : null, }; case "select_project": return { ...state, selectedProjectId: action.projectId }; @@ -95,7 +116,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": @@ -115,6 +140,7 @@ export const initialState: AppState = { wsStatus: "disconnected", streamMappings: {}, claudePanelTerminalId: null, + gitStatuses: null, }; // ── Context ───────────────────────────────────────────────────────────────── From 41a0df239f6e158ff194f442f02a43cbe0071e13 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 14:26:28 +0100 Subject: [PATCH 10/12] fix: review follow-ups for state deferral, attention and web git status MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - mobile: resume state fetches on every return to active; iOS can go background → inactive → active, which left the deferral on and froze the workspace state. - mobile-ffi: do not raise attention stamps for streamed terminals; their bells arrive in the output and were already marked seen. - web: a state response newer than the last git push keeps the status of projects the push does not list (added since). Co-Authored-By: Claude Opus 5.5 --- crates/okena-mobile-ffi/src/client/manager.rs | 10 ++++++++++ mobile/rn/src/state/connectionStore.test.ts | 12 +++++++++++- mobile/rn/src/state/connectionStore.ts | 4 +++- web/src/state/store.ts | 12 +++++++++--- 4 files changed, 33 insertions(+), 5 deletions(-) diff --git a/crates/okena-mobile-ffi/src/client/manager.rs b/crates/okena-mobile-ffi/src/client/manager.rs index 5cba13dfa..376c68cff 100644 --- a/crates/okena-mobile-ffi/src/client/manager.rs +++ b/crates/okena-mobile-ffi/src/client/manager.rs @@ -36,6 +36,10 @@ struct MobileConnection { /// Bumped on every stored usage change so pollers can skip unchanged /// snapshots cheaply (`usage_revision`). usage_revision: AtomicU32, + /// Terminals the UI streams (unprefixed ids). Their bells and + /// notifications arrive in the output itself and are marked seen there, so + /// the state's attention stamps must not raise them again. + streamed: RwLock>, _event_task: Option>, } @@ -564,6 +568,7 @@ impl ConnectionManager { status, state_cache, usage_revision: AtomicU32::new(0), + streamed: RwLock::new(HashSet::new()), _event_task: None, }; @@ -664,6 +669,7 @@ impl ConnectionManager { 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.streamed.write() = terminal_ids.iter().cloned().collect(); conn.client .read() .set_streamed_terminals(Some(terminal_ids)); @@ -1048,7 +1054,11 @@ impl ConnectionManager { new_attention(&previous.terminal_attention, &state.terminal_attention); if !raised.is_empty() { let terminals = conn.handler.terminals().read(); + let streamed = conn.streamed.read(); for (terminal_id, bell, notification) in raised { + if streamed.contains(&terminal_id) { + continue; + } if let Some(holder) = terminals.get(&make_prefixed_id(&conn_id, &terminal_id)) { diff --git a/mobile/rn/src/state/connectionStore.test.ts b/mobile/rn/src/state/connectionStore.test.ts index e3197c984..c0d6f5324 100644 --- a/mobile/rn/src/state/connectionStore.test.ts +++ b/mobile/rn/src/state/connectionStore.test.ts @@ -457,7 +457,17 @@ describe('reconnect handling', () => { expect(setAppInBackground).toHaveBeenLastCalledWith(true); store.onAppStateChange('active'); expect(setAppInBackground).toHaveBeenLastCalledWith(false); - expect(setAppInBackground).toHaveBeenCalledTimes(2); + }); + + 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', () => { diff --git a/mobile/rn/src/state/connectionStore.ts b/mobile/rn/src/state/connectionStore.ts index a0023c2f4..a6639e70c 100644 --- a/mobile/rn/src/state/connectionStore.ts +++ b/mobile/rn/src/state/connectionStore.ts @@ -500,8 +500,10 @@ export const useConnectionStore: UseBoundStore> = 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.setAppInBackground(false); native.onAppForeground(); const { connId } = get(); if (!connId) return; diff --git a/web/src/state/store.ts b/web/src/state/store.ts index 1c182c76f..6c1fa12e3 100644 --- a/web/src/state/store.ts +++ b/web/src/state/store.ts @@ -66,15 +66,21 @@ export interface AppState { 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] ?? null, + git_status: statuses[p.id] ?? (keepMissing ? p.git_status : null), })), }; } @@ -97,7 +103,7 @@ export function appReducer(state: AppState, action: AppAction): AppState { return { ...state, workspace: state.gitStatuses - ? withGitStatuses(action.workspace, state.gitStatuses) + ? withGitStatuses(action.workspace, state.gitStatuses, true) : action.workspace, }; case "set_git_statuses": @@ -106,7 +112,7 @@ export function appReducer(state: AppState, action: AppAction): AppState { ...state, gitStatuses: action.statuses, workspace: state.workspace - ? withGitStatuses(state.workspace, action.statuses) + ? withGitStatuses(state.workspace, action.statuses, false) : null, }; case "select_project": From 832205c13ba140d285f37e94f5ba57d92a2ee61f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 14:33:58 +0100 Subject: [PATCH 11/12] fix(mobile): review follow-ups for liveness, attention and inactive - The connection dot keyed on 3/10 s, but with one streamed terminal and no host metrics a quiet connection only receives the pong to the 15 s ping, so a healthy connection cycled to red. Thresholds are now 20/45 s. - A bell seen on screen right before switching away came back as unseen from the next state's stamp. The holder now records bells and notifications from its own output since the last state, and the stamp skips those, replacing the check of the current stream set. - iOS `inactive` (Control Center, alerts, Face ID) counts as foreground, so the terminal no longer resubscribes with a fresh snapshot. - Document that off-screen bells need a daemon that sends the stamps. Co-Authored-By: Claude Opus 5.5 --- crates/okena-mobile-ffi/src/client/manager.rs | 37 +++++++------- .../src/client/terminal_holder.rs | 51 +++++++++++++++++-- docs/reference/mobile.md | 8 ++- mobile/rn/src/components/ConnectionDot.tsx | 9 ++-- mobile/rn/src/state/workspaceStore.ts | 21 ++++++-- 5 files changed, 94 insertions(+), 32 deletions(-) diff --git a/crates/okena-mobile-ffi/src/client/manager.rs b/crates/okena-mobile-ffi/src/client/manager.rs index 376c68cff..e39384869 100644 --- a/crates/okena-mobile-ffi/src/client/manager.rs +++ b/crates/okena-mobile-ffi/src/client/manager.rs @@ -36,10 +36,6 @@ struct MobileConnection { /// Bumped on every stored usage change so pollers can skip unchanged /// snapshots cheaply (`usage_revision`). usage_revision: AtomicU32, - /// Terminals the UI streams (unprefixed ids). Their bells and - /// notifications arrive in the output itself and are marked seen there, so - /// the state's attention stamps must not raise them again. - streamed: RwLock>, _event_task: Option>, } @@ -568,7 +564,6 @@ impl ConnectionManager { status, state_cache, usage_revision: AtomicU32::new(0), - streamed: RwLock::new(HashSet::new()), _event_task: None, }; @@ -669,7 +664,6 @@ impl ConnectionManager { 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.streamed.write() = terminal_ids.iter().cloned().collect(); conn.client .read() .set_streamed_terminals(Some(terminal_ids)); @@ -1049,23 +1043,28 @@ impl ConnectionManager { 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. + // 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); - if !raised.is_empty() { - let terminals = conn.handler.terminals().read(); - let streamed = conn.streamed.read(); - for (terminal_id, bell, notification) in raised { - if streamed.contains(&terminal_id) { - continue; - } - if let Some(holder) = - terminals.get(&make_prefixed_id(&conn_id, &terminal_id)) - { - holder.raise_attention(bell, notification); - } + 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; diff --git a/crates/okena-mobile-ffi/src/client/terminal_holder.rs b/crates/okena-mobile-ffi/src/client/terminal_holder.rs index 8472be7e6..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,8 +372,20 @@ impl TerminalHolder { } } - /// A bell / notification the server reported for this terminal (it is not - /// streamed, so the phone did not see the bytes itself). + /// 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); @@ -799,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/docs/reference/mobile.md b/docs/reference/mobile.md index 018ef87c6..95265cd8a 100644 --- a/docs/reference/mobile.md +++ b/docs/reference/mobile.md @@ -118,8 +118,12 @@ OSC 9/777 notifications of the others come with the state: the daemon stamps eac 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. The connection dot's liveness counts any inbound frame (pong, -state events), not terminal output. +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 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/state/workspaceStore.ts b/mobile/rn/src/state/workspaceStore.ts index 8b0d145dc..f97b133b5 100644 --- a/mobile/rn/src/state/workspaceStore.ts +++ b/mobile/rn/src/state/workspaceStore.ts @@ -75,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; @@ -238,7 +247,11 @@ function deps(): WorkspaceDeps { resolvedDeps = { native: injectedDeps.native ?? getOkenaNative(), isForeground: - injectedDeps.isForeground ?? (() => AppState.currentState === 'active'), + // `inactive` (Control Center, a system alert, Face ID) is a transient + // overlay over the app: still foreground, so the terminal keeps + // streaming instead of resubscribing with a fresh snapshot. + injectedDeps.isForeground ?? + (() => AppState.currentState !== 'background'), }; } return resolvedDeps; @@ -659,12 +672,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; } From 72b776ddddcdb720c133e9685923424994abfc8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Honza=20Sla=CC=81dek?= Date: Sat, 14 Nov 2026 14:40:31 +0100 Subject: [PATCH 12/12] fix(mobile): count only active as seen, stream until background The previous change made iOS `inactive` count as foreground so the terminal keeps streaming under Control Center or Face ID, but the same check marks the terminal on screen seen. A bell or a finished agent during Notification Center or the app switcher was acknowledged unseen. Streaming now stops only in the background; seen still needs `active`. Co-Authored-By: Claude Opus 5.5 --- mobile/rn/src/state/workspaceStore.test.ts | 18 +++++++++++++++++- mobile/rn/src/state/workspaceStore.ts | 21 +++++++++++++-------- 2 files changed, 30 insertions(+), 9 deletions(-) diff --git a/mobile/rn/src/state/workspaceStore.test.ts b/mobile/rn/src/state/workspaceStore.test.ts index d08c3a0ad..b3a3c4b37 100644 --- a/mobile/rn/src/state/workspaceStore.test.ts +++ b/mobile/rn/src/state/workspaceStore.test.ts @@ -798,12 +798,13 @@ describe('terminal streaming', () => { }, }, isForeground: () => foreground, + isInBackground: () => !foreground, }); }); afterEach(() => { useWorkspaceStore.getState().stop(); - configureWorkspaceStore({ isForeground: undefined }); + configureWorkspaceStore({ isForeground: undefined, isInBackground: undefined }); jest.useRealTimers(); }); @@ -844,6 +845,21 @@ describe('terminal streaming', () => { 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(); diff --git a/mobile/rn/src/state/workspaceStore.ts b/mobile/rn/src/state/workspaceStore.ts index f97b133b5..fd765d981 100644 --- a/mobile/rn/src/state/workspaceStore.ts +++ b/mobile/rn/src/state/workspaceStore.ts @@ -125,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; } /** @@ -247,11 +253,10 @@ function deps(): WorkspaceDeps { resolvedDeps = { native: injectedDeps.native ?? getOkenaNative(), isForeground: - // `inactive` (Control Center, a system alert, Face ID) is a transient - // overlay over the app: still foreground, so the terminal keeps - // streaming instead of resubscribing with a fresh snapshot. - injectedDeps.isForeground ?? - (() => AppState.currentState !== 'background'), + injectedDeps.isForeground ?? (() => AppState.currentState === 'active'), + isInBackground: + injectedDeps.isInBackground ?? + (() => AppState.currentState === 'background'), }; } return resolvedDeps; @@ -507,7 +512,7 @@ function syncStreams(get: StoreApi['getState']): void { if (!connId) return; const { dashboardOpen, selectedTerminalId } = get(); const ids = - !dashboardOpen && selectedTerminalId !== null && deps().isForeground() + !dashboardOpen && selectedTerminalId !== null && !deps().isInBackground() ? [selectedTerminalId] : []; const key = ids.join('\n');