From 8962c6a7e8224cb2c0dede0247d8094ef117d311 Mon Sep 17 00:00:00 2001 From: Andrei Hasna Date: Fri, 24 Jul 2026 23:42:46 +0300 Subject: [PATCH] fix(tui): stop losing queued messages and dropping the wrong steer on flush Follow-ups to #398 (chat.flush_queued_messages), from the close analysis of the superseded #399. Bug 1 (data loss): flush_queued_messages drained the rejected-steer and queued-message queues *before* submitting and discarded the submit result. Every refusal path (unavailable thread model, blocked image attachments, closed op channel) therefore destroyed the queued follow-ups and pasted the merged text over the user's live composer draft. The merge is now built from clones, the queues are drained only once the submission is accepted, and the composer draft is snapshotted/restored around a refused submit. An accepted submit can still be re-queued at the front (session not yet configured, auth-profile switch, usage-limit self-heal), so only the originally merged entries are removed. Bug 2 (wrong entry dropped): enqueue_rejected_steer() popped the *front* pending steer, but after a merged flush the rejected steer usually sits behind older pending steers. Add enqueue_rejected_steer_matching_items(), which matches on the compare_key the steer was recorded with, and use it from the thread-routing path that has the rejected items. The notification path keeps the front-pop fallback because that error carries no items. Also: - apply MAX_USER_INPUT_TEXT_CHARS to the merged steer instead of letting an oversized merge fail server-side after the queues were drained; - render a " submit all now" hint in the queued-input preview, driven by the resolved runtime keymap (so a remapped binding shows) and re-plumbed from apply_keymap_update; - fix the two pre-existing clippy::byte_char_slices errors in core/src/unified_exec/process_manager_tests.rs that fail whole-workspace clippy -D warnings (from #327), rather than silencing them. Regression tests cover the refused-submit path (queues + draft intact), the accepted-but-requeued path, a non-front rejected steer, the no-match fallback, the length cap, and the hint rendering/visibility. --- .../src/unified_exec/process_manager_tests.rs | 4 +- codex-rs/tui/src/app/thread_routing.rs | 5 +- codex-rs/tui/src/bottom_pane/chat_composer.rs | 24 ++ codex-rs/tui/src/bottom_pane/mod.rs | 21 ++ .../src/bottom_pane/pending_input_preview.rs | 62 +++++ ...d_message_with_remapped_flush_binding.snap | 24 ++ codex-rs/tui/src/chatwidget.rs | 1 + codex-rs/tui/src/chatwidget/constructor.rs | 3 + codex-rs/tui/src/chatwidget/input_flow.rs | 111 +++++++-- codex-rs/tui/src/chatwidget/input_restore.rs | 33 ++- codex-rs/tui/src/chatwidget/keymap_picker.rs | 3 + ...compact_queues_user_messages_snapshot.snap | 2 +- ..._review_queues_user_messages_snapshot.snap | 2 +- .../tui/src/chatwidget/tests/review_mode.rs | 219 ++++++++++++++++++ 14 files changed, 492 insertions(+), 22 deletions(-) create mode 100644 codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_queued_message_with_remapped_flush_binding.snap diff --git a/codex-rs/core/src/unified_exec/process_manager_tests.rs b/codex-rs/core/src/unified_exec/process_manager_tests.rs index b3debc9c71..b17c136c93 100644 --- a/codex-rs/core/src/unified_exec/process_manager_tests.rs +++ b/codex-rs/core/src/unified_exec/process_manager_tests.rs @@ -148,7 +148,7 @@ async fn output_collection_stays_bounded_across_repeated_drains() { Instant::now() + Duration::from_secs(5), ); let produce = async { - for byte in [b'a', b'b', b'c'] { + for byte in *b"abc" { output_buffer.lock().await.push_chunk( vec![byte; crate::unified_exec::UNIFIED_EXEC_OUTPUT_MAX_BYTES], ); @@ -173,7 +173,7 @@ async fn output_collection_stays_bounded_across_repeated_drains() { let (collected, ()) = tokio::join!(collect, produce); let mut expected = HeadTailBuffer::default(); - for byte in [b'a', b'b', b'c'] { + for byte in *b"abc" { expected.push_chunk(vec![ byte; crate::unified_exec::UNIFIED_EXEC_OUTPUT_MAX_BYTES diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 8da7f06ba1..6d63161539 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -678,7 +678,10 @@ impl App { if let Some(turn_error) = active_turn_not_steerable_turn_error(&error) { - if !self.chat_widget.enqueue_rejected_steer() { + if !self + .chat_widget + .enqueue_rejected_steer_matching_items(items) + { self.chat_widget.add_error_message(turn_error.message); } return Ok(true); diff --git a/codex-rs/tui/src/bottom_pane/chat_composer.rs b/codex-rs/tui/src/bottom_pane/chat_composer.rs index 484556891a..6f4989482a 100644 --- a/codex-rs/tui/src/bottom_pane/chat_composer.rs +++ b/codex-rs/tui/src/bottom_pane/chat_composer.rs @@ -418,6 +418,9 @@ pub(crate) struct ComposerDraftSnapshot { pub(crate) remote_image_urls: Vec, pub(crate) mention_bindings: Vec, pub(crate) pending_pastes: Vec<(String, String)>, + /// Caret position, kept private so the snapshot can only round-trip through + /// [`ChatComposer::restore_draft_snapshot`]. + cursor: usize, } const FOOTER_SPACING_HEIGHT: u16 = 0; @@ -1484,9 +1487,30 @@ impl ChatComposer { remote_image_urls: self.remote_image_urls(), mention_bindings: self.mention_bindings(), pending_pastes: self.pending_pastes(), + cursor: self.current_cursor(), } } + /// Put a previously captured draft back into the composer, caret included. + /// + /// Used to undo a speculative composer write when the submission it belonged to was + /// refused, so the user's in-progress draft is not clobbered. + pub(crate) fn restore_draft_snapshot(&mut self, snapshot: ComposerDraftSnapshot) { + self.restore_draft(ComposerDraft { + text: snapshot.text, + text_elements: snapshot.text_elements, + local_image_paths: snapshot + .local_images + .into_iter() + .map(|image| image.path) + .collect(), + remote_image_urls: snapshot.remote_image_urls, + mention_bindings: snapshot.mention_bindings, + pending_pastes: snapshot.pending_pastes, + cursor: snapshot.cursor, + }); + } + #[cfg(test)] pub(crate) fn local_image_paths(&self) -> Vec { self.attachments.local_image_paths() diff --git a/codex-rs/tui/src/bottom_pane/mod.rs b/codex-rs/tui/src/bottom_pane/mod.rs index 7e48d944c4..a8806deea9 100644 --- a/codex-rs/tui/src/bottom_pane/mod.rs +++ b/codex-rs/tui/src/bottom_pane/mod.rs @@ -473,6 +473,13 @@ impl BottomPane { self.request_redraw(); } + /// Update the key hint for merging every queued follow-up into the active turn so it + /// matches the binding that `ChatWidget` actually listens for. + pub(crate) fn set_queued_message_flush_binding(&mut self, binding: Option) { + self.pending_input_preview.set_flush_binding(binding); + self.request_redraw(); + } + pub(crate) fn set_vim_enabled(&mut self, enabled: bool) { self.composer.set_vim_enabled(enabled); self.request_redraw(); @@ -858,6 +865,14 @@ impl BottomPane { self.composer.draft_snapshot() } + pub(crate) fn restore_composer_draft_snapshot( + &mut self, + snapshot: chat_composer::ComposerDraftSnapshot, + ) { + self.composer.restore_draft_snapshot(snapshot); + self.request_redraw(); + } + #[cfg(test)] pub(crate) fn composer_text_elements(&self) -> Vec { self.composer.text_elements() @@ -1212,10 +1227,13 @@ impl BottomPane { queued: Vec, pending_steers: Vec, rejected_steers: Vec, + flush_available: bool, ) { self.pending_input_preview.pending_steers = pending_steers; self.pending_input_preview.rejected_steers = rejected_steers; self.pending_input_preview.queued_messages = queued; + self.pending_input_preview + .set_flush_available(flush_available); self.request_redraw(); } @@ -2422,6 +2440,7 @@ mod tests { vec!["Queued follow-up question".to_string()], Vec::new(), Vec::new(), + /*flush_available*/ false, ); let width = 48; @@ -2453,6 +2472,7 @@ mod tests { vec!["Queued follow-up question".to_string()], Vec::new(), Vec::new(), + /*flush_available*/ false, ); pane.hide_status_indicator(); @@ -2485,6 +2505,7 @@ mod tests { vec!["Queued follow-up question".to_string()], Vec::new(), Vec::new(), + /*flush_available*/ false, ); let width = 48; diff --git a/codex-rs/tui/src/bottom_pane/pending_input_preview.rs b/codex-rs/tui/src/bottom_pane/pending_input_preview.rs index 96cc1bdfd8..62e90887e5 100644 --- a/codex-rs/tui/src/bottom_pane/pending_input_preview.rs +++ b/codex-rs/tui/src/bottom_pane/pending_input_preview.rs @@ -27,6 +27,12 @@ pub(crate) struct PendingInputPreview { /// Key combination rendered in the hint line. Defaults to Alt+Up but may /// be overridden for terminals where that chord is unavailable. edit_binding: Option, + /// Key combination rendered for merging every queued follow-up into the active turn. + /// Supplied by the resolved runtime keymap so a remapped binding is displayed. + flush_binding: Option, + /// Whether a flush would actually do something right now. The flush only applies to a + /// running turn whose queue head is mergeable, so the hint stays hidden otherwise. + flush_available: bool, /// Key combination rendered for immediately interrupting and sending steers. interrupt_binding: Option, } @@ -40,6 +46,8 @@ impl PendingInputPreview { rejected_steers: Vec::new(), queued_messages: Vec::new(), edit_binding: Some(key_hint::alt(KeyCode::Up)), + flush_binding: None, + flush_available: false, interrupt_binding: Some(key_hint::plain(KeyCode::Esc)), } } @@ -51,6 +59,14 @@ impl PendingInputPreview { self.edit_binding = binding; } + pub(crate) fn set_flush_binding(&mut self, binding: Option) { + self.flush_binding = binding; + } + + pub(crate) fn set_flush_available(&mut self, available: bool) { + self.flush_available = available; + } + pub(crate) fn set_interrupt_binding(&mut self, binding: Option) { self.interrupt_binding = binding; } @@ -151,6 +167,19 @@ impl PendingInputPreview { } } + if self.flush_available + && let Some(flush_binding) = self.flush_binding + { + lines.push( + Line::from(vec![ + " ".into(), + flush_binding.into(), + " submit all now".into(), + ]) + .dim(), + ); + } + if !self.queued_messages.is_empty() && let Some(edit_binding) = self.edit_binding { @@ -348,6 +377,39 @@ mod tests { ); } + #[test] + fn render_queued_message_with_remapped_flush_binding() { + let mut queue = PendingInputPreview::new(); + queue.queued_messages.push("Please continue.".to_string()); + queue.set_flush_binding(Some(key_hint::plain(KeyCode::F(12)))); + queue.set_flush_available(true); + let width = 48; + let height = queue.desired_height(width); + let mut buf = Buffer::empty(Rect::new(0, 0, width, height)); + queue.render(Rect::new(0, 0, width, height), &mut buf); + assert_snapshot!( + "render_queued_message_with_remapped_flush_binding", + format!("{buf:?}") + ); + } + + #[test] + fn flush_hint_is_hidden_when_flush_is_unavailable() { + let mut queue = PendingInputPreview::new(); + queue.queued_messages.push("Please continue.".to_string()); + queue.set_flush_binding(Some(key_hint::shift(KeyCode::Enter))); + let width = 48; + let without_hint = queue.desired_height(width); + + queue.set_flush_available(true); + + assert_eq!( + queue.desired_height(width), + without_hint + 1, + "the flush hint should only add a row once a flush would do something" + ); + } + #[test] fn render_pending_steers_above_queued_messages() { let mut queue = PendingInputPreview::new(); diff --git a/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_queued_message_with_remapped_flush_binding.snap b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_queued_message_with_remapped_flush_binding.snap new file mode 100644 index 0000000000..5bd6498da1 --- /dev/null +++ b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_queued_message_with_remapped_flush_binding.snap @@ -0,0 +1,24 @@ +--- +source: tui/src/bottom_pane/pending_input_preview.rs +expression: "format!(\"{buf:?}\")" +--- +Buffer { + area: Rect { x: 0, y: 0, width: 48, height: 4 }, + content: [ + "• Queued follow-up inputs ", + " ↳ Please continue. ", + " f12 submit all now ", + " ⌥ + ↑ edit last queued message ", + ], + styles: [ + x: 0, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: DIM, + x: 2, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: NONE, + x: 0, y: 1, fg: Reset, bg: Reset, underline: Reset, modifier: DIM, + x: 4, y: 1, fg: Reset, bg: Reset, underline: Reset, modifier: DIM | ITALIC, + x: 20, y: 1, fg: Reset, bg: Reset, underline: Reset, modifier: NONE, + x: 0, y: 2, fg: Reset, bg: Reset, underline: Reset, modifier: DIM, + x: 22, y: 2, fg: Reset, bg: Reset, underline: Reset, modifier: NONE, + x: 0, y: 3, fg: Reset, bg: Reset, underline: Reset, modifier: DIM, + x: 34, y: 3, fg: Reset, bg: Reset, underline: Reset, modifier: NONE, + ] +} diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 335ad17fcb..4bc695d10f 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -179,6 +179,7 @@ use codex_protocol::plan_tool::PlanItemArg as UpdatePlanItemArg; use codex_protocol::plan_tool::StepStatus as UpdatePlanItemStatus; use codex_protocol::request_permissions::RequestPermissionsEvent; use codex_protocol::user_input::ByteRange; +use codex_protocol::user_input::MAX_USER_INPUT_TEXT_CHARS; use codex_protocol::user_input::TextElement; use codex_terminal_detection::Multiplexer; use codex_terminal_detection::TerminalInfo; diff --git a/codex-rs/tui/src/chatwidget/constructor.rs b/codex-rs/tui/src/chatwidget/constructor.rs index b6c8c3ac4c..1fd796a9be 100644 --- a/codex-rs/tui/src/chatwidget/constructor.rs +++ b/codex-rs/tui/src/chatwidget/constructor.rs @@ -301,6 +301,9 @@ impl ChatWidget { widget .bottom_pane .set_queued_message_edit_binding(widget.queued_message_edit_hint_binding); + widget.bottom_pane.set_queued_message_flush_binding( + widget.chat_keymap.flush_queued_messages.first().copied(), + ); #[cfg(target_os = "windows")] widget.bottom_pane.set_windows_degraded_sandbox_active( crate::legacy_core::windows_sandbox::ELEVATED_SANDBOX_NUX_ENABLED diff --git a/codex-rs/tui/src/chatwidget/input_flow.rs b/codex-rs/tui/src/chatwidget/input_flow.rs index b8dac9b54c..9e530070bf 100644 --- a/codex-rs/tui/src/chatwidget/input_flow.rs +++ b/codex-rs/tui/src/chatwidget/input_flow.rs @@ -6,6 +6,20 @@ use super::*; +/// Remove the `count` entries that a flush merged, skipping any entries the submission +/// itself pushed onto the front of `queue` after the merge snapshot was taken. +fn drain_flushed_entries( + queue: &mut std::collections::VecDeque, + len_before_submit: usize, + count: usize, +) { + let added = queue.len().saturating_sub(len_before_submit); + let end = added.saturating_add(count).min(queue.len()); + if added < end { + queue.drain(added..end); + } +} + impl ChatWidget { pub(super) fn handle_composer_input_result( &mut self, @@ -105,30 +119,45 @@ impl ChatWidget { /// order, which preserves each queue's message/history alignment. Queued `/slash` and /// `!shell` entries are dispatch actions rather than prose, so the merge stops at the /// first one and leaves it (and everything behind it) queued for the normal drain. - /// Already-submitted pending steers and the live composer draft are left untouched. + /// Already-submitted pending steers are left untouched. + /// + /// The merge is built from clones and the queues are only drained once the submission is + /// accepted. Submission can legitimately be refused (unavailable thread model, blocked + /// image attachments, a closed op channel), and those refusal paths push the attempted + /// message back into the composer, so the live draft is snapshotted and restored too. + /// Draining up front would silently destroy every queued follow-up and overwrite whatever + /// the user was still typing. pub(super) fn flush_queued_messages(&mut self) { - let rejected_messages = std::mem::take(&mut self.input_queue.rejected_steers_queue); - let mut rejected_history_records = - std::mem::take(&mut self.input_queue.rejected_steer_history_records); - rejected_history_records.resize( - rejected_messages.len(), - UserMessageHistoryRecord::UserMessageText, - ); + let rejected_count = self.input_queue.rejected_steers_queue.len(); + let mut rejected_history_records = self + .input_queue + .rejected_steer_history_records + .iter() + .cloned() + .collect::>(); + rejected_history_records.resize(rejected_count, UserMessageHistoryRecord::UserMessageText); - let mut messages = rejected_messages - .into_iter() + let mut messages = self + .input_queue + .rejected_steers_queue + .iter() + .cloned() .zip(rejected_history_records) .collect::>(); - while self.input_queue.next_queued_message_is_plain() { - let Some(queued_message) = self.input_queue.queued_user_messages.pop_front() else { + let mut queued_count = 0usize; + while let Some(queued_message) = self.input_queue.queued_user_messages.get(queued_count) { + if !matches!(queued_message.action, QueuedInputAction::Plain) { break; - }; + } + let user_message = queued_message.clone().into_user_message(); let history_record = self .input_queue .queued_user_message_history_records - .pop_front() + .get(queued_count) + .cloned() .unwrap_or(UserMessageHistoryRecord::UserMessageText); - messages.push((queued_message.into_user_message(), history_record)); + messages.push((user_message, history_record)); + queued_count += 1; } if messages.is_empty() { @@ -136,11 +165,58 @@ impl ChatWidget { } let (message, history_record) = merge_user_messages_with_history_record(messages); - self.resubmit_queued_user_message_with_history_record( + let actual_chars = message.text.chars().count(); + if actual_chars > MAX_USER_INPUT_TEXT_CHARS { + self.add_error_message(format!( + "Queued messages exceed the maximum combined length of \ + {MAX_USER_INPUT_TEXT_CHARS} characters ({actual_chars} provided). \ + Edit or remove queued messages before submitting them together." + )); + return; + } + + let composer_snapshot = self.bottom_pane.composer_draft_snapshot(); + let rejected_len_before = self.input_queue.rejected_steers_queue.len(); + let rejected_history_len_before = self.input_queue.rejected_steer_history_records.len(); + let queued_len_before = self.input_queue.queued_user_messages.len(); + let queued_history_len_before = self.input_queue.queued_user_message_history_records.len(); + + let accepted = self.resubmit_queued_user_message_with_history_record( message, history_record, ShellEscapePolicy::Disallow, ); + + if accepted { + // An accepted submission can still be re-queued at the *front* while session, + // auth-profile, or usage-limit state resolves. Remove only the original entries + // that were merged, skipping anything the submission itself pushed on top. + drain_flushed_entries( + &mut self.input_queue.rejected_steers_queue, + rejected_len_before, + rejected_count, + ); + drain_flushed_entries( + &mut self.input_queue.rejected_steer_history_records, + rejected_history_len_before, + rejected_count, + ); + drain_flushed_entries( + &mut self.input_queue.queued_user_messages, + queued_len_before, + queued_count, + ); + drain_flushed_entries( + &mut self.input_queue.queued_user_message_history_records, + queued_history_len_before, + queued_count, + ); + } else { + // The refusal paths (`restore_user_message_to_composer`, + // `restore_blocked_image_submission`) replaced the draft with the merged text. + self.bottom_pane + .restore_composer_draft_snapshot(composer_snapshot); + } self.refresh_pending_input_preview(); } @@ -206,10 +282,13 @@ impl ChatWidget { /// Rebuild and update the bottom-pane pending-input preview. pub(super) fn refresh_pending_input_preview(&mut self) { let preview = self.input_queue.preview(); + let flush_available = self.turn_lifecycle.agent_turn_running + && self.input_queue.has_flushable_queued_messages(); self.bottom_pane.set_pending_input_preview( preview.queued_messages, preview.pending_steers, preview.rejected_steers, + flush_available, ); } diff --git a/codex-rs/tui/src/chatwidget/input_restore.rs b/codex-rs/tui/src/chatwidget/input_restore.rs index 391cdf596e..989c4e2b70 100644 --- a/codex-rs/tui/src/chatwidget/input_restore.rs +++ b/codex-rs/tui/src/chatwidget/input_restore.rs @@ -114,8 +114,39 @@ impl ChatWidget { } } + /// Requeue the pending steer that the server actually rejected. + /// + /// A rejection response is not necessarily for the oldest in-flight steer: merged + /// flushes and interleaved steers routinely leave the rejected one behind other pending + /// entries, so popping the front would requeue the wrong message and drop the rejected + /// one. Match on the same `compare_key` the pending steer was recorded with, and fall + /// back to the caller's error path when nothing matches. + pub(crate) fn enqueue_rejected_steer_matching_items(&mut self, items: &[UserInput]) -> bool { + let compare_key = Self::pending_steer_compare_key_from_items(items); + let Some(index) = self + .input_queue + .pending_steers + .iter() + .position(|pending| pending.compare_key == compare_key) + else { + tracing::warn!( + "received active-turn-not-steerable response without a matching pending steer" + ); + return false; + }; + self.enqueue_rejected_steer_at(index) + } + + /// Requeue the oldest pending steer. + /// + /// Used by the notification path, which reports that the active turn is not steerable + /// without echoing back the rejected items. pub(crate) fn enqueue_rejected_steer(&mut self) -> bool { - let Some(pending_steer) = self.input_queue.pending_steers.pop_front() else { + self.enqueue_rejected_steer_at(/*index*/ 0) + } + + fn enqueue_rejected_steer_at(&mut self, index: usize) -> bool { + let Some(pending_steer) = self.input_queue.pending_steers.remove(index) else { tracing::warn!( "received active-turn-not-steerable error without a matching pending steer" ); diff --git a/codex-rs/tui/src/chatwidget/keymap_picker.rs b/codex-rs/tui/src/chatwidget/keymap_picker.rs index 4546ff114e..db6a16af8f 100644 --- a/codex-rs/tui/src/chatwidget/keymap_picker.rs +++ b/codex-rs/tui/src/chatwidget/keymap_picker.rs @@ -176,6 +176,9 @@ impl ChatWidget { ); self.bottom_pane .set_queued_message_edit_binding(self.queued_message_edit_hint_binding); + self.bottom_pane.set_queued_message_flush_binding( + self.chat_keymap.flush_queued_messages.first().copied(), + ); self.bottom_pane.set_keymap_bindings(runtime_keymap); self.request_redraw(); } diff --git a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__compact_queues_user_messages_snapshot.snap b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__compact_queues_user_messages_snapshot.snap index c2698129c9..4291731d91 100644 --- a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__compact_queues_user_messages_snapshot.snap +++ b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__compact_queues_user_messages_snapshot.snap @@ -11,11 +11,11 @@ expression: normalize_snapshot_paths(term.backend().vt100().screen().contents()) - • Working (0s • esc to interrupt) • Messages to be submitted at end of turn ↳ Steer submitted while /compact was running. + shift + enter submit all now › Ask Codewith to do anything diff --git a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__review_queues_user_messages_snapshot.snap b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__review_queues_user_messages_snapshot.snap index d76a7b21a7..c43fbaf733 100644 --- a/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__review_queues_user_messages_snapshot.snap +++ b/codex-rs/tui/src/chatwidget/snapshots/codex_tui__chatwidget__tests__review_queues_user_messages_snapshot.snap @@ -11,11 +11,11 @@ expression: normalize_snapshot_paths(term.backend().vt100().screen().contents()) - • Working (0s • esc to interrupt) • Messages to be submitted at end of turn ↳ Steer submitted while /review was running. + shift + enter submit all now › Ask Codewith to do anything diff --git a/codex-rs/tui/src/chatwidget/tests/review_mode.rs b/codex-rs/tui/src/chatwidget/tests/review_mode.rs index 406c0c8e55..4d9b672b83 100644 --- a/codex-rs/tui/src/chatwidget/tests/review_mode.rs +++ b/codex-rs/tui/src/chatwidget/tests/review_mode.rs @@ -958,6 +958,225 @@ async fn shift_enter_flushes_plain_messages_before_a_queued_shell_command() { assert_eq!(chat.input_queue.pending_steers.len(), 1); } +/// Regression: the flush used to drain the queues before submitting and then throw away the +/// submit result, so a refused submission destroyed every queued follow-up and pasted the +/// merged text over whatever the user was still typing. +#[tokio::test] +async fn unavailable_model_flush_preserves_queued_messages_and_composer_draft() { + let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + chat.on_task_started(); + chat.set_model(""); + chat.input_queue + .pending_steers + .push_back(pending_steer("already pending")); + chat.input_queue + .rejected_steers_queue + .push_back(UserMessage::from("rejected first")); + chat.input_queue + .rejected_steer_history_records + .push_back(UserMessageHistoryRecord::Override( + UserMessageHistoryOverride { + text: "rejected history".to_string(), + text_elements: Vec::new(), + }, + )); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("queued second").into()); + chat.input_queue + .queued_user_message_history_records + .push_back(UserMessageHistoryRecord::UserMessageText); + chat.bottom_pane + .set_composer_text("still editing".to_string(), Vec::new(), Vec::new()); + drain_insert_history(&mut rx); + + let input_before = chat.capture_thread_input_state(); + + chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::SHIFT)); + + assert_no_submit_op(&mut op_rx); + assert_eq!( + chat.capture_thread_input_state(), + input_before, + "a refused flush must leave every queue and the composer draft untouched" + ); + assert_eq!(chat.bottom_pane.composer_text(), "still editing"); + let inserted = drain_insert_history(&mut rx); + assert_eq!(inserted.len(), 1); + assert!( + lines_to_single_string(&inserted[0]).contains("Thread model is unavailable."), + "expected the unavailable-model error" + ); +} + +/// Regression: a submit that never reaches core (closed op channel) must not eat the queue. +#[tokio::test] +async fn failed_flush_submit_preserves_queued_messages_and_composer_draft() { + let (mut chat, _rx, op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + chat.on_task_started(); + chat.input_queue + .rejected_steers_queue + .push_back(UserMessage::from("rejected first")); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("queued second").into()); + chat.bottom_pane + .set_composer_text("still editing".to_string(), Vec::new(), Vec::new()); + drop(op_rx); + + chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::SHIFT)); + + assert_eq!( + chat.queued_user_message_texts(), + vec!["rejected first".to_string(), "queued second".to_string()] + ); + assert!(chat.input_queue.pending_steers.is_empty()); + assert_eq!(chat.bottom_pane.composer_text(), "still editing"); +} + +/// An accepted flush can still be re-queued at the front while the session is being +/// configured. Only the merged originals may be removed, never the accepted replacement. +#[tokio::test] +async fn accepted_flush_preserves_front_requeued_merged_message() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = None; + chat.input_queue + .rejected_steers_queue + .push_back(UserMessage::from("rejected first")); + chat.input_queue + .rejected_steer_history_records + .push_back(UserMessageHistoryRecord::UserMessageText); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("queued second").into()); + chat.input_queue + .queued_user_message_history_records + .push_back(UserMessageHistoryRecord::UserMessageText); + + chat.flush_queued_messages(); + + assert_no_submit_op(&mut op_rx); + assert!(chat.input_queue.rejected_steers_queue.is_empty()); + assert!(chat.input_queue.rejected_steer_history_records.is_empty()); + assert_eq!( + chat.input_queue.queued_user_messages, + VecDeque::from([QueuedUserMessage::new_with_shell_escape_policy( + UserMessage::from("rejected first\nqueued second"), + QueuedInputAction::Plain, + ShellEscapePolicy::Disallow, + )]) + ); + assert_eq!( + chat.input_queue.queued_user_message_history_records, + VecDeque::from([UserMessageHistoryRecord::UserMessageText]) + ); + assert!(chat.input_queue.pending_steers.is_empty()); +} + +/// Regression: the rejected steer is identified by its compare key. After a merged flush the +/// rejected steer sits *behind* older pending steers, so popping the front requeued the wrong +/// message and silently dropped the rejected one. +#[tokio::test] +async fn rejected_steer_is_matched_by_compare_key_not_queue_position() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + chat.on_task_started(); + chat.input_queue + .pending_steers + .push_back(pending_steer("already pending")); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("first queued").into()); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("second queued").into()); + + chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::SHIFT)); + + let Op::UserTurn { items, .. } = next_submit_op(&mut op_rx) else { + panic!("expected one merged queued-message steer"); + }; + assert_eq!( + chat.input_queue + .pending_steers + .iter() + .map(|pending| pending.user_message.text.as_str()) + .collect::>(), + vec!["already pending", "first queued\nsecond queued"], + "the merged steer must be behind the older pending steer" + ); + + assert!(chat.enqueue_rejected_steer_matching_items(&items)); + + assert_eq!( + chat.input_queue + .pending_steers + .iter() + .map(|pending| pending.user_message.text.as_str()) + .collect::>(), + vec!["already pending"], + "the untouched older steer must stay pending" + ); + assert_eq!( + chat.input_queue.rejected_steers_queue, + VecDeque::from([UserMessage::from("first queued\nsecond queued")]) + ); + assert_eq!( + chat.input_queue.rejected_steer_history_records, + VecDeque::from([UserMessageHistoryRecord::UserMessageText]) + ); +} + +#[tokio::test] +async fn rejected_steer_matching_items_reports_no_match() { + let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.input_queue + .pending_steers + .push_back(pending_steer("already pending")); + + assert!( + !chat.enqueue_rejected_steer_matching_items(&[UserInput::Text { + text: "never submitted".to_string(), + text_elements: Vec::new(), + }]) + ); + assert_eq!(chat.input_queue.pending_steers.len(), 1); + assert!(chat.input_queue.rejected_steers_queue.is_empty()); +} + +/// The merged steer is a single user input and must respect the same length cap as any other +/// user input instead of being rejected server-side after the queues were already drained. +#[tokio::test] +async fn oversized_flush_preserves_queued_messages_and_composer_draft() { + let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + chat.on_task_started(); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("a".repeat(MAX_USER_INPUT_TEXT_CHARS)).into()); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("b").into()); + chat.bottom_pane + .set_composer_text("still editing".to_string(), Vec::new(), Vec::new()); + drain_insert_history(&mut rx); + + chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::SHIFT)); + + assert_no_submit_op(&mut op_rx); + assert_eq!(chat.input_queue.queued_user_messages.len(), 2); + assert!(chat.input_queue.pending_steers.is_empty()); + assert_eq!(chat.bottom_pane.composer_text(), "still editing"); + let inserted = drain_insert_history(&mut rx); + assert_eq!(inserted.len(), 1); + assert!( + lines_to_single_string(&inserted[0]).contains("maximum combined length"), + "expected an actionable combined-length error" + ); +} + #[tokio::test] async fn manual_interrupt_restores_pending_steers_to_composer() { let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;