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 {