fix: mark a thread read only once its conversation loads on screen

crates/core/src/settings.rs

@@ -947,7 +947,8 @@ pub struct Settings {
     pub remote_host_name: Option<String>,
     /// 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<String, u64>,

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, .. }

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)]

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)

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()),

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

crates/runtime/src/app/test_support.rs

@@ -194,14 +194,18 @@ pub(super) struct TestEntity(Rc<RefCell<TestClientState>>);
 
 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(),
             &mut host_cx,
             ClientMessage {
                 key: None,
                 id,
-                payload: ClientPayload::Command(command),
+                payload,
             },
         );
     }

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()

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)));
         }

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::<Vec<_>>()
+        };
+        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::<String>::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

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();
     }