diff --git a/codex-rs/tui/src/app/event_dispatch.rs b/codex-rs/tui/src/app/event_dispatch.rs index 4a97366d41..248baccee8 100644 --- a/codex-rs/tui/src/app/event_dispatch.rs +++ b/codex-rs/tui/src/app/event_dispatch.rs @@ -574,7 +574,7 @@ impl App { { Ok(()) => { // Keep local input without replacing the fork's running state. - self.chat_widget.restore_reconnected_input(retained_input); + self.chat_widget.restore_reconnected_input(retained_input, &[]); if let Some(err) = name_error { self.chat_widget.add_error_message(err); } diff --git a/codex-rs/tui/src/app/reconnect.rs b/codex-rs/tui/src/app/reconnect.rs index bf07b73720..61f7edc1e3 100644 --- a/codex-rs/tui/src/app/reconnect.rs +++ b/codex-rs/tui/src/app/reconnect.rs @@ -1,6 +1,6 @@ //! TUI-only out-of-process reconnection. Each attempt initializes a fresh client and rejoins existing -//! threads using ordinary resume/history semantics; no user operation is retried. -//! Offline input and old async completions are quarantined. +//! threads using ordinary resume/history semantics. Confirmed submissions are reconciled before +//! unsent queues resume; uncertain submissions and old async completions remain quarantined. use super::*; use crate::app_server_session::ResumeModelSettings; @@ -118,7 +118,7 @@ pub(super) async fn reconnect( } // Transport errors can contain endpoint credentials. Do not render or log them. } - color_eyre::eyre::bail!("app-server session could not be restored") + color_eyre::eyre::bail!("Server connection could not be restored") } impl App { @@ -261,7 +261,7 @@ impl App { let displayed = self.current_displayed_thread_id(); let mut input = self.chat_widget.capture_thread_input_state(); if let Some(input) = input.as_mut() { - input.recovered_queue = true; + input.reconnect_pending = true; } self.store_active_thread_receiver().await; // Old request handles stay attached to the dead connection. Rotating the event channel @@ -347,7 +347,7 @@ impl App { self.agents_overview.request_id = None; self.agents_overview.refresh_pending = false; for input in self.agents_overview.input_states.values_mut() { - input.recovered_queue = true; + input.reconnect_pending = true; } self.pending_startup_thread_start = false; // Move cached UI state into fresh channels. Old producers retain the old sender/store, @@ -370,7 +370,7 @@ impl App { .buffer .retain(|event| matches!(event, ThreadBufferedEvent::Notification(_))); if let Some(input) = store.input_state.as_mut() { - input.recovered_queue = true; + input.reconnect_pending = true; } *replacement.store.lock().await = store; *channel = replacement; @@ -378,7 +378,7 @@ impl App { if let Some(id) = self.current_displayed_thread_id() && let Some(mut input) = input.clone() { - input.recovered_queue = true; + input.reconnect_pending = true; self.agents_overview.input_states.insert(id, input); } if let Some(mut started) = thread { @@ -442,7 +442,7 @@ impl App { /*initial_user_message*/ None, ); self.replace_chat_widget(ChatWidget::new_with_app_event(init)); - self.chat_widget.restore_reconnected_input(input); + self.chat_widget.restore_reconnected_input(input, &[]); } // Discover tasks whose notifications were missed, without clearing retained rows. // A hidden overview performs this discovery when it is next opened. @@ -485,9 +485,9 @@ impl App { ); } self.feedback_audience = bootstrap.feedback_audience; - self.chat_widget.add_info_message( - "Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused.".into(), /*hint*/ None, - ); + if displayed.is_some_and(|id| !self.thread_unavailable(id)) { + self.chat_widget.maybe_send_next_queued_input(); + } let connected_notice_key = crate::status::remote_connection::server_version_notice_key( &self.app_server_target, app_server.server_codex_home(), diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 16091cdb1a..123a46e291 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -1432,6 +1432,15 @@ async fn replay_only_thread_keeps_restored_queue_visible() { let (mut app, _app_event_rx, _op_rx) = make_test_app_with_channels().await; let thread_id = ThreadId::new(); let session = test_thread_session(thread_id, test_path_buf("/tmp/project")); + app.thread_event_channels.insert( + thread_id, + ThreadEventChannel::new_with_session( + THREAD_EVENT_CHANNEL_CAPACITY, + session.clone(), + Vec::new(), + ), + ); + app.activate_thread_channel(thread_id).await; app.chat_widget.handle_thread_session(session.clone()); app.chat_widget.handle_server_notification( turn_started_notification(thread_id, "turn-1"), @@ -1445,10 +1454,11 @@ async fn replay_only_thread_keeps_restored_queue_visible() { .apply_external_edit("queued follow-up".to_string()); app.chat_widget .handle_key_event(KeyEvent::new(KeyCode::Tab, KeyModifiers::NONE)); - let input_state = app + let mut input_state = app .chat_widget .capture_thread_input_state() .expect("expected queued follow-up state"); + input_state.reconnect_pending = true; let (chat_widget, _app_event_tx, _rx, mut new_op_rx) = make_chatwidget_manual_with_sender().await; @@ -1620,14 +1630,25 @@ async fn replay_thread_snapshot_does_not_submit_queue_before_replay_catches_up() agent_message_delta_notification(thread_id, "turn-1", "agent-1", "streaming"), /*replay_kind*/ None, ); + app.chat_widget + .apply_external_edit("accepted steer".to_string()); + app.chat_widget + .handle_key_event(KeyEvent::from(KeyCode::Enter)); app.chat_widget .apply_external_edit("queued follow-up".to_string()); app.chat_widget .handle_key_event(KeyEvent::new(KeyCode::Tab, KeyModifiers::NONE)); - let input_state = app + let mut input_state = app .chat_widget .capture_thread_input_state() .expect("expected queued follow-up state"); + let client_id = input_state + .pending_steers + .front() + .unwrap() + .client_id + .clone(); + input_state.reconnect_pending = true; let (chat_widget, _app_event_tx, _rx, mut new_op_rx) = make_chatwidget_manual_with_sender().await; @@ -1641,6 +1662,21 @@ async fn replay_thread_snapshot_does_not_submit_queue_before_replay_catches_up() session: None, turns: Vec::new(), events: vec![ + ThreadBufferedEvent::Notification(Box::new(ServerNotification::ItemCompleted( + codex_app_server_protocol::ItemCompletedNotification { + thread_id: thread_id.to_string(), + turn_id: "turn-1".into(), + completed_at_ms: 0, + item: ThreadItem::UserMessage { + id: "accepted".into(), + client_id: Some(client_id), + content: vec![UserInput::Text { + text: "accepted steer".into(), + text_elements: Vec::new(), + }], + }, + }, + ))), ThreadBufferedEvent::Notification(Box::new(turn_completed_notification( thread_id, "turn-0", 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 36f1054bdb..a0e061ba1b 100644 --- a/codex-rs/tui/src/app/tests/active_reconnect_tests.rs +++ b/codex-rs/tui/src/app/tests/active_reconnect_tests.rs @@ -12,7 +12,7 @@ use tokio::net::TcpListener; use super::disconnect::serve_reconnect_requests; #[tokio::test] -async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> Result<()> { +async fn reconnect_restores_history_permissions_and_resumes_unsent_input() -> Result<()> { for (recovered_queue, edit_offline, resume_error_code, deferred_notice, notice_enabled) in [ (true, false, -32603, false, false), (true, false, -32603, false, true), @@ -143,9 +143,9 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> "thread/list" | "thread/loaded/list" => Some(json!({"result": {"data": [], "nextCursor": null}})), "thread/goal/get" => Some(json!({"result": {"goal": null}})), "turn/start" => { - assert!(!recovered_queue); + assert!(!edit_offline); let params = request.params.as_ref().unwrap(); - assert_eq!(params["input"][0]["text"], "fresh follow-up"); + assert_eq!(params["input"][0]["text"], if recovered_queue { "old queued input" } else { "fresh follow-up" }); assert_eq!(params["approvalPolicy"], if pending_profile { "never" } else { "on-request" }); assert_eq!(params["sandboxPolicy"]["type"], if pending_profile { json!(null) } else { json!("readOnly") }); assert_eq!(params["collaborationMode"], json!(expected_submitted_mode)); @@ -379,42 +379,41 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> let notices = history .lines() .filter(|line| { - line.contains("Reconnected.") || line.contains("background Codex service") + line.contains("Couldn't confirm whether") + || line.contains("background Codex service") }) .collect::>() .join("\n"); if deferred_notice { assert_snapshot!(notices, @r###" -• Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused. ⚠ A background Codex service is running v2.0.0, older than your Codex CLI "###); + } else if edit_offline { + assert!(notices.contains("unacknowledged prompt")); } else { - insta::allow_duplicates! { - assert_snapshot!(notices, @"• Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused."); - } + assert_eq!(notices, ""); } assert_eq!(history.matches("only once").count(), 1); - if recovered_queue { + if edit_offline { assert!(app.chat_widget.has_queued_follow_up_messages()); assert!(!app.chat_widget.maybe_send_next_queued_input()); - if edit_offline { - app.handle_tui_event( - &mut tui, - &mut session, - TuiEvent::Key(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT)), - ) - .await?; - assert_eq!( - app.chat_widget.composer_text_with_pending(), - "unacknowledged prompt" - ); - assert!(ops.try_recv().is_err()); - } else { - assert_snapshot!( - "restored_conversation", - render_bottom_popup(&app.chat_widget, /*width*/ 80) - ); - } + app.handle_tui_event( + &mut tui, + &mut session, + TuiEvent::Key(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT)), + ) + .await?; + assert_eq!( + app.chat_widget.composer_text_with_pending(), + "unacknowledged prompt" + ); + assert!(ops.try_recv().is_err()); + } else if recovered_queue { + assert!(!app.chat_widget.has_queued_follow_up_messages()); + assert_snapshot!( + "restored_conversation", + render_bottom_popup(&app.chat_widget, /*width*/ 80) + ); } else { app.chat_widget .handle_key_event(KeyEvent::new(KeyCode::Char('u'), KeyModifiers::CONTROL)); @@ -426,6 +425,9 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> app.handle_event(&mut tui, &mut session, event).await?; } } + while let Ok(event) = events.try_recv() { + app.handle_event(&mut tui, &mut session, event).await?; + } session.shutdown().await?; let methods = server.await??; assert_eq!( @@ -440,7 +442,7 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> .iter() .filter(|method| *method == "turn/start") .count(), - usize::from(!recovered_queue) + usize::from(!edit_offline) ); } Ok(()) @@ -742,13 +744,10 @@ pub(super) async fn drain_history( events: &mut mpsc::UnboundedReceiver, ) -> Result { while let Ok(event) = events.try_recv() { - assert!(!matches!( - event, - AppEvent::CodexOp(AppCommand::UserTurn { .. }) - )); if matches!( event, AppEvent::InsertHistoryCell(_) + | AppEvent::CodexOp(AppCommand::UserTurn { .. }) | AppEvent::BeginThreadSwitchHistoryReplayBuffer | AppEvent::EndInitialHistoryReplayBuffer ) { diff --git a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__restored_conversation.snap b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__restored_conversation.snap index 3f6cf1fb82..d7b02a4d9b 100644 --- a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__restored_conversation.snap +++ b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__restored_conversation.snap @@ -2,10 +2,6 @@ source: tui/src/app/tests/active_reconnect_tests.rs expression: "render_bottom_popup(&app.chat_widget, 80)" --- -• Queued follow-up inputs - ↳ old queued input - ⌥+↑ edit last queued message - › kept draft gpt-test default Plan mode (shift+tab to cycle) diff --git a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__unavailable_conversation.snap b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__unavailable_conversation.snap index 44faef27ad..335071a077 100644 --- a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__unavailable_conversation.snap +++ b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__unavailable_conversation.snap @@ -3,7 +3,6 @@ source: tui/src/app/tests/active_reconnect_tests.rs expression: "format!(\"{history}\\n{}\",\nrender_bottom_popup(&app.chat_widget,\n80)).replace(&test_path_buf(\"/tmp/project\").display().to_string(),\n\"/tmp/project\")" --- • This conversation is unavailable. Its cached transcript and draft remain here; input is paused. Open the agent picker or return to the parent to continue. -• Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused. ⚠ A background Codex service is running v2.0.0, different from your Codex CLI v0.0.0. › unavailable draft diff --git a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__disconnect__offline_draft.snap b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__disconnect__offline_draft.snap index a7f5173a83..fcd2963fdb 100644 --- a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__disconnect__offline_draft.snap +++ b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__disconnect__offline_draft.snap @@ -2,8 +2,10 @@ source: tui/src/app/tests/disconnect_tests.rs expression: "format!(\"{}\\n{}\",\nlines_to_single_string(&app.transcript_cells.last().unwrap().display_lines(80)),\nrender_bottom_popup(&app.chat_widget, 80))" --- -■ Connection lost. Attempting to reconnect… -• Reconnecting to app-server… (0s) + +› only once + +• Reconnecting to server… (0s) • Queued follow-up inputs ↳ queued before disconnect diff --git a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__navigation_reconnect__reconnected_unavailable_conversation.snap b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__navigation_reconnect__reconnected_unavailable_conversation.snap index 88842c333e..432d5cf464 100644 --- a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__navigation_reconnect__reconnected_unavailable_conversation.snap +++ b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__navigation_reconnect__reconnected_unavailable_conversation.snap @@ -5,7 +5,6 @@ expression: "format!(\"{content}\\n{}\", render_bottom_popup(&app.chat_widget, 8 Cached previous conversation • This conversation is unavailable. Its cached transcript and draft remain here; input is paused. Open the agent picker or return to the parent to continue. -• Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused. • The rename target disappeared. Unsubmitted title: Keep this task draft! › latest draft diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 854a318e9e..78178575b1 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -1878,10 +1878,38 @@ impl App { self.chat_widget .remember_realtime_delegated_reasoning_turn(turn_id); } + let confirmed_message_ids = snapshot + .turns + .iter() + .flat_map(|turn| &turn.items) + .chain(snapshot.events.iter().filter_map(|event| { + let ThreadBufferedEvent::Notification(notification) = event else { + return None; + }; + if let ServerNotification::ItemStarted(notification) = notification.as_ref() { + Some(¬ification.item) + } else if let ServerNotification::ItemCompleted(notification) = + notification.as_ref() + { + Some(¬ification.item) + } else { + None + } + })) + .filter_map(|item| { + if let ThreadItem::UserMessage { client_id, .. } = item { + client_id.clone() + } else { + None + } + }) + .collect::>(); let recovered_input = snapshot .input_state .as_ref() - .is_some_and(|input| input.recovered_queue) + .is_some_and(|input| { + input.recovered_queue || input.reconnect_pending || input.has_unconfirmed_messages() + }) .then(|| snapshot.input_state.take()) .flatten(); self.chat_widget.restore_thread_input_state( @@ -1908,10 +1936,12 @@ impl App { self.app_event_tx .send(AppEvent::EndInitialHistoryReplayBuffer); } - if recovered_input.is_some() { + let recovered = recovered_input.is_some(); + if recovered { let mode = has_resumed_collaboration_mode .then(|| self.chat_widget.effective_collaboration_mode()); - self.chat_widget.restore_reconnected_input(recovered_input); + self.chat_widget + .restore_reconnected_input(recovered_input, &confirmed_message_ids); if let Some(mode) = mode { self.chat_widget.set_effective_collaboration_mode(mode); } diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index eb9c9cdfa2..021694a5c5 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -402,6 +402,7 @@ use self::turn_lifecycle::TurnLifecycleState; mod usage; mod user_messages; mod working_directory; +use self::user_messages::MessageDelivery; use self::user_messages::PendingSteer; #[cfg(test)] use self::user_messages::PendingSteerCompareKey; @@ -1350,6 +1351,14 @@ impl ChatWidget { { self.on_user_message_display(display); } + if let Some(client_id) = client_id + && self.input_queue.queued_user_messages.iter().any(|message| { + matches!(&message.delivery, MessageDelivery::Unconfirmed(Some(id)) if id == client_id) + }) + { + self.reconcile_recovered_messages(&[client_id.to_string()]); + self.maybe_send_next_queued_input(); + } } fn on_user_message_display(&mut self, display: UserMessageDisplay) { diff --git a/codex-rs/tui/src/chatwidget/input_flow.rs b/codex-rs/tui/src/chatwidget/input_flow.rs index a6b5cdd9a5..b4d45daf13 100644 --- a/codex-rs/tui/src/chatwidget/input_flow.rs +++ b/codex-rs/tui/src/chatwidget/input_flow.rs @@ -206,6 +206,7 @@ impl ChatWidget { .push_back(QueuedUserMessage { user_message, action, + delivery: MessageDelivery::Unsent, pending_pastes, source, }); @@ -238,6 +239,7 @@ impl ChatWidget { || self.input_queue.suppress_queue_autosend || self.input_queue.rate_limit_recovery_pending || self.input_queue.recovered_queue + || self.input_queue.has_unconfirmed_messages() { return false; } diff --git a/codex-rs/tui/src/chatwidget/input_queue.rs b/codex-rs/tui/src/chatwidget/input_queue.rs index 4ef1fe6e12..13600b2d86 100644 --- a/codex-rs/tui/src/chatwidget/input_queue.rs +++ b/codex-rs/tui/src/chatwidget/input_queue.rs @@ -5,6 +5,7 @@ use std::collections::VecDeque; +use super::MessageDelivery; use super::PendingSteer; use super::QueuedUserMessage; use super::UserMessage; @@ -32,6 +33,7 @@ pub(super) struct InputQueueState { pub(super) queued_user_message_history_records: VecDeque, /// A user turn has been submitted to core, but `TurnStarted` has not arrived yet. pub(super) user_turn_pending_start: bool, + pub(super) pending_user_message_client_id: Option, /// User messages that tried to steer a non-regular turn and must be retried first. pub(super) rejected_steers_queue: VecDeque, /// The origin of each rejected steer, kept in lockstep with its message. @@ -49,10 +51,17 @@ pub(super) struct InputQueueState { pub(super) suppress_queue_autosend: bool, /// Hold submissions while a usage failure or backend-directed model fallback is resolved. pub(super) rate_limit_recovery_pending: bool, + /// Pause for user recovery (for example an image or permission failure), independent of delivery. pub(super) recovered_queue: bool, } impl InputQueueState { + pub(super) fn has_unconfirmed_messages(&self) -> bool { + self.queued_user_messages + .iter() + .any(|message| matches!(message.delivery, MessageDelivery::Unconfirmed(_))) + } + pub(super) fn has_queued_follow_up_messages(&self) -> bool { !self.rejected_steers_queue.is_empty() || !self.queued_user_messages.is_empty() } @@ -63,6 +72,7 @@ impl InputQueueState { self.queued_user_messages.clear(); self.queued_user_message_history_records.clear(); self.user_turn_pending_start = false; + self.pending_user_message_client_id = None; self.rejected_steers_queue.clear(); self.rejected_steer_sources.clear(); self.rejected_steer_history_records.clear(); diff --git a/codex-rs/tui/src/chatwidget/input_restore.rs b/codex-rs/tui/src/chatwidget/input_restore.rs index 3c47f6dcfa..e1a798b522 100644 --- a/codex-rs/tui/src/chatwidget/input_restore.rs +++ b/codex-rs/tui/src/chatwidget/input_restore.rs @@ -584,7 +584,9 @@ impl ChatWidget { .queued_user_message_history_records .clone(), recovered_queue: self.input_queue.recovered_queue, + reconnect_pending: false, user_turn_pending_start: self.input_queue.user_turn_pending_start, + pending_user_message_client_id: self.input_queue.pending_user_message_client_id.clone(), submit_pending_steers_after_interrupt: self .input_queue .submit_pending_steers_after_interrupt, @@ -618,6 +620,8 @@ impl ChatWidget { ); 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 = + input_state.pending_user_message_client_id; self.input_queue.submit_pending_steers_after_interrupt = preserve_in_flight_turn && input_state.submit_pending_steers_after_interrupt; self.update_collaboration_mode_indicator(); @@ -634,6 +638,11 @@ impl ChatWidget { for pending in pending_steers.into_iter().rev() { queued_user_messages.push_front(QueuedUserMessage { source: pending.source, + delivery: if input_state.reconnect_pending { + MessageDelivery::Unconfirmed(Some(pending.client_id)) + } else { + MessageDelivery::Unsent + }, ..QueuedUserMessage::from(pending.user_message) }); queued_user_message_history_records.push_front(pending.history_record); diff --git a/codex-rs/tui/src/chatwidget/input_submission.rs b/codex-rs/tui/src/chatwidget/input_submission.rs index a37e938cdd..5184f1c6ff 100644 --- a/codex-rs/tui/src/chatwidget/input_submission.rs +++ b/codex-rs/tui/src/chatwidget/input_submission.rs @@ -489,6 +489,7 @@ impl ChatWidget { self.note_realtime_typed_input(&submitted_message.text); if render_in_history { self.input_queue.user_turn_pending_start = true; + self.input_queue.pending_user_message_client_id = Some(client_user_message_id.clone()); } // Persist the submitted text to cross-session message history. Mentions are encoded into diff --git a/codex-rs/tui/src/chatwidget/reconnect.rs b/codex-rs/tui/src/chatwidget/reconnect.rs index 310e875e1d..864922c31d 100644 --- a/codex-rs/tui/src/chatwidget/reconnect.rs +++ b/codex-rs/tui/src/chatwidget/reconnect.rs @@ -1,4 +1,4 @@ -//! Disconnects preserve editable input but never automatically retry queued submissions. +//! Disconnects preserve input and reconcile confirmed submissions before resuming unsent queues. //! Unavailable threads retain uncertain prompts in recovered queues before clearing activity. use super::realtime::RealtimeConversationPhase; @@ -22,7 +22,6 @@ impl ChatWidget { if let Some(questions) = &mut self.bottom_pane.questions { questions.delivery_enabled = false; } - self.input_queue.recovered_queue = true; self.input_queue.suppress_queue_autosend = true; self.set_initial_user_message_submit_suppressed(/*suppressed*/ true); if let Some(message) = self.initial_user_message.take() { @@ -31,28 +30,42 @@ impl ChatWidget { self.bottom_pane.ensure_status_indicator(); self.bottom_pane .set_interrupt_hint_visible(/*visible*/ false); - self.set_status_header("Reconnecting to app-server…".to_string()); + self.set_status_header("Reconnecting to server…".to_string()); self.set_footer_hint_override(Some(vec![("ctrl+c".into(), "quit".into())])); - self.add_error_message("Connection lost. Attempting to reconnect…".into()); } /// Restore local input only after replay, which can otherwise move interrupted queues into the draft. - pub(crate) fn restore_reconnected_input(&mut self, input: Option) { + pub(crate) fn restore_reconnected_input( + &mut self, + input: Option, + confirmed_message_ids: &[String], + ) { let running = self.turn_lifecycle.agent_turn_running; + let reconnect_pending = input.as_ref().is_some_and(|input| input.reconnect_pending); if let Some(mut input) = input { - // Its acceptance is unknown. Keep a local copy for manual recovery without - // comparing against partial history or automatically submitting it again. + // Navigation can continue an unresolved recovery while new steers are in flight. + input.reconnect_pending |= input.has_unconfirmed_messages(); + // Pending compact/review requests have no message receipt to reconcile. + input.recovered_queue |= + input.user_turn_pending_start && input.safety_buffering_prompt.is_none(); + // Only an exact submission ID confirms receipt; missing or legacy history + // must never cause us to retry an already submitted message automatically. if input.user_turn_pending_start && let Some(prompt) = input.safety_buffering_prompt.take() { input.queued_user_messages.push_front(QueuedUserMessage { source: input.safety_buffering_source, + delivery: if input.reconnect_pending { + MessageDelivery::Unconfirmed(input.pending_user_message_client_id.take()) + } else { + MessageDelivery::Unsent + }, ..QueuedUserMessage::from(prompt) }); input .queued_user_message_history_records .push_front(UserMessageHistoryRecord::UserMessageText); - input.recovered_queue = true; + input.recovered_queue |= !input.reconnect_pending; } input.current_collaboration_mode = self.current_collaboration_mode.clone(); // Keep the local selection for older servers. Replay reapplies a supplied mode. @@ -67,10 +80,47 @@ impl ChatWidget { }, ); } + if reconnect_pending || self.input_queue.has_unconfirmed_messages() { + self.reconcile_recovered_messages(confirmed_message_ids); + } + if reconnect_pending + && let Some(message) = self + .input_queue + .queued_user_messages + .iter() + .find(|message| matches!(message.delivery, MessageDelivery::Unconfirmed(_))) + { + let preview = user_message_preview_text(message, /*history_record*/ None); + let preview = if preview.trim().is_empty() { + "message with attachments".into() + } else { + crate::text_formatting::truncate_text(&preview, /*max_graphemes*/ 80) + }; + let notice = + format!("Couldn't confirm whether “{preview}” was sent. It hasn't been resent."); + self.add_info_message(notice, /*hint*/ None); + } self.turn_lifecycle.restore_running(running, Instant::now()); self.update_task_running_state(); } + pub(super) fn reconcile_recovered_messages(&mut self, confirmed_message_ids: &[String]) { + let mut index = 0; + self.input_queue.queued_user_messages.retain(|message| { + let confirmed = matches!(&message.delivery, + MessageDelivery::Unconfirmed(Some(id)) if confirmed_message_ids.contains(id)); + if confirmed { + self.input_queue + .queued_user_message_history_records + .remove(index); + } else { + index += 1; + } + !confirmed + }); + self.refresh_pending_input_preview(); + } + pub(crate) fn pause_unavailable_thread(&mut self) { if let Some(questions) = &mut self.bottom_pane.questions { questions.delivery_enabled = false; @@ -87,6 +137,9 @@ impl ChatWidget { .queued_user_messages .push_front(QueuedUserMessage { source: self.safety_buffering_source, + delivery: MessageDelivery::Unconfirmed( + self.input_queue.pending_user_message_client_id.clone(), + ), ..QueuedUserMessage::from(prompt) }); self.input_queue diff --git a/codex-rs/tui/src/chatwidget/startup_submission.rs b/codex-rs/tui/src/chatwidget/startup_submission.rs index 6fe0493a6f..6d9deb7856 100644 --- a/codex-rs/tui/src/chatwidget/startup_submission.rs +++ b/codex-rs/tui/src/chatwidget/startup_submission.rs @@ -54,6 +54,7 @@ impl ChatWidget { || self.input_queue.suppress_queue_autosend || self.input_queue.rate_limit_recovery_pending || self.input_queue.recovered_queue + || self.input_queue.has_unconfirmed_messages() { return; } diff --git a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs index 0646bb1bf6..099806b8a6 100644 --- a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs +++ b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs @@ -389,6 +389,7 @@ async fn parent_owned_thread_preserves_queued_input_before_draining() { let queued_message = QueuedUserMessage { user_message: UserMessage::from("keep this queued prompt"), action: QueuedInputAction::Plain, + delivery: MessageDelivery::Unsent, pending_pastes: vec![("[Image 1]".to_string(), "pasted contents".to_string())], source: UserMessageSource::Prompt, }; @@ -1826,7 +1827,9 @@ async fn restore_thread_input_state_applies_running_state_policy() { queued_user_messages: VecDeque::from([UserMessage::from("already queued").into()]), queued_user_message_history_records: VecDeque::from([queued_history.clone()]), recovered_queue: false, + reconnect_pending: false, user_turn_pending_start: true, + pending_user_message_client_id: None, submit_pending_steers_after_interrupt: true, current_collaboration_mode: chat.current_collaboration_mode.clone(), active_collaboration_mask: chat.active_collaboration_mask.clone(), @@ -1867,7 +1870,8 @@ async fn restore_thread_input_state_applies_running_state_policy() { ); assert!(!chat.has_queued_follow_up_messages()); // Editing the last queued draft must not release the uncertain steer for replay. - assert!(chat.capture_thread_input_state().unwrap().recovered_queue); + assert_eq!(chat.input_queue.pending_steers.len(), 1); + assert!(chat.input_queue.suppress_queue_autosend); chat.handle_restricted_key( KeyEvent::new(KeyCode::Up, KeyModifiers::ALT), RestrictedInputMode::Disconnected, @@ -2359,46 +2363,106 @@ async fn interrupt_prepends_queued_messages_before_existing_composer_text() { } #[tokio::test] -async fn reconnect_holds_only_recovered_input_until_manually_edited() { - for (recovered, pending_start) in [ - (None, false), - (Some("review this old input"), false), - (Some("unacknowledged prompt"), true), - ] { +async fn reconnect_resumes_unsent_input_and_reconciles_confirmed_submissions() { + for pending_start in [false, true] { let (mut chat, _rx, _ops) = make_chatwidget_manual(/*model_override*/ None).await; - if let Some(text) = recovered { - if pending_start { - chat.input_queue.user_turn_pending_start = true; - chat.safety_buffering_prompt = Some(UserMessage::from(text)); - } else { - chat.input_queue - .queued_user_messages - .push_back(UserMessage::from(text).into()); - } - } - chat.pause_for_disconnect(); - let input = chat.capture_thread_input_state(); - let (mut chat, _rx, mut ops) = make_chatwidget_manual(/*model_override*/ None).await; chat.thread_id = Some(ThreadId::new()); - chat.restore_reconnected_input(input); - chat.set_queue_autosend_suppressed(/*suppressed*/ false); - if let Some(text) = recovered { - assert!(!chat.maybe_send_next_queued_input()); - chat.handle_key_event(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT)); - assert_eq!(chat.bottom_pane.composer_text(), text); - assert_no_submit_op(&mut ops); - chat.handle_key_event(KeyEvent::from(KeyCode::Enter)); - assert_matches!(next_submit_op(&mut ops), Op::UserTurn { .. }); - chat.input_queue.user_turn_pending_start = false; + if !pending_start { + handle_turn_started(&mut chat, "turn-1"); } + chat.submit_user_message("first message".into()); + let client_id = if pending_start { + chat.input_queue + .pending_user_message_client_id + .clone() + .unwrap() + } else { + chat.input_queue.pending_steers[0].client_id.clone() + }; chat.input_queue .queued_user_messages - .push_back(UserMessage::from("new follow-up").into()); - assert!(chat.maybe_send_next_queued_input()); - assert_matches!(next_submit_op(&mut ops), Op::UserTurn { .. }); + .push_back(UserMessage::from("follow-up").into()); + chat.pause_for_disconnect(); + let mut input = chat.capture_thread_input_state(); + // Missing history and unrelated receipts never authorize resubmitting uncertain input. + for ids in [ + vec![], + vec!["different-submission".into()], + vec![client_id.clone()], + ] { + let confirmed = ids.contains(&client_id); + input.as_mut().unwrap().reconnect_pending = true; + let (mut restored, mut rx, mut ops) = + make_chatwidget_manual(/*model_override*/ None).await; + restored.thread_id = Some(ThreadId::new()); + if confirmed && pending_start { + // A fresh steer submitted during recovery must remain uncertain after navigation. + let steer = pending_steer("later steer"); + let later_id = steer.client_id.clone(); + input.as_mut().unwrap().pending_steers.push_back(steer); + input.as_mut().unwrap().reconnect_pending = false; + restored.restore_reconnected_input(input, &[]); + for (text, id) in [("first message", &client_id), ("later steer", &later_id)] { + assert_no_submit_op(&mut ops); + restored.on_committed_user_message( + &[UserInput::Text { + text: text.into(), + text_elements: Vec::new(), + }], + Some(id), + /*from_replay*/ false, + "turn-1", + ); + } + } else { + restored.restore_reconnected_input(input, &ids); + assert_eq!(restored.maybe_send_next_queued_input(), confirmed); + } + if confirmed { + assert_matches!(next_submit_op(&mut ops), Op::UserTurn { items, .. } + if items == vec![UserInput::Text { + text: "follow-up".into(), + text_elements: Vec::new(), + }]); + assert!(!restored.has_queued_follow_up_messages()); + } else { + assert_eq!( + restored.queued_user_message_texts(), + vec!["first message", "follow-up"] + ); + assert_no_submit_op(&mut ops); + let notice = drain_insert_history(&mut rx) + .into_iter() + .flatten() + .map(|line| line.to_string()) + .collect::>() + .join("\n"); + insta::allow_duplicates! { + assert_snapshot!(notice, @"• Couldn't confirm whether “first message” was sent. It hasn't been resent."); + } + } + input = restored.capture_thread_input_state(); + } } } +#[tokio::test] +async fn reconnect_keeps_queue_paused_after_pending_compact() { + let (mut chat, _rx, mut ops) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + chat.prepare_local_op_submission(&AppCommand::Compact); + chat.queue_user_message("follow-up".into()); + chat.pause_for_disconnect(); + let mut input = chat.capture_thread_input_state(); + input.as_mut().unwrap().reconnect_pending = true; + chat.restore_reconnected_input(input, &[]); + chat.set_queue_autosend_suppressed(/*suppressed*/ false); + + assert!(!chat.maybe_send_next_queued_input()); + assert_eq!(chat.queued_user_message_texts(), vec!["follow-up"]); + assert_no_submit_op(&mut ops); +} + #[tokio::test(flavor = "multi_thread")] async fn image_submission_is_portable_for_new_turns_and_steers() { let dir = tempfile::tempdir().unwrap(); @@ -2522,6 +2586,9 @@ async fn image_preparation_failure_restores_full_input_without_submitting() { chat.input_queue .pending_steers .push_back(pending_steer("already sent")); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("follow-up").into()); } let pending_steers = chat.input_queue.pending_steers.clone(); let dir = tempfile::tempdir().unwrap(); @@ -2596,6 +2663,26 @@ async fn image_preparation_failure_restores_full_input_without_submitting() { normalize_snapshot_paths(render_bottom_popup(&chat, /*width*/ 80)) ); } + if running { + // A transport recovery must preserve the pause for the failed image draft. + chat.pause_for_disconnect(); + let mut input = chat.capture_thread_input_state(); + input.as_mut().unwrap().reconnect_pending = true; + chat.restore_reconnected_input(input, &[pending_steers[0].client_id.clone()]); + chat.on_committed_user_message( + &[UserInput::Text { + text: "already sent".into(), + text_elements: Vec::new(), + }], + Some(&pending_steers[0].client_id), + /*from_replay*/ false, + "turn", + ); + chat.input_queue.suppress_queue_autosend = false; + handle_turn_completed(&mut chat, "turn", /*duration_ms*/ None); + assert_eq!(chat.queued_user_message_texts(), vec!["follow-up"]); + assert_no_submit_op(&mut op_rx); + } pixels.save(&path).unwrap(); chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)); while let Some(event) = rx.recv().await { diff --git a/codex-rs/tui/src/chatwidget/tests/questions_tests.rs b/codex-rs/tui/src/chatwidget/tests/questions_tests.rs index a7cd7a2d2e..ec82df314b 100644 --- a/codex-rs/tui/src/chatwidget/tests/questions_tests.rs +++ b/codex-rs/tui/src/chatwidget/tests/questions_tests.rs @@ -363,7 +363,7 @@ async fn reconnect_preserves_buffered_question_answer_source() { let (mut restored, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; restored.thread_id = Some(ThreadId::new()); - restored.restore_reconnected_input(state); + restored.restore_reconnected_input(state, &[]); assert_eq!( restored .input_queue @@ -801,7 +801,7 @@ async fn question_drafts_survive_navigation_and_snapshot_replay() { Some(ReplayKind::ThreadSnapshot), ); assert_eq!(question_count(&chat), 1); - chat.restore_reconnected_input(saved); + chat.restore_reconnected_input(saved, &[]); assert!(render_bottom_popup(&chat, /*width*/ 80).contains("second draft")); chat.pause_unavailable_thread(); chat.handle_question_key(KeyEvent::new(KeyCode::Char(']'), KeyModifiers::CONTROL)); diff --git a/codex-rs/tui/src/chatwidget/tests/review_mode.rs b/codex-rs/tui/src/chatwidget/tests/review_mode.rs index 65d6acb0eb..710c3a9bdf 100644 --- a/codex-rs/tui/src/chatwidget/tests/review_mode.rs +++ b/codex-rs/tui/src/chatwidget/tests/review_mode.rs @@ -390,7 +390,9 @@ async fn restore_thread_input_state_restores_pending_steers_without_downgrading_ queued_user_messages, queued_user_message_history_records: VecDeque::new(), recovered_queue: false, + reconnect_pending: false, user_turn_pending_start: false, + pending_user_message_client_id: None, submit_pending_steers_after_interrupt: false, current_collaboration_mode: chat.current_collaboration_mode.clone(), active_collaboration_mask: chat.active_collaboration_mask.clone(), diff --git a/codex-rs/tui/src/chatwidget/user_messages.rs b/codex-rs/tui/src/chatwidget/user_messages.rs index 84756b682f..ed997253bc 100644 --- a/codex-rs/tui/src/chatwidget/user_messages.rs +++ b/codex-rs/tui/src/chatwidget/user_messages.rs @@ -66,10 +66,18 @@ pub(super) enum UserMessageSource { QuestionAnswer, } +/// Whether a recovered message can be submitted without risking a duplicate. +#[derive(Debug, Clone, PartialEq)] +pub(super) enum MessageDelivery { + Unsent, + Unconfirmed(Option), +} + #[derive(Debug, Clone, PartialEq)] pub(super) struct QueuedUserMessage { pub(super) user_message: UserMessage, pub(super) action: QueuedInputAction, + pub(super) delivery: MessageDelivery, pub(super) pending_pastes: Vec<(String, String)>, pub(super) source: UserMessageSource, } @@ -79,6 +87,7 @@ impl QueuedUserMessage { Self { user_message, action, + delivery: MessageDelivery::Unsent, pending_pastes: Vec::new(), source: UserMessageSource::Prompt, } @@ -145,7 +154,10 @@ 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. + pub(crate) reconnect_pending: bool, pub(super) user_turn_pending_start: bool, + pub(super) pending_user_message_client_id: Option, pub(super) submit_pending_steers_after_interrupt: bool, pub(super) current_collaboration_mode: CollaborationMode, pub(super) active_collaboration_mask: Option, @@ -838,3 +850,11 @@ impl ChatWidget { ) } } + +impl ThreadInputState { + pub(crate) fn has_unconfirmed_messages(&self) -> bool { + self.queued_user_messages + .iter() + .any(|message| matches!(message.delivery, MessageDelivery::Unconfirmed(_))) + } +}