diff --git a/crates/core/src/settings.rs b/crates/core/src/settings.rs index 5bd961c0d..bbfb9549b 100644 --- a/crates/core/src/settings.rs +++ b/crates/core/src/settings.rs @@ -947,7 +947,8 @@ pub struct Settings { pub remote_host_name: Option, /// Per-session last-visited time (unix secs), keyed by session id. A session /// whose `updated_at` exceeds its last-visited time (and isn't active) shows - /// an unread dot. Opening a thread refreshes it; "Mark unread" clears it. + /// an unread dot. A client that has loaded a thread advances it to the + /// `updated_at` it showed; "Mark unread" sets it just below. /// UI state; absent in legacy files. #[serde(default, skip_serializing_if = "HashMap::is_empty")] pub last_visited: HashMap, diff --git a/crates/protocol/src/command.rs b/crates/protocol/src/command.rs index d8bd5fc8c..ac077c665 100644 --- a/crates/protocol/src/command.rs +++ b/crates/protocol/src/command.rs @@ -322,6 +322,14 @@ pub enum Command { MarkSessionUnread { session_id: String, }, + /// The client has loaded the thread's conversation and shown it as of + /// the thread's `updated_at` value `through`. A subscription alone never + /// marks a thread read: it may arrive long after the user gave up on a + /// view that never loaded. + MarkSessionRead { + session_id: String, + through: u64, + }, StartDraft { project_id: String, cwd: PathBuf, @@ -487,6 +495,7 @@ impl Command { | Self::DeleteSession { session_id, .. } | Self::MergeWorktree { session_id, .. } | Self::MarkSessionUnread { session_id, .. } + | Self::MarkSessionRead { session_id, .. } | Self::SetDraftWorkspace { session_id, .. } | Self::SendTurn { session_id, .. } | Self::ScheduleTurn { session_id, .. } diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 52f032c65..d846aeb03 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -52,7 +52,9 @@ pub use wire::{ // // Unreleased: provider plugin catalogs, their commands, challenges and switches; // provider update availability merged per check round, UpdateProvider replaced -// by UpdateProviders, and sequential update run status with toast progress. +// by UpdateProviders, and sequential update run status with toast progress; +// MarkSessionRead, the client's acknowledgement that a thread's conversation +// loaded, replaces marking a thread read when it is subscribed. pub const PROTOCOL_VERSION: u32 = 9; #[cfg(test)] diff --git a/crates/runtime/src/app/command_validation.rs b/crates/runtime/src/app/command_validation.rs index 2be2f726c..f0d955521 100644 --- a/crates/runtime/src/app/command_validation.rs +++ b/crates/runtime/src/app/command_validation.rs @@ -134,6 +134,7 @@ impl AppState { | Command::RegenerateSessionTitle { .. } | Command::DeleteSession { .. } | Command::MarkSessionUnread { .. } + | Command::MarkSessionRead { .. } | Command::ForkThread { .. } ) { return if self.sessions.iter().any(|meta| meta.id == session_id) diff --git a/crates/runtime/src/app/lifecycle.rs b/crates/runtime/src/app/lifecycle.rs index 5ed840c3c..b20267044 100644 --- a/crates/runtime/src/app/lifecycle.rs +++ b/crates/runtime/src/app/lifecycle.rs @@ -308,19 +308,6 @@ impl AppState { } pub(super) fn persist_meta(&mut self, meta: &SessionMeta, cx: &mut HostCx) { - // An update landing on the conversation the user is currently viewing - // is already read: advance the last-visited watermark alongside it so - // switching away later does not surface a stale unread dot. Threads the - // user is not viewing keep their watermark (and their dot), as does an - // explicit "mark unread" (which only rewrites the watermark). - if self.residents.live.contains_key(meta.id.as_str()) { - let visited = self.settings.last_visited.entry(meta.id.clone()); - let visited = visited.or_insert(meta.updated_at); - if *visited < meta.updated_at { - *visited = meta.updated_at; - self.persist_settings(cx); - } - } self.enqueue_store_write( StoreWrite::UpsertMeta { meta: Box::new(meta.clone()), diff --git a/crates/runtime/src/app/sessions.rs b/crates/runtime/src/app/sessions.rs index ac3d6bbca..37086315d 100644 --- a/crates/runtime/src/app/sessions.rs +++ b/crates/runtime/src/app/sessions.rs @@ -1086,11 +1086,20 @@ impl AppState { .is_some_and(ActiveSession::has_work) } - /// Record that a thread has been visited now (clears its unread dot). - pub(super) fn mark_visited(&mut self, session_id: &str, cx: &mut HostCx) { + /// Advance a thread's last-visited watermark to the `updated_at` a client + /// has shown. A late acknowledgement never rewinds a newer one. + pub fn mark_session_read(&mut self, session_id: &str, through: u64, cx: &mut HostCx) { + if self + .settings + .last_visited + .get(session_id) + .is_some_and(|&visited| visited >= through) + { + return; + } self.settings .last_visited - .insert(session_id.to_string(), now_secs()); + .insert(session_id.to_string(), through); self.persist_settings(cx); } @@ -1556,7 +1565,6 @@ impl AppState { let Some(meta) = self.find_meta(session_id) else { return; }; - self.mark_visited(session_id, cx); // A parked session is re-adopted, not replayed cold: its process, pump // and queue come back as they were, and the timeline is rebuilt from the diff --git a/crates/runtime/src/app/test_support.rs b/crates/runtime/src/app/test_support.rs index a05eda36c..5e3147068 100644 --- a/crates/runtime/src/app/test_support.rs +++ b/crates/runtime/src/app/test_support.rs @@ -194,6 +194,10 @@ pub(super) struct TestEntity(Rc>); impl TestEntity { pub(super) fn dispatch_command(&self, cx: &mut TestAppContext, id: u64, command: Command) { + self.deliver(cx, id, ClientPayload::Command(command)); + } + + pub(super) fn deliver(&self, cx: &mut TestAppContext, id: u64, payload: ClientPayload) { let mut host_cx = cx.host_cx(); crate::pipe::handle_client_message( &mut self.0.borrow_mut(), @@ -201,7 +205,7 @@ impl TestEntity { ClientMessage { key: None, id, - payload: ClientPayload::Command(command), + payload, }, ); } diff --git a/crates/runtime/src/app/tests.rs b/crates/runtime/src/app/tests.rs index b7aa76527..477a3c0d1 100644 --- a/crates/runtime/src/app/tests.rs +++ b/crates/runtime/src/app/tests.rs @@ -2746,52 +2746,56 @@ fn orchestrate_dispatch_access_maps_known_values() { ); } +/// A remote client on a bad network can queue a thread's subscription and its +/// unsubscription together and deliver both after it reconnects, although the +/// conversation never loaded on its screen. Only the client's acknowledgement +/// that it showed the thread marks it read. #[test] -fn updates_on_the_viewed_thread_do_not_mark_it_unread() { +fn only_a_loaded_conversation_marks_a_thread_read() { + use tcode_protocol::{ClientPayload, Subscription, Topic}; let cx = &mut TestAppContext::default(); - let test_store = TestStore::new("tcode-viewed-unread-test"); + let test_store = TestStore::new("tcode-read-on-load-test"); let store = (*test_store).clone(); - let mut first = SessionMeta::new(ProviderKind::Codex, PathBuf::from("/tmp/a"), None); - first.updated_at = 100; - let second = SessionMeta::new(ProviderKind::Codex, PathBuf::from("/tmp/b"), None); - store.upsert_meta(&first).unwrap(); - store.upsert_meta(&second).unwrap(); - let first_id = first.id.clone(); - let second_id = second.id.clone(); + let mut meta = SessionMeta::new(ProviderKind::Codex, PathBuf::from("/tmp/a"), None); + meta.updated_at = 100; + store.upsert_meta(&meta).unwrap(); + let id = meta.id.clone(); let state = cx.new_entity(TestClientState::new(store)); + state.update(cx, |state, _| { + state.settings.last_visited.insert(id.clone(), 50); + }); + let unread = |state: &TestEntity| state.read(|state| state.session_unread(&id)); + assert!(unread(&state), "updated after the last visit"); - state.update(cx, |state, cx| { - assert!(!state.session_unread(&first_id), "never visited"); - state.settings.last_visited.insert(first_id.clone(), 50); - assert!(state.session_unread(&first_id), "newer than the last visit"); - state.settings.last_visited.insert(first_id.clone(), 100); - assert!(!state.session_unread(&first_id), "equal to the last visit"); - state.select_session(&first_id, cx); - - // A turn finishes while the user is watching: updated_at moves past - // the watermark stamped on entry, and the meta is persisted. - let active = state.selected_session_mut().unwrap(); - active.meta.updated_at = now_secs() + 10; - let meta = active.meta.clone(); - state.persist_meta(&meta, cx); + let subscription = Subscription { + topic: Topic::SessionEvents { + session_id: id.clone(), + }, + after: None, + }; + state.deliver(cx, 1, ClientPayload::Subscribe(subscription.clone())); + state.deliver(cx, 2, ClientPayload::Unsubscribe(subscription)); + cx.run_until_parked(); + assert!(unread(&state), "a late subscription is not a read"); - // Switching away must not surface an unread dot for what the user - // already saw happen on screen. - state.select_session(&second_id, cx); - assert!(!state.session_unread(&first_id)); + let read = |through| Command::MarkSessionRead { + session_id: id.clone(), + through, + }; + state.dispatch_command(cx, 3, read(100)); + assert!(!unread(&state), "read through what the client showed"); + state.dispatch_command(cx, 4, read(80)); + assert!( + !unread(&state), + "a delayed older acknowledgement does not rewind" + ); - // But an update landing on a thread the user is NOT viewing still - // marks it unread. - let mut parked = state - .sessions - .iter() - .find(|m| m.id == first_id) - .cloned() - .unwrap(); - parked.updated_at = now_secs() + 20; + state.update(cx, |state, cx| { + let mut parked = state.sessions.iter().find(|m| m.id == id).cloned().unwrap(); + parked.updated_at = 120; state.persist_meta(&parked, cx); - assert!(state.session_unread(&first_id)); }); + assert!(unread(&state), "newer than anything the client showed"); } #[test] @@ -8272,7 +8276,14 @@ fn index_and_visit_changes_cross_the_wire_one_thread_at_a_time() { ); }); - state.update(cx, |state, cx| state.mark_visited("kept", cx)); + state.dispatch_command( + cx, + 2, + Command::MarkSessionRead { + session_id: "kept".into(), + through: 1, + }, + ); cx.run_until_parked(); let settings: Vec<_> = cx .drain_outgoing() diff --git a/crates/runtime/src/pipe.rs b/crates/runtime/src/pipe.rs index 34898d6ee..17a63ac6a 100644 --- a/crates/runtime/src/pipe.rs +++ b/crates/runtime/src/pipe.rs @@ -573,6 +573,10 @@ fn dispatch_command(app: &mut AppState, cx: &mut HostCx, command: Command) -> Co } => app.delete_session(&session_id, remove_worktree, cx), Command::DeleteProject { project_id } => app.delete_project(&project_id, cx), Command::MarkSessionUnread { session_id } => app.mark_session_unread(&session_id, cx), + Command::MarkSessionRead { + session_id, + through, + } => app.mark_session_read(&session_id, through, cx), Command::StartDraft { project_id, cwd } => { response = CommandResponse::SessionId(Some(app.start_draft(project_id, cwd, cx))); } diff --git a/crates/ui/src/shell.rs b/crates/ui/src/shell.rs index fc8c080c7..48a61ffcf 100644 --- a/crates/ui/src/shell.rs +++ b/crates/ui/src/shell.rs @@ -896,6 +896,17 @@ impl AppShell { }); self.mounted.truncate(common); self.mounted.extend(pushed); + let state = self.window_state.read(cx); + let on_screen = if state.compact { + state.destination() == Destination::Thread + } else { + state.route() == Route::Chat + }; + if let Some(attachment) = &self.attachment { + attachment.link.store.update(cx, |store, _| { + store.set_conversation_on_screen(on_screen); + }); + } } /// A software keyboard the IME dismissed on its own (its own hide or Back @@ -2795,6 +2806,43 @@ mod tests { await_restore_update(shell, cx, WorkspaceStore::baseline_ready); } + /// A phone that goes back to the thread list before the thread's + /// conversation arrives keeps the thread selected, so the conversation + /// still loads behind the list. That is not a read; showing the thread + /// page again is. + #[gpui::test] + fn a_conversation_that_loads_behind_the_thread_list_is_not_read(cx: &mut TestAppContext) { + let _locale_guard = crate::settings::TestLocaleGuard::acquire(); + let (shell, host, _, cx) = mount_restored(cx, &["hosts", "threads", "thread"], true); + let reads = |host: &MountedShell| { + sent(host) + .into_iter() + .filter_map(|payload| match payload { + ClientPayload::Command(Command::MarkSessionRead { session_id, .. }) => { + Some(session_id) + } + _ => None, + }) + .collect::>() + }; + host.states + .try_send(tcode_client::ConnectionState::Syncing { path: None }) + .unwrap(); + draw(cx); + assert!(cx.debug_bounds("compact-thread-page").is_some()); + assert!(cx.update(|window, cx| shell.update(cx, |shell, cx| shell.back(window, cx)))); + draw(cx); + + restore_index(&shell, &host, true, cx); + restore_status(&shell, &host, cx); + restore_settings_and_events(&shell, &host, cx); + assert_eq!(reads(&host), Vec::::new()); + + shell.update(cx, |shell, cx| shell.go(Destination::Thread, cx)); + draw(cx); + assert_eq!(reads(&host), ["thread-a"]); + } + /// The phone's direct path dying is reported by the transport as a new /// value of the state it is in naming the relay; that alone, with no /// host line and no touch, changes what the nav bar says about the diff --git a/crates/ui/src/store/intents.rs b/crates/ui/src/store/intents.rs index 2356cec5f..dbf9d3d15 100644 --- a/crates/ui/src/store/intents.rs +++ b/crates/ui/src/store/intents.rs @@ -264,6 +264,7 @@ impl WorkspaceStore { } self.session_status_replica = None; self.session_replica = None; + self.read_acknowledged = None; self.active_destination = None; self.git_status_replica = Default::default(); } diff --git a/crates/ui/src/store/mod.rs b/crates/ui/src/store/mod.rs index 6b924e3fb..9ed740bc9 100644 --- a/crates/ui/src/store/mod.rs +++ b/crates/ui/src/store/mod.rs @@ -167,6 +167,13 @@ pub struct WorkspaceStore { baseline_topics: HashSet, index_hydrated: bool, hydrated_sessions: HashSet, + /// The `updated_at` this view last reported read, so an acknowledgement + /// is sent once per change rather than once per event until the host's + /// visit echo arrives. + read_acknowledged: Option<(String, u64)>, + /// Whether the shell shows the selected thread's conversation. A compact + /// window keeps the thread selected on the thread list it returned to. + conversation_on_screen: bool, selected_session_id: Option, session_records: HashMap>, /// The log cursors `session_records` stands for. The host may merge @@ -347,6 +354,8 @@ impl WorkspaceStore { baseline_topics: HashSet::new(), index_hydrated: false, hydrated_sessions: HashSet::new(), + read_acknowledged: None, + conversation_on_screen: false, selected_session_id: None, session_records: HashMap::new(), session_from: HashMap::new(), @@ -1135,6 +1144,54 @@ impl WorkspaceStore { self.reconcile_destination(cx); self.removed_session = None; } + self.acknowledge_read(); + } + + pub(crate) fn set_conversation_on_screen(&mut self, on_screen: bool) { + self.conversation_on_screen = on_screen; + self.acknowledge_read(); + } + + /// Report the thread on screen read through its current `updated_at` + /// once its conversation has loaded. Only this marks a thread read: a + /// subscription can reach the host long after the user left a view that + /// never loaded. + fn acknowledge_read(&mut self) { + if !self.conversation_on_screen { + return; + } + let Some(session_id) = self + .session_replica + .as_ref() + .map(|(id, _)| id) + .filter(|id| self.selected_session_id.as_ref() == Some(id)) + .cloned() + else { + return; + }; + let Some(through) = self + .index_replica + .0 + .iter() + .find(|meta| meta.id == session_id) + .map(|meta| meta.updated_at) + else { + return; + }; + let acknowledged = self + .read_acknowledged + .as_ref() + .filter(|(id, _)| *id == session_id) + .map(|(_, at)| *at) + .max(self.settings_replica.last_visited.get(&session_id).copied()); + if acknowledged.is_some_and(|at| at >= through) { + return; + } + self.read_acknowledged = Some((session_id.clone(), through)); + self.dispatch(Command::MarkSessionRead { + session_id, + through, + }); } /// Decide what the workspace shows after the index changed. @@ -3586,6 +3643,70 @@ mod tests { }); } + /// A thread is read only once its conversation has loaded and is on + /// screen, and stays read through updates that land while it is shown. A + /// view left before its conversation arrived reports nothing. + #[gpui::test] + fn a_thread_is_reported_read_once_its_conversation_loads(cx: &mut TestAppContext) { + let (to_host, outgoing) = async_channel::unbounded(); + let (_incoming, from_host) = async_channel::unbounded(); + let link = tcode_client::HostLink::new(to_host, from_host); + let workspace = cx.new(|cx| { + WorkspaceStore::new_attached(link, WorkspaceAttachment::Local, None, None, false, cx) + }); + let reads = || { + std::iter::from_fn(|| outgoing.try_recv().ok()) + .filter_map(|line| { + match tcode_protocol::decode_client_line(&line).unwrap().payload { + tcode_protocol::ClientPayload::Command(Command::MarkSessionRead { + session_id, + through, + }) => Some((session_id, through)), + _ => None, + } + }) + .collect::>() + }; + let upsert = |id: &str, updated_at| { + let mut meta = + SessionMeta::new(ProviderKind::Codex, std::path::PathBuf::from("/tmp"), None); + meta.id = id.into(); + meta.updated_at = updated_at; + EventEnvelope { + request_id: None, + topic: Topic::Index, + event: ServerEvent::IndexUpsertSession(meta), + } + }; + workspace.update(cx, |store, cx| { + store.apply_domain_event(&upsert("left", 100), cx); + store.apply_domain_event(&upsert("shown", 100), cx); + store.settings_replica.last_visited = + std::collections::HashMap::from([("left".into(), 50), ("shown".into(), 50)]); + + store.set_conversation_on_screen(true); + store.select_session("left".into()); + store.leave_session(); + store.apply_domain_event(&session_snapshot("left", 0, vec![reply(1)]), cx); + assert_eq!(reads(), [], "left before the conversation loaded"); + + store.select_session("shown".into()); + assert_eq!(reads(), [], "nothing has loaded yet"); + // A compact window went back to the thread list, which keeps the + // thread selected, before the conversation arrived. + store.set_conversation_on_screen(false); + store.apply_domain_event(&session_snapshot("shown", 0, vec![reply(1)]), cx); + assert_eq!(reads(), [], "loaded behind the thread list"); + store.set_conversation_on_screen(true); + assert_eq!(reads(), [("shown".to_string(), 100)]); + store.apply_domain_event(&upsert("left", 110), cx); + assert_eq!(reads(), [], "unchanged for the thread on screen"); + + store.apply_domain_event(&upsert("shown", 120), cx); + assert_eq!(reads(), [("shown".to_string(), 120)]); + }); + } + #[gpui::test] fn history_prefetch_keeps_one_bounded_page_in_flight(cx: &mut TestAppContext) { let root = scratch_root("tcode-prefetch-test");