diff --git a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs index f661fdd312..1c9f1668da 100644 --- a/codex-rs/app-server/src/request_processors/thread_lifecycle.rs +++ b/codex-rs/app-server/src/request_processors/thread_lifecycle.rs @@ -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() { diff --git a/codex-rs/app-server/src/request_processors/thread_processor.rs b/codex-rs/app-server/src/request_processors/thread_processor.rs index ed72571f40..b296a4e33d 100644 --- a/codex-rs/app-server/src/request_processors/thread_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_processor.rs @@ -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(), diff --git a/codex-rs/app-server/src/request_processors/token_usage_replay.rs b/codex-rs/app-server/src/request_processors/token_usage_replay.rs index 8a44d886e8..50a59aa659 100644 --- a/codex-rs/app-server/src/request_processors/token_usage_replay.rs +++ b/codex-rs/app-server/src/request_processors/token_usage_replay.rs @@ -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(); diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index a4410da23b..341e367d00 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -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(()) } diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index 60649f0d16..047cdd1736 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -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) => { diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index 2e7d60e26b..26034a4d8e 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -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, diff --git a/codex-rs/tui/src/app/history_pagination.rs b/codex-rs/tui/src/app/history_pagination.rs new file mode 100644 index 0000000000..e83296d632 --- /dev/null +++ b/codex-rs/tui/src/app/history_pagination.rs @@ -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, + ) -> 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::>(); + 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::>(); + 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::() else { + continue; + }; + let Some(position) = persisted_user_cells.iter().rposition(|(_, projected)| { + projected + .as_any() + .downcast_ref::() + .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::>(); + 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::() && 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(()) + } +} diff --git a/codex-rs/tui/src/app/session_lifecycle.rs b/codex-rs/tui/src/app/session_lifecycle.rs index 6486679e3d..608febaa27 100644 --- a/codex-rs/tui/src/app/session_lifecycle.rs +++ b/codex-rs/tui/src/app/session_lifecycle.rs @@ -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") }) } diff --git a/codex-rs/tui/src/app/tests/safety_buffering.rs b/codex-rs/tui/src/app/tests/safety_buffering.rs index 6e0e710f21..42fc6bce46 100644 --- a/codex-rs/tui/src/app/tests/safety_buffering.rs +++ b/codex-rs/tui/src/app/tests/safety_buffering.rs @@ -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::() + == prompt + }) .count() } diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index b3aa550b09..d8af260df4 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -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::) + .collect::, _>>()?; + 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::>() + .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::()) + .map(|user| user.message.as_str()) + .collect::>(), + 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::()) + .map(|user| user.message.clone()) + .collect::>(); + 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::(); + 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; diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index f1765b8171..433f3fb371 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -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); diff --git a/codex-rs/tui/src/app_backtrack.rs b/codex-rs/tui/src/app_backtrack.rs index 609e3e6b1e..b85769b45b 100644 --- a/codex-rs/tui/src/app_backtrack.rs +++ b/codex-rs/tui/src/app_backtrack.rs @@ -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 { + 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::(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]) -> user_count(cells) > 0 } -fn nth_user_position( +pub(crate) fn nth_user_position( cells: &[Arc], nth: usize, ) -> Option { diff --git a/codex-rs/tui/src/app_event.rs b/codex-rs/tui/src/app_event.rs index 6037b7225b..21b2e61199 100644 --- a/codex-rs/tui/src/app_event.rs +++ b/codex-rs/tui/src/app_event.rs @@ -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, + }, + /// Persist a submitted prompt in the cross-session message history. AppendMessageHistoryEntry { thread_id: ThreadId, diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 75c5c16392..3c8db59e47 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -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, remote_cwd_override: Option, 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 { + 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 { let request_id = self.next_request_id(); - let response: ThreadReadResponse = self + let response = self .client - .request_typed(ClientRequest::ThreadRead { + .request_typed::(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"); diff --git a/codex-rs/tui/src/app_server_session/history.rs b/codex-rs/tui/src/app_server_session/history.rs new file mode 100644 index 0000000000..01cf943da3 --- /dev/null +++ b/codex-rs/tui/src/app_server_session/history.rs @@ -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, + next_item_cursor: Option, + 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 { + 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, + ) -> Result> { + 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, + ) -> Result { + 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, + ) -> Result> { + 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, + item_cursor: Option, + 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()) + }) +} diff --git a/codex-rs/tui/src/bottom_pane/chat_composer.rs b/codex-rs/tui/src/bottom_pane/chat_composer.rs index af5ec97452..fe3cd178c5 100644 --- a/codex-rs/tui/src/bottom_pane/chat_composer.rs +++ b/codex-rs/tui/src/bottom_pane/chat_composer.rs @@ -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 { diff --git a/codex-rs/tui/src/bottom_pane/chat_composer/footer_state.rs b/codex-rs/tui/src/bottom_pane/chat_composer/footer_state.rs index 3ad87a1462..fd63d2d000 100644 --- a/codex-rs/tui/src/bottom_pane/chat_composer/footer_state.rs +++ b/codex-rs/tui/src/bottom_pane/chat_composer/footer_state.rs @@ -23,6 +23,7 @@ pub(super) struct FooterState { pub(super) flash: Option, pub(super) context_window_percent: Option, pub(super) context_window_used_tokens: Option, + pub(super) context_window_pending: bool, pub(super) collaboration_mode_indicator: Option, pub(super) goal_status_indicator: Option, pub(super) ide_context_active: bool, diff --git a/codex-rs/tui/src/bottom_pane/mod.rs b/codex-rs/tui/src/bottom_pane/mod.rs index e6455e2eb3..527e510aa7 100644 --- a/codex-rs/tui/src/bottom_pane/mod.rs +++ b/codex-rs/tui/src/bottom_pane/mod.rs @@ -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, diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 92eae23843..7151ed0d50 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -564,6 +564,7 @@ pub(crate) struct ChatWidget { runtime_model_provider_base_url: Option, pub(crate) remote_connection: Option, token_info: Option, + token_usage_pending: bool, rate_limit_snapshots_by_limit_id: BTreeMap, 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); diff --git a/codex-rs/tui/src/chatwidget/constructor.rs b/codex-rs/tui/src/chatwidget/constructor.rs index 99430339bc..06ccca7313 100644 --- a/codex-rs/tui/src/chatwidget/constructor.rs +++ b/codex-rs/tui/src/chatwidget/constructor.rs @@ -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, diff --git a/codex-rs/tui/src/chatwidget/status_controls.rs b/codex-rs/tui/src/chatwidget/status_controls.rs index 8bae8b64fb..8a78601ea7 100644 --- a/codex-rs/tui/src/chatwidget/status_controls.rs +++ b/codex-rs/tui/src/chatwidget/status_controls.rs @@ -358,6 +358,9 @@ impl ChatWidget { } pub(super) fn status_line_context_remaining_percent(&self) -> Option { + 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 { - 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 { diff --git a/codex-rs/tui/src/chatwidget/status_surfaces.rs b/codex-rs/tui/src/chatwidget/status_surfaces.rs index fa17bb385b..5c9343c697 100644 --- a/codex-rs/tui/src/chatwidget/status_surfaces.rs +++ b/codex-rs/tui/src/chatwidget/status_surfaces.rs @@ -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()) { diff --git a/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs b/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs index 1fe5065eeb..7532551d55 100644 --- a/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs +++ b/codex-rs/tui/src/chatwidget/tests/status_and_layout.rs @@ -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; diff --git a/codex-rs/tui/src/pager_overlay.rs b/codex-rs/tui/src/pager_overlay.rs index f149bd0744..c32c1b4ced 100644 --- a/codex-rs/tui/src/pager_overlay.rs +++ b/codex-rs/tui/src/pager_overlay.rs @@ -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>, 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::()) + .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] diff --git a/codex-rs/tui/src/snapshots/codex_tui__pager_overlay__tests__transcript_overlay_prepended_history.snap b/codex-rs/tui/src/snapshots/codex_tui__pager_overlay__tests__transcript_overlay_prepended_history.snap new file mode 100644 index 0000000000..e7602f1517 --- /dev/null +++ b/codex-rs/tui/src/snapshots/codex_tui__pager_overlay__tests__transcript_overlay_prepended_history.snap @@ -0,0 +1,7 @@ +--- +source: tui/src/pager_overlay.rs +expression: visible_before.trim() +--- +line2 + +line3 diff --git a/codex-rs/tui/src/thread_transcript.rs b/codex-rs/tui/src/thread_transcript.rs index c7b7a2bb8a..a2f74e7933 100644 --- a/codex-rs/tui/src/thread_transcript.rs +++ b/codex-rs/tui/src/thread_transcript.rs @@ -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, + cwd: &AbsolutePathBuf, + items: impl IntoIterator, + 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 }