openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-31 19:28:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Show unread results in the TUI command center (#58263)

Browse Source 
## What changed

- Display blue filled dots for unread results and gray hollow dots for read idle tasks, preserving status text on narrow rows.
- Load unread state with thread listings and update markers from read-state notifications and refreshes.
- Acknowledge the observed receipt revision when a conversation is rendered in a focused, unobstructed view, including read-only snapshots, without clearing newer results.

## Testing

Extend rendering snapshots and tests for stale unread markers and read-only snapshot receipts.

GitOrigin-RevId: 2dc676c3ffe1e887a5bbf131c88891057386ffc8
This commit is contained in:
Eric Traut
2026-10-30 06:43:51 +00:00
committed by copyberry
parent 8b5ce2fd61
commit 025f6d715d
27 changed files with 420 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app.rs
+45
View File
@@ -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());
codex-rs/tui/src/app/agent_center/rows.rs
+15 -6
View File
@@ -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(
codex-rs/tui/src/app/agent_center_tests.rs
+4 -4
View File
@@ -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(),
codex-rs/tui/src/app/agents_overview.rs
+11 -1
View File
@@ -42,6 +42,7 @@ pub(super) struct AgentsOverviewState {
pub(super) removed_threads: HashSet<ThreadId>,
/// Local visibility only; activity and metadata refreshes never reveal hidden roots.
pub(super) hidden_threads: HashSet<ThreadId>,
pub(super) unread: HashSet<ThreadId>,
pub(super) last_messages: HashMap<ThreadId, String>,
pub(super) usage: HashMap<ThreadId, super::agents_overview_usage::AgentsOverviewUsage>,
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,
codex-rs/tui/src/app/agents_overview_actions_tests.rs
+9 -3
View File
@@ -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,
codex-rs/tui/src/app/agents_overview_discovery.rs
+25 -9
View File
@@ -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<Item = (Thread, bool)> {
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<Option<Vec<Thread>>, String> {
) -> Result<Option<Vec<(Thread, bool)>>, 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<String>,
rows: VecDeque<Thread>,
rows: VecDeque<(Thread, bool)>,
exhausted: bool,
sort_key: ThreadSortKey,
}
@@ -125,7 +138,7 @@ impl AgentsOverviewDiscovery {
&mut self,
handle: &AppServerRequestHandle,
limit: usize,
) -> (Vec<Thread>, 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) => {
codex-rs/tui/src/app/agents_overview_discovery_tests.rs
+8 -1
View File
@@ -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::<HashSet<_>>(),
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] {
codex-rs/tui/src/app/agents_overview_tests.rs
+22 -14
View File
@@ -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<i64> =
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,
codex-rs/tui/src/app/agents_overview_threads.rs
+53 -8
View File
@@ -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,
codex-rs/tui/src/app/agents_overview_usage_tests.rs
+2 -1
View File
@@ -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,
codex-rs/tui/src/app/agents_overview_view.rs
+1
View File
@@ -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,
}
codex-rs/tui/src/app/app_server_event_targets.rs
+3 -1
View File
@@ -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(_)
codex-rs/tui/src/app/event_dispatch.rs
+7
View File
@@ -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;
}
codex-rs/tui/src/app/session_lifecycle.rs
+3
View File
@@ -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)
}
};
codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_columns.snap
+1 -1
View File
@@ -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
│
codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__command_center__live_center_selected_status_styles.snap
+4 -5
View File
@@ -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()
codex-rs/tui/src/app/snapshots/codex_tui__app__agents_overview__tests__overview_thread_colors_match_footer_and_respect_color_suppression.snap
+4 -2
View File
@@ -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()
codex-rs/tui/src/app/tests/active_reconnect_tests.rs
+7 -4
View File
@@ -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"}}))
},
)
codex-rs/tui/src/app/tests/external_writer_fork_tests.rs
+20
View File
@@ -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");
codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs
+6 -1
View File
@@ -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,
codex-rs/tui/src/app/tests/session_lifecycle_requests.rs
+3 -2
View File
@@ -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 {
codex-rs/tui/src/app_event.rs
+7 -1
View File
@@ -279,7 +279,8 @@ pub(crate) enum RecapTrigger {
#[derive(Debug)]
pub(crate) struct AgentsOverviewThreadRefresh {
pub(crate) threads: std::collections::HashMap<ThreadId, Option<Thread>>,
pub(crate) cleared_unread: Vec<ThreadId>,
pub(crate) threads: std::collections::HashMap<ThreadId, Option<(Thread, bool)>>,
pub(crate) last_messages: std::collections::HashMap<ThreadId, String>,
pub(crate) recent_seed_complete: bool,
pub(crate) discovery: Option<crate::app::agents_overview_discovery::AgentsOverviewDiscovery>,
@@ -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<AbsolutePathBuf>,
codex-rs/tui/src/app_server_session.rs
+28 -8
View File
@@ -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<ThreadId, history::ThreadHistoryPagination>,
pub(crate) focused_conversation: Option<ThreadId>,
pub(crate) attempted_revision: Option<String>,
pub(crate) frozen_receipt: Option<(ThreadId, String)>,
pub(crate) unread_revisions: HashMap<ThreadId, String>,
task_tool_threads: HashSet<ThreadId>,
task_tool_capabilities_dir: Option<AbsolutePathBuf>,
task_search_generation: Arc<AtomicU64>,
@@ -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<AppServerSessionEvent> {
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
codex-rs/tui/src/app_server_session/read_state.rs
+87
View File
@@ -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<ThreadReadState>) -> Option<String> {
state.and_then(|s| s.first_unread.map(|_| s.revision))
}
impl AppServerSession {
pub(crate) fn remember_read_state(&mut self, id: ThreadId, state: Option<ThreadReadState>) {
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<String> {
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<Output = Result<ThreadReadState>> + 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::<ThreadReadStateUpdateResponse>(
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()),
}
}
}
}
codex-rs/tui/src/app_server_session/rollout_history.rs
+22 -21
View File
@@ -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;
}
codex-rs/tui/src/app_server_session/rollout_history_tests.rs
+1 -1
View File
@@ -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
codex-rs/tui/src/style.rs
+22
View File
@@ -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,