openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Paginate transcript history in the TUI (#36948)

Browse Source 
## Why

Paginated threads should not require the TUI to load and render their entire
history when a session is resumed.

## What changed

- Hydrate a bounded initial history page for paginated resumes and forks,
  respecting the configured terminal reflow row budget.
- Load older transcript items in the background as users navigate upward, while
  preserving the viewport, live output, and backtrack selection.
- Reconcile review-mode prompts across page boundaries and keep restored token
  usage hidden until it can be attributed to a persisted turn.

## Testing

- Added coverage for paginated resume and fork hydration, older-page merging,
  transcript prepending, backtrack behavior, review prompt filtering, and token
  usage restoration.

GitOrigin-RevId: 42160b943a9f6a1ba0cfe4b195c24ddf410b747d
This commit is contained in:
Felipe Coury
2026-08-04 18:59:32 +00:00
committed by copyberry
parent 989a0b053e
commit 8bfa49e350
26 changed files with 1360 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/thread_lifecycle.rs
+6 -3
View File
@@ -581,9 +581,6 @@ pub(super) async fn handle_pending_thread_resume_request(
thread_status.clone(),
has_live_in_progress_turn,
);
let token_usage_turn_id = pending
.include_turns
.then(|| restored_token_usage_turn_id(&pending.history_items, thread.turns.as_slice()));
let mut initial_turns_page = if let Some(mut page) = pending.paginated_initial_turns_page.take()
{
if let (Some(active_turn), Some(params)) =
@@ -623,6 +620,12 @@ pub(super) async fn handle_pending_thread_resume_request(
} else {
None
};
let token_usage_turn_id = pending
.include_turns
.then(|| restored_token_usage_turn_id(&pending.history_items, thread.turns.as_slice()));
if pending.initial_turns_page.is_none() {
initial_turns_page = None;
}
if pending.redact_resume_payloads {
redact_thread_resume_payloads(&mut thread.turns);
if let Some(initial_turns_page) = initial_turns_page.as_mut() {
codex-rs/app-server/src/request_processors/thread_processor.rs
+24 -21
View File
@@ -3156,8 +3156,6 @@ impl ThreadRequestProcessor {
matches!(thread.history_mode, ThreadHistoryMode::Paginated).then_some(thread.thread_id)
});
let paginated_resume = paginated_thread_id.is_some();
let needs_paginated_projection =
paginated_resume && (include_turns || initial_turns_page.is_some());
let history_cwd = thread_history.session_cwd();
let runtime_workspace_roots = runtime_workspace_roots.map(resolve_runtime_workspace_roots);
@@ -3246,7 +3244,7 @@ impl ThreadRequestProcessor {
// Paginated JSONL is canonical, but its SQLite projection can lag after a
// previous write failure. Persist after reopening the live writer so legacy
// response hydration reads the latest durable turns and items.
if needs_paginated_projection
if paginated_resume
&& let Err(error) = self
.thread_store
.persist_thread(thread_id)
@@ -3342,12 +3340,6 @@ impl ThreadRequestProcessor {
let active_permission_profile = thread_response_active_permission_profile(
config_snapshot.active_permission_profile,
);
let token_usage_turn_id = include_turns.then(|| {
restored_token_usage_turn_id(
response_history.get_rollout_items(),
thread.turns.as_slice(),
)
});
let mut initial_turns_page = if let Some(params) = initial_turns_page.as_ref() {
let initial_turns_page_result = if paginated_resume {
self.paginated_resume_initial_turns_page(thread_id, params)
@@ -3371,6 +3363,18 @@ impl ThreadRequestProcessor {
} else {
None
};
let token_usage_turn_id = (include_turns || paginated_resume)
.then(|| {
let turns = if thread.turns.is_empty() {
initial_turns_page
.as_ref()
.map_or(&[][..], |page| page.data.as_slice())
} else {
thread.turns.as_slice()
};
restored_token_usage_turn_id(response_history.get_rollout_items(), turns)
})
.filter(|turn_id| !turn_id.is_empty());
if redact_resume_payloads {
redact_thread_resume_payloads(&mut thread.turns);
if let Some(initial_turns_page) = initial_turns_page.as_mut() {
@@ -3583,6 +3587,12 @@ impl ThreadRequestProcessor {
)
.await?;
}
if paginated_resume && (include_turns || params.initial_turns_page.is_some()) {
self.thread_store
.persist_thread(existing_thread_id)
.await
.map_err(thread_store_resume_read_error)?;
}
let history_items = if needs_history {
source_thread
.history
@@ -3641,15 +3651,6 @@ impl ThreadRequestProcessor {
.thread_goal_processor
.pending_resume_goal_state(existing_thread.as_ref())
.await;
if paginated_resume && (include_turns || params.initial_turns_page.is_some()) {
// Paginated JSONL is canonical, but its SQLite projection can lag after a
// previous write failure. Persist before legacy response hydration reads the
// latest durable turns and items.
self.thread_store
.persist_thread(existing_thread_id)
.await
.map_err(thread_store_resume_read_error)?;
}
let paginated_turns = if paginated_resume && include_turns {
Some(self.paginated_thread_full_turns(existing_thread_id).await?)
} else {
@@ -4229,7 +4230,7 @@ impl ThreadRequestProcessor {
};
let ephemeral_token_usage_turn_id = (ephemeral && include_turns)
.then(|| restored_token_usage_turn_id(&history_items, ephemeral_turns.as_slice()));
let paginated_history_items = paginated_source.then(|| Arc::clone(&history_items));
let token_usage_history_items = paginated_source.then(|| Arc::clone(&history_items));
let new_thread = if let Some(prepared_fork) = prepared_fork {
self.thread_manager
@@ -4358,7 +4359,9 @@ impl ThreadRequestProcessor {
restored_token_usage_turn_id(
history
.as_ref()
.map_or(&[], |history| history.items.as_slice()),
.map(|history| history.items.as_slice())
.or_else(|| token_usage_history_items.as_deref().map(Vec::as_slice))
.unwrap_or(&[]),
thread.turns.as_slice(),
)
});
@@ -4379,7 +4382,7 @@ impl ThreadRequestProcessor {
if paginated_source && include_turns {
thread.turns = self.paginated_thread_full_turns(thread_id).await?;
token_usage_turn_id = Some(restored_token_usage_turn_id(
paginated_history_items
token_usage_history_items
.as_deref()
.map_or(&[], Vec::as_slice),
thread.turns.as_slice(),
codex-rs/app-server/src/request_processors/token_usage_replay.rs
+14
View File
@@ -79,6 +79,10 @@ fn latest_token_usage_turn_id_from_rollout_items(
builder.handle_rollout_item(item);
}
if turns.is_empty() {
return builder.active_turn_id_if_explicit();
}
let active_turn_id = builder.active_turn_id()?;
if turns.iter().any(|turn| turn.id == active_turn_id) {
Some(active_turn_id.to_string())
@@ -137,6 +141,16 @@ mod tests {
);
}
#[test]
fn replay_attribution_rejects_suffix_generated_turn_ids() {
let rollout_items = token_usage_history();
assert_eq!(
latest_token_usage_turn_id_from_rollout_items(&rollout_items, /*turns*/ &[]),
None
);
}
#[test]
fn replay_attribution_uses_latest_token_count_and_ignores_tail_turn() {
let mut rollout_items = token_usage_history();
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+237 -15
View File
@@ -57,6 +57,8 @@ use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::ThreadStatusChangedNotification;
use codex_app_server_protocol::ThreadTurnsListParams;
use codex_app_server_protocol::ThreadTurnsListResponse;
use codex_app_server_protocol::ThreadUnsubscribeParams;
use codex_app_server_protocol::TurnItemsView;
use codex_app_server_protocol::TurnStartParams;
@@ -105,6 +107,8 @@ use core_test_support::responses;
use core_test_support::skip_if_no_network;
use core_test_support::skip_if_remote;
use core_test_support::skip_if_wine_exec;
use core_test_support::streaming_sse::StreamingSseChunk;
use core_test_support::streaming_sse::start_streaming_sse_server;
use pretty_assertions::assert_eq;
use serde_json::json;
use std::fs::FileTimes;
@@ -114,6 +118,7 @@ use std::path::PathBuf;
use std::process::Command;
use std::time::Duration;
use tempfile::TempDir;
use tokio::sync::oneshot;
use tokio::time::timeout;
use uuid::Uuid;
use wiremock::Mock;
@@ -2392,6 +2397,183 @@ async fn thread_resume_emits_restored_token_usage_before_next_turn() -> Result<(
Ok(())
}
#[tokio::test]
async fn cold_paginated_resume_restores_usage_without_loading_turns() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
mock_responses_config(&server.uri()).write(codex_home.path())?;
let conversation_id = create_fake_paginated_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let path = rollout_path(codex_home.path(), "2025-01-05T12-00-00", &conversation_id);
let canonical_turn_id = "persisted-token-usage-turn";
append_rollout_item_to_path(
&path,
&RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
turn_id: canonical_turn_id.to_string(),
trace_id: None,
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
})),
)
.await?;
append_rollout_item_to_path(
&path,
&RolloutItem::EventMsg(EventMsg::TokenCount(TokenCountEvent {
info: Some(TokenUsageInfo {
total_token_usage: TokenUsage {
input_tokens: 120,
output_tokens: 30,
total_tokens: 150,
..Default::default()
},
last_token_usage: TokenUsage {
input_tokens: 70,
output_tokens: 20,
total_tokens: 90,
..Default::default()
},
model_context_window: Some(200_000),
}),
rate_limits: None,
})),
)
.await?;
let mut app_server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let resume_id = app_server
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
exclude_turns: true,
..Default::default()
})
.await?;
let ThreadResumeResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(resume_id)).await??;
assert!(thread.turns.is_empty());
let notification = timeout(
DEFAULT_READ_TIMEOUT,
app_server.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await??;
let ServerNotification::ThreadTokenUsageUpdated(notification) = notification.try_into()? else {
panic!("expected thread/tokenUsage/updated notification");
};
assert_eq!(notification.thread_id, thread.id);
assert_eq!(notification.turn_id, canonical_turn_id);
assert_eq!(notification.token_usage.total.total_tokens, 150);
let turns_id = app_server
.send_thread_turns_list_request(ThreadTurnsListParams {
thread_id: thread.id,
cursor: None,
limit: Some(1),
sort_direction: Some(SortDirection::Desc),
items_view: Some(TurnItemsView::NotLoaded),
})
.await?;
let turns: ThreadTurnsListResponse =
timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(turns_id)).await??;
assert_eq!(notification.turn_id, turns.data[0].id);
Ok(())
}
#[tokio::test]
async fn cold_paginated_resume_omits_usage_when_its_turn_is_ambiguous() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
mock_responses_config(&server.uri()).write(codex_home.path())?;
let filename_ts = "2025-01-05T12-00-00";
let conversation_id = create_fake_paginated_rollout(
codex_home.path(),
filename_ts,
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
/*git_info*/ None,
)?;
let path = rollout_path(codex_home.path(), filename_ts, &conversation_id);
append_rollout_item_to_path(
&path,
&RolloutItem::EventMsg(EventMsg::TokenCount(TokenCountEvent {
info: Some(TokenUsageInfo {
total_token_usage: TokenUsage {
total_tokens: 150,
..Default::default()
},
last_token_usage: TokenUsage {
total_tokens: 90,
..Default::default()
},
model_context_window: Some(200_000),
}),
rate_limits: None,
})),
)
.await?;
let interrupted_turn_id = "interrupted-turn-after-token-usage";
append_rollout_item_to_path(
&path,
&RolloutItem::EventMsg(EventMsg::TurnStarted(TurnStartedEvent {
turn_id: interrupted_turn_id.to_string(),
trace_id: None,
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
})),
)
.await?;
let mut app_server = TestAppServer::builder()
.with_codex_home(codex_home.path())
.build_initialized()
.await?;
let resume_id = app_server
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
exclude_turns: true,
..Default::default()
})
.await?;
let ThreadResumeResponse { thread, .. } =
timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(resume_id)).await??;
assert!(thread.turns.is_empty());
let turns_id = app_server
.send_thread_turns_list_request(ThreadTurnsListParams {
thread_id: thread.id,
cursor: None,
limit: Some(1),
sort_direction: Some(SortDirection::Desc),
items_view: Some(TurnItemsView::NotLoaded),
})
.await?;
let turns: ThreadTurnsListResponse =
timeout(DEFAULT_READ_TIMEOUT, app_server.read_response(turns_id)).await??;
assert_eq!(turns.data[0].id, interrupted_turn_id);
assert!(
timeout(
Duration::from_millis(100),
app_server.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await
.is_err(),
"usage owned by an implicit turn must not be attributed to {interrupted_turn_id}"
);
Ok(())
}
#[tokio::test]
async fn thread_resume_skips_restored_token_usage_when_turns_are_excluded() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
@@ -3437,22 +3619,28 @@ async fn thread_resume_rejects_mismatched_path_for_running_thread_id() -> Result
#[tokio::test]
async fn thread_resume_rejoins_running_paginated_thread_with_initial_page() -> Result<()> {
let server = responses::start_mock_server().await;
let first_response = responses::sse_response(responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_assistant_message("msg-1", "Done"),
responses::ev_completed("resp-1"),
]));
let second_response = responses::sse_response(responses::sse(vec![
responses::ev_response_created("resp-2"),
responses::ev_assistant_message("msg-2", "Done"),
responses::ev_completed("resp-2"),
]))
.set_delay(std::time::Duration::from_secs(2));
let _response_mock =
responses::mount_response_sequence(&server, vec![first_response, second_response]).await;
let (release_running_turn, running_turn_gate) = oneshot::channel();
let (server, _response_completions) = start_streaming_sse_server(vec![
vec![StreamingSseChunk {
gate: None,
body: responses::sse(vec![
responses::ev_response_created("resp-1"),
responses::ev_assistant_message("msg-1", "Done"),
responses::ev_completed("resp-1"),
]),
}],
vec![StreamingSseChunk {
gate: Some(running_turn_gate),
body: responses::sse(vec![
responses::ev_response_created("resp-2"),
responses::ev_assistant_message("msg-2", "Done"),
responses::ev_completed("resp-2"),
]),
}],
])
.await;
let codex_home = TempDir::new()?;
mock_responses_config(&server.uri()).write(codex_home.path())?;
mock_responses_config(server.uri()).write(codex_home.path())?;
let mut primary = TestAppServer::builder()
.with_codex_home(codex_home.path())
@@ -3546,6 +3734,36 @@ async fn thread_resume_rejoins_running_paginated_thread_with_initial_page() -> R
assert!(initial_turns_page.backwards_cursor.is_some());
assert!(initial_turns_page.next_cursor.is_some());
timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await??;
let metadata_resume_id = primary
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id.clone(),
exclude_turns: true,
..Default::default()
})
.await?;
let metadata_resume: ThreadResumeResponse = timeout(
DEFAULT_READ_TIMEOUT,
primary.read_response(metadata_resume_id),
)
.await??;
assert!(metadata_resume.thread.turns.is_empty());
assert!(metadata_resume.initial_turns_page.is_none());
assert!(
timeout(
Duration::from_millis(100),
primary.read_stream_until_notification_message("thread/tokenUsage/updated"),
)
.await
.is_err(),
"hot paginated resume should wait for a real token usage update"
);
let asc_resume_id = primary
.send_thread_resume_request(ThreadResumeParams {
thread_id: thread.id.clone(),
@@ -3573,11 +3791,15 @@ async fn thread_resume_rejoins_running_paginated_thread_with_initial_page() -> R
status => panic!("unexpected thread status after running resume: {status:?}"),
}
release_running_turn
.send(())
.expect("release the running model response");
timeout(
DEFAULT_READ_TIMEOUT,
primary.read_stream_until_notification_message("turn/completed"),
)
.await??;
server.shutdown().await;
Ok(())
}
codex-rs/tui/src/app.rs
+4 -1
View File
@@ -211,6 +211,7 @@ pub(crate) mod app_server_requests;
mod background_requests;
mod config_persistence;
mod event_dispatch;
mod history_pagination;
mod history_ui;
mod input;
mod loaded_threads;
@@ -1308,7 +1309,9 @@ See the Codex keymap documentation for supported actions and examples."
};
if self.overlay.is_some() {
let _ = self.handle_backtrack_overlay_event(tui, event).await?;
let _ = self
.handle_backtrack_overlay_event(tui, app_server, event)
.await?;
} else {
match event {
TuiEvent::Key(key_event) => {
codex-rs/tui/src/app/event_dispatch.rs
+21 -3
View File
@@ -33,6 +33,19 @@ impl App {
self.handle_startup_thread_started(app_server, result)
.await?;
}
AppEvent::OlderThreadHistoryLoaded {
thread_id,
cursor,
result,
} => {
if let Err(err) = self
.handle_older_history_page(tui, app_server, thread_id, &cursor, result)
.await
{
app_server.cancel_older_history_page(thread_id);
tracing::warn!(%thread_id, error = %err, "failed to load older transcript history");
}
}
AppEvent::ClearUi { name } => {
self.clear_terminal_ui(tui, /*redraw_header*/ false)?;
self.reset_app_ui_state_after_clear();
@@ -278,18 +291,23 @@ impl App {
nth_user_message,
&mut prompt,
) {
Ok(Some(before_turn_id)) => {
Ok(before_turn_id)
if before_turn_id.is_some()
|| app_server.has_older_history(thread_id) =>
{
let before_turn_id = before_turn_id
.or_else(|| thread.turns.first().map(|turn| turn.id.clone()));
app_server
.fork_thread_at(
config.clone(),
thread_id,
/*last_turn_id*/ None,
/*before_turn_id*/ Some(before_turn_id),
before_turn_id,
ForkGoalContinuation::StartIfIdle,
)
.await
}
Ok(None) => {
Ok(_) => {
app_server
.start_thread_with_session_start_source(
&config, /*session_start_source*/ None,
codex-rs/tui/src/app/history_pagination.rs
+186
View File
@@ -0,0 +1,186 @@
//! Load older transcript pages without rewriting terminal-native scrollback.
use std::collections::HashSet;
use super::*;
use crate::history_cell::UserHistoryCell;
use crate::thread_transcript::RawReasoningVisibility;
use crate::thread_transcript::thread_items_to_transcript_cells;
use codex_app_server_protocol::ThreadItemsListResponse;
impl App {
pub(super) async fn handle_older_history_page(
&mut self,
tui: &mut tui::Tui,
app_server: &mut AppServerSession,
thread_id: ThreadId,
cursor: &str,
result: Result<ThreadItemsListResponse, String>,
) -> Result<()> {
if self.chat_widget.thread_id() != Some(thread_id)
|| !matches!(self.overlay, Some(Overlay::Transcript(_)))
{
app_server.cancel_older_history_page(thread_id);
return Ok(());
}
let page = result.map_err(|err| color_eyre::eyre::eyre!(err))?;
let Some(store) = self
.thread_event_channels
.get(&thread_id)
.map(|channel| Arc::clone(&channel.store))
else {
app_server.cancel_older_history_page(thread_id);
return Ok(());
};
let (cwd, mut turns) = {
let store = store.lock().await;
(
store
.session
.as_ref()
.map_or_else(|| self.config.cwd.clone(), |session| session.cwd.clone()),
store.turns.clone(),
)
};
let mut items = app_server
.apply_older_history_page(thread_id, cursor, page, &mut turns)
.await?;
let mut hidden_item_ids = HashSet::new();
let mut review_mode = false;
for (index, turn) in turns.iter().enumerate() {
let hidden_nested_review_turn = index
.checked_sub(/*rhs*/ 1)
.and_then(|previous| turns.get(previous))
.is_some_and(|previous| {
crate::app_backtrack::is_hidden_nested_review_turn(previous, turn)
});
for item in &turn.items {
match item {
ThreadItem::EnteredReviewMode { .. } => review_mode = true,
ThreadItem::ExitedReviewMode { .. } => review_mode = false,
ThreadItem::UserMessage { .. } if review_mode || hidden_nested_review_turn => {
hidden_item_ids.insert(item.id());
}
_ => {}
}
}
}
let visibility = if self.config.show_raw_agent_reasoning {
RawReasoningVisibility::Visible
} else {
RawReasoningVisibility::Hidden
};
let width = tui.terminal.last_known_screen_size.width;
if !hidden_item_ids.is_empty() {
let user_items = turns
.iter()
.flat_map(|turn| turn.items.iter())
.filter(|item| matches!(item, ThreadItem::UserMessage { .. }))
.map(|item| (item.id().to_string(), item.clone()))
.collect::<Vec<_>>();
let projected_user_cells = thread_items_to_transcript_cells(
Some(thread_id),
&cwd,
user_items.iter().map(|(_, item)| item.clone()),
visibility,
Some(self.config.codex_home.as_path()),
);
let mut persisted_user_cells = user_items
.into_iter()
.zip(projected_user_cells)
.map(|((item_id, _), cell)| (item_id, cell))
.collect::<Vec<_>>();
let mut hidden_transcript_indices = Vec::new();
for (index, cell) in self.transcript_cells.iter().enumerate().rev() {
let Some(user_cell) = cell.as_any().downcast_ref::<UserHistoryCell>() else {
continue;
};
let Some(position) = persisted_user_cells.iter().rposition(|(_, projected)| {
projected
.as_any()
.downcast_ref::<UserHistoryCell>()
.is_some_and(|projected| {
projected.message == user_cell.message
&& projected.text_elements == user_cell.text_elements
&& projected.local_image_paths == user_cell.local_image_paths
&& projected.remote_image_urls == user_cell.remote_image_urls
})
}) else {
continue;
};
if hidden_item_ids.contains(persisted_user_cells[position].0.as_str()) {
hidden_transcript_indices.push(index);
}
persisted_user_cells.truncate(position);
}
if !hidden_transcript_indices.is_empty() {
if self.backtrack.overlay_preview_active {
let selected_index = crate::app_backtrack::nth_user_position(
&self.transcript_cells,
self.backtrack.nth_user_message,
);
let removed_visible_users = hidden_transcript_indices
.iter()
.filter(|&&index| {
selected_index.is_some_and(|selected| index < selected)
&& self.transcript_cells[index].desired_height(width) != 0
})
.count();
self.backtrack.nth_user_message = self
.backtrack
.nth_user_message
.saturating_sub(removed_visible_users);
}
for index in hidden_transcript_indices {
self.transcript_cells.remove(index);
}
if let Some(Overlay::Transcript(overlay)) = self.overlay.as_mut() {
overlay.replace_cells(self.transcript_cells.clone());
}
}
}
items.retain(|item| !hidden_item_ids.contains(item.id()));
{
let mut store = store.lock().await;
turns.retain_mut(|turn| {
let Some(current) = store.turns.iter_mut().find(|current| current.id == turn.id)
else {
return true;
};
let items = std::mem::take(&mut turn.items)
.into_iter()
.filter(|item| !current.items.iter().any(|known| known.id() == item.id()))
.collect::<Vec<_>>();
current.items.splice(0..0, items);
false
});
store.turns.splice(0..0, turns);
}
let cells = thread_items_to_transcript_cells(
Some(thread_id),
&cwd,
items,
visibility,
Some(self.config.codex_home.as_path()),
);
if self.backtrack.overlay_preview_active {
self.backtrack.nth_user_message = self.backtrack.nth_user_message.saturating_add(
cells
.iter()
.filter(|cell| {
cell.as_any().is::<UserHistoryCell>() && cell.desired_height(width) != 0
})
.count(),
);
}
if let Some(Overlay::Transcript(overlay)) = self.overlay.as_mut() {
let index = overlay.prepend(cells.clone(), width);
self.transcript_cells.splice(index..index, cells);
}
if self.backtrack.overlay_preview_active {
self.apply_backtrack_selection_internal(self.backtrack.nth_user_message);
}
tui.frame_requester().schedule_frame();
Ok(())
}
}
codex-rs/tui/src/app/session_lifecycle.rs
+1
View File
@@ -235,6 +235,7 @@ impl App {
err.chain().any(|cause| {
let message = cause.to_string();
message.contains("includeTurns is unavailable before first user message")
|| message.contains("thread/turns/list is unavailable before first user message")
|| message.contains("ephemeral threads do not support includeTurns")
})
}
codex-rs/tui/src/app/tests/safety_buffering.rs
+10 -2
View File
@@ -314,8 +314,16 @@ fn user_message_count(thread: &Thread, prompt: &str) -> usize {
ThreadItem::UserMessage { content, .. } => Some(content),
_ => None,
})
.flatten()
.filter(|item| matches!(item, AppServerUserInput::Text { text, .. } if text == prompt))
.filter(|content| {
content
.iter()
.filter_map(|item| match item {
AppServerUserInput::Text { text, .. } => Some(text.as_str()),
_ => None,
})
.collect::<String>()
== prompt
})
.count()
}
codex-rs/tui/src/app/tests/session_lifecycle_requests.rs
+208
View File
@@ -1,4 +1,5 @@
use super::*;
use app_test_support::create_fake_paginated_rollout;
use app_test_support::create_fake_parented_rollout_with_source;
use app_test_support::create_fake_rollout;
use app_test_support::rollout_path;
@@ -9,8 +10,19 @@ use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadItemsListParams;
use codex_app_server_protocol::ThreadItemsListResponse;
use codex_app_server_protocol::ThreadStatus;
use codex_protocol::AgentPath;
use codex_protocol::items::AgentMessageContent;
use codex_protocol::items::AgentMessageItem;
use codex_protocol::items::EnteredReviewModeItem;
use codex_protocol::items::ExitedReviewModeItem;
use codex_protocol::items::TurnItem;
use codex_protocol::items::UserMessageItem;
use codex_protocol::protocol::ItemCompletedEvent;
use codex_protocol::protocol::ReviewTarget;
use codex_protocol::user_input::UserInput as CoreUserInput;
use codex_state::SqliteConfig;
use futures::SinkExt;
use futures::StreamExt;
@@ -168,6 +180,202 @@ async fn start_recording_app_server(
))
}
#[tokio::test]
async fn older_pagination_reconciles_review_prompts_across_page_boundaries() -> Result<()> {
let mut app = make_test_app().await;
let codex_home = tempdir()?;
app.config.codex_home = codex_home.path().to_path_buf().abs();
app.config.sqlite = SqliteConfig::new_for_testing(codex_home.path().abs());
let thread_id = create_fake_paginated_rollout(
codex_home.path(),
"2026-01-02T00-00-00",
"2026-01-02T00:00:00Z",
"older visible prompt",
Some(app.config.model_provider_id.as_str()),
/*git_info*/ None,
)
.map_err(|error| color_eyre::eyre::eyre!("failed to create paginated rollout: {error}"))?;
let thread_id = ThreadId::from_string(&thread_id)?;
let path = rollout_path(
codex_home.path(),
"2026-01-02T00-00-00",
&thread_id.to_string(),
);
let mut records = std::fs::read_to_string(&path)?
.lines()
.map(serde_json::from_str::<serde_json::Value>)
.collect::<Result<Vec<_>, _>>()?;
let user_item = |id: &str, text: &str| {
TurnItem::UserMessage(UserMessageItem {
id: id.to_string(),
client_id: None,
content: vec![CoreUserInput::Text {
text: text.to_string(),
text_elements: Vec::new(),
}],
})
};
let mut items = vec![
user_item("older-visible-prompt", "older visible prompt"),
TurnItem::EnteredReviewMode(EnteredReviewModeItem {
id: "cross-page-review-start".to_string(),
target: ReviewTarget::UncommittedChanges,
user_facing_hint: "review started".to_string(),
}),
user_item("hidden-review-prompt", "hidden cross-page review prompt"),
];
items.extend((0..97).map(|index| {
TurnItem::AgentMessage(AgentMessageItem {
id: format!("review-output-{index}"),
content: vec![AgentMessageContent::Text {
text: format!("review output {index}"),
}],
phase: None,
memory_citation: None,
})
}));
items.extend([
TurnItem::ExitedReviewMode(ExitedReviewModeItem {
id: "cross-page-review-end".to_string(),
review_output: None,
}),
user_item("newer-visible-prompt", "newer visible prompt"),
]);
let events = std::iter::once(EventMsg::TurnStarted(TurnStartedEvent {
turn_id: "cross-page-review-turn".to_string(),
trace_id: None,
started_at: None,
model_context_window: None,
collaboration_mode_kind: Default::default(),
}))
.chain(items.into_iter().map(|item| {
EventMsg::ItemCompleted(ItemCompletedEvent {
thread_id,
turn_id: "cross-page-review-turn".to_string(),
item,
started_at_ms: None,
completed_at_ms: 0,
})
}));
for event in events {
records.push(serde_json::json!({
"timestamp": "2026-01-02T00:00:00Z",
"ordinal": records.len(),
"type": "event_msg",
"payload": serde_json::to_value(event)?,
}));
}
let records = records
.into_iter()
.map(|record| record.to_string())
.collect::<Vec<_>>()
.join("\n");
std::fs::write(path, format!("{records}\n"))?;
let (mut app_server, _requests, proxy) = start_recording_app_server(
&app.config,
/*blocked_thread_list*/ None,
/*failed_thread_name*/ None,
)
.await?;
let started = app_server
.resume_thread(
app.config.clone(),
thread_id,
crate::app_server_session::ResumeModelSettings::RestoreFromThread,
)
.await?;
let initial_cells = crate::thread_transcript::thread_items_to_transcript_cells(
Some(thread_id),
&app.config.cwd,
started.turns.iter().flat_map(|turn| turn.items.clone()),
crate::thread_transcript::RawReasoningVisibility::Hidden,
Some(app.config.codex_home.as_path()),
);
app.enqueue_primary_thread_session(started.session, started.turns)
.await?;
app.transcript_cells = initial_cells;
assert_eq!(
app.transcript_cells
.iter()
.filter_map(|cell| cell.as_any().downcast_ref::<UserHistoryCell>())
.map(|user| user.message.as_str())
.collect::<Vec<_>>(),
vec!["hidden cross-page review prompt", "newer visible prompt"]
);
app.backtrack.overlay_preview_active = true;
app.backtrack.nth_user_message = 1;
let mut tui = crate::tui::test_support::make_test_tui()?;
app.open_transcript_overlay(&mut tui);
app.apply_backtrack_selection_internal(app.backtrack.nth_user_message);
let cursor = app_server
.begin_older_history_page(thread_id)
.expect("review-mode marker should remain on an older page");
let request_id = app_server.next_request_id();
let page: ThreadItemsListResponse = app_server
.request_handle()
.request_typed(ClientRequest::ThreadItemsList {
request_id,
params: ThreadItemsListParams {
thread_id: thread_id.to_string(),
turn_id: None,
cursor: Some(cursor.clone()),
limit: Some(crate::app_server_session::HISTORY_ITEM_PAGE_LIMIT),
sort_direction: Some(SortDirection::Desc),
},
})
.await?;
app.handle_older_history_page(&mut tui, &mut app_server, thread_id, &cursor, Ok(page))
.await?;
let visible_user_messages = app
.transcript_cells
.iter()
.filter_map(|cell| cell.as_any().downcast_ref::<UserHistoryCell>())
.map(|user| user.message.clone())
.collect::<Vec<_>>();
assert_eq!(
visible_user_messages,
vec!["older visible prompt", "newer visible prompt"]
);
assert_eq!(app.backtrack.nth_user_message, 1);
let area = ratatui::layout::Rect::new(
/*x*/ 0, /*y*/ 0, /*width*/ 80, /*height*/ 12,
);
let mut buffer = Buffer::empty(area);
let Some(Overlay::Transcript(overlay)) = app.overlay.as_mut() else {
panic!("expected a transcript overlay");
};
overlay.render(area, &mut buffer);
let highlighted_output = area
.positions()
.filter(|position| {
buffer[*position]
.style()
.add_modifier
.contains(ratatui::style::Modifier::REVERSED)
})
.map(|position| buffer[position].symbol())
.collect::<String>();
let highlighted_message = visible_user_messages
.iter()
.find(|message| highlighted_output.contains(message.as_str()))
.expect("the selected user message should remain highlighted");
insta::assert_snapshot!(
format!(
"{}\nhighlighted: {highlighted_message}",
visible_user_messages.join("\n")
),
@r"
older visible prompt
newer visible prompt
highlighted: newer visible prompt
");
app_server.shutdown().await?;
proxy.await??;
Ok(())
}
#[test]
fn fresh_session_applies_requested_name() -> Result<()> {
const TEST_STACK_SIZE_BYTES: usize = 8 * 1024 * 1024;
codex-rs/tui/src/app/thread_routing.rs
+3
View File
@@ -1246,6 +1246,9 @@ impl App {
self.activate_thread_channel(thread_id).await;
self.chat_widget
.set_initial_user_message_submit_suppressed(/*suppressed*/ true);
if !turns.is_empty() {
self.chat_widget.set_token_info(/*info*/ None);
}
match presentation {
ThreadAttachPresentation::SessionLineage => {
self.chat_widget.handle_thread_session(session);
codex-rs/tui/src/app_backtrack.rs
+43 -2
View File
@@ -26,6 +26,8 @@ use std::sync::Arc;
use crate::app::App;
use crate::app_event::AppEvent;
use crate::app_server_session::AppServerSession;
use crate::app_server_session::HISTORY_ITEM_PAGE_LIMIT;
use crate::bottom_pane::LocalImageAttachment;
use crate::chatwidget::ChatWidget;
use crate::chatwidget::UserMessage;
@@ -37,7 +39,11 @@ use crate::history_cell::UserHistoryCell;
use crate::pager_overlay::Overlay;
use crate::tui;
use crate::tui::TuiEvent;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadItemsListParams;
use codex_app_server_protocol::ThreadItemsListResponse;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnStatus;
use codex_protocol::ThreadId;
@@ -88,8 +94,43 @@ impl App {
pub(crate) async fn handle_backtrack_overlay_event(
&mut self,
tui: &mut tui::Tui,
app_server: &mut AppServerSession,
event: TuiEvent,
) -> Result<bool> {
if let TuiEvent::Key(key_event) = &event
&& let Some(Overlay::Transcript(overlay)) = self.overlay.as_ref()
&& (overlay.should_load_older(*key_event)
|| (self.backtrack.overlay_preview_active
&& self.backtrack.nth_user_message == 0
&& matches!(key_event.code, KeyCode::Esc | KeyCode::Left)
&& matches!(key_event.kind, KeyEventKind::Press | KeyEventKind::Repeat)))
&& let Some(thread_id) = self.chat_widget.thread_id()
&& let Some(cursor) = app_server.begin_older_history_page(thread_id)
{
let request_id = app_server.next_request_id();
let request_handle = app_server.request_handle();
let app_event_tx = self.app_event_tx.clone();
tokio::spawn(async move {
let result = request_handle
.request_typed::<ThreadItemsListResponse>(ClientRequest::ThreadItemsList {
request_id,
params: ThreadItemsListParams {
thread_id: thread_id.to_string(),
turn_id: None,
cursor: Some(cursor.clone()),
limit: Some(HISTORY_ITEM_PAGE_LIMIT),
sort_direction: Some(SortDirection::Desc),
},
})
.await
.map_err(|err| err.to_string());
app_event_tx.send(AppEvent::OlderThreadHistoryLoaded {
thread_id,
cursor,
result,
});
});
}
if self.backtrack.overlay_preview_active {
match event {
TuiEvent::Key(KeyEvent {
@@ -313,7 +354,7 @@ impl App {
}
/// Apply a computed backtrack selection to the overlay and internal counter.
fn apply_backtrack_selection_internal(&mut self, nth_user_message: usize) {
pub(crate) fn apply_backtrack_selection_internal(&mut self, nth_user_message: usize) {
if let Some(cell_idx) = nth_user_position(&self.transcript_cells, nth_user_message) {
self.backtrack.nth_user_message = nth_user_message;
if let Some(Overlay::Transcript(t)) = &mut self.overlay {
@@ -584,7 +625,7 @@ fn has_backtrack_target(cells: &[Arc<dyn crate::history_cell::HistoryCell>]) ->
user_count(cells) > 0
}
fn nth_user_position(
pub(crate) fn nth_user_position(
cells: &[Arc<dyn crate::history_cell::HistoryCell>],
nth: usize,
) -> Option<usize> {
codex-rs/tui/src/app_event.rs
+8
View File
@@ -32,6 +32,7 @@ use codex_app_server_protocol::PluginUninstallResponse;
use codex_app_server_protocol::SkillsListResponse;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadGoalStatus;
use codex_app_server_protocol::ThreadItemsListResponse;
use codex_connectors::AppInfo;
use codex_file_search::FileMatch;
use codex_message_history::HistoryBatchCursor;
@@ -224,6 +225,13 @@ pub(crate) enum AppEvent {
event: HistoryLookupResponse,
},
/// One background-loaded page of older Ctrl+T transcript history.
OlderThreadHistoryLoaded {
thread_id: ThreadId,
cursor: String,
result: Result<ThreadItemsListResponse, String>,
},
/// Persist a submitted prompt in the cross-session message history.
AppendMessageHistoryEntry {
thread_id: ThreadId,
codex-rs/tui/src/app_server_session.rs
+114 -20
View File
@@ -4,6 +4,9 @@
//! request/response plumbing out of `App` and `ChatWidget`.
mod fs;
mod history;
pub(crate) use history::HISTORY_ITEM_PAGE_LIMIT;
use crate::bottom_pane::FeedbackAudience;
use crate::legacy_core::config::Config;
@@ -70,6 +73,7 @@ use codex_app_server_protocol::ThreadGoalGetResponse;
use codex_app_server_protocol::ThreadGoalSetParams;
use codex_app_server_protocol::ThreadGoalSetResponse;
use codex_app_server_protocol::ThreadGoalStatus;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadInjectItemsParams;
use codex_app_server_protocol::ThreadInjectItemsResponse;
use codex_app_server_protocol::ThreadListParams;
@@ -186,6 +190,7 @@ pub(crate) struct AppServerBootstrap {
pub(crate) struct AppServerSession {
client: AppServerClient,
next_request_id: i64,
history_pagination: HashMap<ThreadId, history::ThreadHistoryPagination>,
remote_cwd_override: Option<PathBuf>,
thread_params_mode: ThreadParamsMode,
thread_settings_update_supported: bool,
@@ -258,6 +263,7 @@ impl AppServerSession {
Self {
client,
next_request_id: 1,
history_pagination: HashMap::new(),
remote_cwd_override: None,
thread_params_mode,
thread_settings_update_supported: true,
@@ -544,22 +550,36 @@ impl AppServerSession {
} else {
self.session_config_with_effective_service_tier(&config)
};
let response: ThreadResumeResponse = self
let exclude_turns = self
.history_pagination
.get(&thread_id)
.is_some_and(|state| state.history_mode == ThreadHistoryMode::Paginated);
let mut response: ThreadResumeResponse = self
.client
.request_typed(ClientRequest::ThreadResume {
request_id,
params: thread_resume_params_from_config(
session_config,
thread_id,
self.thread_params_mode(),
self.remote_cwd_override.as_deref(),
model_settings,
),
params: ThreadResumeParams {
exclude_turns,
..thread_resume_params_from_config(
session_config,
thread_id,
self.thread_params_mode(),
self.remote_cwd_override.as_deref(),
model_settings,
)
},
})
.await
.map_err(|err| {
bootstrap_request_error("thread/resume failed during TUI bootstrap", err)
})?;
self.hydrate_initial_thread_history(
&mut response.thread,
response.turns_backwards_cursor.clone(),
response.items_backwards_cursor.clone(),
Some(&config),
)
.await?;
let fork_parent_title = self
.fork_parent_title_from_app_server(response.thread.forked_from_id.as_deref())
.await;
@@ -629,6 +649,17 @@ impl AppServerSession {
goal_continuation: ForkGoalContinuation,
presentation: ForkPresentation,
) -> Result<AppServerStartedThread> {
let fork_parent = match presentation {
ForkPresentation::Regular => self
.thread_read(thread_id, /*include_turns*/ false)
.await
.ok(),
ForkPresentation::SideConversation => None,
};
let exclude_turns = fork_parent
.as_ref()
.is_some_and(|thread| thread.history_mode == ThreadHistoryMode::Paginated)
|| presentation == ForkPresentation::SideConversation;
let request_id = self.next_request_id();
let session_config = self.session_config_with_effective_service_tier(&config);
let response: ThreadForkResponse = self
@@ -640,7 +671,7 @@ impl AppServerSession {
before_turn_id,
defer_goal_continuation: goal_continuation
== ForkGoalContinuation::DeferUntilNextTurn,
exclude_turns: presentation == ForkPresentation::SideConversation,
exclude_turns,
..thread_fork_params_from_config(
session_config,
thread_id,
@@ -653,15 +684,19 @@ impl AppServerSession {
.map_err(|err| {
bootstrap_request_error("thread/fork failed during TUI bootstrap", err)
})?;
let fork_parent_title = if presentation == ForkPresentation::SideConversation {
None
} else {
self.fork_parent_title_from_app_server(response.thread.forked_from_id.as_deref())
.await
};
let mut response = response;
if presentation == ForkPresentation::Regular && !response.thread.ephemeral {
self.hydrate_initial_thread_history(
&mut response.thread,
/*turn_cursor*/ None,
/*item_cursor*/ None,
Some(&config),
)
.await?;
}
let mut started =
started_thread_from_fork_response(response, &config, self.thread_params_mode()).await?;
started.session.fork_parent_title = fork_parent_title;
started.session.fork_parent_title = fork_parent.and_then(|thread| thread.name);
Ok(started)
}
@@ -753,17 +788,44 @@ impl AppServerSession {
include_turns: bool,
) -> Result<Thread> {
let request_id = self.next_request_id();
let response: ThreadReadResponse = self
let response = self
.client
.request_typed(ClientRequest::ThreadRead {
.request_typed::<ThreadReadResponse>(ClientRequest::ThreadRead {
request_id,
params: ThreadReadParams {
thread_id: thread_id.to_string(),
include_turns,
},
})
.await
.wrap_err("thread/read failed during TUI session lookup")?;
.await;
let mut response: ThreadReadResponse = match response {
Ok(response) => return Ok(response.thread),
Err(TypedRequestError::Server { source, .. })
if include_turns
&& source.message
== "paginated threads do not support thread/read(includeTurns=true)" =>
{
let request_id = self.next_request_id();
self.client
.request_typed(ClientRequest::ThreadRead {
request_id,
params: ThreadReadParams {
thread_id: thread_id.to_string(),
include_turns: false,
},
})
.await
.wrap_err("thread/read failed during TUI session lookup")?
}
Err(err) => return Err(err).wrap_err("thread/read failed during TUI session lookup"),
};
self.hydrate_initial_thread_history(
&mut response.thread,
/*turn_cursor*/ None,
/*item_cursor*/ None,
/*config*/ None,
)
.await?;
Ok(response.thread)
}
@@ -1895,6 +1957,7 @@ mod tests {
use super::*;
use crate::legacy_core::config::ConfigBuilder;
use crate::legacy_core::config::ConfigOverrides;
use app_test_support::create_fake_paginated_rollout;
use app_test_support::create_fake_rollout;
use codex_app_server_protocol::ThreadStatus;
use codex_app_server_protocol::Turn;
@@ -2509,6 +2572,37 @@ mod tests {
Ok(())
}
#[tokio::test]
async fn ephemeral_paginated_fork_skips_unsupported_history_hydration() -> Result<()> {
let codex_home = tempfile::tempdir()?;
let config = build_config(&codex_home).await;
let source_thread_id = ThreadId::from_string(
&create_fake_paginated_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some(config.model_provider_id.as_str()),
/*git_info*/ None,
)
.map_err(|error| {
color_eyre::eyre::eyre!("failed to create paginated rollout: {error}")
})?,
)?;
let mut app_server = crate::start_embedded_app_server_for_picker(&config).await?;
let mut ephemeral_config = config;
ephemeral_config.ephemeral = true;
let fork = app_server
.fork_thread(ephemeral_config, source_thread_id)
.await?;
assert_eq!(fork.session.forked_from_id, Some(source_thread_id));
assert!(fork.turns.is_empty());
app_server.shutdown().await?;
Ok(())
}
#[tokio::test]
async fn config_request_overrides_preserve_implicit_personality_default() {
let temp_dir = tempfile::tempdir().expect("tempdir");
codex-rs/tui/src/app_server_session/history.rs
+239
View File
@@ -0,0 +1,239 @@
//! Bounded app-server transcript loading for resume, fork, and transcript views.
use super::AppServerSession;
use crate::history_cell::HistoryRenderMode;
use crate::legacy_core::config::Config;
use crate::resize_reflow_cap::resize_reflow_max_rows;
use crate::thread_transcript::RawReasoningVisibility;
use crate::thread_transcript::thread_to_transcript_cells;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::SortDirection;
use codex_app_server_protocol::Thread;
use codex_app_server_protocol::ThreadHistoryMode;
use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::ThreadItemsListParams;
use codex_app_server_protocol::ThreadItemsListResponse;
use codex_app_server_protocol::ThreadTurnsListParams;
use codex_app_server_protocol::ThreadTurnsListResponse;
use codex_app_server_protocol::Turn;
use codex_app_server_protocol::TurnItemsView;
use codex_protocol::ThreadId;
use color_eyre::eyre::Result;
use color_eyre::eyre::WrapErr;
pub(crate) const INITIAL_HISTORY_TURN_LIMIT: u32 = 5;
pub(crate) const HISTORY_ITEM_PAGE_LIMIT: u32 = 100;
#[derive(Clone, Debug, Default)]
pub(crate) struct ThreadHistoryPagination {
pub(super) history_mode: ThreadHistoryMode,
next_turn_cursor: Option<String>,
next_item_cursor: Option<String>,
loading_older: bool,
}
impl AppServerSession {
pub(crate) fn has_older_history(&self, thread_id: ThreadId) -> bool {
self.history_pagination
.get(&thread_id)
.is_some_and(|page| page.next_item_cursor.is_some())
}
pub(crate) fn begin_older_history_page(&mut self, thread_id: ThreadId) -> Option<String> {
let page = self.history_pagination.get_mut(&thread_id)?;
if page.loading_older {
return None;
}
let cursor = page.next_item_cursor.clone()?;
page.loading_older = true;
Some(cursor)
}
pub(crate) fn cancel_older_history_page(&mut self, thread_id: ThreadId) {
if let Some(page) = self.history_pagination.get_mut(&thread_id) {
page.loading_older = false;
}
}
pub(crate) async fn apply_older_history_page(
&mut self,
thread_id: ThreadId,
cursor: &str,
page: ThreadItemsListResponse,
turns: &mut Vec<Turn>,
) -> Result<Vec<ThreadItem>> {
let Some(mut state) = self.history_pagination.get(&thread_id).cloned() else {
return Ok(Vec::new());
};
if !state.loading_older || state.next_item_cursor.as_deref() != Some(cursor) {
return Ok(Vec::new());
}
let items = self
.merge_thread_item_page(thread_id, page, &mut state, turns)
.await?;
state.loading_older = false;
self.history_pagination.insert(thread_id, state);
Ok(items)
}
async fn thread_turns_page(
&mut self,
thread_id: ThreadId,
cursor: Option<String>,
) -> Result<ThreadTurnsListResponse> {
let request_id = self.next_request_id();
self.client
.request_typed(ClientRequest::ThreadTurnsList {
request_id,
params: ThreadTurnsListParams {
thread_id: thread_id.to_string(),
cursor,
limit: Some(INITIAL_HISTORY_TURN_LIMIT),
sort_direction: Some(SortDirection::Desc),
items_view: Some(TurnItemsView::NotLoaded),
},
})
.await
.wrap_err("failed to load a bounded thread history page")
}
async fn merge_thread_item_page(
&mut self,
thread_id: ThreadId,
page: ThreadItemsListResponse,
state: &mut ThreadHistoryPagination,
turns: &mut Vec<Turn>,
) -> Result<Vec<ThreadItem>> {
state.next_item_cursor = page.next_cursor;
let mut items = Vec::new();
for entry in page.data {
while !turns.iter().any(|turn| turn.id == entry.turn_id) {
let Some(cursor) = state.next_turn_cursor.take() else {
break;
};
let page = self.thread_turns_page(thread_id, Some(cursor)).await?;
state.next_turn_cursor = page.next_cursor;
turns.splice(0..0, page.data.into_iter().rev());
}
if let Some(turn) = turns.iter_mut().find(|turn| turn.id == entry.turn_id)
&& !turn.items.iter().any(|item| item.id() == entry.item.id())
{
items.push(entry.item.clone());
turn.items.insert(/*index*/ 0, entry.item);
turn.items_view = TurnItemsView::Summary;
}
}
items.reverse();
Ok(items)
}
pub(crate) async fn hydrate_initial_thread_history(
&mut self,
thread: &mut Thread,
turn_cursor: Option<String>,
item_cursor: Option<String>,
config: Option<&Config>,
) -> Result<()> {
let thread_id = ThreadId::from_string(&thread.id)
.wrap_err("invalid thread id in bounded history response")?;
if thread.history_mode == ThreadHistoryMode::Legacy {
if thread.turns.is_empty() {
thread.turns = Box::pin(self.thread_read(thread_id, /*include_turns*/ true))
.await?
.turns;
}
self.history_pagination.entry(thread_id).or_default();
return Ok(());
}
let page = self.thread_turns_page(thread_id, turn_cursor).await?;
thread.turns = page.data.into_iter().rev().collect();
let mut state = ThreadHistoryPagination {
history_mode: ThreadHistoryMode::Paginated,
next_turn_cursor: page.next_cursor,
next_item_cursor: item_cursor,
..ThreadHistoryPagination::default()
};
let row_budget =
config.and_then(|config| resize_reflow_max_rows(config.terminal_resize_reflow));
let item_budget = if config.is_some() && row_budget.is_none() {
None
} else {
Some(HISTORY_ITEM_PAGE_LIMIT as usize)
};
let width = crossterm::terminal::size()
.map(|(width, _)| width.max(/*other*/ 1))
.unwrap_or(/*default*/ 80);
let mut scanned_items = 0;
loop {
let remaining_rows = row_budget.map(|budget| {
budget.saturating_sub(config.map_or(/*default*/ 0, |config| {
rendered_history_rows(thread, config, width)
}))
});
let remaining_items = item_budget.map(|budget| budget.saturating_sub(scanned_items));
if remaining_rows == Some(0) || remaining_items == Some(0) {
break;
}
let limit = remaining_rows
.unwrap_or(HISTORY_ITEM_PAGE_LIMIT as usize)
.min(remaining_items.unwrap_or(HISTORY_ITEM_PAGE_LIMIT as usize))
.min(HISTORY_ITEM_PAGE_LIMIT as usize) as u32;
let request_id = self.next_request_id();
let page: ThreadItemsListResponse = self
.client
.request_typed(ClientRequest::ThreadItemsList {
request_id,
params: ThreadItemsListParams {
thread_id: thread_id.to_string(),
turn_id: None,
cursor: state.next_item_cursor.clone(),
limit: Some(limit),
sort_direction: Some(SortDirection::Desc),
},
})
.await
.wrap_err("failed to load a bounded thread item page")?;
if page.data.is_empty() {
state.next_item_cursor = None;
break;
}
scanned_items = scanned_items.saturating_add(page.data.len());
self.merge_thread_item_page(thread_id, page, &mut state, &mut thread.turns)
.await?;
if state.next_item_cursor.is_none() {
break;
}
}
if config.is_some() {
self.history_pagination.insert(thread_id, state);
}
Ok(())
}
}
fn rendered_history_rows(thread: &Thread, config: &Config, width: u16) -> usize {
if thread.turns.iter().all(|turn| turn.items.is_empty()) {
return 0;
}
let visibility = if config.show_raw_agent_reasoning {
RawReasoningVisibility::Visible
} else {
RawReasoningVisibility::Hidden
};
let mode = if config.tui_raw_output_mode {
HistoryRenderMode::Raw
} else {
HistoryRenderMode::Rich
};
thread_to_transcript_cells(
thread.clone(),
visibility,
Some(config.codex_home.as_path()),
)
.into_iter()
.fold(/*init*/ 0, |rows, cell| {
let height = usize::from(cell.desired_height_for_mode(width, mode));
rows + height + usize::from(height != 0 && rows != 0 && !cell.is_stream_continuation())
})
}
codex-rs/tui/src/bottom_pane/chat_composer.rs
+13 -4
View File
@@ -620,6 +620,7 @@ impl ChatComposer {
flash: None,
context_window_percent: None,
context_window_used_tokens: None,
context_window_pending: false,
collaboration_mode_indicator: None,
goal_status_indicator: None,
ide_context_active: false,
@@ -1353,10 +1354,14 @@ impl ChatComposer {
}
fn right_footer_line_with_context(&self) -> Line<'static> {
let mut line = context_window_line(
self.footer.context_window_percent,
self.footer.context_window_used_tokens,
);
let mut line = if self.footer.context_window_pending {
Line::default()
} else {
context_window_line(
self.footer.context_window_percent,
self.footer.context_window_used_tokens,
)
};
if let Some(vim_mode) = self.vim_mode_indicator_span() {
line.spans.push(" | ".dim());
line.spans.push(vim_mode);
@@ -4218,6 +4223,10 @@ impl ChatComposer {
self.footer.context_window_used_tokens = used_tokens;
}
pub(crate) fn set_context_window_pending(&mut self, pending: bool) {
self.footer.context_window_pending = pending;
}
pub(crate) fn set_esc_backtrack_hint(&mut self, show: bool) {
self.footer.esc_backtrack_hint = show;
if show {
codex-rs/tui/src/bottom_pane/chat_composer/footer_state.rs
+1
View File
@@ -23,6 +23,7 @@ pub(super) struct FooterState {
pub(super) flash: Option<FooterFlash>,
pub(super) context_window_percent: Option<i64>,
pub(super) context_window_used_tokens: Option<i64>,
pub(super) context_window_pending: bool,
pub(super) collaboration_mode_indicator: Option<CollaborationModeIndicator>,
pub(super) goal_status_indicator: Option<GoalStatusIndicator>,
pub(super) ide_context_active: bool,
codex-rs/tui/src/bottom_pane/mod.rs
+5
View File
@@ -1123,6 +1123,11 @@ impl BottomPane {
self.request_redraw();
}
pub(crate) fn set_context_window_pending(&mut self, pending: bool) {
self.composer.set_context_window_pending(pending);
self.request_redraw();
}
/// Show a generic list selection view with the provided items.
pub(crate) fn show_selection_view(
&mut self,
codex-rs/tui/src/chatwidget.rs
+7
View File
@@ -564,6 +564,7 @@ pub(crate) struct ChatWidget {
runtime_model_provider_base_url: Option<String>,
pub(crate) remote_connection: Option<RemoteConnectionStatus>,
token_info: Option<TokenUsageInfo>,
token_usage_pending: bool,
rate_limit_snapshots_by_limit_id: BTreeMap<String, RateLimitSnapshotDisplay>,
refreshing_status_outputs: Vec<(u64, StatusHistoryHandle)>,
next_status_refresh_request_id: u64,
@@ -1144,6 +1145,9 @@ impl ChatWidget {
match info {
Some(info) => self.apply_token_info(info),
None => {
self.token_usage_pending = true;
self.bottom_pane
.set_context_window_pending(/*pending*/ true);
self.bottom_pane
.set_context_window(/*percent*/ None, /*used_tokens*/ None);
self.token_info = None;
@@ -1152,6 +1156,9 @@ impl ChatWidget {
}
fn apply_token_info(&mut self, info: TokenUsageInfo) {
self.token_usage_pending = false;
self.bottom_pane
.set_context_window_pending(/*pending*/ false);
let percent = self.context_remaining_percent(&info);
let used_tokens = self.context_used_tokens(&info, percent.is_some());
self.bottom_pane.set_context_window(percent, used_tokens);
codex-rs/tui/src/chatwidget/constructor.rs
+1
View File
@@ -128,6 +128,7 @@ impl ChatWidget {
runtime_model_provider_base_url,
remote_connection: None,
token_info: None,
token_usage_pending: false,
rate_limit_snapshots_by_limit_id: BTreeMap::new(),
refreshing_status_outputs: Vec::new(),
next_status_refresh_request_id: 0,
codex-rs/tui/src/chatwidget/status_controls.rs
+5 -2
View File
@@ -358,6 +358,9 @@ impl ChatWidget {
}
pub(super) fn status_line_context_remaining_percent(&self) -> Option<i64> {
if self.token_usage_pending {
return None;
}
let Some(context_window) = self.status_line_context_window_size() else {
return Some(100);
};
@@ -375,8 +378,8 @@ impl ChatWidget {
}
pub(super) fn status_line_context_used_percent(&self) -> Option<i64> {
let remaining = self.status_line_context_remaining_percent().unwrap_or(100);
Some((100 - remaining).clamp(0, 100))
self.status_line_context_remaining_percent()
.map(|remaining| (100 - remaining).clamp(0, 100))
}
pub(super) fn status_line_total_usage(&self) -> TokenUsage {
codex-rs/tui/src/chatwidget/status_surfaces.rs
+12 -8
View File
@@ -715,14 +715,18 @@ impl ChatWidget {
StatusLineItem::ContextWindowSize => self
.status_line_context_window_size()
.map(|cws| format!("{} window", format_tokens_compact(cws))),
StatusLineItem::TotalInputTokens => Some(format!(
"{} in",
format_tokens_compact(self.status_line_total_usage().input_tokens)
)),
StatusLineItem::TotalOutputTokens => Some(format!(
"{} out",
format_tokens_compact(self.status_line_total_usage().output_tokens)
)),
StatusLineItem::TotalInputTokens => (!self.token_usage_pending).then(|| {
format!(
"{} in",
format_tokens_compact(self.status_line_total_usage().input_tokens)
)
}),
StatusLineItem::TotalOutputTokens => (!self.token_usage_pending).then(|| {
format!(
"{} out",
format_tokens_compact(self.status_line_total_usage().output_tokens)
)
}),
StatusLineItem::SessionId => self.thread_id.map(|id| id.to_string()),
StatusLineItem::FastMode => Some(
if self.current_service_tier() == Some(ServiceTier::Fast.request_value()) {
codex-rs/tui/src/chatwidget/tests/status_and_layout.rs
+46
View File
@@ -41,6 +41,52 @@ async fn token_count_none_resets_context_indicator() {
assert_eq!(chat.bottom_pane.context_window_percent(), None);
}
#[tokio::test]
async fn resumed_session_hides_unknown_token_usage_until_an_update_arrives() {
let (mut chat, _rx, _ops) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
chat.set_token_info(/*info*/ None);
let width = 80;
let height = chat.desired_height(width);
let mut terminal =
ratatui::Terminal::new(TestBackend::new(width, height)).expect("create terminal");
terminal
.draw(|frame| chat.render(frame.area(), frame.buffer_mut()))
.expect("render resumed session");
let rendered = normalized_backend_snapshot(terminal.backend());
assert!(!rendered.contains("100% context left"));
insta::assert_snapshot!(
rendered
.lines()
.find(|line| line.contains("context left"))
.map(str::trim)
.unwrap_or("(hidden)"),
@"(hidden)"
);
chat.config.tui_status_line = Some(vec![
"context-remaining".to_string(),
"context-used".to_string(),
"total-input-tokens".to_string(),
"total-output-tokens".to_string(),
]);
chat.refresh_status_line();
assert_eq!(status_line_text(&chat), None);
handle_token_count(
&mut chat,
Some(make_token_info(
/*total_tokens*/ 12_700, /*context_window*/ 13_000,
)),
);
chat.refresh_status_line();
assert_eq!(
status_line_text(&chat),
Some("Context 30% left · Context 70% used · 0 in · 0 out".to_string())
);
}
#[tokio::test]
async fn app_server_cyber_policy_error_renders_dedicated_notice() {
let (mut chat, mut rx, _ops) = make_chatwidget_manual(/*model_override*/ None).await;
codex-rs/tui/src/pager_overlay.rs
+119
View File
@@ -20,6 +20,7 @@ use std::sync::Arc;
use crate::chatwidget::ActiveCellTranscriptKey;
use crate::history_cell::HistoryCell;
use crate::history_cell::SessionInfoCell;
use crate::history_cell::UserHistoryCell;
use crate::key_hint;
use crate::key_hint::KeyBinding;
@@ -597,6 +598,57 @@ impl TranscriptOverlay {
}
}
/// Returns whether an upward navigation is close enough to request older history.
pub(crate) fn should_load_older(&self, key_event: KeyEvent) -> bool {
self.view.keymap.jump_top.is_pressed(key_event)
|| (self.view.scroll_offset
<= self.view.last_content_height.unwrap_or(/*default*/ 0)
&& (self.view.keymap.scroll_up.is_pressed(key_event)
|| self.view.keymap.page_up.is_pressed(key_event)
|| self.view.keymap.half_page_up.is_pressed(key_event)))
}
/// Prepends history without moving visible content and returns its insertion index.
pub(crate) fn prepend(&mut self, cells: Vec<Arc<dyn HistoryCell>>, width: u16) -> usize {
if cells.is_empty() {
return 0;
}
let follow_bottom = self.view.is_scrolled_to_bottom();
let previous_height = self.view.content_height(width);
let live_tail = self.take_live_tail_renderable();
let added_cells = cells.len();
let insert_at = self
.cells
.iter()
.rposition(|cell| cell.as_any().is::<SessionInfoCell>())
.map_or(/*default*/ 0, |index| index.saturating_add(/*rhs*/ 1));
self.cells.splice(insert_at..insert_at, cells);
for index in [
&mut self.highlight_cell,
&mut self.view.pending_scroll_chunk,
] {
if let Some(index) = index.as_mut()
&& *index >= insert_at
{
*index = index.saturating_add(added_cells);
}
}
self.rebuild_renderables();
if let Some(live_tail) = live_tail {
self.view.renderables.push(live_tail);
}
self.view.scroll_offset = if follow_bottom {
usize::MAX
} else {
self.view.scroll_offset.saturating_add(
self.view
.content_height(width)
.saturating_sub(previous_height),
)
};
insert_at
}
/// Replace committed transcript cells while keeping any cached in-progress output that is
/// currently shown at the end of the overlay.
///
@@ -1059,6 +1111,18 @@ mod tests {
);
}
#[test]
fn jump_top_requests_older_history_from_the_bottom() {
let overlay = transcript_overlay(vec![Arc::new(TestCell {
lines: vec![Line::from("recent")],
})]);
assert!(overlay.should_load_older(KeyEvent::new(
KeyCode::Home,
crossterm::event::KeyModifiers::NONE,
)));
}
#[test]
fn edit_next_hint_is_visible_when_highlighted() {
let mut overlay = transcript_overlay(vec![Arc::new(TestCell {
@@ -1140,6 +1204,38 @@ mod tests {
assert_snapshot!(term.backend());
}
#[test]
fn transcript_overlay_preserves_live_tail_when_prepending_history() {
let mut overlay = transcript_overlay(vec![Arc::new(TestCell {
lines: vec![Line::from("recent")],
})]);
overlay.sync_live_tail(
/*width*/ 40,
Some(ActiveCellTranscriptKey {
revision: 1,
is_stream_continuation: false,
animation_tick: None,
}),
|_| Some(vec![HyperlinkLine::from("live tail")]),
);
overlay.prepend(
vec![Arc::new(TestCell {
lines: vec![Line::from("older")],
})],
/*width*/ 40,
);
let area = Rect::new(
/*x*/ 0, /*y*/ 0, /*width*/ 40, /*height*/ 10,
);
let mut buffer = Buffer::empty(area);
overlay.render(area, &mut buffer);
let rendered = buffer_to_text(&buffer, area);
assert!(rendered.contains("older"));
assert!(rendered.contains("recent"));
assert!(rendered.contains("live tail"));
}
#[test]
fn transcript_overlay_live_tail_preserves_semantic_web_links() {
let destination = "https://example.com/a/streamed/path";
@@ -1326,6 +1422,29 @@ mod tests {
}));
assert_eq!(overlay.view.scroll_offset, 0);
overlay.view.scroll_offset = 3;
term.draw(|frame| overlay.render(frame.area(), frame.buffer_mut()))
.expect("draw");
let content_area = Rect::new(
/*x*/ 0, /*y*/ 1, /*width*/ 40, /*height*/ 4,
);
let visible_before = buffer_to_text(term.backend().buffer(), content_area);
overlay.prepend(
vec![Arc::new(TestCell {
lines: vec!["older first".into(), "older second".into()],
})],
/*width*/ 40,
);
term.draw(|frame| overlay.render(frame.area(), frame.buffer_mut()))
.expect("draw");
assert_eq!(
buffer_to_text(term.backend().buffer(), content_area),
visible_before
);
assert_snapshot!(
"transcript_overlay_prepended_history",
visible_before.trim()
);
}
#[test]
codex-rs/tui/src/snapshots/codex_tui__pager_overlay__tests__transcript_overlay_prepended_history.snap
+7
View File
@@ -0,0 +1,7 @@
---
source: tui/src/pager_overlay.rs
expression: visible_before.trim()
---
line2
line3
codex-rs/tui/src/thread_transcript.rs
+26 -9
View File
@@ -17,6 +17,7 @@ use codex_app_server_protocol::ThreadItem;
use codex_app_server_protocol::UserInput;
use codex_protocol::ThreadId;
use codex_protocol::items::UserMessageItem;
use codex_utils_absolute_path::AbsolutePathBuf;
use ratatui::style::Stylize as _;
use ratatui::text::Line;
@@ -51,13 +52,34 @@ pub(crate) fn thread_to_transcript_cells(
codex_home: Option<&std::path::Path>,
) -> TranscriptCells {
let cwd = thread.cwd;
let thread_id = ThreadId::from_string(&thread.id).ok();
let mut cells = thread_items_to_transcript_cells(
thread_id,
&cwd,
thread.turns.into_iter().flat_map(|turn| turn.items),
raw_reasoning_visibility,
codex_home,
);
if cells.is_empty() {
cells.push(Arc::new(PlainHistoryCell::new(vec![
"No transcript content available".italic().dim().into(),
])));
}
cells
}
pub(crate) fn thread_items_to_transcript_cells(
thread_id: Option<ThreadId>,
cwd: &AbsolutePathBuf,
items: impl IntoIterator<Item = ThreadItem>,
raw_reasoning_visibility: RawReasoningVisibility,
codex_home: Option<&std::path::Path>,
) -> TranscriptCells {
let inline_visualization_context = codex_home.and_then(|codex_home| {
ThreadId::from_string(&thread.id)
.ok()
.and_then(|thread_id| InlineVisualizationContext::new(codex_home, thread_id))
thread_id.and_then(|thread_id| InlineVisualizationContext::new(codex_home, thread_id))
});
let mut cells: TranscriptCells = Vec::new();
for item in thread.turns.into_iter().flat_map(|turn| turn.items) {
for item in items {
match item {
ThreadItem::UserMessage {
id,
@@ -135,11 +157,6 @@ pub(crate) fn thread_to_transcript_cells(
}
}
}
if cells.is_empty() {
cells.push(Arc::new(PlainHistoryCell::new(vec![
"No transcript content available".italic().dim().into(),
])));
}
cells
}