tui: reconcile pending steers on item completion

codex-rs/tui/src/chatwidget.rs

@@ -65,7 +65,6 @@ use codex_core::git_info::get_git_repo_root;
 use codex_core::git_info::local_git_branches;
 use codex_core::mcp::McpManager;
 use codex_core::models_manager::manager::ModelsManager;
-use codex_core::parse_turn_item;
 use codex_core::plugins::PluginsManager;
 use codex_core::project_doc::DEFAULT_PROJECT_DOC_FILENAME;
 use codex_core::skills::model::SkillMetadata;
@@ -623,8 +622,10 @@ pub(crate) struct ChatWidget {
     // Steers already submitted to core but not yet committed into history.
     //
     // The bottom pane shows these above queued drafts until core records the
-    // corresponding user message and emits the matching raw response item.
+    // corresponding user message item.
     pending_steers: VecDeque<RenderedUserMessageEvent>,
+    // Whether a steer was submitted during the current turn.
+    steer_submitted_this_turn: bool,
     /// Terminal-appropriate keybinding for popping the most-recently queued
     /// message back into the composer.  Determined once at construction time via
     /// [`queued_message_edit_binding_for_terminal`] and propagated to
@@ -1428,6 +1429,7 @@ impl ChatWidget {
         self.turn_sleep_inhibitor.set_turn_running(true);
         self.saw_plan_update_this_turn = false;
         self.saw_plan_item_this_turn = false;
+        self.steer_submitted_this_turn = false;
         self.plan_delta_buffer.clear();
         self.plan_item_active = false;
         self.adaptive_chunking.reset();
@@ -1496,7 +1498,29 @@ impl ChatWidget {
         self.unified_exec_wait_streak = None;
         self.request_redraw();
 
-        if !from_replay && self.queued_user_messages.is_empty() {
+        let had_pending_steers = !self.pending_steers.is_empty();
+        while let Some(pending_steer) = self.pending_steers.pop_front() {
+            self.last_rendered_user_message_event = Some(pending_steer.clone());
+            if !pending_steer.message.trim().is_empty()
+                || !pending_steer.text_elements.is_empty()
+                || !pending_steer.remote_image_urls.is_empty()
+            {
+                self.add_to_history(history_cell::new_user_prompt(
+                    pending_steer.message,
+                    pending_steer.text_elements,
+                    pending_steer.local_images,
+                    pending_steer.remote_image_urls,
+                ));
+            }
+            self.needs_final_message_separator = false;
+        }
+        self.refresh_pending_input_preview();
+
+        if !from_replay
+            && self.queued_user_messages.is_empty()
+            && !self.steer_submitted_this_turn
+            && !had_pending_steers
+        {
             self.maybe_prompt_plan_implementation();
         }
         // Keep this flag for replayed completion events so a subsequent live TurnComplete can
@@ -2938,6 +2962,7 @@ impl ChatWidget {
             forked_from: None,
             queued_user_messages: VecDeque::new(),
             pending_steers: VecDeque::new(),
+            steer_submitted_this_turn: false,
             queued_message_edit_binding,
             show_welcome_banner: is_first_run,
             startup_tooltip_override,
@@ -3122,6 +3147,7 @@ impl ChatWidget {
             plan_item_active: false,
             queued_user_messages: VecDeque::new(),
             pending_steers: VecDeque::new(),
+            steer_submitted_this_turn: false,
             queued_message_edit_binding,
             show_welcome_banner: is_first_run,
             startup_tooltip_override,
@@ -3287,6 +3313,7 @@ impl ChatWidget {
             forked_from: None,
             queued_user_messages: VecDeque::new(),
             pending_steers: VecDeque::new(),
+            steer_submitted_this_turn: false,
             queued_message_edit_binding,
             show_welcome_banner: false,
             startup_tooltip_override: None,
@@ -3462,13 +3489,10 @@ impl ChatWidget {
                     else {
                         return;
                     };
-                    let should_preview_as_pending_steer = self.is_session_configured()
-                        && self.bottom_pane.is_task_running()
-                        && !self.is_review_mode;
-                    if should_preview_as_pending_steer {
-                        self.submit_user_message_internal(user_message, false);
-                    } else if self.is_session_configured() {
-                        // Submitted is only emitted when steer is enabled.
+                    let should_submit_now =
+                        self.is_session_configured() && !self.is_plan_streaming_in_tui();
+                    if should_submit_now {
+                        // Submitted is emitted when user submits.
                         // Reset any reasoning header only when we are actually submitting a turn.
                         self.reasoning_buffer.clear();
                         self.full_reasoning_buffer.clear();
@@ -4124,6 +4148,7 @@ impl ChatWidget {
             return;
         }
 
+        let render_in_history = render_in_history && !self.agent_turn_running;
         let mut items: Vec<UserInput> = Vec::new();
 
         // Special-case: "!cmd" executes a local shell command instead of sending to the model.
@@ -4295,6 +4320,7 @@ impl ChatWidget {
 
         if let Some(pending_steer) = pending_steer {
             self.pending_steers.push_back(pending_steer);
+            self.steer_submitted_this_turn = true;
             self.refresh_pending_input_preview();
         }
 
@@ -4552,24 +4578,7 @@ impl ChatWidget {
                     self.on_user_message_event(ev);
                 }
             }
-            EventMsg::RawResponseItem(event) => {
-                if !from_replay
-                    && let Some(TurnItem::UserMessage(item)) = parse_turn_item(&event.item)
-                    && let EventMsg::UserMessage(user_message) = item.as_legacy_event()
-                {
-                    let rendered = Self::rendered_user_message_event_from_event(&user_message);
-                    let should_render = if self.pending_steers.front() == Some(&rendered) {
-                        self.pending_steers.pop_front();
-                        self.refresh_pending_input_preview();
-                        true
-                    } else {
-                        self.last_rendered_user_message_event.as_ref() != Some(&rendered)
-                    };
-                    if should_render {
-                        self.on_user_message_event(user_message);
-                    }
-                }
-            }
+            EventMsg::RawResponseItem(_) => {}
             EventMsg::EnteredReviewMode(review_request) => {
                 self.on_entered_review_mode(review_request, from_replay)
             }
@@ -4623,6 +4632,16 @@ impl ChatWidget {
             }
             EventMsg::ItemCompleted(event) => {
                 let item = event.item;
+                if !from_replay && let codex_protocol::items::TurnItem::UserMessage(item) = &item {
+                    let EventMsg::UserMessage(event) = item.as_legacy_event() else {
+                        unreachable!("user message item should convert to a legacy user message");
+                    };
+                    let rendered = Self::rendered_user_message_event_from_event(&event);
+                    if self.pending_steers.front() == Some(&rendered) {
+                        self.pending_steers.pop_front();
+                        self.refresh_pending_input_preview();
+                    }
+                }
                 if let codex_protocol::items::TurnItem::Plan(plan_item) = &item {
                     self.on_plan_item_completed(plan_item.text.clone());
                 }

codex-rs/tui/src/chatwidget/realtime.rs

@@ -51,10 +51,10 @@ impl RealtimeConversationUiState {
 
 #[derive(Clone, Debug, PartialEq)]
 pub(super) struct RenderedUserMessageEvent {
-    message: String,
-    remote_image_urls: Vec<String>,
-    local_images: Vec<PathBuf>,
-    text_elements: Vec<TextElement>,
+    pub(super) message: String,
+    pub(super) remote_image_urls: Vec<String>,
+    pub(super) local_images: Vec<PathBuf>,
+    pub(super) text_elements: Vec<TextElement>,
 }
 
 impl RenderedUserMessageEvent {

codex-rs/tui/src/chatwidget/tests.rs

@@ -43,6 +43,7 @@ use codex_protocol::items::AgentMessageContent;
 use codex_protocol::items::AgentMessageItem;
 use codex_protocol::items::PlanItem;
 use codex_protocol::items::TurnItem;
+use codex_protocol::items::UserMessageItem;
 use codex_protocol::models::MessagePhase;
 use codex_protocol::openai_models::ModelPreset;
 use codex_protocol::openai_models::ReasoningEffortPreset;
@@ -78,7 +79,6 @@ use codex_protocol::protocol::PatchApplyBeginEvent;
 use codex_protocol::protocol::PatchApplyEndEvent;
 use codex_protocol::protocol::PatchApplyStatus as CorePatchApplyStatus;
 use codex_protocol::protocol::RateLimitWindow;
-use codex_protocol::protocol::RawResponseItemEvent;
 use codex_protocol::protocol::ReviewRequest;
 use codex_protocol::protocol::ReviewTarget;
 use codex_protocol::protocol::SessionSource;
@@ -1710,6 +1710,7 @@ async fn make_chatwidget_manual(
         startup_tooltip_override: None,
         queued_user_messages: VecDeque::new(),
         pending_steers: VecDeque::new(),
+        steer_submitted_this_turn: false,
         queued_message_edit_binding: crate::key_hint::alt(KeyCode::Up),
         suppress_session_configured_redraw: false,
         pending_notification: None,
@@ -1831,27 +1832,6 @@ fn drain_insert_history(
     out
 }
 
-fn raw_user_message_event(text: &str) -> Event {
-    raw_user_message_event_for_inputs(
-        format!("raw-user-{text}"),
-        vec![UserInput::Text {
-            text: text.to_string(),
-            text_elements: Vec::new(),
-        }],
-    )
-}
-
-fn raw_user_message_event_for_inputs(id: String, items: Vec<UserInput>) -> Event {
-    let response_item: codex_protocol::models::ResponseItem =
-        codex_protocol::models::ResponseInputItem::from(items).into();
-    Event {
-        id,
-        msg: EventMsg::RawResponseItem(RawResponseItemEvent {
-            item: response_item,
-        }),
-    }
-}
-
 fn lines_to_single_string(lines: &[ratatui::text::Line<'static>]) -> String {
     let mut s = String::new();
     for line in lines {
@@ -3075,6 +3055,31 @@ fn complete_assistant_message(
     });
 }
 
+fn complete_user_message(chat: &mut ChatWidget, item_id: &str, text: &str) {
+    complete_user_message_for_inputs(
+        chat,
+        item_id,
+        vec![UserInput::Text {
+            text: text.to_string(),
+            text_elements: Vec::new(),
+        }],
+    );
+}
+
+fn complete_user_message_for_inputs(chat: &mut ChatWidget, item_id: &str, content: Vec<UserInput>) {
+    chat.handle_codex_event(Event {
+        id: format!("raw-{item_id}"),
+        msg: EventMsg::ItemCompleted(ItemCompletedEvent {
+            thread_id: ThreadId::new(),
+            turn_id: "turn-1".to_string(),
+            item: TurnItem::UserMessage(UserMessageItem {
+                id: item_id.to_string(),
+                content,
+            }),
+        }),
+    });
+}
+
 fn begin_exec(chat: &mut ChatWidget, call_id: &str, raw_cmd: &str) -> ExecCommandBeginEvent {
     begin_exec_with_source(chat, call_id, raw_cmd, ExecCommandSource::Agent)
 }
@@ -3470,7 +3475,7 @@ async fn unified_exec_begin_restores_working_status_snapshot() {
 }
 
 #[tokio::test]
-async fn steer_enter_uses_pending_steers_while_plan_stream_is_active() {
+async fn steer_enter_queues_while_plan_stream_is_active() {
     let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
     chat.thread_id = Some(ThreadId::new());
     chat.set_feature_enabled(Feature::CollaborationModes, true);
@@ -3487,30 +3492,14 @@ async fn steer_enter_uses_pending_steers_while_plan_stream_is_active() {
     chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
 
     assert_eq!(chat.active_collaboration_mode_kind(), ModeKind::Plan);
-    assert!(chat.queued_user_messages.is_empty());
-    assert_eq!(chat.pending_steers.len(), 1);
+    assert_eq!(chat.queued_user_messages.len(), 1);
     assert_eq!(
-        chat.pending_steers.front().unwrap().message(),
+        chat.queued_user_messages.front().unwrap().text,
         "queued submission"
     );
-    match next_submit_op(&mut op_rx) {
-        Op::UserTurn { items, .. } => assert_eq!(
-            items,
-            vec![UserInput::Text {
-                text: "queued submission".to_string(),
-                text_elements: Vec::new(),
-            }]
-        ),
-        other => panic!("expected Op::UserTurn, got {other:?}"),
-    }
-    assert!(drain_insert_history(&mut rx).is_empty());
-
-    chat.handle_codex_event(raw_user_message_event("queued submission"));
-
     assert!(chat.pending_steers.is_empty());
-    let inserted = drain_insert_history(&mut rx);
-    assert_eq!(inserted.len(), 1);
-    assert!(lines_to_single_string(&inserted[0]).contains("queued submission"));
+    assert_no_submit_op(&mut op_rx);
+    assert!(drain_insert_history(&mut rx).is_empty());
 }
 
 #[tokio::test]
@@ -3535,12 +3524,9 @@ async fn steer_enter_uses_pending_steers_while_turn_is_running_without_streaming
     }
     assert!(drain_insert_history(&mut rx).is_empty());
 
-    chat.handle_codex_event(raw_user_message_event("queued while running"));
+    complete_user_message(&mut chat, "user-1", "queued while running");
 
     assert!(chat.pending_steers.is_empty());
-    let inserted = drain_insert_history(&mut rx);
-    assert_eq!(inserted.len(), 1);
-    assert!(lines_to_single_string(&inserted[0]).contains("queued while running"));
 }
 
 #[tokio::test]
@@ -3571,12 +3557,9 @@ async fn steer_enter_uses_pending_steers_while_final_answer_stream_is_active() {
     }
     assert!(drain_insert_history(&mut rx).is_empty());
 
-    chat.handle_codex_event(raw_user_message_event("queued while streaming"));
+    complete_user_message(&mut chat, "user-1", "queued while streaming");
 
     assert!(chat.pending_steers.is_empty());
-    let inserted = drain_insert_history(&mut rx);
-    assert_eq!(inserted.len(), 1);
-    assert!(lines_to_single_string(&inserted[0]).contains("queued while streaming"));
 }
 
 #[tokio::test]
@@ -3598,52 +3581,6 @@ async fn failed_pending_steer_submit_does_not_add_pending_preview() {
     assert!(drain_insert_history(&mut rx).is_empty());
 }
 
-#[tokio::test]
-async fn raw_response_item_with_canonicalized_user_payload_clears_pending_steer() {
-    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
-    let items = vec![
-        UserInput::Text {
-            text: "hello".to_string(),
-            text_elements: vec![TextElement::new((1..4).into(), Some("ell".to_string()))],
-        },
-        UserInput::Text {
-            text: " world".to_string(),
-            text_elements: vec![TextElement::new((1..6).into(), Some("world".to_string()))],
-        },
-    ];
-    chat.pending_steers
-        .push_back(ChatWidget::rendered_user_message_event_from_inputs(&items));
-    chat.refresh_pending_input_preview();
-
-    chat.handle_codex_event(raw_user_message_event_for_inputs(
-        "raw-user-canonicalized".to_string(),
-        items,
-    ));
-
-    assert!(chat.pending_steers.is_empty());
-    let inserted = drain_insert_history(&mut rx);
-    assert_eq!(inserted.len(), 1);
-    assert!(lines_to_single_string(&inserted[0]).contains("hello world"));
-}
-
-#[tokio::test]
-async fn raw_response_item_does_not_duplicate_locally_rendered_user_message() {
-    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
-    chat.thread_id = Some(ThreadId::new());
-
-    chat.bottom_pane
-        .set_composer_text("hello".to_string(), Vec::new(), Vec::new());
-    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
-
-    let inserted = drain_insert_history(&mut rx);
-    assert_eq!(inserted.len(), 1);
-    assert!(lines_to_single_string(&inserted[0]).contains("hello"));
-
-    chat.handle_codex_event(raw_user_message_event("hello"));
-
-    assert!(drain_insert_history(&mut rx).is_empty());
-}
-
 #[tokio::test]
 async fn live_legacy_agent_message_after_item_completed_does_not_duplicate_assistant_message() {
     let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
@@ -3670,8 +3607,8 @@ async fn live_legacy_agent_message_after_item_completed_does_not_duplicate_assis
 }
 
 #[tokio::test]
-async fn raw_response_item_only_pops_front_pending_steer() {
-    let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(None).await;
+async fn item_completed_only_pops_front_pending_steer() {
+    let (mut chat, _rx, _op_rx) = make_chatwidget_manual(None).await;
     chat.pending_steers
         .push_back(ChatWidget::rendered_user_message_event_from_inputs(&[
             UserInput::Text {
@@ -3688,21 +3625,15 @@ async fn raw_response_item_only_pops_front_pending_steer() {
         ]));
     chat.refresh_pending_input_preview();
 
-    chat.handle_codex_event(raw_user_message_event("other"));
+    complete_user_message(&mut chat, "user-other", "other");
 
     assert_eq!(chat.pending_steers.len(), 2);
     assert_eq!(chat.pending_steers.front().unwrap().message(), "first");
-    let inserted = drain_insert_history(&mut rx);
-    assert_eq!(inserted.len(), 1);
-    assert!(lines_to_single_string(&inserted[0]).contains("other"));
 
-    chat.handle_codex_event(raw_user_message_event("first"));
+    complete_user_message(&mut chat, "user-first", "first");
 
     assert_eq!(chat.pending_steers.len(), 1);
     assert_eq!(chat.pending_steers.front().unwrap().message(), "second");
-    let inserted = drain_insert_history(&mut rx);
-    assert_eq!(inserted.len(), 1);
-    assert!(lines_to_single_string(&inserted[0]).contains("first"));
 }
 
 #[tokio::test]
@@ -3756,23 +3687,17 @@ async fn steer_enter_during_final_stream_preserves_follow_up_prompts_in_order()
     );
     assert!(drain_insert_history(&mut rx).is_empty());
 
-    chat.handle_codex_event(raw_user_message_event("first follow-up"));
+    complete_user_message(&mut chat, "user-1", "first follow-up");
 
     assert_eq!(chat.pending_steers.len(), 1);
     assert_eq!(
         chat.pending_steers.front().unwrap().message(),
         "second follow-up"
     );
-    let first_insert = drain_insert_history(&mut rx);
-    assert_eq!(first_insert.len(), 1);
-    assert!(lines_to_single_string(&first_insert[0]).contains("first follow-up"));
 
-    chat.handle_codex_event(raw_user_message_event("second follow-up"));
+    complete_user_message(&mut chat, "user-2", "second follow-up");
 
     assert!(chat.pending_steers.is_empty());
-    let second_insert = drain_insert_history(&mut rx);
-    assert_eq!(second_insert.len(), 1);
-    assert!(lines_to_single_string(&second_insert[0]).contains("second follow-up"));
 }
 
 #[tokio::test]
@@ -3811,12 +3736,33 @@ async fn manual_interrupt_keeps_pending_steers_pending_until_core_commits_them()
     );
     assert_no_submit_op(&mut op_rx);
 
-    chat.handle_codex_event(raw_user_message_event("queued while streaming"));
+    complete_user_message(&mut chat, "user-1", "queued while streaming");
+
+    assert!(chat.pending_steers.is_empty());
+}
+
+#[tokio::test]
+async fn task_complete_renders_unacknowledged_pending_steer() {
+    let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(None).await;
+    chat.thread_id = Some(ThreadId::new());
+    chat.on_task_started();
+
+    chat.bottom_pane
+        .set_composer_text("late steer".to_string(), Vec::new(), Vec::new());
+    chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
+
+    match next_submit_op(&mut op_rx) {
+        Op::UserTurn { .. } => {}
+        other => panic!("expected Op::UserTurn, got {other:?}"),
+    }
+    assert!(drain_insert_history(&mut rx).is_empty());
+
+    chat.on_task_complete(None, false);
 
     assert!(chat.pending_steers.is_empty());
     let inserted = drain_insert_history(&mut rx);
-    assert_eq!(inserted.len(), 2);
-    assert!(lines_to_single_string(&inserted[1]).contains("queued while streaming"));
+    assert_eq!(inserted.len(), 1);
+    assert!(lines_to_single_string(&inserted[0]).contains("late steer"));
 }
 
 #[tokio::test]