diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index a164b5a24a..afa11fb269 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/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, + // 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 = 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()); } diff --git a/codex-rs/tui/src/chatwidget/realtime.rs b/codex-rs/tui/src/chatwidget/realtime.rs index 38c3d1c4ea..99aefdf4b8 100644 --- a/codex-rs/tui/src/chatwidget/realtime.rs +++ b/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, - local_images: Vec, - text_elements: Vec, + pub(super) message: String, + pub(super) remote_image_urls: Vec, + pub(super) local_images: Vec, + pub(super) text_elements: Vec, } impl RenderedUserMessageEvent { diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index 014850d145..3cf3f9e73a 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/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) -> 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) { + 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]