diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index cef1a77f0b..2d31721bd2 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -1109,6 +1109,7 @@ impl App { match &event { TuiEvent::FocusLost => { + app_server.focused_conversation = None; self.chat_widget .set_sparkle_terminal_focus(/*focused*/ false); let now = Instant::now(); @@ -1126,6 +1127,10 @@ impl App { _ => {} } + let draws_history = matches!( + event, + TuiEvent::Draw | TuiEvent::Resume | TuiEvent::Resize(_) | TuiEvent::FocusGained + ); if self.overlay.is_some() { let _ = self .handle_backtrack_overlay_event(tui, app_server, event) @@ -1230,6 +1235,46 @@ impl App { TuiEvent::FocusLost => {} } } + if draws_history { + let frozen = self.chat_widget.is_external_writer_view(); + let visible = self.current_displayed_thread_id().filter(|&id| { + tui.is_terminal_focused() + && matches!(self.overlay, None | Some(Overlay::Transcript(_))) + && self.chat_widget.no_modal_or_popup_active() + && self.initial_history_replay_buffer.is_none() + && (!self.thread_unavailable(id) || frozen) + }); + let previous = std::mem::replace(&mut app_server.focused_conversation, visible); + if previous != visible { + app_server.attempted_revision = None; + } + // Opening a full view counts immediately. Within the same view a new result's + // receipt can arrive just before the terminal event that finishes rendering it. + if let Some(id) = visible + && (previous != visible || !self.chat_widget.is_agent_turn_running()) + && let Some(revision) = app_server.viewed_revision(id, frozen) + { + // A read-only snapshot does not subscribe to receipt notifications. + let refresh = frozen.then(|| { + ( + self.app_event_tx.clone(), + AppEvent::RefreshViewedThread { + backend_id: app_server.backend_id(), + thread_id: id, + revision: revision.clone(), + }, + ) + }); + let update = app_server.acknowledge_read(id, revision); + tokio::spawn(async move { + if let Err(error) = update.await { + tracing::warn!(%id, %error, "Could not acknowledge viewed conversation"); + } else if let Some((tx, event)) = refresh { + tx.send(event); + } + }); + } + } // Both transcript owners must consume completions before automatic work advances. if idle_draw { tui.clipboard.advance(tui.frame_requester()); diff --git a/codex-rs/tui/src/app/agent_center/rows.rs b/codex-rs/tui/src/app/agent_center/rows.rs index 07aaff3ebf..c06300a8c2 100644 --- a/codex-rs/tui/src/app/agent_center/rows.rs +++ b/codex-rs/tui/src/app/agent_center/rows.rs @@ -202,7 +202,13 @@ impl AgentsOverviewView { style, ); let (status, mut dot) = Self::status(task); - if index == self.selected { + let idle = matches!( + task.group, + AgentsOverviewGroup::Ready | AgentsOverviewGroup::Finished + ); + if task.unread || idle { + dot = crate::style::task_marker(task.unread, style); + } else if index == self.selected { dot.style = style; } line( @@ -236,11 +242,14 @@ impl AgentsOverviewView { } else { self.title_style(task.thread_id) }; - line( - Line::from(display_title(&task.thread).to_owned()).style(title_style), - title, - buf, - ); + let task_title = display_title(&task.thread); + // On narrow rows the result dot shares the only slot for the working or alert icon. + let task_title = if task.unread && !idle && status_area.is_empty() { + format!("{status}: {task_title}") + } else { + task_title.to_owned() + }; + line(Line::from(task_title).style(title_style), title, buf); line(Line::from(status).style(style), status_area, buf); if !updated.is_empty() { let age = format_relative_time( diff --git a/codex-rs/tui/src/app/agent_center_tests.rs b/codex-rs/tui/src/app/agent_center_tests.rs index 40ec900d39..7fd5fafc93 100644 --- a/codex-rs/tui/src/app/agent_center_tests.rs +++ b/codex-rs/tui/src/app/agent_center_tests.rs @@ -332,7 +332,11 @@ async fn live_center_columns() { )); crate::chatwidget::activate_voice_for_thread(&mut app.chat_widget, child); app.agents_overview.pinned_thread_ids = Some(vec![current, ThreadId::from_u128(/*value*/ 44)]); + app.agents_overview + .unread + .insert(ThreadId::from_u128(/*value*/ 47)); let mut view = app.agents_overview_view(threads, Some(current)); + assert!(screen(&view, /*width*/ 50, /*height*/ 24).contains("● Error: Failed task")); insta::assert_snapshot!(screen(&view, /*width*/ 160, /*height*/ 22)); view.pinned_thread_ranks = None; let mut selected_status_styles = Vec::new(); @@ -351,10 +355,6 @@ async fn live_center_columns() { .iter() .position(|cell| cell.symbol() == "›") .unwrap(); - assert_eq!( - selected_row[marker + 2].style(), - selected_row[marker].style() - ); selected_status_styles.push(format!( "{} | {:?}", selected_row[marker + 2].symbol(), diff --git a/codex-rs/tui/src/app/agents_overview.rs b/codex-rs/tui/src/app/agents_overview.rs index 79fcc543cd..2b3fc447f8 100644 --- a/codex-rs/tui/src/app/agents_overview.rs +++ b/codex-rs/tui/src/app/agents_overview.rs @@ -42,6 +42,7 @@ pub(super) struct AgentsOverviewState { pub(super) removed_threads: HashSet, /// Local visibility only; activity and metadata refreshes never reveal hidden roots. pub(super) hidden_threads: HashSet, + pub(super) unread: HashSet, pub(super) last_messages: HashMap, pub(super) usage: HashMap, pub(super) pending_usage: Option<(ThreadId, Uuid)>, @@ -145,6 +146,9 @@ impl App { } match result { Ok(refresh) => { + for id in refresh.cleared_unread { + self.agents_overview.unread.remove(&id); + } if let Some(pinned_thread_ids) = refresh.pinned_thread_ids { self.agents_overview.pinned_thread_ids = pinned_thread_ids; } @@ -169,7 +173,12 @@ impl App { if self.agents_overview.removed_threads.contains(&thread_id) { continue; } - if let Some(mut thread) = thread { + if let Some((mut thread, unread)) = thread { + if unread { + self.agents_overview.unread.insert(thread_id); + } else { + self.agents_overview.unread.remove(&thread_id); + } if thread.ephemeral { self.agents_overview.threads.remove(&thread_id); self.agents_overview.last_messages.remove(&thread_id); @@ -340,6 +349,7 @@ impl App { continue; } rows.push(AgentsOverviewRow { + unread: self.agents_overview.unread.contains(&thread_id), details: self.agents_overview_details(root, &children), thread: root.clone(), thread_id, diff --git a/codex-rs/tui/src/app/agents_overview_actions_tests.rs b/codex-rs/tui/src/app/agents_overview_actions_tests.rs index 5dc6e6e170..406e7ea04d 100644 --- a/codex-rs/tui/src/app/agents_overview_actions_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_actions_tests.rs @@ -48,7 +48,7 @@ async fn pin_action_updates_shared_section_state() -> Result<()> { assert_eq!( threads .iter() - .any(|thread| thread.id == started.session.thread_id.to_string()), + .any(|(thread, _)| thread.id == started.session.thread_id.to_string()), pinned ); } @@ -654,7 +654,8 @@ async fn hidden_task_stays_hidden_through_activity_and_seed_until_explicit_resum &app_server, request_id, Ok(AgentsOverviewThreadRefresh { - threads: HashMap::from([(id, Some(thread.clone()))]), + cleared_unread: Vec::new(), + threads: HashMap::from([(id, Some((thread.clone(), false)))]), last_messages: HashMap::new(), recent_seed_complete: true, discovery: None, @@ -935,7 +936,11 @@ async fn lifecycle_removes_background_and_current_tasks_without_losing_the_dashb &app_server, request_id, Ok(AgentsOverviewThreadRefresh { - threads: stale_threads, + cleared_unread: Vec::new(), + threads: stale_threads + .into_iter() + .map(|(id, thread)| (id, thread.map(|thread| (thread, false)))) + .collect(), last_messages: HashMap::new(), recent_seed_complete: true, discovery: None, @@ -1038,6 +1043,7 @@ async fn lifecycle_removes_background_and_current_tasks_without_losing_the_dashb .send(AppEvent::AgentsOverviewThreadsLoaded { request_id, result: Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), threads: HashMap::new(), last_messages: HashMap::new(), recent_seed_complete: true, diff --git a/codex-rs/tui/src/app/agents_overview_discovery.rs b/codex-rs/tui/src/app/agents_overview_discovery.rs index fb6aedf93b..6463a86447 100644 --- a/codex-rs/tui/src/app/agents_overview_discovery.rs +++ b/codex-rs/tui/src/app/agents_overview_discovery.rs @@ -18,9 +18,21 @@ use uuid::Uuid; const PAGE_SIZE: u32 = 10; +// Pair each row with its batched receipt before buffering or merging source pages. +fn listed_threads(page: ThreadListResponse) -> impl Iterator { + page.data.into_iter().map(move |thread| { + let unread = page + .read_states + .as_ref() + .and_then(|states| states.get(&thread.id)) + .is_some_and(|state| state.first_unread.is_some()); + (thread, unread) + }) +} + pub(crate) async fn list_pinned_threads( handle: &AppServerRequestHandle, -) -> Result>, String> { +) -> Result>, String> { let mut threads = Vec::new(); for source_kinds in [ Vec::new(), @@ -54,10 +66,11 @@ pub(crate) async fn list_pinned_threads( .await; match result { Ok(page) => { - threads.extend(page.data.into_iter().filter(|thread| { + let next_cursor = page.next_cursor.clone(); + threads.extend(listed_threads(page).filter(|(thread, _)| { is_overview_thread(thread) && supports_shared_pinning(&thread.source) })); - let Some(next_cursor) = page.next_cursor else { + let Some(next_cursor) = next_cursor else { break; }; cursor = Some(next_cursor); @@ -93,7 +106,7 @@ pub(super) fn supports_shared_pinning(source: &SessionSource) -> bool { #[derive(Clone, Debug)] struct SourcePage { cursor: Option, - rows: VecDeque, + rows: VecDeque<(Thread, bool)>, exhausted: bool, sort_key: ThreadSortKey, } @@ -125,7 +138,7 @@ impl AgentsOverviewDiscovery { &mut self, handle: &AppServerRequestHandle, limit: usize, - ) -> (Vec, bool) { + ) -> (Vec<(Thread, bool)>, bool) { let [interactive, other] = &mut self.sources; let (left, right) = tokio::join!( interactive.fill(handle, Vec::new()), @@ -140,7 +153,9 @@ impl AgentsOverviewDiscovery { .sources .iter() .enumerate() - .filter_map(|(index, source)| source.rows.front().map(|row| (index, row))) + .filter_map(|(index, source)| { + source.rows.front().map(|(thread, _)| (index, thread)) + }) .max_by(|(_, left), (_, right)| { left.recency_at .unwrap_or(left.updated_at) @@ -202,9 +217,10 @@ impl SourcePage { } Ok(page) => { self.exhausted = page.next_cursor.is_none(); - self.cursor = page.next_cursor; - self.rows - .extend(page.data.into_iter().filter(is_overview_thread)); + self.cursor = page.next_cursor.clone(); + self.rows.extend( + listed_threads(page).filter(|(thread, _)| is_overview_thread(thread)), + ); return true; } Err(error) => { diff --git a/codex-rs/tui/src/app/agents_overview_discovery_tests.rs b/codex-rs/tui/src/app/agents_overview_discovery_tests.rs index 9b81dd30c3..82830c7e93 100644 --- a/codex-rs/tui/src/app/agents_overview_discovery_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_discovery_tests.rs @@ -127,7 +127,11 @@ async fn check_discovery(mixed_sources: bool) -> Result<()> { } } "thread/read" => { - json!({"thread": rows.iter().find(|thread| thread.id == params["threadId"]).unwrap()}) + let thread = rows + .iter() + .find(|thread| thread.id == params["threadId"]) + .unwrap(); + json!({"thread": thread, "readState": {"revision": "read", "firstUnread": null}}) } "thread/turns/list" => json!({"data": [], "nextCursor": null}), method => panic!("unexpected request: {method}"), @@ -189,10 +193,13 @@ async fn check_discovery(mixed_sources: bool) -> Result<()> { .collect::>(), expected_first_twenty ); + // Another client read this older, unloaded task while the center was closed. + app.agents_overview.unread.insert(ids[19]); app.remove_agents_overview_thread(ids[1]); expected.remove(&ids[1]); app.refresh_agents_overview_threads(&session); finish_overview_refresh(&mut app, &session, &mut rx).await; + assert!(!app.agents_overview.unread.contains(&ids[19])); assert_eq!(app.agents_overview.threads.len(), 20); assert!(app.agents_overview.threads.contains_key(&ids[21])); for &id in &ids[2..22] { diff --git a/codex-rs/tui/src/app/agents_overview_tests.rs b/codex-rs/tui/src/app/agents_overview_tests.rs index 562271c9b5..a43460a8d2 100644 --- a/codex-rs/tui/src/app/agents_overview_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_tests.rs @@ -42,6 +42,8 @@ async fn overview_worktree_creation_busy_state() { async fn overview_thread_colors_match_footer_and_respect_color_suppression() { let mut app = make_test_app().await; let id = ThreadId::from_u128(/*value*/ 42); + let selected_id = ThreadId::from_u128(/*value*/ 43); + app.agents_overview.unread.extend([id, selected_id]); let mut snapshot = Vec::new(); for enabled in [true, false] { app.local_settings.tui.status_line_use_colors = enabled; @@ -53,12 +55,9 @@ async fn overview_thread_colors_match_footer_and_respect_color_suppression() { ); thread.name = Some("Named task".into()); let mut selected = thread.clone(); - selected.id = ThreadId::from_u128(/*value*/ 43).to_string(); + selected.id = selected_id.to_string(); selected.name = Some("Selected task".into()); - let view = app.agents_overview_view( - vec![thread, selected], - Some(ThreadId::from_u128(/*value*/ 43)), - ); + let view = app.agents_overview_view(vec![thread, selected], Some(selected_id)); let mut terminal = Terminal::new(TestBackend::new(100, 30)).unwrap(); terminal .draw(|frame| view.render(frame.area(), frame.buffer_mut())) @@ -66,15 +65,20 @@ async fn overview_thread_colors_match_footer_and_respect_color_suppression() { let buffer = terminal.backend().buffer(); for row in buffer.content.chunks(100) { let text: String = row.iter().map(ratatui::buffer::Cell::symbol).collect(); + if let Some(x) = text.find("● Selected task") { + let x = text[..x].chars().count(); + snapshot.push(format!( + "selected dot {:?} | title {:?}", + row[x].style(), + row[x + 2].style(), + )); + } if let Some(x) = text.find("Named task") { let x = text[..x].chars().count(); - let dot = row.iter().find(|cell| cell.symbol() == "○").unwrap(); - let mut expected = ratatui::buffer::Cell::default(); - expected.set_symbol("○").set_style(Style::default().cyan()); - assert_eq!(dot, &expected); snapshot.push(format!( - "{} | title style: {:?}", + "{} | unread: {:?} | title style: {:?}", normalize_agent_center_snapshot(&text), + row[x - 2].fg, row[x].style() )); } @@ -158,7 +162,6 @@ use pretty_assertions::assert_eq; use ratatui::Terminal; use ratatui::backend::TestBackend; use ratatui::layout::Rect; -use ratatui::style::Style; static OVERVIEW_TIMESTAMP: std::sync::LazyLock = std::sync::LazyLock::new(|| chrono::Utc::now().timestamp() - 120); @@ -371,11 +374,12 @@ async fn shared_overview_keeps_rows_and_replays_changes_over_stale_reads() -> Re &app_server, request_id, Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), last_messages: HashMap::new(), threads: HashMap::from([ - (retained, Some(retained_thread.clone())), - (archived, Some(archived_thread)), - (deleted, Some(deleted_thread)), + (retained, Some((retained_thread.clone(), false))), + (archived, Some((archived_thread, false))), + (deleted, Some((deleted_thread, false))), ]), recent_seed_complete: true, discovery: None, @@ -396,6 +400,7 @@ async fn shared_overview_keeps_rows_and_replays_changes_over_stale_reads() -> Re &app_server, request_id, Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), last_messages: HashMap::new(), threads: HashMap::from([(retained, None)]), recent_seed_complete: true, @@ -446,6 +451,7 @@ async fn shared_overview_keeps_rows_and_replays_changes_over_stale_reads() -> Re &app_server, request_id, Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), threads: HashMap::new(), last_messages, recent_seed_complete: true, @@ -2737,6 +2743,7 @@ async fn command_center_refresh_failure_is_inline_and_clears_on_success() -> Res for result in [ Err("unavailable".into()), Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), threads: HashMap::new(), last_messages: HashMap::new(), recent_seed_complete: false, @@ -2769,6 +2776,7 @@ async fn command_center_refresh_failure_is_inline_and_clears_on_success() -> Res &server, request_id, Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), threads: HashMap::new(), last_messages: HashMap::new(), recent_seed_complete: true, diff --git a/codex-rs/tui/src/app/agents_overview_threads.rs b/codex-rs/tui/src/app/agents_overview_threads.rs index 084f86ed97..1f6927bf68 100644 --- a/codex-rs/tui/src/app/agents_overview_threads.rs +++ b/codex-rs/tui/src/app/agents_overview_threads.rs @@ -27,6 +27,7 @@ use codex_app_server_protocol::ThreadTurnsListParams; use codex_app_server_protocol::ThreadTurnsListResponse; use codex_protocol::ThreadId; use codex_protocol::protocol::SubAgentSource; +use futures::StreamExt; use std::collections::HashMap; use uuid::Uuid; @@ -57,6 +58,7 @@ impl App { pub(super) fn remove_agents_overview_thread(&mut self, thread_id: ThreadId) { self.prepare_agents_overview_removal(&std::collections::HashSet::from([thread_id])); self.agents_overview.removed_threads.insert(thread_id); + self.agents_overview.unread.remove(&thread_id); if let Some(Some(thread)) = self.agents_overview.threads.remove(&thread_id) && !thread.ephemeral && thread.parent_thread_id.is_none() @@ -100,6 +102,14 @@ impl App { .get_mut(&thread_id) .and_then(Option::as_mut); match notification { + ServerNotification::ThreadReadStateChanged(change) => { + if change.read_state.first_unread.is_some() { + self.agents_overview.unread.insert(thread_id); + } else { + self.agents_overview.unread.remove(&thread_id); + } + self.repaint_agents_overview(); + } ServerNotification::ThreadTokenUsageUpdated(usage) => { if self.agents_overview.threads.contains_key(&thread_id) { self.agents_overview @@ -197,7 +207,9 @@ impl App { } if !matches!( notification, - ServerNotification::ThreadReverted(_) | ServerNotification::ThreadTokenUsageUpdated(_) + ServerNotification::ThreadReverted(_) + | ServerNotification::ThreadTokenUsageUpdated(_) + | ServerNotification::ThreadReadStateChanged(_) ) && self.agents_overview.threads.contains_key(&thread_id) { self.agents_overview.refresh_thread_ids.insert(thread_id); @@ -269,7 +281,16 @@ impl App { return; } - self.agents_overview.pin_refresh_requested = false; + let mut unread_rows = Vec::new(); + if std::mem::take(&mut self.agents_overview.pin_refresh_requested) { + unread_rows.extend( + self.agents_overview + .visible_thread_ids + .iter() + .copied() + .filter(|id| self.agents_overview.unread.contains(id)), + ); + } let request_id = Uuid::new_v4(); self.agents_overview.request_id = Some(request_id); let initialized = self.agents_overview.initialized; @@ -309,10 +330,10 @@ impl App { let pinned_thread_ids = match list_pinned_threads(&request_handle).await { Ok(Some(pinned_threads)) => { let mut ids = Vec::with_capacity(pinned_threads.len()); - for thread in pinned_threads { + for (thread, is_unread) in pinned_threads { if let Ok(thread_id) = ThreadId::from_string(&thread.id) { ids.push(thread_id); - threads.insert(thread_id, Some(thread)); + threads.insert(thread_id, Some((thread, is_unread))); } } Some(Some(ids)) @@ -353,14 +374,34 @@ impl App { .into_iter() .filter_map(|id| ThreadId::from_string(&id).ok()), ); - thread_ids.extend(detail_thread_ids(recent.iter())); - for thread in recent { + thread_ids.extend(detail_thread_ids(recent.iter().map(|(thread, _)| thread))); + for (thread, is_unread) in recent { if let Ok(thread_id) = ThreadId::from_string(&thread.id) { - threads.insert(thread_id, Some(thread)); + threads.insert(thread_id, Some((thread, is_unread))); } } } + // Recheck dots on older rows without fetching their transcript previews. + unread_rows.retain(|id| !threads.contains_key(id) && !thread_ids.contains(id)); + let cleared_unread = futures::stream::iter(unread_rows) + .map(async |id| { + let response: ThreadReadResponse = request_handle + .request_typed(ClientRequest::ThreadRead { + request_id: RequestId::String(Uuid::new_v4().to_string()), + params: ThreadReadParams { + thread_id: id.to_string(), + include_turns: false, + }, + }) + .await + .ok()?; + response.read_state?.first_unread.is_none().then_some(id) + }) + .buffer_unordered(16) + .filter_map(std::future::ready) + .collect() + .await; let mut reads = tokio::task::JoinSet::new(); for thread_id in thread_ids { threads.entry(thread_id).or_default(); @@ -412,7 +453,10 @@ impl App { _ => None, }); } - Some((thread_id, response.thread, last_message)) + let unread = response + .read_state + .is_some_and(|state| state.first_unread.is_some()); + Some((thread_id, (response.thread, unread), last_message)) } Err(error) => { tracing::warn!(%thread_id, %error, "failed to read agent thread"); @@ -439,6 +483,7 @@ impl App { } } Ok(AgentsOverviewThreadRefresh { + cleared_unread, threads, last_messages, recent_seed_complete, diff --git a/codex-rs/tui/src/app/agents_overview_usage_tests.rs b/codex-rs/tui/src/app/agents_overview_usage_tests.rs index af459f856d..c03658e79e 100644 --- a/codex-rs/tui/src/app/agents_overview_usage_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_usage_tests.rs @@ -99,7 +99,8 @@ async fn selected_usage_is_cached_and_account_changes_discard_old_results() -> R &app_server, request_id, Ok(AgentsOverviewThreadRefresh { - threads: HashMap::from([(selected, Some(threads[0].clone()))]), + cleared_unread: Vec::new(), + threads: HashMap::from([(selected, Some((threads[0].clone(), false)))]), last_messages: HashMap::new(), recent_seed_complete: true, discovery: None, diff --git a/codex-rs/tui/src/app/agents_overview_view.rs b/codex-rs/tui/src/app/agents_overview_view.rs index c3843db7f1..6d6c6e438d 100644 --- a/codex-rs/tui/src/app/agents_overview_view.rs +++ b/codex-rs/tui/src/app/agents_overview_view.rs @@ -112,6 +112,7 @@ pub(super) struct AgentsOverviewRow { pub(super) thread_id: ThreadId, pub(super) group: AgentsOverviewGroup, pub(super) is_current: bool, + pub(super) unread: bool, pub(super) has_voice: bool, } diff --git a/codex-rs/tui/src/app/app_server_event_targets.rs b/codex-rs/tui/src/app/app_server_event_targets.rs index 13d30c273f..bf2112f50d 100644 --- a/codex-rs/tui/src/app/app_server_event_targets.rs +++ b/codex-rs/tui/src/app/app_server_event_targets.rs @@ -185,6 +185,9 @@ pub(super) fn server_notification_thread_target( | ServerNotification::AuthRecoveryCompleted(notification) => { Some(notification.thread_id.as_str()) } + ServerNotification::ThreadReadStateChanged(notification) => { + Some(notification.thread_id.as_str()) + } ServerNotification::Warning(notification) => notification.thread_id.as_deref(), ServerNotification::GuardianWarning(notification) => Some(notification.thread_id.as_str()), ServerNotification::McpServerOauthLoginCompleted(notification) => { @@ -202,7 +205,6 @@ pub(super) fn server_notification_thread_target( ServerNotification::ProjectChanged(_) | ServerNotification::SkillsChanged(_) | ServerNotification::AccountUpdated(_) - | ServerNotification::ThreadReadStateChanged(_) | ServerNotification::GatewayOAuthChanged(_) | ServerNotification::AccountRateLimitsUpdated(_) | ServerNotification::AppListUpdated(_) diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index 27fd797c1e..a9c64f516b 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -2976,6 +2976,13 @@ impl App { } AppEvent::OpenAgentsOverview => self.open_agents_overview(app_server), AppEvent::ShowMoreAgentsOverview => self.show_more_agents_overview(app_server), + AppEvent::RefreshViewedThread { backend_id, thread_id, revision } => { + if app_server.backend_id() == backend_id { + app_server.frozen_receipt.take_if(|(id, rev)| *id == thread_id && *rev == revision); + self.agents_overview.refresh_thread_ids.insert(thread_id); + self.refresh_changed_agents_overview_threads(app_server); + } + } AppEvent::NewAgentsOverviewSession { cwd } => { return Box::pin(self.new_agents_overview_session(tui, app_server, cwd)).await; } diff --git a/codex-rs/tui/src/app/session_lifecycle.rs b/codex-rs/tui/src/app/session_lifecycle.rs index 724ae6a5ff..8f8416f545 100644 --- a/codex-rs/tui/src/app/session_lifecycle.rs +++ b/codex-rs/tui/src/app/session_lifecycle.rs @@ -451,6 +451,9 @@ impl App { // `thread/read` can seed replay state, but it does not attach the app-server // listener that `thread/resume` establishes, so treat this path as replay-only. session.model.clear(); + if was_external_writer { + app_server.freeze_viewed_receipt(thread_id); + } (session, turns, false) } }; diff --git a/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_columns.snap b/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_columns.snap index 8f331bb8dc..346345fe7f 100644 --- a/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_columns.snap +++ b/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_columns.snap @@ -12,7 +12,7 @@ expression: "screen(&view, 160, 22)" │ /tmp/project 4 │ Model: Unknown ○ A longer task title that uses the available column voice Ready 5m ago │ Reasoning: Unknown - ! Failed task Error 1h ago │ + ● Failed task Error 1h ago │ ● Working task Working 1d ago │ Project ○ Unloaded task Inactive 2d ago │ /tmp/project │ diff --git a/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_selected_status_styles.snap b/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_selected_status_styles.snap index 7d804f48b7..06d7109bcf 100644 --- a/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_selected_status_styles.snap +++ b/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_selected_status_styles.snap @@ -1,11 +1,10 @@ --- source: tui/src/app/agent_center_tests.rs -assertion_line: 98 expression: "selected_status_styles.join(\"\\n\")" --- -○ | Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold().reversed() -○ | Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold().reversed() -! | Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold().reversed() +○ | Style::new().fg(Color::Reset).bg(Color::Rgb(128, 128, 128)).underline_color(Color::Reset).bold().reversed() +○ | Style::new().fg(Color::Reset).bg(Color::Rgb(128, 128, 128)).underline_color(Color::Reset).bold().reversed() +● | Style::new().fg(Color::Reset).bg(Color::Rgb(59, 130, 246)).underline_color(Color::Reset).bold().reversed() ● | Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold().reversed() ● | Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold().reversed() -○ | Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold().reversed() +○ | Style::new().fg(Color::Reset).bg(Color::Rgb(128, 128, 128)).underline_color(Color::Reset).bold().reversed() diff --git a/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__overview_thread_colors_match_footer_and_respect_color_suppression.snap b/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__overview_thread_colors_match_footer_and_respect_color_suppression.snap index 06ae317732..1128e68fae 100644 --- a/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__overview_thread_colors_match_footer_and_respect_color_suppression.snap +++ b/codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__overview_thread_colors_match_footer_and_respect_color_suppression.snap @@ -2,5 +2,7 @@ source: tui/src/app/agents_overview_tests.rs expression: "snapshot.join(\"\\n\")" --- - ○ Named task │ Selected task | title style: Style::new().fg(Color::Rgb(242, 205, 205)).bg(Color::Reset).underline_color(Color::Reset) - ○ Named task │ Selected task | title style: Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset) + ● Named task │ Selected task | unread: Rgb(59, 130, 246) | title style: Style::new().fg(Color::Rgb(242, 205, 205)).bg(Color::Reset).underline_color(Color::Reset) +selected dot Style::new().fg(Color::Reset).bg(Color::Rgb(59, 130, 246)).underline_color(Color::Reset).bold().reversed() | title Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold().reversed() + ● Named task │ Selected task | unread: Rgb(59, 130, 246) | title style: Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset) +selected dot Style::new().fg(Color::Reset).bg(Color::Rgb(59, 130, 246)).underline_color(Color::Reset).bold().reversed() | title Style::new().fg(Color::Reset).bg(Color::Reset).underline_color(Color::Reset).bold().reversed() diff --git a/codex-rs/tui/src/app/tests/active_reconnect_tests.rs b/codex-rs/tui/src/app/tests/active_reconnect_tests.rs index eee993a6dd..a56853b2fc 100644 --- a/codex-rs/tui/src/app/tests/active_reconnect_tests.rs +++ b/codex-rs/tui/src/app/tests/active_reconnect_tests.rs @@ -686,10 +686,13 @@ async fn reconnect_allows_slow_hydration_but_bounds_a_stalled_server() -> Result serve_reconnect_requests( tokio_tungstenite::accept_async(stream).await?, move |request| async move { - assert_eq!(request.method, "thread/resume"); - tokio::time::pause(); - tokio::time::sleep(Duration::from_secs(delay)).await; - tokio::time::resume(); + if request.method == "thread/resume" { + tokio::time::pause(); + tokio::time::sleep(Duration::from_secs(delay)).await; + tokio::time::resume(); + } else { + assert_eq!(request.method, "thread/read"); + } Some(json!({"error": {"code": -32600, "message": "thread no longer exists"}})) }, ) diff --git a/codex-rs/tui/src/app/tests/external_writer_fork_tests.rs b/codex-rs/tui/src/app/tests/external_writer_fork_tests.rs index a7e0ad810b..7351ed2a1b 100644 --- a/codex-rs/tui/src/app/tests/external_writer_fork_tests.rs +++ b/codex-rs/tui/src/app/tests/external_writer_fork_tests.rs @@ -153,6 +153,26 @@ async fn external_writer_fork_opens_editable_thread_without_taking_source_lease( app.enqueue_primary_thread_session(view.session, view.turns) .await?; app.ensure_thread_channel(thread_id).mark_external_writer(); + // Navigation back to this read-only root must acknowledge its fresh snapshot. + let db = crate::init_state_db_for_app_server_target(&app.config, &AppServerTarget::Embedded) + .await? + .expect("state database"); + assert!( + db.publish_thread_attention(thread_id, "parent-result") + .await + .map_err(std::io::Error::other)? + ); + assert!(!Box::pin(app.attach_live_thread_for_selection(&mut server, thread_id)).await?); + let revision = db + .thread_read_states(&[thread_id]) + .await + .map_err(std::io::Error::other)?[&thread_id] + .revision + .clone(); + assert_eq!( + server.viewed_revision(thread_id, /*frozen*/ true), + Some(revision) + ); app.chat_widget .set_queue_autosend_suppressed(/*suppressed*/ true); app.chat_widget.insert_str("Retained queued prompt"); diff --git a/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs b/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs index 6125e94028..e40950562c 100644 --- a/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs +++ b/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs @@ -287,6 +287,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv &session, stale_request, Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), last_messages: HashMap::new(), threads: HashMap::new(), recent_seed_complete: false, @@ -433,8 +434,12 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv &session, stale_request, Ok(AgentsOverviewThreadRefresh { + cleared_unread: Vec::new(), last_messages: HashMap::new(), - threads: stale_threads, + threads: stale_threads + .into_iter() + .map(|(id, thread)| (id, thread.map(|thread| (thread, false)))) + .collect(), recent_seed_complete: true, discovery: None, pinned_thread_ids: None, diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index 341b995861..ed82fc15a5 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -3322,7 +3322,7 @@ async fn paginated_workflows_never_request_full_thread_history() -> Result<()> { )) .await?; assert_eq!(resumed.session.thread_id, paginated_thread_id); - assert!(recorded_params(&requests, "thread/read").is_empty()); + assert_eq!(recorded_params(&requests, "thread/read").len(), 1); let resume_requests = recorded_params(&requests, "thread/resume"); assert_eq!(resume_requests.len(), 1); assert_eq!(resume_requests[0]["excludeTurns"], true); @@ -4952,7 +4952,8 @@ fn session_lifecycle_avoids_redundant_subagent_metadata_reads() -> Result<()> { assert!(matches!(control, AppRunControl::Continue)); assert_eq!(app.chat_widget.thread_id(), Some(root_thread_id)); - assert_eq!(take_backfill_counts(&requests), (1, expected_reads)); + // Include the selected root’s receipt as well as the loaded child metadata. + assert_eq!(take_backfill_counts(&requests), (1, expected_reads + 1)); assert_eq!( app.agent_navigation.get(&child_thread_id), Some(&AgentPickerThreadEntry { diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index 94f1ee31d2..883c2516e3 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -279,7 +279,8 @@ pub(crate) enum RecapTrigger { #[derive(Debug)] pub(crate) struct AgentsOverviewThreadRefresh { - pub(crate) threads: std::collections::HashMap>, + pub(crate) cleared_unread: Vec, + pub(crate) threads: std::collections::HashMap>, pub(crate) last_messages: std::collections::HashMap, pub(crate) recent_seed_complete: bool, pub(crate) discovery: Option, @@ -316,6 +317,11 @@ pub(crate) enum AppEvent { /// Open the live command center for recent and locally retained root sessions. OpenAgentsOverview, ShowMoreAgentsOverview, + RefreshViewedThread { + backend_id: crate::backend_id::BackendId, + thread_id: ThreadId, + revision: String, + }, /// Create an empty thread from the command center. NewAgentsOverviewSession { cwd: Option, diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index f6ce3f4cd4..0e9b8f989d 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -13,6 +13,7 @@ pub(crate) mod provider_selection; #[cfg(test)] #[path = "app_server_session/provider_selection_tests.rs"] mod provider_selection_tests; +mod read_state; mod realtime; mod rollout_history; mod startup_launch; @@ -328,6 +329,10 @@ pub(crate) struct AppServerSession { client: AppServerClient, next_request_id: i64, history_pagination: HashMap, + pub(crate) focused_conversation: Option, + pub(crate) attempted_revision: Option, + pub(crate) frozen_receipt: Option<(ThreadId, String)>, + pub(crate) unread_revisions: HashMap, task_tool_threads: HashSet, task_tool_capabilities_dir: Option, task_search_generation: Arc, @@ -484,6 +489,10 @@ impl AppServerSession { client, next_request_id: 1, history_pagination: HashMap::new(), + focused_conversation: None, + attempted_revision: None, + frozen_receipt: None, + unread_revisions: HashMap::new(), task_tool_threads: HashSet::new(), task_tool_capabilities_dir: None, task_search_generation: Arc::new(AtomicU64::new(0)), @@ -930,13 +939,18 @@ impl AppServerSession { } pub(crate) async fn next_event(&mut self) -> Option { - self.client - .next_event() - .await - .map(|event| AppServerSessionEvent { - backend_id: self.backend_id, - event, - }) + let event = self.client.next_event().await?; + if let codex_app_server_client::AppServerEvent::ServerNotification(notification) = &event + && let codex_app_server_protocol::ServerNotification::ThreadReadStateChanged(change) = + notification.as_ref() + && let Ok(id) = ThreadId::from_string(&change.thread_id) + { + self.remember_read_state(id, Some(change.read_state.clone())); + } + Some(AppServerSessionEvent { + backend_id: self.backend_id, + event, + }) } #[cfg(test)] @@ -1370,7 +1384,13 @@ impl AppServerSession { }) .await; let mut response: ThreadReadResponse = match response { - Ok(response) => return Ok(response.thread), + Ok(response) => { + // Freeze the metadata revision observed before history hydration. + if !include_turns { + self.remember_read_state(thread_id, response.read_state); + } + return Ok(response.thread); + } Err(TypedRequestError::Server { source, .. }) if include_turns && source.message diff --git a/codex-rs/tui/src/app_server_session/read_state.rs b/codex-rs/tui/src/app_server_session/read_state.rs new file mode 100644 index 0000000000..fdf4b76138 --- /dev/null +++ b/codex-rs/tui/src/app_server_session/read_state.rs @@ -0,0 +1,87 @@ +//! A focused conversation acknowledges only the receipt revision it observed from this backend. + +use super::*; +use codex_app_server_protocol::ThreadReadState; + +pub(crate) fn unread_revision(state: Option) -> Option { + state.and_then(|s| s.first_unread.map(|_| s.revision)) +} + +impl AppServerSession { + pub(crate) fn remember_read_state(&mut self, id: ThreadId, state: Option) { + if let Some(revision) = unread_revision(state) { + self.unread_revisions.insert(id, revision); + } else { + self.unread_revisions.remove(&id); + } + } + + pub(crate) fn freeze_viewed_receipt(&mut self, thread_id: ThreadId) { + self.frozen_receipt = self + .unread_revisions + .remove(&thread_id) + .map(|revision| (thread_id, revision)); + } + + /// Keep the observed revision for a later visit if the write fails, but send it + /// only once per focused view. The server notification clears successful live reads. + pub(crate) fn viewed_revision(&mut self, id: ThreadId, frozen: bool) -> Option { + let revision = if frozen { + self.frozen_receipt + .as_ref() + .filter(|(thread, _)| *thread == id) + .map(|(_, revision)| revision) + } else { + self.unread_revisions.get(&id) + }?; + if self.attempted_revision.as_ref() == Some(revision) { + return None; + } + self.attempted_revision = Some(revision.clone()); + Some(revision.clone()) + } + + /// A conflict returns the current receipt. A newer result requires a fresh view. + pub(crate) fn acknowledge_read( + &self, + thread_id: ThreadId, + expected_revision: String, + ) -> impl Future> + use<> { + use codex_app_server_protocol::ThreadReadStateOperation; + use codex_app_server_protocol::ThreadReadStateUpdateParams; + use codex_app_server_protocol::ThreadReadStateUpdateResponse; + let handle = self.request_handle(); + let params = ThreadReadStateUpdateParams { + thread_id: thread_id.to_string(), + expected_revision, + operation: ThreadReadStateOperation::Read, + }; + async move { + let send = || { + handle.request_typed::( + ClientRequest::ThreadReadStateUpdate { + request_id: RequestId::String(Uuid::new_v4().to_string()), + params: params.clone(), + }, + ) + }; + let mut result = send().await; + // Retrying the same revision cannot clear a result published in the meantime. + if matches!(&result, Err(TypedRequestError::Server { source, .. }) if matches!(source.code, -32603 | -32001)) + { + tokio::time::sleep(Duration::from_millis(250)).await; + result = send().await; + } + match result { + Ok(response) => Ok(response.read_state), + Err(TypedRequestError::Server { ref source, .. }) + if let Some(data) = source.data.as_ref() + && data["reason"] == "readStateConflict" => + { + Ok(serde_json::from_value(data["readState"].clone())?) + } + Err(error) => Err(error.into()), + } + } + } +} diff --git a/codex-rs/tui/src/app_server_session/rollout_history.rs b/codex-rs/tui/src/app_server_session/rollout_history.rs index 2ca421d4ed..07d22565fd 100644 --- a/codex-rs/tui/src/app_server_session/rollout_history.rs +++ b/codex-rs/tui/src/app_server_session/rollout_history.rs @@ -97,6 +97,7 @@ impl AppServerSession { ) .await .map_err(color_eyre::eyre::Report::msg)?; + self.freeze_viewed_receipt(thread_id); Ok(( AppServerStartedThread { session, @@ -184,31 +185,31 @@ impl AppServerSession { self.thread_tool_transport() .configure_mcp(&mut params.config); } - let mut rollout_maintenance_guard = None; - params.exclude_turns = if self.history_support == ThreadHistorySupport::Paginated { - let known_legacy_history = self + let paginated = self.history_support == ThreadHistorySupport::Paginated; + let cached_legacy = paginated + && self .history_pagination .get(&thread_id) - .is_some_and(|state| state.history_mode == ThreadHistoryMode::Legacy) - && (!self.uses_embedded_app_server() - || ({ - // The guard prevents migration through the full resume, - // regardless of the server's migration feature settings. - rollout_maintenance_guard = - codex_rollout::try_acquire_rollout_maintenance_lock( - config.codex_home.as_path(), - ) - .ok() - .flatten(); - rollout_maintenance_guard.is_some() - } && self - .thread_read(thread_id, /*include_turns*/ false) - .await - .is_ok_and(|thread| thread.history_mode == ThreadHistoryMode::Legacy))); - !known_legacy_history + .is_some_and(|state| state.history_mode == ThreadHistoryMode::Legacy); + // Hold the maintenance lock through a full legacy resume to prevent migration. + let mut rollout_maintenance_guard = if cached_legacy && self.uses_embedded_app_server() { + codex_rollout::try_acquire_rollout_maintenance_lock(config.codex_home.as_path()) + .ok() + .flatten() } else { - false + None }; + // Capture the receipt before the resume-time history boundary. Later results are delivered live. + self.unread_revisions.remove(&thread_id); + let metadata = Box::pin(self.thread_read(thread_id, /*include_turns*/ false)) + .await + .ok(); + let known_legacy = cached_legacy + && (!self.uses_embedded_app_server() + || (rollout_maintenance_guard.is_some() + && metadata + .is_some_and(|thread| thread.history_mode == ThreadHistoryMode::Legacy))); + params.exclude_turns = paginated && !known_legacy; if params.exclude_turns { rollout_maintenance_guard = None; } diff --git a/codex-rs/tui/src/app_server_session/rollout_history_tests.rs b/codex-rs/tui/src/app_server_session/rollout_history_tests.rs index fb315da585..71a652ad23 100644 --- a/codex-rs/tui/src/app_server_session/rollout_history_tests.rs +++ b/codex-rs/tui/src/app_server_session/rollout_history_tests.rs @@ -414,7 +414,7 @@ async fn rollout_maintenance_contention_disables_cached_legacy_resume_shortcut() ) .await?; - assert_eq!(app_server.next_request_id, next_request_id + 3); + assert_eq!(app_server.next_request_id, next_request_id + 4); assert_eq!(resumed.session.thread_id, thread_id); assert_eq!( app_server diff --git a/codex-rs/tui/src/style.rs b/codex-rs/tui/src/style.rs index 8008042c1c..bf54571c59 100644 --- a/codex-rs/tui/src/style.rs +++ b/codex-rs/tui/src/style.rs @@ -13,6 +13,7 @@ use crate::terminal_palette::effective_stdout_color_level; use crate::terminal_palette::rgb_color; use crate::terminal_palette::stdout_color_level; use ratatui::style::Color; +use ratatui::style::Modifier; use ratatui::style::Style; const LIGHT_BG_ACCENT_RGB: (u8, u8, u8) = (28, 100, 200); @@ -26,6 +27,27 @@ pub(crate) const CHATGPT_BLUE_200: (u8, u8, u8) = (99, 168, 248); /// Shared accent for picker selection backgrounds and transcript foreground emphasis. pub(crate) const UI_ACCENT: (u8, u8, u8) = CHATGPT_BLUE_200; +/// The result marker inherits the row fill, including the reverse-video selection fallback. +pub(crate) fn task_marker(unread: bool, row: Style) -> ratatui::text::Span<'static> { + let (symbol, rgb, ansi) = if unread { + ("●", (0x3b, 0x82, 0xf6), Color::Blue) + } else { + ("○", (128, 128, 128), Color::DarkGray) + }; + let color = match effective_stdout_color_level() { + StdoutColorLevel::Ansi16 => ansi, + StdoutColorLevel::Ansi256 => best_color(rgb), + StdoutColorLevel::TrueColor | StdoutColorLevel::Unknown => rgb_color(rgb), + }; + // In reverse video the terminal paints glyphs with the background color. + let style = if row.add_modifier.contains(Modifier::REVERSED) { + row.bg(color) + } else { + row.fg(color) + }; + ratatui::text::Span::styled(symbol, style) +} + #[derive(Clone, Copy)] pub(crate) enum StatusTone { Success,