From d048e90404c255aee816cdb77ab98d4ce6b56747 Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Tue, 10 Nov 2026 10:58:52 +0800 Subject: [PATCH 1/2] fix(remote-connect): keep browser controllers subscribed across presence changes Mobile web did not show a session created on the desktop until the page was reloaded. The host drops every stream-hint subscriber that is absent from each relay presence update, but the relay only reports machine-scoped sockets. Browser controllers connect user-scoped and never appear, so any unrelated presence change (another device connecting, the phone reconnecting after the screen locks) ended their catalog and session subscriptions. Later catalog changes sent them no hint until the 4-minute renewal, a return to the tab, or a reload. Presence now decides only for devices it has reported at least once: a device that was online and is now absent is dropped as before. Other subscribers are governed by the existing 10-minute read lease, which mobile web renews every 4 minutes. No relay or controller change is required. --- .../src/remote_connect/host_stream.rs | 59 +++++++++++++++++-- 1 file changed, 53 insertions(+), 6 deletions(-) diff --git a/src/crates/services/services-integrations/src/remote_connect/host_stream.rs b/src/crates/services/services-integrations/src/remote_connect/host_stream.rs index 0a8fe6548d..962af0e8f7 100644 --- a/src/crates/services/services-integrations/src/remote_connect/host_stream.rs +++ b/src/crates/services/services-integrations/src/remote_connect/host_stream.rs @@ -413,6 +413,9 @@ fn estimate_bytes(value: &Value) -> usize { struct HubState { streams: HashMap, dirty: HashSet, + /// Devices the relay has reported in presence at least once. Only these + /// take part in presence; other controllers are governed by their lease. + presence_participants: HashSet, } /// Account-scoped owner of every host stream while account routing is alive. @@ -431,6 +434,7 @@ impl HostStreamHub { let state = Arc::new(std::sync::Mutex::new(HubState { streams: HashMap::new(), dirty: HashSet::new(), + presence_participants: HashSet::new(), })); let wake = Arc::new(tokio::sync::Notify::new()); let hub = Arc::new(Self { @@ -792,15 +796,28 @@ impl HostStreamHub { } } - /// Presence is authoritative for who can still receive hints. A device that - /// went offline is dropped now; it resubscribes on its next read. + /// Presence is authoritative for who can still receive hints, but only for + /// devices that take part in it. The relay reports machine-scoped sockets + /// (Desktop, CLI); browser controllers such as mobile web connect + /// user-scoped and never appear, so dropping every absent device would end + /// their subscriptions on any unrelated presence change and silence their + /// hints until the next renewal. A device that was reported online and is + /// now absent is dropped; it resubscribes on its next read. Others keep + /// their subscription until the read lease expires. pub fn retain_online(&self, online_device_ids: &[String]) { - let mut state = self.lock(); + let mut guard = self.lock(); + let state = &mut *guard; + state + .presence_participants + .extend(online_device_ids.iter().cloned()); + let participants = &state.presence_participants; + let went_offline = |device: &str| { + participants.contains(device) && !online_device_ids.iter().any(|id| id == device) + }; for log in state.streams.values_mut() { - log.subscribers - .retain(|device, _| online_device_ids.iter().any(|id| id == device)); + log.subscribers.retain(|device, _| !went_offline(device)); log.turn_snapshots - .retain(|_, snapshot| online_device_ids.contains(&snapshot.device)); + .retain(|_, snapshot| !went_offline(&snapshot.device)); } } @@ -1463,6 +1480,36 @@ mod tests { hub.retain_online(&[]); } + #[tokio::test] + async fn presence_drops_only_devices_that_take_part_in_it() { + let recorder = Arc::new(Recorder(Default::default())); + let hub = HostStreamHub::start(recorder.clone()); + hub.read("peer-desktop", &request(HOST_CATALOG_ID)).unwrap(); + // Browser controllers connect user-scoped and are never in presence. + hub.read("mobile-web", &request(HOST_CATALOG_ID)).unwrap(); + hub.retain_online(&["peer-desktop".into(), "host".into()]); + assert_eq!(hub.subscriber_count(HOST_CATALOG_ID), 2); + // An unrelated presence change keeps the browser subscribed and drops + // only the device that was reported online and is now absent. + hub.retain_online(&["host".into()]); + assert_eq!(hub.subscriber_count(HOST_CATALOG_ID), 1); + hub.append( + HOST_CATALOG_ID.into(), + "host-catalog-changed".into(), + serde_json::json!({"sessionsRevision":1}), + ) + .await + .unwrap(); + tokio::time::sleep(Duration::from_millis(150)).await; + let hints = recorder.0.lock().unwrap().clone(); + assert_eq!(hints.len(), 1); + assert_eq!(hints[0].0, "mobile-web"); + assert_eq!(hints[0].1["stream_id"], HOST_CATALOG_ID); + // An empty presence still keeps lease-governed controllers. + hub.retain_online(&[]); + assert_eq!(hub.subscriber_count(HOST_CATALOG_ID), 1); + } + #[tokio::test] async fn records_are_compacted_by_identity_and_paged_both_ways() { let hub = HostStreamHub::start(Arc::new(Recorder(Default::default()))); From ae47c8d017dc6aa7dd685866517c36d973f9362d Mon Sep 17 00:00:00 2001 From: Bob Lee Date: Tue, 10 Nov 2026 11:23:20 +0800 Subject: [PATCH 2/2] docs(remote-connect): document the host stream hint lease and presence rule Name the affected clients as hidden controllers (mobile web, KMP, HarmonyOS), which all connect user-scoped, and record in the Peer Device Mode host stream section how long a hint lease lasts, how controllers renew it, and when presence may end it early. --- docs/architecture/peer-device-mode.md | 7 ++++++- .../src/remote_connect/host_stream.rs | 6 +++--- 2 files changed, 9 insertions(+), 4 deletions(-) diff --git a/docs/architecture/peer-device-mode.md b/docs/architecture/peer-device-mode.md index 28fa6c7eaa..201a38a262 100644 --- a/docs/architecture/peer-device-mode.md +++ b/docs/architecture/peer-device-mode.md @@ -105,7 +105,12 @@ Permission and other control events remain separate from transcript content. The relay stores none of this. A controller opens a stream with the `read_stream` device RPC (`{stream_id, after|before, epoch, subscribe}`), which answers a `stream_page` with `epoch`, `cursor`, `events`, `has_more` and -`oldest_seq`; `unsubscribe_stream` releases the hint lease. While subscribed, +`oldest_seq`; `unsubscribe_stream` releases the hint lease. A subscribing read +holds the lease for 10 minutes and controllers renew it with a keepalive read +every 4 minutes. Account presence ends a lease early only for a device that +presence has reported and that is now absent. Hidden (user-scoped) controllers +such as mobile web, KMP and HarmonyOS never appear in presence, so they keep +their lease until they unsubscribe or stop renewing. While subscribed, the host fans out an encrypted `host-stream-changed` device event naming only the stream id, epoch and newest sequence; controllers treat it as a nudge and read the missing range themselves. The catalog is the `@host/catalog` stream, diff --git a/src/crates/services/services-integrations/src/remote_connect/host_stream.rs b/src/crates/services/services-integrations/src/remote_connect/host_stream.rs index 962af0e8f7..0c2852c36d 100644 --- a/src/crates/services/services-integrations/src/remote_connect/host_stream.rs +++ b/src/crates/services/services-integrations/src/remote_connect/host_stream.rs @@ -798,12 +798,12 @@ impl HostStreamHub { /// Presence is authoritative for who can still receive hints, but only for /// devices that take part in it. The relay reports machine-scoped sockets - /// (Desktop, CLI); browser controllers such as mobile web connect + /// (Desktop, CLI); hidden controllers (mobile web, KMP, HarmonyOS) connect /// user-scoped and never appear, so dropping every absent device would end /// their subscriptions on any unrelated presence change and silence their /// hints until the next renewal. A device that was reported online and is /// now absent is dropped; it resubscribes on its next read. Others keep - /// their subscription until the read lease expires. + /// their subscription until they unsubscribe or the read lease expires. pub fn retain_online(&self, online_device_ids: &[String]) { let mut guard = self.lock(); let state = &mut *guard; @@ -1485,7 +1485,7 @@ mod tests { let recorder = Arc::new(Recorder(Default::default())); let hub = HostStreamHub::start(recorder.clone()); hub.read("peer-desktop", &request(HOST_CATALOG_ID)).unwrap(); - // Browser controllers connect user-scoped and are never in presence. + // Hidden controllers connect user-scoped and are never in presence. hub.read("mobile-web", &request(HOST_CATALOG_ID)).unwrap(); hub.retain_online(&["peer-desktop".into(), "host".into()]); assert_eq!(hub.subscriber_count(HOST_CATALOG_ID), 2);