From 5183e1dcb2c35870ed7a966b69e66ed438f00c11 Mon Sep 17 00:00:00 2001 From: Benjamin Carlsson Date: Thu, 12 Nov 2026 23:45:36 +0000 Subject: [PATCH] Preserve thread input across TUI session resumes (#61216) ## Why Resuming another session must keep drafts, queued follow-ups, and interrupted steer retries with their original thread. Submitted input whose delivery is still uncertain must wait for matching receipts before the queue can advance. ## What changed - Pause submissions while draining outgoing events, then capture input before switching threads or backends. - Restore saved input through a shared resume path, reconcile receipts from resumed history and buffered events, preserve active-turn state, and dispatch eligible queued input when the resumed thread is idle. - Preserve whether a draft's leading `!` is literal text or a shell command. - Reject `!` shell commands in Cloud conversations with an error and clear the pending submission so the next prompt can proceed. ## Testing Add regression coverage for session round trips with buffered completion and interruption events, receipt reconciliation across read-only and writable resumes, draft shell mode, and submitting a prompt after a rejected Cloud shell command. GitOrigin-RevId: 9a823a398a371e689b8ae743de494323ae95ef57 --- codex-rs/tui/src/app/agents_overview.rs | 2 + codex-rs/tui/src/app/agents_overview_tests.rs | 294 +++++++++++++++++- codex-rs/tui/src/app/backend_selection.rs | 35 +-- codex-rs/tui/src/app/session_lifecycle.rs | 170 ++++++++-- codex-rs/tui/src/app/session_picker.rs | 23 -- .../src/app/tests/active_reconnect_tests.rs | 2 +- .../src/app/tests/backend_ownership_tests.rs | 206 ++++++++++++ .../src/app/tests/history_hydration_tests.rs | 4 +- .../app/tests/navigation_reconnect_tests.rs | 18 +- .../tui/src/app/tests/new_session_tests.rs | 13 + .../app/tests/session_lifecycle_requests.rs | 26 +- .../app/tests/unavailable_commands_tests.rs | 53 ++++ codex-rs/tui/src/app/thread_routing.rs | 12 +- codex-rs/tui/src/chatwidget/input_queue.rs | 6 +- codex-rs/tui/src/chatwidget/input_restore.rs | 56 +++- .../tui/src/chatwidget/input_submission.rs | 7 + codex-rs/tui/src/chatwidget/questions.rs | 1 + .../chatwidget/tests/composer_submission.rs | 35 ++- codex-rs/tui/src/chatwidget/tests/helpers.rs | 7 + codex-rs/tui/src/chatwidget/user_messages.rs | 4 +- 20 files changed, 841 insertions(+), 133 deletions(-) diff --git a/codex-rs/tui/src/app/agents_overview.rs b/codex-rs/tui/src/app/agents_overview.rs index 7f0410ed82..14faa045b0 100644 --- a/codex-rs/tui/src/app/agents_overview.rs +++ b/codex-rs/tui/src/app/agents_overview.rs @@ -1055,6 +1055,8 @@ impl App { && self.active_turn_id_for_thread(thread_id).await.is_some(); self.reconcile_saved_session_input(thread_id, &mut input) .await; + // Missing receipts must not turn a previously submitted steer into unsent input. + input.reconnect_pending |= !preserve_in_flight_turn && !input.pending_steers.is_empty(); self.chat_widget.restore_thread_input_state( Some(input), ThreadInputStateRestoreMode { diff --git a/codex-rs/tui/src/app/agents_overview_tests.rs b/codex-rs/tui/src/app/agents_overview_tests.rs index 9fc0d68905..0c521e5fb6 100644 --- a/codex-rs/tui/src/app/agents_overview_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_tests.rs @@ -2581,13 +2581,99 @@ async fn resume_failure_keeps_command_center_available() { } #[tokio::test] -async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> { - let mut app = Box::pin(make_test_app()).await; +async fn resume_round_trip_preserves_input_and_dispatches_idle_queue() -> Result<()> { + check_resume_round_trip(ResumeRoundTrip::CompleteWhileAway).await +} + +#[tokio::test] +async fn resume_round_trip_drains_buffered_completion_without_sending_on_other_thread() -> Result<()> +{ + check_resume_round_trip(ResumeRoundTrip::BufferCompletion).await +} + +#[tokio::test] +async fn resume_round_trip_keeps_buffered_esc_retry_on_original_thread() -> Result<()> { + check_resume_round_trip(ResumeRoundTrip::BufferInterruption).await +} + +#[derive(PartialEq, Eq)] +enum ResumeRoundTrip { + CompleteWhileAway, + BufferCompletion, + BufferInterruption, +} + +async fn check_resume_round_trip(round_trip: ResumeRoundTrip) -> Result<()> { + use codex_app_server_protocol::RequestId; + use codex_app_server_protocol::TurnStartParams; + use codex_app_server_protocol::TurnStartResponse; + use codex_app_server_protocol::TurnStatus; + use codex_model_provider_info::ModelProviderInfo; + use core_test_support::responses; + use core_test_support::streaming_sse::StreamingSseChunk; + use core_test_support::streaming_sse::start_streaming_sse_server; + + async fn wait_for_turn_status( + server: &mut AppServerSession, + outgoing: ThreadId, + status: TurnStatus, + ) -> Result { + tokio::time::timeout(Duration::from_secs(/*secs*/ 5), async { + loop { + let thread = server.thread_read(outgoing, /*include_turns*/ true).await?; + if let Some(turn) = thread.turns.into_iter().last() + && turn.status == status + { + break Ok::<_, color_eyre::eyre::Report>(turn); + } + tokio::task::yield_now().await; + } + }) + .await? + } + + let interrupted = round_trip == ResumeRoundTrip::BufferInterruption; + let requests_before_return = if interrupted { 1 } else { 2 }; + let steer_text = "Steer for the first task"; + let (release, gate) = tokio::sync::oneshot::channel(); + let (responses_server, _completions) = start_streaming_sse_server(vec![ + vec![ + StreamingSseChunk { + gate: None, + body: responses::sse(vec![responses::ev_response_created("running")]), + }, + StreamingSseChunk { + gate: Some(gate), + body: responses::sse(vec![responses::ev_completed("running")]), + }, + ], + vec![StreamingSseChunk { + gate: None, + body: responses::sse(vec![responses::ev_completed("steer")]), + }], + vec![StreamingSseChunk { + gate: None, + body: responses::sse(vec![responses::ev_completed("follow-up")]), + }], + ]) + .await; + let (mut app, mut events, _ops) = Box::pin(make_test_app_with_channels()).await; trust_fixture_folders(&mut app); - std::fs::write( - app.config.codex_home.join("config.toml"), - "[tui]\nresume_cwd = \"current\"\n", - )?; + app.config.model = Some("gpt-5.2".into()); + app.config.model_provider_id = "resume-test".into(); + app.config.model_provider = ModelProviderInfo { + name: "Resume test".into(), + base_url: Some(format!("{}/v1", responses_server.uri())), + request_max_retries: Some(0), + stream_max_retries: Some(0), + ..ModelProviderInfo::default() + }; + app_test_support::MockResponsesConfig::new(responses_server.uri()) + .with_model("gpt-5.2") + .with_model_provider("resume-test") + .with_provider_name("Resume test") + .with_extra_config("[tui]\nresume_cwd = \"current\"") + .write(app.config.codex_home.as_path())?; let mut server = Box::pin(crate::start_embedded_app_server_for_picker(&app.config)).await?; let backend_id = server.backend_id(); let mut targets = Vec::new(); @@ -2618,7 +2704,68 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> { } let mut tui = crate::tui::test_support::make_test_tui()?; let mut expected_states = Vec::new(); + let mut release = Some(release); for target in targets.iter().chain(&targets) { + if round_trip != ResumeRoundTrip::CompleteWhileAway + && expected_states.len() == targets.len() + && target.thread_id == targets[1].thread_id + { + let outgoing = targets[0].thread_id; + assert_eq!(app.chat_widget.thread_id(), Some(outgoing)); + assert_eq!( + app.chat_widget + .capture_thread_input_state() + .unwrap() + .pending_steers + .len(), + 1 + ); + tokio::time::timeout( + Duration::from_secs(/*secs*/ 5), + responses_server.wait_for_request_count(requests_before_return), + ) + .await?; + assert_eq!( + responses_server.requests().await.len(), + requests_before_return + ); + let status = if interrupted { + let turn_id = app + .chat_widget + .last_turn_id() + .expect("running turn") + .to_owned(); + app.chat_widget + .handle_key_event(KeyEvent::from(KeyCode::Esc)); + assert!( + std::iter::from_fn(|| events.try_recv().ok()) + .any(|event| matches!(event, AppEvent::CodexOp(AppCommand::Interrupt))) + ); + server.turn_interrupt(outgoing, turn_id).await?; + TurnStatus::Interrupted + } else { + release.take().unwrap().send(()).unwrap(); + TurnStatus::Completed + }; + let turn = wait_for_turn_status(&mut server, outgoing, status).await?; + drop(release.take()); + // Route the real completion to A's receiver, leaving it buffered across the switch. + app.enqueue_thread_notification( + outgoing, + ServerNotification::TurnCompleted( + codex_app_server_protocol::TurnCompletedNotification { + thread_id: outgoing.to_string(), + turn, + }, + ), + ) + .await?; + assert!( + app.active_thread_rx + .as_ref() + .is_some_and(|rx| !rx.is_empty()) + ); + } let view = app.agents_overview_view(Vec::new(), /*selected_thread_id*/ None); app.chat_widget.show_bottom_pane_view(Box::new(view)); Box::pin(app.apply_resume_picker_selection( @@ -2631,11 +2778,35 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> { if expected_states.len() < targets.len() { assert_eq!(app.chat_widget.composer_text_with_pending(), ""); assert!(app.chat_widget.queued_user_message_texts().is_empty()); + let running_turn = if target.thread_id == targets[0].thread_id { + let response = server + .request_handle() + .request_typed::(ClientRequest::TurnStart { + request_id: RequestId::String(Uuid::new_v4().to_string()), + params: TurnStartParams { + thread_id: target.thread_id.to_string(), + input: vec![codex_app_server_protocol::UserInput::Text { + text: "Keep working".into(), + text_elements: Vec::new(), + }], + ..Default::default() + }, + }) + .await?; + tokio::time::timeout( + Duration::from_secs(/*secs*/ 5), + responses_server.wait_for_request_count(/*count*/ 1), + ) + .await?; + Some(response.turn) + } else { + None + }; app.chat_widget.handle_server_notification( ServerNotification::TurnStarted( codex_app_server_protocol::TurnStartedNotification { thread_id: target.thread_id.to_string(), - turn: codex_app_server_protocol::Turn { + turn: running_turn.unwrap_or_else(|| codex_app_server_protocol::Turn { id: "turn-with-follow-up".to_string(), root_turn_id: None, items_view: codex_app_server_protocol::TurnItemsView::Full, @@ -2645,7 +2816,7 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> { started_at: None, completed_at: None, duration_ms: None, - }, + }), }, ), /*replay_kind*/ None, @@ -2657,15 +2828,17 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> { assert_eq!(app.chat_widget.queued_user_message_texts(), vec![follow_up]); app.chat_widget .apply_external_edit(format!("Draft for {}", target.thread_id)); - let mut input_state = app.chat_widget.capture_thread_input_state().unwrap(); - // Recovered follow-ups stay paused until the user explicitly submits them. - input_state.recovered_queue = true; - app.chat_widget.restore_thread_input_state( - Some(input_state), - crate::chatwidget::ThreadInputStateRestoreMode { - preserve_in_flight_turn: false, - }, - ); + if target.thread_id != targets[0].thread_id { + let mut input_state = app.chat_widget.capture_thread_input_state().unwrap(); + // Recovered follow-ups stay paused until the user explicitly submits them. + input_state.recovered_queue = true; + app.chat_widget.restore_thread_input_state( + Some(input_state), + crate::chatwidget::ThreadInputStateRestoreMode { + preserve_in_flight_turn: false, + }, + ); + } expected_states.push(app.chat_widget.capture_thread_input_state()); } else { let index = targets @@ -2676,9 +2849,96 @@ async fn resume_picker_round_trip_preserves_each_threads_input() -> Result<()> { app.chat_widget.capture_thread_input_state(), expected_states[index] ); + if target.thread_id == targets[0].thread_id { + app.chat_widget.apply_external_edit(steer_text.to_string()); + app.chat_widget + .handle_key_event(KeyEvent::from(KeyCode::Enter)); + let steer_event = std::iter::from_fn(|| events.try_recv().ok()) + .find(|event| matches!(event, AppEvent::CodexOp(AppCommand::UserTurn { .. }))) + .expect("steer submission"); + // Leave the steer uncommitted when exercising Esc's retry path. + if !interrupted { + Box::pin(app.handle_event(&mut tui, &mut server, steer_event)).await?; + } + app.chat_widget + .apply_external_edit(format!("Draft for {}", target.thread_id)); + } } } + assert_eq!(app.chat_widget.thread_id(), Some(targets[1].thread_id)); + if let Some(release) = release { + let outgoing = targets[0].thread_id; + assert_eq!( + app.agents_overview.input_states[&outgoing] + .pending_steers + .len(), + 1 + ); + tokio::time::timeout( + Duration::from_secs(/*secs*/ 5), + responses_server.wait_for_request_count(/*count*/ 2), + ) + .await?; + // A is unsubscribed now. Leave the completion out of the TUI event queue so + // resuming must discover it from the server's thread history. + release.send(()).unwrap(); + wait_for_turn_status(&mut server, outgoing, TurnStatus::Completed).await?; + } + assert_eq!( + responses_server.requests().await.len(), + requests_before_return + ); + while let Ok(event) = events.try_recv() { + assert!(!matches!( + event, + AppEvent::CodexOp(AppCommand::UserTurn { .. }) + )); + } + Box::pin(app.handle_event( + &mut tui, + &mut server, + AppEvent::ResumeSessionByIdOrName(targets[0].thread_id.to_string()), + )) + .await?; + assert_eq!(app.active_thread_id, Some(targets[0].thread_id)); + let mut submitted = Vec::new(); + while let Ok(event) = events.try_recv() { + if let AppEvent::CodexOp(AppCommand::UserTurn { ref items, .. }) = event { + submitted.push(items.clone()); + Box::pin(app.handle_event(&mut tui, &mut server, event)).await?; + } + } + let follow_up = format!("Follow-up for {}", targets[0].thread_id); + let expected_text = if interrupted { + steer_text.to_string() + } else { + follow_up.clone() + }; + assert_eq!( + submitted, + vec![vec![codex_app_server_protocol::UserInput::Text { + text: expected_text, + text_elements: Vec::new(), + }]] + ); + tokio::time::timeout( + Duration::from_secs(/*secs*/ 5), + responses_server.wait_for_request_count(requests_before_return + 1), + ) + .await?; + assert_eq!( + responses_server.requests().await.len(), + requests_before_return + 1 + ); + if interrupted { + assert_eq!(app.chat_widget.queued_user_message_texts(), [follow_up]); + } + assert_eq!( + app.chat_widget.composer_text_with_pending(), + format!("Draft for {}", targets[0].thread_id) + ); server.shutdown().await?; + responses_server.shutdown().await; Ok(()) } diff --git a/codex-rs/tui/src/app/backend_selection.rs b/codex-rs/tui/src/app/backend_selection.rs index e41deb0e47..a83a621843 100644 --- a/codex-rs/tui/src/app/backend_selection.rs +++ b/codex-rs/tui/src/app/backend_selection.rs @@ -491,6 +491,8 @@ impl App { Ok(prepared) => prepared, Err(control) => return Ok(control), }; + let handoff = self.chat_widget.pause_input_for_resume_handoff(); + let _handoff = self.prepare_resume_handoff(tui, handoff).await?; self.switch_backend_context( backends, destination_slot, @@ -500,35 +502,16 @@ impl App { bootstrap, ) .await; - let thread_id = target.thread_id; - let confirmed = super::thread_events::confirmed_user_message_ids( - &prepared.thread.turns, - std::iter::empty(), - ); - let restored_blank_session = prepared.restored_blank_session; - let has_resumed_mode = prepared.thread.session.collaboration_mode.is_some(); + backends.switch_displayed(destination_slot, &mut self.agents_overview.input_states); let destination = backends.session_at_mut(destination_slot); let control = self - .activate_prepared_session(tui, destination, prepared) + .activate_prepared_session_after_clear( + tui, + destination, + prepared, + /*switched_backend*/ true, + ) .await?; - backends.switch_displayed(destination_slot, &mut self.agents_overview.input_states); - if !restored_blank_session { - self.agents_overview.blank_sessions.remove(&thread_id); - } - if let Some(mut input) = self.agents_overview.input_states.remove(&thread_id) { - input.provider_error = None; - input - .pending_steers - .retain(|steer| !confirmed.contains(&steer.client_id)); - if !restored_blank_session { - self.chat_widget - .update_input_collaboration_mode(&mut input, has_resumed_mode); - } - let settings = restored_blank_session - .then(|| input.pending_thread_settings.take()) - .flatten(); - self.restore_session_draft(thread_id, input, settings).await; - } tui.frame_requester().schedule_frame(); Ok(control) }) diff --git a/codex-rs/tui/src/app/session_lifecycle.rs b/codex-rs/tui/src/app/session_lifecycle.rs index f853f3c6e0..b68e011730 100644 --- a/codex-rs/tui/src/app/session_lifecycle.rs +++ b/codex-rs/tui/src/app/session_lifecycle.rs @@ -33,6 +33,8 @@ pub(super) enum ThreadAttachPresentation { /// A fresh thread whose startup composer is already visible. FreshWithDraft, SessionLineage, + /// A resume whose outgoing events were drained and terminal cleared before capture. + SessionLineageAfterClear, } pub(super) struct PreparedFreshSession { @@ -1474,6 +1476,16 @@ impl App { .await } + async fn drain_outgoing_thread_events(&mut self) { + if let Some(mut receiver) = self.active_thread_rx.take() { + while let Ok(event) = receiver.try_recv() { + self.handle_thread_event_now_recovering_file_changes(event) + .await; + } + self.active_thread_rx = Some(receiver); + } + } + async fn replace_chat_widget_with_account( &mut self, tui: &mut tui::Tui, @@ -1485,18 +1497,20 @@ impl App { // Initial messages are for freshly attached primary threads only. Thread switches and // resume/fork flows pass `None` so they cannot replay old history and then auto-submit a new // user turn by accident. - if let Some(mut receiver) = self.active_thread_rx.take() { - while let Ok(event) = receiver.try_recv() { - self.handle_thread_event_now_recovering_file_changes(event) - .await; - } - self.active_thread_rx = Some(receiver); + if !matches!( + presentation, + ThreadAttachPresentation::SessionLineageAfterClear + ) { + self.drain_outgoing_thread_events().await; } self.store_active_thread_receiver().await; self.reset_transcript_state_after_clear(); tui.clear_pending_history_lines(); let presentation = match presentation { ThreadAttachPresentation::FreshAfterClear => ThreadAttachPresentation::Fresh, + ThreadAttachPresentation::SessionLineageAfterClear => { + ThreadAttachPresentation::SessionLineage + } ThreadAttachPresentation::FreshWithDraft => { tui.defer_thread_switch_clear(); presentation @@ -1742,6 +1756,43 @@ impl App { tui: &mut tui::Tui, app_server: &mut AppServerSession, prepared: super::resume_preparation::PreparedResume, + ) -> Result { + if prepared.backend_id != app_server.backend_id() { + self.add_session_picker_error( + "This session belongs to a different backend. Open the session picker again." + .to_string(), + ); + return Ok(AppRunControl::Continue); + } + let handoff = self.chat_widget.pause_input_for_resume_handoff(); + let _handoff = self.prepare_resume_handoff(tui, handoff).await?; + self.activate_prepared_session_after_clear( + tui, app_server, prepared, /*switched_backend*/ false, + ) + .await + } + + pub(super) async fn prepare_resume_handoff( + &mut self, + tui: &mut tui::Tui, + handoff: std::sync::Arc<()>, + ) -> Result> { + self.drain_outgoing_thread_events().await; + // Keep the outgoing receiver and session in place if the terminal clear fails. + if let Err(error) = tui.clear_for_thread_switch() { + drop(handoff); + self.chat_widget.maybe_send_next_queued_input(); + return Err(error.into()); + } + Ok(handoff) + } + + pub(super) async fn activate_prepared_session_after_clear( + &mut self, + tui: &mut tui::Tui, + app_server: &mut AppServerSession, + prepared: super::resume_preparation::PreparedResume, + switched_backend: bool, ) -> Result { let super::resume_preparation::PreparedResume { backend_id, @@ -1751,24 +1802,19 @@ impl App { read_only, history_notice, provider_error, - .. + restored_blank_session, } = prepared; - if backend_id != app_server.backend_id() { - self.add_session_picker_error( - "This session belongs to a different backend. Open the session picker again." - .to_string(), - ); - return Ok(AppRunControl::Continue); - } - self.backend_registry - .observe(backend_id, resumed.session.thread_id); + debug_assert_eq!(backend_id, app_server.backend_id()); let resumed_thread_id = resumed.session.thread_id; - let was_reconnecting = self.chat_widget.thread_stream_disconnected; + self.backend_registry.observe(backend_id, resumed_thread_id); + let has_resumed_mode = resumed.session.collaboration_mode.is_some(); + let was_reconnecting = !switched_backend && self.chat_widget.thread_stream_disconnected; let owned_channel = self .thread_event_channels .get(&resumed_thread_id) .filter(|channel| channel.backend_id.is_none_or(|owner| owner == backend_id)); - let same_thread = self.chat_widget.thread_id() == Some(resumed_thread_id); + let same_thread = + !switched_backend && self.chat_widget.thread_id() == Some(resumed_thread_id); let refreshing_history = same_thread && !was_reconnecting && owned_channel.is_some_and(|channel| channel.history_may_be_incomplete); @@ -1780,9 +1826,37 @@ impl App { || self.chat_widget.has_provider_error())) .then(|| self.chat_widget.capture_thread_input_state()) .flatten(); - let confirmed_message_ids = refreshing_history.then(|| { - super::thread_events::confirmed_user_message_ids(&resumed.turns, std::iter::empty()) - }); + let mut confirmed_message_ids = + super::thread_events::confirmed_user_message_ids(&resumed.turns, std::iter::empty()); + if let Some(channel) = owned_channel { + let store = channel.store.lock().await; + confirmed_message_ids.extend(super::thread_events::confirmed_user_message_ids( + &store.turns, + store.buffer.iter(), + )); + } + // A backend switch has already parked the outgoing inputs. + if !switched_backend { + let displayed_thread_id = self.current_displayed_thread_id(); + for (id, channel) in &self.thread_event_channels { + if Some(*id) != displayed_thread_id + && channel.backend_id.is_none_or(|owner| owner == backend_id) + && let Some(input) = channel.store.lock().await.input_state.clone() + { + self.agents_overview.input_states.insert(*id, input); + } + } + if displayed_thread_id != Some(resumed_thread_id) + && let Some(id) = displayed_thread_id + && self + .backend_registry + .owner(id) + .is_none_or(|owner| owner == backend_id) + && let Some(input) = self.chat_widget.capture_thread_input_state() + { + self.agents_overview.input_states.insert(id, input); + } + } self.detach_current_thread_for_navigation(app_server, Some(resumed_thread_id)) .await; self.local_settings = local_settings; @@ -1798,7 +1872,7 @@ impl App { .replace_chat_widget_with_app_server_thread( tui, resumed, - ThreadAttachPresentation::SessionLineage, + ThreadAttachPresentation::SessionLineageAfterClear, /*initial_user_message*/ None, ) .await @@ -1807,21 +1881,58 @@ impl App { self.ensure_thread_channel(resumed_thread_id) .history_may_be_incomplete = false; self.remember_attached_host_project(app_server); - if let Some(mut input) = retained_input { + let resumed_turn_running = !read_only + && self + .active_turn_id_for_thread(resumed_thread_id) + .await + .is_some(); + let preserve_in_flight_turn = resumed_turn_running; + if switched_backend && !restored_blank_session { + self.agents_overview + .blank_sessions + .remove(&resumed_thread_id); + } + let keep_blank_settings = self + .agents_overview + .blank_sessions + .contains_key(&resumed_thread_id); + if let Some(mut input) = + retained_input.or(self.agents_overview.input_states.remove(&resumed_thread_id)) + { // A successful resume is the authority for clearing the old rejection. input.provider_error = None; - if let Some(confirmed_message_ids) = confirmed_message_ids { - input.reconnect_pending = true; + if !restored_blank_session { + self.chat_widget + .update_input_collaboration_mode(&mut input, has_resumed_mode); + } + // A saved draft includes model settings, so apply newer server settings after + // restoring input for retained blank sessions. + let pending_settings = keep_blank_settings + .then(|| input.pending_thread_settings.take()) + .flatten(); + let reconcile_input = refreshing_history + || !input.pending_steers.is_empty() + || input.has_unconfirmed_messages(); + if reconcile_input { + input.reconcile_confirmed_messages(&confirmed_message_ids); + input.reconnect_pending |= refreshing_history || !resumed_turn_running; + } + if refreshing_history + || (reconcile_input && !read_only && !resumed_turn_running) + { self.chat_widget .restore_reconnected_input(Some(input), &confirmed_message_ids); } else { self.chat_widget.restore_thread_input_state( Some(input), ThreadInputStateRestoreMode { - preserve_in_flight_turn: false, + preserve_in_flight_turn, }, ); } + if let Some(settings) = pending_settings { + self.chat_widget.on_thread_settings_updated(settings); + } if was_reconnecting && !read_only && provider_error.is_none() { self.chat_widget.maybe_send_next_queued_input(); } @@ -1864,6 +1975,13 @@ impl App { resumed_thread_id, ) .await; + if self + .active_turn_id_for_thread(resumed_thread_id) + .await + .is_none() + { + self.chat_widget.maybe_send_next_queued_input(); + } } } Err(err) => return Err(err), diff --git a/codex-rs/tui/src/app/session_picker.rs b/codex-rs/tui/src/app/session_picker.rs index 063a447eec..3c26b5aeb4 100644 --- a/codex-rs/tui/src/app/session_picker.rs +++ b/codex-rs/tui/src/app/session_picker.rs @@ -1,7 +1,6 @@ //! In-app session picker transitions, preserving per-thread input and the view on cancellation. use super::*; -use crate::chatwidget::ThreadInputStateRestoreMode; impl App { pub(super) async fn capture_session_inputs(&mut self, owner: crate::backend_id::BackendId) { @@ -93,34 +92,12 @@ impl App { return Ok(AppRunControl::Continue); } let thread_id = target_session.thread_id; - let switching_threads = self.current_displayed_thread_id() != Some(thread_id); - if switching_threads { - self.capture_session_inputs(app_server.backend_id()).await; - } if let AppRunControl::Exit(reason) = self .resume_target_session(tui, app_server, target_session) .await? { return Ok(AppRunControl::Exit(reason)); } - if switching_threads - && self.current_displayed_thread_id() == Some(thread_id) - && let Some(mut input_state) = - self.agents_overview.input_states.remove(&thread_id) - { - // The resumed widget already reflects the latest server result. - input_state.provider_error = None; - self.reconcile_saved_session_input(thread_id, &mut input_state) - .await; - let preserve_in_flight_turn = - self.active_turn_id_for_thread(thread_id).await.is_some(); - self.chat_widget.restore_thread_input_state( - Some(input_state), - ThreadInputStateRestoreMode { - preserve_in_flight_turn, - }, - ); - } if self.active_thread_id == Some(thread_id) && self .chat_widget diff --git a/codex-rs/tui/src/app/tests/active_reconnect_tests.rs b/codex-rs/tui/src/app/tests/active_reconnect_tests.rs index e6d3e5cc52..ad94b93e71 100644 --- a/codex-rs/tui/src/app/tests/active_reconnect_tests.rs +++ b/codex-rs/tui/src/app/tests/active_reconnect_tests.rs @@ -799,7 +799,7 @@ pub(super) async fn drain_history( | AppEvent::BeginThreadSwitchHistoryReplayBuffer | AppEvent::EndInitialHistoryReplayBuffer ) { - app.handle_event(tui, session, event).await?; + crate::boxed_future!(app.handle_event(tui, session, event)).await?; } } Ok(lines_to_single_string( diff --git a/codex-rs/tui/src/app/tests/backend_ownership_tests.rs b/codex-rs/tui/src/app/tests/backend_ownership_tests.rs index 2ef1b5777f..16b5c67247 100644 --- a/codex-rs/tui/src/app/tests/backend_ownership_tests.rs +++ b/codex-rs/tui/src/app/tests/backend_ownership_tests.rs @@ -4,6 +4,7 @@ use super::start_recording_app_server; use super::*; use crate::app_server_session::ResumeModelSettings; use crate::backend_id::BackendId; +use crate::chatwidget::tests::helpers::assert_pending_steers_unadmitted; use crate::session_start::SessionStartAction; use crate::session_start::SessionStartConfig; use crate::session_start::complete_session_start; @@ -351,3 +352,208 @@ async fn prepared_resume_cannot_activate_through_another_backend() -> Result<()> proxy.abort(); Ok(()) } + +#[tokio::test] +async fn read_only_resume_reconciles_receipts_before_writable_retry() -> Result<()> { + let (mut app, mut events, _ops) = make_test_app_with_channels().await; + let mut tui = crate::tui::test_support::make_test_tui()?; + let (mut server, _requests, proxy) = start_recording_app_server( + &app.config, + /*blocked_thread_list*/ None, + /*failed_thread_name*/ None, + ) + .await?; + let current_thread = ThreadId::new(); + let mut current_session = test_thread_session(current_thread, app.config.cwd.to_path_buf()); + current_session.model = app.config.model.clone().expect("configured model"); + current_session.model_provider_id = app.config.model_provider_id.clone(); + app.enqueue_primary_thread_session(current_session, Vec::new()) + .await?; + app.chat_widget.handle_server_notification( + ServerNotification::TurnStarted(codex_app_server_protocol::TurnStartedNotification { + thread_id: current_thread.to_string(), + turn: test_turn("current", TurnStatus::InProgress, Vec::new()), + }), + /*replay_kind*/ None, + ); + app.chat_widget + .submit_user_message_as_plain_user_turn(crate::chatwidget::UserMessage::from( + "accepted steer", + )) + .expect("submit steer"); + app.chat_widget + .submit_user_message_as_plain_user_turn(crate::chatwidget::UserMessage::from( + "delayed steer", + )) + .expect("submit delayed steer"); + app.chat_widget + .apply_external_edit("retained draft".to_string()); + let input = app + .chat_widget + .capture_thread_input_state() + .expect("pending input"); + assert_eq!(input.pending_steers.len(), 2); + assert!(!input.reconnect_pending); + assert_pending_steers_unadmitted(&input); + let command_center_input = input.clone(); + let accepted_client_id = input.pending_steers[0].client_id.clone(); + let delayed_client_id = input.pending_steers[1].client_id.clone(); + let resumed_thread = ThreadId::new(); + app.agents_overview + .input_states + .insert(resumed_thread, input); + + let receipt = |id: &str, client_id: String, text: &str| ThreadItemPayload::UserMessage { + id: id.into(), + client_id: Some(client_id), + content: vec![UserInput::Text { + text: text.into(), + text_elements: Vec::new(), + }], + }; + let accepted_steer = receipt("accepted-steer", accepted_client_id, "accepted steer"); + let delayed_steer = receipt("delayed-steer", delayed_client_id, "delayed steer"); + let backend_id = server.backend_id(); + let resumed_mode = CollaborationMode { + mode: ModeKind::Plan, + settings: Settings { + model: "resumed-model".into(), + reasoning_effort: Some(ReasoningEffortConfig::High), + developer_instructions: None, + }, + }; + let prepared = |app: &App, read_only, status, items| { + let mut session = test_thread_session(resumed_thread, app.config.cwd.to_path_buf()); + session.model = resumed_mode.model().to_string(); + session.reasoning_effort = resumed_mode.reasoning_effort(); + session.collaboration_mode = Some(Box::new(resumed_mode.clone())); + session.model_provider_id = app.config.model_provider_id.clone(); + super::super::resume_preparation::PreparedResume { + backend_id, + config: app.config.clone(), + local_settings: app.local_settings.clone(), + thread: AppServerStartedThread { + session, + turns: vec![test_turn("resumed", status, items)], + blocks_direct_input: false, + task_tools_available: true, + }, + restored_blank_session: false, + read_only, + history_notice: None, + provider_error: None, + } + }; + + // Exercise the Command Center restore boundary with the raw saved input, + // rather than the uncertainty already marked by direct resume. + let idle_resume = prepared( + &app, + /*read_only*/ false, + TurnStatus::Completed, + vec![accepted_steer.clone()], + ); + app.activate_prepared_session(&mut tui, &mut server, idle_resume) + .await?; + assert_eq!(app.active_turn_id_for_thread(resumed_thread).await, None); + assert!(!app.chat_widget.is_external_writer_view()); + app.restore_session_draft( + resumed_thread, + command_center_input.clone(), + /*pending_settings*/ None, + ) + .await; + assert_eq!( + app.chat_widget.composer_text_with_pending(), + "retained draft" + ); + assert_eq!( + app.chat_widget.queued_user_message_texts(), + ["delayed steer"] + ); + assert!( + app.chat_widget + .capture_thread_input_state() + .unwrap() + .has_unconfirmed_messages() + ); + for item in [ + receipt("wrong-id", "unrelated-client-id".into(), "delayed steer"), + delayed_steer.clone(), + ] { + assert_eq!( + app.chat_widget.queued_user_message_texts(), + ["delayed steer"] + ); + app.chat_widget.handle_server_notification( + ServerNotification::ItemCompleted( + codex_app_server_protocol::ItemCompletedNotification::new( + resumed_thread.to_string(), + "resumed".into(), + item, + /*started_at_ms*/ None, + /*completed_at_ms*/ 1, + ), + ), + /*replay_kind*/ None, + ); + } + assert!(app.chat_widget.queued_user_message_texts().is_empty()); + app.agents_overview + .input_states + .insert(resumed_thread, command_center_input); + + let read_only_resume = prepared( + &app, + /*read_only*/ true, + TurnStatus::Completed, + vec![accepted_steer.clone()], + ); + app.activate_prepared_session(&mut tui, &mut server, read_only_resume) + .await?; + assert!(app.chat_widget.is_external_writer_view()); + assert_eq!(app.chat_widget.effective_collaboration_mode(), resumed_mode); + assert_eq!( + app.chat_widget.composer_text_with_pending(), + "retained draft" + ); + assert_eq!( + app.chat_widget.queued_user_message_texts(), + vec!["delayed steer"] + ); + let input = app + .chat_widget + .capture_thread_input_state() + .expect("unconfirmed steer"); + assert!(input.has_unconfirmed_messages()); + + let writable_resume = prepared( + &app, + /*read_only*/ false, + TurnStatus::InProgress, + vec![accepted_steer, delayed_steer], + ); + app.activate_prepared_session(&mut tui, &mut server, writable_resume) + .await?; + assert!(!app.chat_widget.is_external_writer_view()); + assert!(app.chat_widget.is_agent_turn_running()); + assert_eq!( + app.chat_widget.composer_text_with_pending(), + "retained draft" + ); + assert!(app.chat_widget.queued_user_message_texts().is_empty()); + let input = app + .chat_widget + .capture_thread_input_state() + .expect("retained draft"); + assert!(input.pending_steers.is_empty()); + while let Ok(event) = events.try_recv() { + assert!(!matches!( + event, + AppEvent::CodexOp(AppCommand::UserTurn { .. }) + )); + } + server.shutdown().await?; + proxy.abort(); + Ok(()) +} diff --git a/codex-rs/tui/src/app/tests/history_hydration_tests.rs b/codex-rs/tui/src/app/tests/history_hydration_tests.rs index bc0c65508b..749f2e9f56 100644 --- a/codex-rs/tui/src/app/tests/history_hydration_tests.rs +++ b/codex-rs/tui/src/app/tests/history_hydration_tests.rs @@ -222,7 +222,7 @@ async fn history_hydration_metadata_tracks_missing_turns_and_stale_completions_k backwards_cursor: None, }), ] { - app.handle_event( + crate::boxed_future!(app.handle_event( &mut tui, &mut server, AppEvent::OlderThreadHistoryLoaded { @@ -230,7 +230,7 @@ async fn history_hydration_metadata_tracks_missing_turns_and_stale_completions_k cursor: old_cursor.clone(), result, }, - ) + )) .await?; assert!(server.is_older_history_page_pending(target.thread_id, &cursor)); assert_eq!( diff --git a/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs b/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs index afff9d14b7..b048790c9c 100644 --- a/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs +++ b/codex-rs/tui/src/app/tests/navigation_reconnect_tests.rs @@ -331,13 +331,13 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv ) .await?; let old_event_tx = app.app_event_tx.app_event_tx.clone(); - app.finish_reconnect( + crate::boxed_future!(app.finish_reconnect( &mut tui, &mut session, &mut events, connected, CODEX_CLI_VERSION, - ) + )) .await?; assert_eq!(app.app_server_backend_id, Some(session.backend_id())); assert!(old_event_tx.is_closed()); @@ -373,7 +373,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv while app.agents_overview.active_refresh.is_some() { let event = events.recv().await.expect("inventory refresh event"); if matches!(event, AppEvent::AgentsOverviewThreadsLoaded { .. }) { - Box::pin(app.handle_event(&mut tui, &mut session, event)).await?; + crate::boxed_future!(app.handle_event(&mut tui, &mut session, event)).await?; } else { deferred.push(event); } @@ -507,8 +507,12 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv app.chat_widget.composer_text_with_pending(), "latest draft!" ); - Box::pin(app.handle_event(&mut tui, &mut session, AppEvent::OpenAgentsOverview)) - .await?; + crate::boxed_future!(app.handle_event( + &mut tui, + &mut session, + AppEvent::OpenAgentsOverview + )) + .await?; assert!(app.chat_widget.has_active_view()); app.handle_tui_event( &mut tui, @@ -554,7 +558,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv ); } available.store(true, std::sync::atomic::Ordering::SeqCst); - Box::pin(app.handle_event( + crate::boxed_future!(app.handle_event( &mut tui, &mut session, AppEvent::SelectAgentsOverviewThread { thread_id: id }, @@ -604,7 +608,7 @@ async fn reconnect_daemon_command_center_after_socket_replacement_without_a_conv .await?; while let Ok(event) = events.try_recv() { if matches!(event, AppEvent::CodexOp(AppCommand::UserTurn { .. })) { - Box::pin(app.handle_event(&mut tui, &mut session, event)).await?; + crate::boxed_future!(app.handle_event(&mut tui, &mut session, event)).await?; } } let channel = app.thread_event_channels.get(&id).unwrap(); diff --git a/codex-rs/tui/src/app/tests/new_session_tests.rs b/codex-rs/tui/src/app/tests/new_session_tests.rs index fdfec26849..84494877ad 100644 --- a/codex-rs/tui/src/app/tests/new_session_tests.rs +++ b/codex-rs/tui/src/app/tests/new_session_tests.rs @@ -1199,6 +1199,9 @@ async fn check_retained_draft_handoff( let mode = server_mode.as_mut().expect("test has a mode to update"); mode.mode = ModeKind::Plan; mode.settings.developer_instructions = Some("Instructions from the parked task".into()); + if parked_policy.is_some() { + mode.settings.reasoning_effort = Some(ReasoningEffortConfig::High); + } Some(mode.clone()) } else { None @@ -1220,6 +1223,7 @@ async fn check_retained_draft_handoff( let settings = ThreadSettingsUpdateParams { thread_id: thread_id.to_string(), collaboration_mode, + effort: parked_policy.map(|_| ReasoningEffortConfig::High), approval_policy: parked_policy, ..Default::default() }; @@ -1236,6 +1240,15 @@ async fn check_retained_draft_handoff( let (destination, parked_inputs) = backends.session_and_parked_inputs(destination_slot); app.handle_polled_backend_event(destination, outgoing, parked_inputs, event) .await; + } else if saved_history && retained_blank { + // A formerly blank thread must ignore settings cached before it acquired history. + let mut stale_update = update; + stale_update.thread_settings.model = "old-model".into(); + backends + .parked_inputs_mut(destination_slot) + .get_mut(&thread_id) + .unwrap() + .pending_thread_settings = Some(stale_update); } } let settings = app.backend_settings(); 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 aa54f17c81..0f52a18283 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -1133,7 +1133,7 @@ async fn complete_managed_worktree_creation( .await? .ok_or_else(|| color_eyre::eyre::eyre!("worktree creation event channel closed"))?; if matches!(event, AppEvent::ManagedWorktreeCreated(_)) { - app.handle_event(tui, server, event).await?; + crate::boxed_future!(app.handle_event(tui, server, event)).await?; if let Some(created) = app.pending_managed_worktree_created.take() { app.finish_managed_worktree(*created).await; } @@ -4796,14 +4796,14 @@ model_reasoning_effort = "low" ); let mut browser_entries = Vec::new(); requests.lock().expect("request recorder lock").clear(); - app.handle_event( + crate::boxed_future!(app.handle_event( &mut tui, &mut server, AppEvent::StartManagedWorktree { mode: ManagedWorktreeMode::Fork, name: None, }, - ) + )) .await?; complete_managed_worktree_creation(&mut app, &mut tui, &mut server, &mut events).await?; assert_eq!(app.chat_widget.thread_id(), Some(original)); @@ -4869,14 +4869,14 @@ terminal_visualization_instructions = true )?; app.config = app.rebuild_config_for_cwd(source.clone()).await?; requests.lock().expect("request recorder lock").clear(); - app.handle_event( + crate::boxed_future!(app.handle_event( &mut tui, &mut server, AppEvent::StartManagedWorktree { mode: ManagedWorktreeMode::Fork, name: None, }, - ) + )) .await?; complete_managed_worktree_creation(&mut app, &mut tui, &mut server, &mut events).await?; assert_eq!(app.chat_widget.thread_id(), Some(original)); @@ -4893,14 +4893,14 @@ terminal_visualization_instructions = true owner: crate::worktree_browser::Owner::None, }); while events.try_recv().is_ok() {} - app.handle_event( + crate::boxed_future!(app.handle_event( &mut tui, &mut server, AppEvent::StartManagedWorktree { mode: ManagedWorktreeMode::New, name: None, }, - ) + )) .await?; assert_eq!(app.chat_widget.thread_id(), Some(original)); drain_managed_worktree_start(&mut app, &mut server).await; @@ -4928,11 +4928,11 @@ terminal_visualization_instructions = true .cwd .clone(); app.primary_thread_id = Some(ThreadId::new()); - app.handle_event( + crate::boxed_future!(app.handle_event( &mut tui, &mut server, AppEvent::ManagedWorktreeCreated(created), - ) + )) .await?; let created = app .pending_managed_worktree_created @@ -4968,14 +4968,14 @@ terminal_visualization_instructions = true requests.lock().expect("request recorder lock").clear(); let previous = app.chat_widget.thread_id(); let previous_rollout = app.chat_widget.rollout_path().expect("previous rollout"); - app.handle_event( + crate::boxed_future!(app.handle_event( &mut tui, &mut server, AppEvent::StartManagedWorktree { mode, name: Some(format!("{mode:?} worktree")), }, - ) + )) .await?; complete_managed_worktree_creation(&mut app, &mut tui, &mut server, &mut events).await?; let replacement = app.chat_widget.thread_id().expect("replacement thread"); @@ -5146,11 +5146,11 @@ terminal_visualization_instructions = true .set_turns(vec![completed]); for mode in [ManagedWorktreeMode::Fork, ManagedWorktreeMode::New] { requests.lock().expect("request recorder lock").clear(); - app.handle_event( + crate::boxed_future!(app.handle_event( &mut tui, &mut server, AppEvent::StartManagedWorktree { mode, name: None }, - ) + )) .await?; drain_managed_worktree_start(&mut app, &mut server).await; let is_new = mode == ManagedWorktreeMode::New; diff --git a/codex-rs/tui/src/app/tests/unavailable_commands_tests.rs b/codex-rs/tui/src/app/tests/unavailable_commands_tests.rs index 8ff73ff2b0..478514670d 100644 --- a/codex-rs/tui/src/app/tests/unavailable_commands_tests.rs +++ b/codex-rs/tui/src/app/tests/unavailable_commands_tests.rs @@ -4,6 +4,59 @@ use super::active_reconnect::drain_history; use super::*; use pretty_assertions::assert_eq; +#[tokio::test] +async fn cloud_shell_command_reports_error_and_allows_next_prompt() -> Result<()> { + let (mut app, mut events, mut ops) = make_test_app_with_channels().await; + let thread_id = ThreadId::new(); + app.active_thread_id = Some(thread_id); + app.app_server_target = AppServerTarget::Cloud; + app.chat_widget + .handle_thread_session_quiet(test_thread_session(thread_id, app.config.cwd.to_path_buf())); + let (client, mut requests, _server_events) = + codex_app_server_client::CloudClient::channel(/*capacity*/ 1); + let mut session = AppServerSession::new( + crate::backend_id::BackendId::new(), + codex_app_server_client::AppServerClient::Cloud(client), + crate::app_server_session::ThreadParamsMode::Cloud, + ); + let mut tui = crate::tui::test_support::make_test_tui()?; + while events.try_recv().is_ok() {} + while ops.try_recv().is_ok() {} + + app.chat_widget + .restore_user_message_to_composer("First prompt".into()); + app.chat_widget.handle_key_event(KeyCode::Enter.into()); + assert!(matches!(ops.try_recv()?, Op::UserTurn { .. })); + app.chat_widget.handle_server_notification( + turn_started_notification(thread_id, "turn-1"), + /*replay_kind*/ None, + ); + app.chat_widget.handle_server_notification( + turn_completed_notification(thread_id, "turn-1", TurnStatus::Completed), + /*replay_kind*/ None, + ); + while events.try_recv().is_ok() {} + + app.chat_widget + .restore_user_message_to_composer("! pwd".into()); + app.chat_widget.handle_key_event(KeyCode::Enter.into()); + let shell = ops.try_recv()?; + assert!(matches!(&shell, Op::RunUserShellCommand { command } if command == "pwd")); + let result = tokio::select! { + result = Box::pin(app.handle_event(&mut tui, &mut session, AppEvent::CodexOp(shell))) => result?, + _ = requests.recv() => panic!("Cloud shell command unexpectedly sent a request"), + }; + assert!(matches!(result, AppRunControl::Continue)); + assert_snapshot!(next_history_message(&mut events), @"■ Shell commands starting with ! are unavailable in Cloud conversations."); + assert_eq!(app.active_thread_id, Some(thread_id)); + + app.chat_widget + .restore_user_message_to_composer("Continue the conversation".into()); + app.chat_widget.handle_key_event(KeyCode::Enter.into()); + assert!(matches!(ops.try_recv()?, Op::UserTurn { .. })); + Ok(()) +} + #[tokio::test] async fn unavailable_thread_routes_local_and_recovery_commands() -> Result<()> { let (mut app, mut events, mut ops) = make_test_app_with_channels().await; diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 4aee17d93f..44a1f64e0e 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -970,6 +970,15 @@ impl App { Ok(true) } AppCommand::RunUserShellCommand { command } => { + if app_server.thread_params_mode() + == crate::app_server_session::ThreadParamsMode::Cloud + { + self.chat_widget.reject_user_shell_command( + "Shell commands starting with ! are unavailable in Cloud conversations." + .into(), + ); + return Ok(true); + } app_server .thread_shell_command(thread_id, command.to_string()) .await?; @@ -1613,7 +1622,8 @@ impl App { ThreadAttachPresentation::Fresh | ThreadAttachPresentation::FreshAfterClear | ThreadAttachPresentation::FreshWithDraft - | ThreadAttachPresentation::SessionLineage => { + | ThreadAttachPresentation::SessionLineage + | ThreadAttachPresentation::SessionLineageAfterClear => { self.chat_widget.handle_thread_session(session); } } diff --git a/codex-rs/tui/src/chatwidget/input_queue.rs b/codex-rs/tui/src/chatwidget/input_queue.rs index dc162e982e..68ceb8c1f0 100644 --- a/codex-rs/tui/src/chatwidget/input_queue.rs +++ b/codex-rs/tui/src/chatwidget/input_queue.rs @@ -50,6 +50,8 @@ pub(super) struct InputQueueState { /// fresh user turn instead of restoring them into the composer. pub(super) submit_pending_steers_after_interrupt: bool, pub(super) suppress_queue_autosend: bool, + /// Independent of settings suppression, which a drained event can set. + pub(super) resume_handoff: std::sync::Weak<()>, pub(super) transcript_copy: std::sync::Weak, /// Hold submissions while a usage failure or backend-directed model fallback is resolved. pub(super) rate_limit_recovery_pending: bool, @@ -67,7 +69,9 @@ impl InputQueueState { } pub(super) fn submissions_paused(&self) -> bool { - self.suppress_queue_autosend || self.transcript_copy.strong_count() > 0 + self.suppress_queue_autosend + || self.transcript_copy.strong_count() > 0 + || self.resume_handoff.strong_count() > 0 } pub(super) fn has_unconfirmed_messages(&self) -> bool { diff --git a/codex-rs/tui/src/chatwidget/input_restore.rs b/codex-rs/tui/src/chatwidget/input_restore.rs index 47359d7632..90c77fbd1a 100644 --- a/codex-rs/tui/src/chatwidget/input_restore.rs +++ b/codex-rs/tui/src/chatwidget/input_restore.rs @@ -462,6 +462,7 @@ impl ChatWidget { // interrupted turn reaches the UI, so any unacknowledged steers still // tracked here must be restored locally instead of waiting for a later commit. if send_pending_steers_immediately { + let resume_handoff = self.input_queue.resume_handoff.strong_count() > 0; let pending_steers = self .input_queue .pending_steers @@ -483,7 +484,9 @@ impl ChatWidget { .rejected_steer_history_records .push_back(pending.history_record); } - if let Some((message, history_record)) = self.pop_next_queued_user_message() { + if !resume_handoff + && let Some((message, history_record)) = self.pop_next_queued_user_message() + { let source = message.source; self.submit_user_message_with_history_and_shell_escape_policy( message.into_user_message(), @@ -499,12 +502,22 @@ impl ChatWidget { .map(|pending| (pending.user_message, pending.history_record)) .collect(), ); - self.submit_user_message_with_history_and_shell_escape_policy( - user_message, - history_record, - ShellEscapePolicy::Allow, - UserMessageSource::Prompt, - ); + if resume_handoff { + // Capture the retry with the outgoing thread's input. + self.input_queue + .queued_user_messages + .push_front(user_message.into()); + self.input_queue + .queued_user_message_history_records + .push_front(history_record); + } else { + self.submit_user_message_with_history_and_shell_escape_policy( + user_message, + history_record, + ShellEscapePolicy::Allow, + UserMessageSource::Prompt, + ); + } } else if let Some(combined) = self.drain_pending_messages_for_restore() { self.restore_composer_state(combined); } @@ -634,7 +647,8 @@ impl ChatWidget { } pub(super) fn restore_composer_state(&mut self, composer: ThreadComposerState) { - self.restore_composer_state_with_shell_mode(composer, /*shell_mode*/ None); + let shell_mode = composer.shell_mode; + self.restore_composer_state_with_shell_mode(composer, shell_mode); } fn restore_composer_state_with_shell_mode( @@ -644,6 +658,7 @@ impl ChatWidget { ) { let ThreadComposerState { text, + shell_mode: _, local_images, remote_image_urls, text_elements, @@ -679,6 +694,7 @@ impl ChatWidget { } = user_message; ThreadComposerState { text, + shell_mode: None, local_images, remote_image_urls, text_elements, @@ -692,6 +708,8 @@ impl ChatWidget { let draft = self.bottom_pane.composer_recovery_snapshot(); let composer = ThreadComposerState { text: draft.text, + // A disabled shell composer also treats a leading bang literally. + shell_mode: Some(draft.shell_mode.unwrap_or(false)), text_elements: draft.text_elements, local_images: draft.local_images, remote_image_urls: draft.remote_image_urls, @@ -763,8 +781,14 @@ impl ChatWidget { .as_ref() .and_then(|state| state.provider_error.clone()); let preserve_in_flight_turn = restore_mode.preserve_in_flight_turn; - let restored_task_running = - preserve_in_flight_turn && input_state.as_ref().is_some_and(|state| state.task_running); + let restored_task_running = preserve_in_flight_turn + && (self.bottom_pane.is_task_running() + || input_state.as_ref().is_some_and(|state| state.task_running)); + let restored_agent_turn_running = preserve_in_flight_turn + && (self.turn_lifecycle.agent_turn_running + || input_state + .as_ref() + .is_some_and(|state| state.agent_turn_running)); if let Some(input_state) = input_state { self.bottom_pane.restore_questions(input_state.questions); self.initial_user_message = input_state.initial_user_message; @@ -782,10 +806,8 @@ impl ChatWidget { self.restore_status_surface_overrides(input_state.status_surface_overrides); self.safety_buffering_prompt = input_state.safety_buffering_prompt; self.safety_buffering_source = input_state.safety_buffering_source; - self.turn_lifecycle.restore_running( - preserve_in_flight_turn && input_state.agent_turn_running, - Instant::now(), - ); + self.turn_lifecycle + .restore_running(restored_agent_turn_running, Instant::now()); self.input_queue.user_turn_pending_start = preserve_in_flight_turn && input_state.user_turn_pending_start; self.input_queue.pending_user_message_client_id = @@ -864,6 +886,12 @@ impl ChatWidget { self.request_redraw(); } + pub(crate) fn pause_input_for_resume_handoff(&mut self) -> std::sync::Arc<()> { + let handoff = std::sync::Arc::new(()); + self.input_queue.resume_handoff = std::sync::Arc::downgrade(&handoff); + handoff + } + pub(crate) fn set_queue_autosend_suppressed(&mut self, suppressed: bool) { self.input_queue.suppress_queue_autosend = suppressed; } diff --git a/codex-rs/tui/src/chatwidget/input_submission.rs b/codex-rs/tui/src/chatwidget/input_submission.rs index a7d2bbdf71..a2da2d3461 100644 --- a/codex-rs/tui/src/chatwidget/input_submission.rs +++ b/codex-rs/tui/src/chatwidget/input_submission.rs @@ -51,6 +51,13 @@ impl ChatWidget { } } + pub(crate) fn reject_user_shell_command(&mut self, message: String) { + // No turn event will arrive to clear the pending shell submission. + self.input_queue.user_turn_pending_start = false; + self.add_error_message(message); + self.maybe_send_next_queued_input(); + } + fn submit_shell_command_with_history( &mut self, command: &str, diff --git a/codex-rs/tui/src/chatwidget/questions.rs b/codex-rs/tui/src/chatwidget/questions.rs index c65a10c216..7e702bf999 100644 --- a/codex-rs/tui/src/chatwidget/questions.rs +++ b/codex-rs/tui/src/chatwidget/questions.rs @@ -178,6 +178,7 @@ impl ChatWidget { let cursor = main.cursor; self.restore_composer_state(ThreadComposerState { text: main.text, + shell_mode: None, text_elements: main.text_elements, local_images: main.local_images, remote_image_urls: main.remote_image_urls, diff --git a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs index 0109ea7011..bc41cfa14e 100644 --- a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs +++ b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs @@ -1382,7 +1382,7 @@ async fn blocked_image_restore_with_remote_images_keeps_local_placeholder_mappin } #[tokio::test] -async fn startup_draft_handoff_preserves_literal_and_shell_bang() { +async fn draft_handoffs_preserve_literal_and_shell_bang() { for text in [ "!echo literal", "!echo literal\nsecond line", @@ -1400,6 +1400,13 @@ async fn startup_draft_handoff_preserves_literal_and_shell_bang() { chat.thread_id = Some(ThreadId::new()); chat.restore_startup_draft(draft.clone()); assert_eq!(chat.bottom_pane.composer_draft_snapshot(), draft); + let input = chat.capture_thread_input_state(); + chat.restore_thread_input_state( + input, + ThreadInputStateRestoreMode { + preserve_in_flight_turn: false, + }, + ); chat.handle_key_event(KeyEvent::from(KeyCode::Enter)); assert_hidden_shell_payload_is_literal(ops.try_recv(), text.trim().to_string()); } @@ -1413,6 +1420,13 @@ async fn startup_draft_handoff_preserves_literal_and_shell_bang() { let (mut chat, _rx, mut ops) = make_chatwidget_manual(/*model_override*/ None).await; chat.thread_id = Some(ThreadId::new()); chat.restore_startup_draft(draft); + let input = chat.capture_thread_input_state(); + chat.restore_thread_input_state( + input, + ThreadInputStateRestoreMode { + preserve_in_flight_turn: false, + }, + ); chat.handle_key_event(KeyEvent::from(KeyCode::Enter)); assert_matches!(ops.try_recv(), Ok(Op::RunUserShellCommand { command }) if command == "pwd"); } @@ -2744,7 +2758,26 @@ async fn reconnect_resumes_unsent_input_and_reconciles_confirmed_submissions() { let mut input = chat.capture_thread_input_state(); let saved_source = input.as_ref().unwrap().safety_buffering_source; assert_eq!(saved_source, UserMessageSource::Prompt); + if !pending_start { + let mut running_input = input.clone().unwrap(); + running_input.reconcile_confirmed_messages(std::slice::from_ref(&client_id)); + let (mut restored, _rx, mut ops) = + make_chatwidget_manual(/*model_override*/ None).await; + restored.restore_thread_input_state( + Some(running_input), + ThreadInputStateRestoreMode { + preserve_in_flight_turn: true, + }, + ); + assert!(restored.turn_lifecycle.agent_turn_running); + assert!(restored.input_queue.pending_steers.is_empty()); + assert!(!restored.input_queue.submit_pending_steers_after_interrupt); + restored.on_interrupted_turn(TurnAbortReason::Interrupted); + assert_no_submit_op(&mut ops); + } // Missing history and unrelated receipts never authorize resubmitting uncertain input. + // Each pass restores state captured from the prior pass, so an exact receipt discovered + // on a later resume still releases the ordinary follow-up exactly once. for ids in [ vec![], vec!["different-submission".into()], diff --git a/codex-rs/tui/src/chatwidget/tests/helpers.rs b/codex-rs/tui/src/chatwidget/tests/helpers.rs index 8fce9e01a2..00bbe1e64b 100644 --- a/codex-rs/tui/src/chatwidget/tests/helpers.rs +++ b/codex-rs/tui/src/chatwidget/tests/helpers.rs @@ -6,6 +6,13 @@ use codex_app_server_protocol::PluginAvailability; use codex_utils_absolute_path::test_support::PathExt; use pretty_assertions::assert_eq; +#[track_caller] +pub(crate) fn assert_pending_steers_unadmitted(input: &ThreadInputState) { + for steer in &input.pending_steers { + assert_eq!(steer.admitted_turn_id, None); + } +} + pub(super) async fn test_config() -> (tempfile::TempDir, Config) { // Start from the built-in defaults so tests do not inherit host/system config. let codex_home = tempfile::Builder::new() diff --git a/codex-rs/tui/src/chatwidget/user_messages.rs b/codex-rs/tui/src/chatwidget/user_messages.rs index eaa2b70c79..9963b63b0c 100644 --- a/codex-rs/tui/src/chatwidget/user_messages.rs +++ b/codex-rs/tui/src/chatwidget/user_messages.rs @@ -133,6 +133,7 @@ pub(super) enum QueueDrain { #[derive(Debug, Clone, PartialEq, Default)] pub(super) struct ThreadComposerState { pub(super) text: String, + pub(super) shell_mode: Option, pub(super) local_images: Vec, pub(super) remote_image_urls: Vec, pub(super) text_elements: Vec, @@ -169,7 +170,7 @@ pub(crate) struct ThreadInputState { pub(super) queued_user_messages: VecDeque, pub(super) queued_user_message_history_records: VecDeque, pub(crate) recovered_queue: bool, - /// Reconcile delivery after transport recovery, not after a fork or ordinary thread switch. + /// Reconcile delivery against resumed history after recovery or an idle thread switch. pub(crate) reconnect_pending: bool, pub(super) user_turn_pending_start: bool, pub(super) pending_user_message_client_id: Option, @@ -886,6 +887,7 @@ impl ThreadInputState { pub(crate) fn reconcile_confirmed_messages(&mut self, confirmed_ids: &[String]) { self.pending_steers .retain(|steer| !confirmed_ids.contains(&steer.client_id)); + self.submit_pending_steers_after_interrupt &= !self.pending_steers.is_empty(); remove_confirmed_queued_messages( &mut self.queued_user_messages, &mut self.queued_user_message_history_records,