From 8b8ee28a9b0df8188fec8d4e3855a7b3af3ed8a2 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Thu, 3 Sep 2026 05:05:00 +0000 Subject: [PATCH] Acknowledge pending TUI steers by submission ID (#42451) ## Why Completion receipts for identical user messages could acknowledge the wrong pending steer. When the receipt arrived while another thread was displayed, the stale pending state could later be restored and resubmitted. ## What changed - Assign each TUI submission a client message ID and pass it through turn start and steer requests. - Preserve pending steer IDs across thread input snapshots and match completion receipts by ID, while retaining content-based matching for older app servers that do not echo IDs. - Acknowledge matching receipts for inactive threads before their notifications can be discarded, including threads stored in the agents overview. ## Testing Added coverage for identical steer receipts and for switching within and between tasks before interrupting an active turn. GitOrigin-RevId: 28bd80a735ff86cd7bf836f24b448e7499f0a902 --- codex-rs/tui/src/app/app_server_events.rs | 29 ++++ .../tui/src/app/tests/safety_buffering.rs | 143 ++++++++++++++++-- codex-rs/tui/src/app/thread_routing.rs | 9 +- codex-rs/tui/src/app_command.rs | 3 + codex-rs/tui/src/app_server_session.rs | 6 +- codex-rs/tui/src/chatwidget.rs | 17 ++- codex-rs/tui/src/chatwidget/input_queue.rs | 1 + codex-rs/tui/src/chatwidget/input_restore.rs | 54 +------ .../tui/src/chatwidget/input_submission.rs | 3 + codex-rs/tui/src/chatwidget/replay.rs | 6 +- .../chatwidget/tests/composer_submission.rs | 7 +- codex-rs/tui/src/chatwidget/tests/helpers.rs | 1 + .../tui/src/chatwidget/tests/review_mode.rs | 53 ++++--- codex-rs/tui/src/chatwidget/user_messages.rs | 15 +- 14 files changed, 250 insertions(+), 97 deletions(-) diff --git a/codex-rs/tui/src/app/app_server_events.rs b/codex-rs/tui/src/app/app_server_events.rs index 5783787dd5..145c7e021a 100644 --- a/codex-rs/tui/src/app/app_server_events.rs +++ b/codex-rs/tui/src/app/app_server_events.rs @@ -19,6 +19,7 @@ use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::SessionSource; +use codex_app_server_protocol::ThreadItem; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadSource; @@ -275,6 +276,34 @@ impl App { match server_notification_thread_target(¬ification) { ServerNotificationThreadTarget::Thread(thread_id) => { + if self.current_displayed_thread_id() != Some(thread_id) + && let ServerNotification::ItemCompleted(item) = ¬ification + && let ThreadItem::UserMessage { + client_id: Some(client_id), + .. + } = &item.item + { + // Acknowledge by ID before routing can discard the receipt. ID-less receipts + // cannot safely distinguish identical pending submissions. + let mut store = match self.thread_event_channels.get(&thread_id) { + Some(channel) => Some(channel.store.lock().await), + None => None, + }; + for input in store + .as_mut() + .and_then(|store| store.input_state.as_mut()) + .into_iter() + .chain(self.agents_overview.input_states.get_mut(&thread_id)) + { + if input + .pending_steers + .front() + .is_some_and(|pending| pending.client_id == *client_id) + { + input.pending_steers.pop_front(); + } + } + } if self.primary_thread_id.is_none() && !self.pending_startup_thread_start { return; } diff --git a/codex-rs/tui/src/app/tests/safety_buffering.rs b/codex-rs/tui/src/app/tests/safety_buffering.rs index 9735915ee7..fd7baef6cd 100644 --- a/codex-rs/tui/src/app/tests/safety_buffering.rs +++ b/codex-rs/tui/src/app/tests/safety_buffering.rs @@ -163,9 +163,26 @@ async fn wait_for_turn_completed( #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn active_turn_interrupt_is_nonblocking_and_coalesces_repeated_requests() -> Result<()> { + Box::pin(interrupt_after_inactive_steer(SteerSwitch::WithinTask)).await +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn agents_overview_acknowledges_inactive_steer_before_interrupt() -> Result<()> { + Box::pin(interrupt_after_inactive_steer(SteerSwitch::BetweenTasks)).await +} + +enum SteerSwitch { + WithinTask, + BetweenTasks, +} + +async fn interrupt_after_inactive_steer(switch: SteerSwitch) -> Result<()> { + let via_overview = matches!(switch, SteerSwitch::BetweenTasks); let (chunks, release_response) = gated_response_chunks("interrupt-response", ev_completed("interrupt-response")); - let (server, _completions) = start_streaming_sse_server(vec![chunks]).await; + let (steered_chunks, release_steered_response) = + gated_response_chunks("steered-response", ev_completed("steered-response")); + let (server, _completions) = start_streaming_sse_server(vec![chunks, steered_chunks]).await; let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await; let codex_home = tempdir()?; std::fs::write( @@ -210,18 +227,117 @@ stream_max_retries = 0 .await?; while app_event_rx.try_recv().is_ok() {} - submit_prompt(&mut app, "Keep this turn running until interrupted"); + submit_prompt(&mut app, COMMITTED_STEER); let turn = next_user_turn_event(&mut app_event_rx); app.submit_thread_op(&mut app_server, thread_id, turn) .await?; let turn_id = next_turn_started(&mut app, &mut app_server, thread_id).await; - drive_until_request_count( - &mut app, - &mut app_server, - &server, - /*expected_request_count*/ 1, + // Leave the original message's completion queued until after submitting the identical steer. + tokio::time::timeout( + std::time::Duration::from_secs(/*secs*/ 5), + server.wait_for_request_count(/*count*/ 1), ) - .await; + .await?; + + let acknowledged_input = app.chat_widget.capture_thread_input_state(); + submit_prompt(&mut app, COMMITTED_STEER); + let steer = next_user_turn_event(&mut app_event_rx); + let AppCommand::UserTurn { + client_user_message_id, + .. + } = &steer + else { + unreachable!("user turn"); + }; + let expected_client_id = client_user_message_id.clone(); + let pending_input = app.chat_widget.capture_thread_input_state(); + app.submit_thread_op(&mut app_server, thread_id, steer) + .await?; + let other_id = ThreadId::from_string( + &app_test_support::create_fake_rollout( + app.config.codex_home.as_path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "Other task", + Some(MODEL_PROVIDER_ID), + /*git_info*/ None, + ) + .expect("materialize other task"), + )?; + let other = app_server + .resume_thread( + &app.local_settings, + app.config.clone(), + other_id, + crate::app_server_session::ResumeModelSettings::PreserveExistingThread, + ) + .await?; + app.thread_event_channels.insert( + other_id, + ThreadEventChannel::new_with_session( + THREAD_EVENT_CHANNEL_CAPACITY, + other.session, + other.turns, + ), + ); + // Receive the real commit while another thread is visible, with no replay capacity. + if via_overview { + Box::pin(app.select_agents_overview_thread(&mut tui, &mut app_server, other_id)).await?; + } else { + Box::pin(app.select_agent_thread(&mut tui, &mut app_server, other_id)).await?; + app.thread_event_channels[&thread_id] + .store + .lock() + .await + .capacity = 0; + } + assert_eq!(app.current_displayed_thread_id(), Some(other_id)); + let mut release_response = Some(release_response); + loop { + let event = tokio::time::timeout( + std::time::Duration::from_secs(/*secs*/ 5), + app_server.next_event(), + ) + .await? + .expect("app-server remains open"); + let committed_id = if let AppServerEvent::ServerNotification(notification) = &event + && let ServerNotification::ItemCompleted(item) = notification.as_ref() + && let ThreadItem::UserMessage { + content, client_id, .. + } = &item.item + && ChatWidget::user_message_display_from_inputs(content).message == COMMITTED_STEER + { + Some(client_id.clone().expect("receipt echoes the submission ID")) + } else { + None + }; + app.handle_app_server_event(&app_server, event).await; + if let Some(client_id) = committed_id { + let saved_input = if via_overview { + assert!(!app.thread_event_channels.contains_key(&thread_id)); + app.agents_overview.input_states.get(&thread_id).cloned() + } else { + let store = app.thread_event_channels[&thread_id].store.lock().await; + assert!(store.buffer.is_empty()); + store.input_state.clone() + }; + if let Some(release_response) = release_response.take() { + assert_ne!(client_id, expected_client_id); + assert_eq!(saved_input, pending_input); + let _ = release_response.send(()); + } else { + assert_eq!(client_id, expected_client_id); + assert_eq!(saved_input, acknowledged_input); + break; + } + } + } + if via_overview { + Box::pin(app.select_agents_overview_thread(&mut tui, &mut app_server, thread_id)).await?; + } else { + Box::pin(app.select_agent_thread(&mut tui, &mut app_server, thread_id)).await?; + } + assert_eq!(app.active_thread_id, Some(thread_id)); let interrupt = AppCommand::interrupt(); assert!( @@ -270,7 +386,16 @@ stream_max_retries = 0 None ); - let _ = release_response.send(()); + insta::assert_snapshot!(app.chat_widget.composer_text_with_pending(), @""); + assert!( + std::iter::from_fn(|| app_event_rx.try_recv().ok()) + .all(|event| !matches!(event, AppEvent::CodexOp(AppCommand::UserTurn { .. }))) + ); + let source = app_server + .thread_read(thread_id, /*include_turns*/ true) + .await?; + assert_eq!(user_message_count(&source, COMMITTED_STEER), 2); + let _ = release_steered_response.send(()); app_server.shutdown().await?; server.shutdown().await; Ok(()) diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index b78ea15dad..a40fe6b273 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -659,6 +659,7 @@ impl App { Ok(true) } AppCommand::UserTurn { + client_user_message_id, items, cwd, approval_policy, @@ -678,7 +679,12 @@ impl App { let mut retried_after_turn_mismatch = false; loop { match app_server - .turn_steer(thread_id, steer_turn_id.clone(), items.to_vec()) + .turn_steer( + thread_id, + steer_turn_id.clone(), + client_user_message_id.clone(), + items.to_vec(), + ) .await { Ok(_) => return Ok(true), @@ -751,6 +757,7 @@ impl App { let response = app_server .turn_start( thread_id, + client_user_message_id.clone(), items.to_vec(), cwd.clone(), *approval_policy, diff --git a/codex-rs/tui/src/app_command.rs b/codex-rs/tui/src/app_command.rs index fef3fbb5d3..f54b00497d 100644 --- a/codex-rs/tui/src/app_command.rs +++ b/codex-rs/tui/src/app_command.rs @@ -30,6 +30,7 @@ pub(crate) enum AppCommand { command: String, }, UserTurn { + client_user_message_id: String, items: Vec, cwd: PathBuf, approval_policy: AskForApproval, @@ -113,6 +114,7 @@ impl AppCommand { #[allow(clippy::too_many_arguments)] pub(crate) fn user_turn( + client_user_message_id: String, items: Vec, cwd: PathBuf, approval_policy: AskForApproval, @@ -126,6 +128,7 @@ impl AppCommand { personality: Option, ) -> Self { Self::UserTurn { + client_user_message_id, items, cwd, approval_policy, diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index b90e88cff1..5a36fa92cd 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -1229,6 +1229,7 @@ impl AppServerSession { pub(crate) async fn turn_start( &mut self, thread_id: ThreadId, + client_user_message_id: String, items: Vec, cwd: PathBuf, approval_policy: AskForApproval, @@ -1252,7 +1253,7 @@ impl AppServerSession { params: TurnStartParams { thread_id: thread_id.to_string(), turn_trigger: None, - client_user_message_id: None, + client_user_message_id: Some(client_user_message_id), input: items, tool_output: None, responsesapi_client_metadata: None, @@ -1310,6 +1311,7 @@ impl AppServerSession { &mut self, thread_id: ThreadId, turn_id: String, + client_user_message_id: String, items: Vec, ) -> std::result::Result { let request_id = self.next_request_id(); @@ -1318,7 +1320,7 @@ impl AppServerSession { request_id, params: TurnSteerParams { thread_id: thread_id.to_string(), - client_user_message_id: None, + client_user_message_id: Some(client_user_message_id), input: items, responsesapi_client_metadata: None, additional_context: None, diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 8cda1b818a..14a2b8b3e2 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -449,6 +449,7 @@ mod usage; mod user_messages; mod working_directory; use self::user_messages::PendingSteer; +#[cfg(test)] use self::user_messages::PendingSteerCompareKey; use self::user_messages::QueueDrain; use self::user_messages::QueuedUserMessage; @@ -1319,7 +1320,12 @@ impl ChatWidget { self.request_redraw(); } - fn on_committed_user_message(&mut self, items: &[UserInput], from_replay: bool) { + fn on_committed_user_message( + &mut self, + items: &[UserInput], + client_id: Option<&str>, + from_replay: bool, + ) { let display = Self::user_message_display_from_inputs(items); if from_replay { if self.review.is_review_mode { @@ -1338,12 +1344,15 @@ impl ChatWidget { return; } - let compare_key = Self::pending_steer_compare_key_from_items(items); if self .input_queue .pending_steers .front() - .is_some_and(|pending| pending.compare_key == compare_key) + .is_some_and(|pending| match client_id { + Some(client_id) => pending.client_id == client_id, + // Older app servers do not echo submission IDs. + None => pending.compare_key == Self::pending_steer_compare_key_from_items(items), + }) { if let Some(pending) = self.input_queue.pending_steers.pop_front() { self.refresh_pending_input_preview(); @@ -1352,7 +1361,7 @@ impl ChatWidget { self.on_user_message_display(pending_display); } else if self.last_rendered_user_message_display.as_ref() != Some(&display) { tracing::warn!( - "pending steer matched compare key but queue was empty when rendering committed user message" + "pending steer matched receipt but queue was empty when rendering committed user message" ); self.on_user_message_display(display); } diff --git a/codex-rs/tui/src/chatwidget/input_queue.rs b/codex-rs/tui/src/chatwidget/input_queue.rs index 249cade908..48a0916c0a 100644 --- a/codex-rs/tui/src/chatwidget/input_queue.rs +++ b/codex-rs/tui/src/chatwidget/input_queue.rs @@ -116,6 +116,7 @@ mod tests { .rejected_steers_queue .push_back(UserMessage::from("rejected")); state.pending_steers.push_back(PendingSteer { + client_id: "test-submission".to_string(), user_message: UserMessage::from("pending"), history_record: UserMessageHistoryRecord::UserMessageText, compare_key: crate::chatwidget::user_messages::PendingSteerCompareKey { diff --git a/codex-rs/tui/src/chatwidget/input_restore.rs b/codex-rs/tui/src/chatwidget/input_restore.rs index 8928dc4ff5..d7e7ae3390 100644 --- a/codex-rs/tui/src/chatwidget/input_restore.rs +++ b/codex-rs/tui/src/chatwidget/input_restore.rs @@ -1,7 +1,6 @@ //! Input queue restore and thread-input snapshot behavior for `ChatWidget`. use std::collections::HashSet; -use std::collections::VecDeque; use crate::bottom_pane::ComposerDraftSnapshot; @@ -446,24 +445,7 @@ impl ChatWidget { Some(ThreadInputState { composer: composer.has_content().then_some(composer), safety_buffering_prompt: self.safety_buffering_prompt.clone(), - pending_steers: self - .input_queue - .pending_steers - .iter() - .map(|pending| pending.user_message.clone()) - .collect(), - pending_steer_history_records: self - .input_queue - .pending_steers - .iter() - .map(|pending| pending.history_record.clone()) - .collect(), - pending_steer_compare_keys: self - .input_queue - .pending_steers - .iter() - .map(|pending| pending.compare_key.clone()) - .collect(), + pending_steers: self.input_queue.pending_steers.clone(), rejected_steers_queue: self.input_queue.rejected_steers_queue.clone(), rejected_steer_history_records: self.input_queue.rejected_steer_history_records.clone(), queued_user_messages: self.input_queue.queued_user_messages.clone(), @@ -507,42 +489,18 @@ impl ChatWidget { self.update_collaboration_mode_indicator(); self.refresh_model_dependent_surfaces(); self.restore_composer_state(input_state.composer.unwrap_or_default()); - let mut pending_steer_history_records = input_state.pending_steer_history_records; - pending_steer_history_records.resize( - input_state.pending_steers.len(), - UserMessageHistoryRecord::UserMessageText, - ); - let mut pending_steer_compare_keys = input_state.pending_steer_compare_keys; let pending_steers = input_state.pending_steers; let mut queued_user_messages = input_state.queued_user_messages; let mut queued_user_message_history_records = input_state.queued_user_message_history_records; if preserve_in_flight_turn { - self.input_queue.pending_steers = pending_steers - .into_iter() - .zip(pending_steer_history_records) - .map(|(user_message, history_record)| PendingSteer { - compare_key: pending_steer_compare_keys.pop_front().unwrap_or_else(|| { - PendingSteerCompareKey { - message: user_message.text.clone(), - image_count: user_message.local_images.len() - + user_message.remote_image_urls.len(), - } - }), - history_record, - user_message, - }) - .collect(); + self.input_queue.pending_steers = pending_steers; } else { self.input_queue.pending_steers.clear(); - let mut safety_retry_follow_ups = pending_steers - .into_iter() - .map(QueuedUserMessage::from) - .collect::>(); - safety_retry_follow_ups.append(&mut queued_user_messages); - queued_user_messages = safety_retry_follow_ups; - pending_steer_history_records.append(&mut queued_user_message_history_records); - queued_user_message_history_records = pending_steer_history_records; + for pending in pending_steers.into_iter().rev() { + queued_user_messages.push_front(pending.user_message.into()); + queued_user_message_history_records.push_front(pending.history_record); + } } self.input_queue.rejected_steers_queue = input_state.rejected_steers_queue; self.input_queue.rejected_steer_history_records = diff --git a/codex-rs/tui/src/chatwidget/input_submission.rs b/codex-rs/tui/src/chatwidget/input_submission.rs index 424f21a544..27a6e7eb51 100644 --- a/codex-rs/tui/src/chatwidget/input_submission.rs +++ b/codex-rs/tui/src/chatwidget/input_submission.rs @@ -347,7 +347,9 @@ impl ChatWidget { } else { None }; + let client_user_message_id = uuid::Uuid::new_v4().to_string(); let pending_steer = (!render_in_history).then(|| PendingSteer { + client_id: client_user_message_id.clone(), user_message: UserMessage { text: text.clone(), local_images: local_images.clone(), @@ -366,6 +368,7 @@ impl ChatWidget { let service_tier = self.service_tier_update_for_core(); let active_permission_profile = self.config.permissions.active_permission_profile(); let op = AppCommand::user_turn( + client_user_message_id, items, self.config.cwd.to_path_buf(), AskForApproval::from(self.config.permissions.approval_policy.value()), diff --git a/codex-rs/tui/src/chatwidget/replay.rs b/codex-rs/tui/src/chatwidget/replay.rs index 1c3f3e6381..61d69d10cb 100644 --- a/codex-rs/tui/src/chatwidget/replay.rs +++ b/codex-rs/tui/src/chatwidget/replay.rs @@ -114,8 +114,10 @@ impl ChatWidget { let from_replay = render_source.is_replay(); let replay_kind = render_source.replay_kind(); match item { - ThreadItem::UserMessage { content, .. } => { - self.on_committed_user_message(&content, from_replay); + ThreadItem::UserMessage { + content, client_id, .. + } => { + self.on_committed_user_message(&content, client_id.as_deref(), from_replay); } ThreadItem::AgentMessage { id, diff --git a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs index 40b917aed5..d763e044a7 100644 --- a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs +++ b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs @@ -1761,9 +1761,10 @@ async fn restore_thread_input_state_applies_running_state_policy() { ..Default::default() }), safety_buffering_prompt: Some(UserMessage::from("buffered prompt")), - pending_steers: VecDeque::from([UserMessage::from("submitted to the interrupted turn")]), - pending_steer_history_records: VecDeque::from([pending_history.clone()]), - pending_steer_compare_keys: VecDeque::new(), + pending_steers: VecDeque::from([PendingSteer { + history_record: pending_history.clone(), + ..pending_steer("submitted to the interrupted turn") + }]), rejected_steers_queue: VecDeque::new(), rejected_steer_history_records: VecDeque::new(), queued_user_messages: VecDeque::from([UserMessage::from("already queued").into()]), diff --git a/codex-rs/tui/src/chatwidget/tests/helpers.rs b/codex-rs/tui/src/chatwidget/tests/helpers.rs index 47da9b3922..1e480395c4 100644 --- a/codex-rs/tui/src/chatwidget/tests/helpers.rs +++ b/codex-rs/tui/src/chatwidget/tests/helpers.rs @@ -940,6 +940,7 @@ pub(super) fn complete_assistant_message( pub(super) fn pending_steer(text: &str) -> PendingSteer { PendingSteer { + client_id: "test-submission".to_string(), user_message: UserMessage::from(text), history_record: UserMessageHistoryRecord::UserMessageText, compare_key: PendingSteerCompareKey { diff --git a/codex-rs/tui/src/chatwidget/tests/review_mode.rs b/codex-rs/tui/src/chatwidget/tests/review_mode.rs index ed3853e919..117a6522ca 100644 --- a/codex-rs/tui/src/chatwidget/tests/review_mode.rs +++ b/codex-rs/tui/src/chatwidget/tests/review_mode.rs @@ -360,14 +360,15 @@ async fn review_restores_context_window_indicator() { #[tokio::test] async fn restore_thread_input_state_restores_pending_steers_without_downgrading_them() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; - let mut pending_steers = VecDeque::new(); - pending_steers.push_back(UserMessage::from("pending steer")); let expected_compare_key = PendingSteerCompareKey { message: "hidden IDE context\npending steer".to_string(), image_count: 0, }; - let mut pending_steer_compare_keys = VecDeque::new(); - pending_steer_compare_keys.push_back(expected_compare_key.clone()); + let expected_pending = PendingSteer { + client_id: "saved-submission".to_string(), + compare_key: expected_compare_key, + ..pending_steer("pending steer") + }; let mut rejected_steers_queue = VecDeque::new(); rejected_steers_queue.push_back(UserMessage::from("already rejected")); let mut queued_user_messages = VecDeque::new(); @@ -377,9 +378,7 @@ async fn restore_thread_input_state_restores_pending_steers_without_downgrading_ Some(ThreadInputState { composer: None, safety_buffering_prompt: None, - pending_steers, - pending_steer_history_records: VecDeque::new(), - pending_steer_compare_keys, + pending_steers: VecDeque::from([expected_pending.clone()]), rejected_steers_queue, rejected_steer_history_records: VecDeque::new(), queued_user_messages, @@ -401,22 +400,38 @@ async fn restore_thread_input_state_restores_pending_steers_without_downgrading_ chat.queued_user_message_texts(), vec!["already rejected", "queued draft"] ); - assert_eq!(chat.input_queue.pending_steers.len(), 1); assert_eq!( - chat.input_queue - .pending_steers - .front() - .unwrap() - .user_message - .text, - "pending steer" - ); - assert_eq!( - chat.input_queue.pending_steers.front().unwrap().compare_key, - expected_compare_key + chat.input_queue.pending_steers, + VecDeque::from([expected_pending]) ); } +#[tokio::test] +async fn identical_steer_receipts_only_acknowledge_the_matching_submission() { + let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let pending = pending_steer("continue"); + chat.input_queue.pending_steers.push_back(pending.clone()); + + for (client_id, expected) in [ + ("older-submission", VecDeque::from([pending.clone()])), + (pending.client_id.as_str(), VecDeque::new()), + ] { + chat.handle_thread_item( + AppServerThreadItem::UserMessage { + id: client_id.to_string(), + client_id: Some(client_id.to_string()), + content: vec![UserInput::Text { + text: "continue".to_string(), + text_elements: Vec::new(), + }], + }, + "turn-1".to_string(), + ThreadItemRenderSource::Live, + ); + assert_eq!(chat.input_queue.pending_steers, expected); + } +} + #[tokio::test] async fn steer_enter_queues_while_plan_stream_is_active() { let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; diff --git a/codex-rs/tui/src/chatwidget/user_messages.rs b/codex-rs/tui/src/chatwidget/user_messages.rs index 50feb09b80..88e62358c0 100644 --- a/codex-rs/tui/src/chatwidget/user_messages.rs +++ b/codex-rs/tui/src/chatwidget/user_messages.rs @@ -124,9 +124,7 @@ impl ThreadComposerState { pub(crate) struct ThreadInputState { pub(super) composer: Option, pub(super) safety_buffering_prompt: Option, - pub(super) pending_steers: VecDeque, - pub(super) pending_steer_history_records: VecDeque, - pub(super) pending_steer_compare_keys: VecDeque, + pub(crate) pending_steers: VecDeque, pub(super) rejected_steers_queue: VecDeque, pub(super) rejected_steer_history_records: VecDeque, pub(super) queued_user_messages: VecDeque, @@ -171,8 +169,10 @@ impl From<&str> for UserMessage { } } -#[derive(Debug)] -pub(super) struct PendingSteer { +#[derive(Debug, Clone, PartialEq)] +pub(crate) struct PendingSteer { + /// Preserved across request retries and thread switches until this submission commits. + pub(crate) client_id: String, pub(super) user_message: UserMessage, pub(super) history_record: UserMessageHistoryRecord, pub(super) compare_key: PendingSteerCompareKey, @@ -696,10 +696,7 @@ impl ChatWidget { } } - /// Build the compare key for a submitted pending steer without invoking the - /// expensive request-serialization path. Pending steers only need to match the - /// committed app-server `UserMessage` item emitted after input drains, which - /// preserves flattened text and total image count. + /// Build the legacy content key for app servers that do not echo submission IDs. pub(super) fn pending_steer_compare_key_from_items( items: &[UserInput], ) -> PendingSteerCompareKey {