From 07b25626b4912b2d5b3f9565d465f96db6b7e536 Mon Sep 17 00:00:00 2001 From: Edward Frazer Date: Fri, 29 May 2026 15:04:46 -0700 Subject: [PATCH] feat: queue TUI follow-ups through app-server --- codex-rs/tui/src/app/tests.rs | 44 +++ codex-rs/tui/src/app/thread_routing.rs | 17 ++ codex-rs/tui/src/app_command.rs | 18 ++ codex-rs/tui/src/app_server_session.rs | 21 ++ codex-rs/tui/src/bottom_pane/mod.rs | 5 + .../src/bottom_pane/pending_input_preview.rs | 20 +- codex-rs/tui/src/chatwidget.rs | 4 +- codex-rs/tui/src/chatwidget/input_flow.rs | 70 ++++- codex-rs/tui/src/chatwidget/input_queue.rs | 261 +++++++++++++++++- codex-rs/tui/src/chatwidget/input_restore.rs | 11 +- .../tui/src/chatwidget/input_submission.rs | 99 ++++--- codex-rs/tui/src/chatwidget/interaction.rs | 2 +- codex-rs/tui/src/chatwidget/protocol.rs | 11 +- ...compact_queues_user_messages_snapshot.snap | 1 + ..._review_queues_user_messages_snapshot.snap | 1 + .../tui/src/chatwidget/tests/app_server.rs | 59 ++++ .../chatwidget/tests/composer_submission.rs | 216 +++++++++++++++ .../tui/src/chatwidget/tests/exec_flow.rs | 1 + .../tui/src/chatwidget/tests/review_mode.rs | 3 + .../src/chatwidget/tests/slash_commands.rs | 22 ++ codex-rs/tui/src/chatwidget/turn_runtime.rs | 9 +- codex-rs/tui/src/chatwidget/user_messages.rs | 4 + 22 files changed, 841 insertions(+), 58 deletions(-) diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 993772ab77..7b87caa574 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -5138,6 +5138,50 @@ async fn interrupt_without_active_turn_is_treated_as_handled() { .await; } +#[test] +fn failed_server_queue_submission_falls_back_without_exiting_tui() -> Result<()> { + const WORKER_THREADS: usize = 1; + const TEST_STACK_SIZE_BYTES: usize = 8 * 1024 * 1024; + + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(WORKER_THREADS) + .thread_stack_size(TEST_STACK_SIZE_BYTES) + .enable_all() + .build()?; + + runtime.block_on(async { + let mut app = make_test_app().await; + let mut app_server = Box::pin(crate::start_embedded_app_server_for_picker( + app.chat_widget.config_ref(), + )) + .await + .expect("embedded app server"); + let thread_id = ThreadId::new(); + let op = AppCommand::queue_turn( + codex_app_server_protocol::TurnSubmission { + input: vec![UserInput::Text { + text: "durable follow-up".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }, + crate::chatwidget::UserMessage::from("durable follow-up"), + ); + + let handled = app + .try_submit_active_thread_op_via_app_server(&mut app_server, thread_id, &op) + .await + .expect("queue failure should fall back locally"); + + assert!(handled); + assert_eq!( + app.chat_widget.queued_user_message_texts(), + vec!["durable follow-up".to_string()] + ); + Ok(()) + }) +} + #[tokio::test] async fn override_turn_context_sends_thread_settings_update() { Box::pin(async { diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index a403dc232c..d92039a59e 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -616,6 +616,23 @@ impl App { } Ok(true) } + AppCommand::QueueTurn { + submission, + fallback_user_message, + } => { + if let Err(err) = app_server + .thread_queue_add(thread_id, submission.clone()) + .await + { + self.chat_widget + .requeue_failed_server_submission(fallback_user_message.clone()); + self.chat_widget.add_error_message(format!( + "Failed to queue follow-up through app-server; queued locally instead: {err}" + )); + self.chat_widget.maybe_send_next_queued_input(); + } + Ok(true) + } AppCommand::ListSkills { cwds, force_reload } => { self.handle_skills_list_result( app_server diff --git a/codex-rs/tui/src/app_command.rs b/codex-rs/tui/src/app_command.rs index 2ecf17f5a3..53b276f96a 100644 --- a/codex-rs/tui/src/app_command.rs +++ b/codex-rs/tui/src/app_command.rs @@ -9,6 +9,7 @@ use codex_app_server_protocol::ReviewTarget; use codex_app_server_protocol::ThreadRealtimeAudioChunk; use codex_app_server_protocol::ThreadRealtimeStartTransport; use codex_app_server_protocol::ToolRequestUserInputResponse; +use codex_app_server_protocol::TurnSubmission; use codex_app_server_protocol::UserInput; use codex_config::types::ApprovalsReviewer; use codex_protocol::approvals::GuardianAssessmentEvent; @@ -23,6 +24,8 @@ use codex_protocol::request_permissions::RequestPermissionsResponse; use serde::Serialize; use serde_json::Value; +use crate::chatwidget::UserMessage; + #[allow(clippy::large_enum_variant)] #[derive(Debug, Clone, PartialEq, Serialize)] pub(crate) enum AppCommand { @@ -51,6 +54,11 @@ pub(crate) enum AppCommand { collaboration_mode: Option, personality: Option, }, + QueueTurn { + submission: TurnSubmission, + #[serde(skip)] + fallback_user_message: UserMessage, + }, OverrideTurnContext { cwd: Option, approval_policy: Option, @@ -169,6 +177,16 @@ impl AppCommand { } } + pub(crate) fn queue_turn( + submission: TurnSubmission, + fallback_user_message: UserMessage, + ) -> Self { + Self::QueueTurn { + submission, + fallback_user_message, + } + } + #[allow(clippy::too_many_arguments)] pub(crate) fn override_turn_context( cwd: Option, diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index e66b5ee8a2..938db049eb 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -73,6 +73,8 @@ use codex_app_server_protocol::ThreadMemoryModeSetResponse; use codex_app_server_protocol::ThreadMetadataGitInfoUpdateParams; use codex_app_server_protocol::ThreadMetadataUpdateParams; use codex_app_server_protocol::ThreadMetadataUpdateResponse; +use codex_app_server_protocol::ThreadQueueAddParams; +use codex_app_server_protocol::ThreadQueueAddResponse; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadRealtimeAppendAudioParams; @@ -106,6 +108,7 @@ use codex_app_server_protocol::TurnStartParams; use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::TurnSteerParams; use codex_app_server_protocol::TurnSteerResponse; +use codex_app_server_protocol::TurnSubmission; use codex_app_server_protocol::UserInput; use codex_otel::TelemetryAuthMode; use codex_protocol::ThreadId; @@ -603,6 +606,24 @@ impl AppServerSession { .wrap_err("thread/metadata/update failed while syncing git branch") } + pub(crate) async fn thread_queue_add( + &mut self, + thread_id: ThreadId, + submission: TurnSubmission, + ) -> Result { + let request_id = self.next_request_id(); + self.client + .request_typed(ClientRequest::ThreadQueueAdd { + request_id, + params: ThreadQueueAddParams { + thread_id: thread_id.to_string(), + submission: submission.into(), + }, + }) + .await + .wrap_err("thread/queue/add failed in TUI") + } + pub(crate) async fn thread_settings_update( &mut self, params: ThreadSettingsUpdateParams, diff --git a/codex-rs/tui/src/bottom_pane/mod.rs b/codex-rs/tui/src/bottom_pane/mod.rs index fa92914670..f1fca6be85 100644 --- a/codex-rs/tui/src/bottom_pane/mod.rs +++ b/codex-rs/tui/src/bottom_pane/mod.rs @@ -1179,10 +1179,12 @@ impl BottomPane { queued: Vec, pending_steers: Vec, rejected_steers: Vec, + has_editable_queued_message: 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.has_editable_queued_message = has_editable_queued_message; self.request_redraw(); } @@ -2431,6 +2433,7 @@ mod tests { vec!["Queued follow-up question".to_string()], Vec::new(), Vec::new(), + /*has_editable_queued_message*/ true, ); let width = 48; @@ -2462,6 +2465,7 @@ mod tests { vec!["Queued follow-up question".to_string()], Vec::new(), Vec::new(), + /*has_editable_queued_message*/ true, ); pane.hide_status_indicator(); @@ -2494,6 +2498,7 @@ mod tests { vec!["Queued follow-up question".to_string()], Vec::new(), Vec::new(), + /*has_editable_queued_message*/ true, ); 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..576837b903 100644 --- a/codex-rs/tui/src/bottom_pane/pending_input_preview.rs +++ b/codex-rs/tui/src/bottom_pane/pending_input_preview.rs @@ -24,6 +24,7 @@ pub(crate) struct PendingInputPreview { pub pending_steers: Vec, pub rejected_steers: Vec, pub queued_messages: Vec, + pub has_editable_queued_message: bool, /// 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, @@ -39,6 +40,7 @@ impl PendingInputPreview { pending_steers: Vec::new(), rejected_steers: Vec::new(), queued_messages: Vec::new(), + has_editable_queued_message: false, edit_binding: Some(key_hint::alt(KeyCode::Up)), interrupt_binding: Some(key_hint::plain(KeyCode::Esc)), } @@ -151,7 +153,7 @@ impl PendingInputPreview { } } - if !self.queued_messages.is_empty() + if self.has_editable_queued_message && let Some(edit_binding) = self.edit_binding { lines.push( @@ -197,13 +199,22 @@ mod tests { #[test] fn desired_height_one_message() { let mut queue = PendingInputPreview::new(); + queue.has_editable_queued_message = true; queue.queued_messages.push("Hello, world!".to_string()); assert_eq!(queue.desired_height(/*width*/ 40), 3); } + #[test] + fn desired_height_server_owned_message_omits_edit_hint() { + let mut queue = PendingInputPreview::new(); + queue.queued_messages.push("Hello, world!".to_string()); + assert_eq!(queue.desired_height(/*width*/ 40), 2); + } + #[test] fn render_one_message() { let mut queue = PendingInputPreview::new(); + queue.has_editable_queued_message = true; queue.queued_messages.push("Hello, world!".to_string()); let width = 40; let height = queue.desired_height(width); @@ -215,6 +226,7 @@ mod tests { #[test] fn render_one_message_with_shift_left_binding() { let mut queue = PendingInputPreview::new(); + queue.has_editable_queued_message = true; queue.queued_messages.push("Hello, world!".to_string()); queue.set_edit_binding(Some(key_hint::shift(KeyCode::Left))); let width = 40; @@ -230,6 +242,7 @@ mod tests { #[test] fn render_two_messages() { let mut queue = PendingInputPreview::new(); + queue.has_editable_queued_message = true; queue.queued_messages.push("Hello, world!".to_string()); queue .queued_messages @@ -244,6 +257,7 @@ mod tests { #[test] fn render_more_than_three_messages() { let mut queue = PendingInputPreview::new(); + queue.has_editable_queued_message = true; queue.queued_messages.push("Hello, world!".to_string()); queue .queued_messages @@ -264,6 +278,7 @@ mod tests { #[test] fn render_wrapped_message() { let mut queue = PendingInputPreview::new(); + queue.has_editable_queued_message = true; queue .queued_messages .push("This is a longer message that should be wrapped".to_string()); @@ -280,6 +295,7 @@ mod tests { #[test] fn render_many_line_message() { let mut queue = PendingInputPreview::new(); + queue.has_editable_queued_message = true; queue .queued_messages .push("This is\na message\nwith many\nlines".to_string()); @@ -293,6 +309,7 @@ mod tests { #[test] fn long_url_like_message_does_not_expand_into_wrapped_ellipsis_rows() { let mut queue = PendingInputPreview::new(); + queue.has_editable_queued_message = true; queue.queued_messages.push( "example.test/api/v1/projects/alpha-team/releases/2026-02-17/builds/1234567890/artifacts/reports/performance/summary/detail/session_id=abc123def456ghi789" .to_string(), @@ -351,6 +368,7 @@ mod tests { #[test] fn render_pending_steers_above_queued_messages() { let mut queue = PendingInputPreview::new(); + queue.has_editable_queued_message = true; queue.pending_steers.push("Please continue.".to_string()); queue .pending_steers diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index 8604229552..092287d93b 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -125,6 +125,7 @@ use codex_app_server_protocol::Turn; use codex_app_server_protocol::TurnCompletedNotification; use codex_app_server_protocol::TurnPlanStepStatus; use codex_app_server_protocol::TurnStatus; +use codex_app_server_protocol::TurnSubmission; use codex_app_server_protocol::UserInput; use codex_config::ConfigLayerStackOrdering; use codex_config::types::ApprovalsReviewer; @@ -346,6 +347,8 @@ mod ide_context; use self::ide_context::IdeContextState; mod input_queue; use self::input_queue::InputQueueState; +#[cfg(test)] +use self::input_queue::ServerQueueBarrier; mod input_flow; mod input_restore; mod input_submission; @@ -428,7 +431,6 @@ use self::user_messages::ThreadComposerState; pub(crate) use self::user_messages::ThreadInputState; pub(crate) use self::user_messages::UserMessage; use self::user_messages::UserMessageDisplay; -#[cfg(test)] use self::user_messages::UserMessageHistoryOverride; use self::user_messages::UserMessageHistoryRecord; use self::user_messages::app_server_text_elements; diff --git a/codex-rs/tui/src/chatwidget/input_flow.rs b/codex-rs/tui/src/chatwidget/input_flow.rs index 3930224e04..28a7ec72fb 100644 --- a/codex-rs/tui/src/chatwidget/input_flow.rs +++ b/codex-rs/tui/src/chatwidget/input_flow.rs @@ -82,14 +82,25 @@ impl ChatWidget { user_message: UserMessage, action: QueuedInputAction, ) { - if !self.is_session_configured() || self.is_user_turn_pending_or_running() { - self.input_queue - .queued_user_messages - .push_back(QueuedUserMessage::new(user_message, action)); - self.input_queue - .queued_user_message_history_records - .push_back(UserMessageHistoryRecord::UserMessageText); - self.refresh_pending_input_preview(); + if self.is_session_configured() + && self.is_user_turn_pending_or_running() + && action == QueuedInputAction::Plain + && !user_message.text.starts_with('!') + && self.config.features.enabled(Feature::AppServerQueue) + && !self.config.ephemeral + && !self.only_user_shell_commands_running() + && self.input_queue.queued_user_messages.is_empty() + && self.input_queue.rejected_steers_queue.is_empty() + && self.input_queue.pending_steers.is_empty() + { + self.input_queue.note_server_queue_submission(); + self.submit_user_message_to_queue(user_message); + } else if !self.is_session_configured() || self.is_user_turn_pending_or_running() { + self.queue_user_message_locally( + user_message, + action, + UserMessageHistoryRecord::UserMessageText, + ); } else { self.submit_user_message(user_message); } @@ -100,6 +111,9 @@ impl ChatWidget { if self.input_queue.suppress_queue_autosend { return false; } + if self.input_queue.blocks_local_queue_autosend() { + return false; + } if self.is_user_turn_pending_or_running() { return false; } @@ -137,6 +151,45 @@ impl ChatWidget { submitted_follow_up } + pub(crate) fn note_server_queue_submission_failed(&mut self) { + self.input_queue.note_server_queue_submission_failed(); + self.refresh_pending_input_preview(); + } + + pub(crate) fn requeue_failed_server_submission(&mut self, user_message: UserMessage) { + self.input_queue.note_server_queue_submission_failed(); + self.input_queue + .queued_user_messages + .push_front(QueuedUserMessage::new( + user_message, + QueuedInputAction::Plain, + )); + self.input_queue + .queued_user_message_history_records + .push_front(UserMessageHistoryRecord::Override( + UserMessageHistoryOverride { + text: String::new(), + text_elements: Vec::new(), + }, + )); + self.refresh_pending_input_preview(); + } + + fn queue_user_message_locally( + &mut self, + user_message: UserMessage, + action: QueuedInputAction, + history_record: UserMessageHistoryRecord, + ) { + self.input_queue + .queued_user_messages + .push_back(QueuedUserMessage::new(user_message, action)); + self.input_queue + .queued_user_message_history_records + .push_back(history_record); + self.refresh_pending_input_preview(); + } + pub(super) fn is_user_turn_pending_or_running(&self) -> bool { self.input_queue.user_turn_pending_start || self.bottom_pane.is_task_running() } @@ -157,6 +210,7 @@ impl ChatWidget { preview.queued_messages, preview.pending_steers, preview.rejected_steers, + preview.has_editable_queued_message, ); } diff --git a/codex-rs/tui/src/chatwidget/input_queue.rs b/codex-rs/tui/src/chatwidget/input_queue.rs index 67a306af22..6b35af924f 100644 --- a/codex-rs/tui/src/chatwidget/input_queue.rs +++ b/codex-rs/tui/src/chatwidget/input_queue.rs @@ -5,6 +5,10 @@ use std::collections::VecDeque; +use codex_app_server_protocol::QueuedTurn; +use codex_app_server_protocol::QueuedTurnStatus; +use codex_app_server_protocol::UserInput; + use super::PendingSteer; use super::QueuedUserMessage; use super::UserMessage; @@ -16,12 +20,26 @@ pub(super) struct PendingInputPreview { pub(super) queued_messages: Vec, pub(super) pending_steers: Vec, pub(super) rejected_steers: Vec, + pub(super) has_editable_queued_message: bool, +} + +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub(super) enum ServerQueueBarrier { + #[default] + Inactive, + WaitingForSnapshot, + WaitingForTurn, + TurnRunning, } #[derive(Debug, Default)] pub(super) struct InputQueueState { /// User inputs queued while a turn is in progress. pub(super) queued_user_messages: VecDeque, + /// User inputs persisted by app-server while a turn is in progress. + pub(super) server_queued_messages: Vec, + pub(super) server_queue_has_pending_turn: bool, + pub(super) server_queue_barrier: ServerQueueBarrier, /// History records for queued user messages. Slash commands such as `/goal` /// can render history that differs from the text submitted to core, so this /// stays in lockstep with `queued_user_messages`, with missing entries @@ -46,11 +64,25 @@ pub(super) struct InputQueueState { impl InputQueueState { pub(super) fn has_queued_follow_up_messages(&self) -> bool { + !self.rejected_steers_queue.is_empty() + || !self.queued_user_messages.is_empty() + || self.server_queue_has_pending_turn + || self.server_queue_barrier != ServerQueueBarrier::Inactive + } + + pub(super) fn has_editable_queued_message(&self) -> bool { + self.has_locally_owned_follow_up_messages() + } + + pub(super) fn has_locally_owned_follow_up_messages(&self) -> bool { !self.rejected_steers_queue.is_empty() || !self.queued_user_messages.is_empty() } pub(super) fn clear(&mut self) { self.queued_user_messages.clear(); + self.server_queued_messages.clear(); + self.server_queue_has_pending_turn = false; + self.server_queue_barrier = ServerQueueBarrier::Inactive; self.queued_user_message_history_records.clear(); self.user_turn_pending_start = false; self.rejected_steers_queue.clear(); @@ -61,15 +93,20 @@ impl InputQueueState { pub(super) fn preview(&self) -> PendingInputPreview { let queued_messages = self - .queued_user_messages + .server_queued_messages .iter() - .enumerate() - .map(|(idx, message)| { - user_message_preview_text( - message, - self.queued_user_message_history_records.get(idx), - ) - }) + .cloned() + .chain( + self.queued_user_messages + .iter() + .enumerate() + .map(|(idx, message)| { + user_message_preview_text( + message, + self.queued_user_message_history_records.get(idx), + ) + }), + ) .collect(); let pending_steers = self .pending_steers @@ -91,8 +128,92 @@ impl InputQueueState { queued_messages, pending_steers, rejected_steers, + has_editable_queued_message: self.has_editable_queued_message(), } } + + pub(super) fn set_server_queued_turns( + &mut self, + queued_turns: Vec, + dispatching_queued_turn_id: Option, + ) { + self.server_queue_has_pending_turn = false; + self.server_queued_messages = queued_turns + .into_iter() + .map(|queued_turn| { + let preview = queued_turn + .submission + .input + .into_iter() + .find_map(|item| match item { + UserInput::Text { text, .. } => Some( + crate::ide_context::extract_prompt_request_with_offset(&text) + .0 + .to_string(), + ), + _ => None, + }) + .unwrap_or_else(|| "[queued input]".to_string()); + match queued_turn.status { + QueuedTurnStatus::Pending => { + self.server_queue_has_pending_turn = true; + preview + } + QueuedTurnStatus::Failed { error } => { + format!("[failed: {}] {preview}", error.message) + } + } + }) + .collect(); + if dispatching_queued_turn_id.is_some() + && self.server_queue_barrier != ServerQueueBarrier::TurnRunning + { + self.server_queue_barrier = ServerQueueBarrier::WaitingForTurn; + } else if !self.server_queue_has_pending_turn + && self.server_queue_barrier != ServerQueueBarrier::TurnRunning + { + self.server_queue_barrier = ServerQueueBarrier::Inactive; + } else if self.server_queue_has_pending_turn + && self.server_queue_barrier != ServerQueueBarrier::TurnRunning + { + self.server_queue_barrier = ServerQueueBarrier::WaitingForTurn; + } + } + + pub(super) fn note_server_queue_submission(&mut self) { + if self.server_queue_barrier == ServerQueueBarrier::Inactive { + self.server_queue_barrier = ServerQueueBarrier::WaitingForSnapshot; + } + } + + pub(super) fn note_server_queue_submission_failed(&mut self) { + if self.server_queue_barrier == ServerQueueBarrier::WaitingForSnapshot { + self.server_queue_barrier = ServerQueueBarrier::Inactive; + } + } + + pub(super) fn note_server_queue_turn_started(&mut self) { + if self.server_queue_barrier == ServerQueueBarrier::WaitingForTurn { + self.server_queue_barrier = ServerQueueBarrier::TurnRunning; + } + } + + pub(super) fn note_server_queue_turn_completed(&mut self) { + if matches!( + self.server_queue_barrier, + ServerQueueBarrier::WaitingForTurn | ServerQueueBarrier::TurnRunning + ) { + self.server_queue_barrier = if self.server_queue_has_pending_turn { + ServerQueueBarrier::WaitingForTurn + } else { + ServerQueueBarrier::Inactive + }; + } + } + + pub(super) fn blocks_local_queue_autosend(&self) -> bool { + self.server_queue_barrier != ServerQueueBarrier::Inactive + } } #[cfg(test)] @@ -107,6 +228,10 @@ mod tests { state .queued_user_messages .push_back(UserMessage::from("queued").into()); + state + .server_queued_messages + .push("server queued".to_string()); + state.server_queue_barrier = ServerQueueBarrier::TurnRunning; state .rejected_steers_queue .push_back(UserMessage::from("rejected")); @@ -122,9 +247,10 @@ mod tests { assert_eq!( state.preview(), PendingInputPreview { - queued_messages: vec!["queued".to_string()], + queued_messages: vec!["server queued".to_string(), "queued".to_string()], pending_steers: vec!["pending".to_string()], rejected_steers: vec!["rejected".to_string()], + has_editable_queued_message: true, } ); } @@ -135,6 +261,9 @@ mod tests { state .queued_user_messages .push_back(UserMessage::from("queued").into()); + state + .server_queued_messages + .push("server queued".to_string()); state .rejected_steers_queue .push_back(UserMessage::from("rejected")); @@ -144,6 +273,9 @@ mod tests { state.clear(); assert!(state.queued_user_messages.is_empty()); + assert!(state.server_queued_messages.is_empty()); + assert!(!state.server_queue_has_pending_turn); + assert_eq!(state.server_queue_barrier, ServerQueueBarrier::Inactive); assert!(state.queued_user_message_history_records.is_empty()); assert!(!state.user_turn_pending_start); assert!(state.rejected_steers_queue.is_empty()); @@ -151,4 +283,115 @@ mod tests { assert!(state.pending_steers.is_empty()); assert!(!state.submit_pending_steers_after_interrupt); } + + #[test] + fn server_queue_barrier_waits_for_snapshot_before_tracking_turn_lifecycle() { + let mut state = InputQueueState::default(); + + state.note_server_queue_submission(); + state.note_server_queue_turn_started(); + assert_eq!( + state.server_queue_barrier, + ServerQueueBarrier::WaitingForSnapshot + ); + + state.set_server_queued_turns( + vec![QueuedTurn { + id: "queued".to_string(), + submission: Default::default(), + status: QueuedTurnStatus::Pending, + }], + /*dispatching_queued_turn_id*/ None, + ); + assert_eq!( + state.server_queue_barrier, + ServerQueueBarrier::WaitingForTurn + ); + + state.set_server_queued_turns(Vec::new(), Some("queued".to_string())); + state.note_server_queue_turn_started(); + assert_eq!(state.server_queue_barrier, ServerQueueBarrier::TurnRunning); + state.note_server_queue_turn_completed(); + assert_eq!(state.server_queue_barrier, ServerQueueBarrier::Inactive); + } + + #[test] + fn failed_server_queue_preview_includes_error() { + let mut state = InputQueueState::default(); + state.set_server_queued_turns( + vec![QueuedTurn { + id: "queued".to_string(), + submission: Default::default(), + status: QueuedTurnStatus::Failed { + error: codex_app_server_protocol::TurnError { + message: "dispatch failed".to_string(), + codex_error_info: None, + additional_details: None, + }, + }, + }], + /*dispatching_queued_turn_id*/ None, + ); + + assert_eq!( + state.server_queued_messages, + vec!["[failed: dispatch failed] [queued input]".to_string()] + ); + assert_eq!(state.server_queue_barrier, ServerQueueBarrier::Inactive); + assert!(!state.has_queued_follow_up_messages()); + } + + #[test] + fn replayed_dispatch_completion_releases_waiting_barrier() { + let mut state = InputQueueState::default(); + state.set_server_queued_turns(Vec::new(), Some("queued".to_string())); + assert_eq!( + state.server_queue_barrier, + ServerQueueBarrier::WaitingForTurn + ); + + state.note_server_queue_turn_completed(); + + assert_eq!(state.server_queue_barrier, ServerQueueBarrier::Inactive); + } + + #[test] + fn empty_server_queue_snapshot_releases_waiting_barrier_without_dispatch() { + let mut state = InputQueueState::default(); + state.note_server_queue_submission(); + state.set_server_queued_turns( + vec![QueuedTurn { + id: "queued".to_string(), + submission: Default::default(), + status: QueuedTurnStatus::Pending, + }], + /*dispatching_queued_turn_id*/ None, + ); + + state.set_server_queued_turns(Vec::new(), /*dispatching_queued_turn_id*/ None); + + assert_eq!(state.server_queue_barrier, ServerQueueBarrier::Inactive); + } + + #[test] + fn server_queue_preview_strips_injected_ide_context() { + let mut state = InputQueueState::default(); + state.set_server_queued_turns( + vec![QueuedTurn { + id: "queued".to_string(), + submission: codex_app_server_protocol::TurnSubmission { + input: vec![UserInput::Text { + text: "# Context from my IDE setup:\n\n## My request for Codex:\nqueue me" + .to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }, + status: QueuedTurnStatus::Pending, + }], + /*dispatching_queued_turn_id*/ None, + ); + + assert_eq!(state.server_queued_messages, vec!["queue me".to_string()]); + } } diff --git a/codex-rs/tui/src/chatwidget/input_restore.rs b/codex-rs/tui/src/chatwidget/input_restore.rs index b7e8ee72e0..9984c50a14 100644 --- a/codex-rs/tui/src/chatwidget/input_restore.rs +++ b/codex-rs/tui/src/chatwidget/input_restore.rs @@ -148,7 +148,9 @@ impl ChatWidget { /// state stays aligned with the merged attachment list. Returns `None` when there is nothing to /// restore. fn drain_pending_messages_for_restore(&mut self) -> Option { - if self.input_queue.pending_steers.is_empty() && !self.has_queued_follow_up_messages() { + if self.input_queue.pending_steers.is_empty() + && !self.input_queue.has_locally_owned_follow_up_messages() + { return None; } @@ -273,6 +275,9 @@ impl ChatWidget { .input_queue .queued_user_message_history_records .clone(), + server_queued_messages: self.input_queue.server_queued_messages.clone(), + server_queue_has_pending_turn: self.input_queue.server_queue_has_pending_turn, + server_queue_barrier: self.input_queue.server_queue_barrier, user_turn_pending_start: self.input_queue.user_turn_pending_start, current_collaboration_mode: self.current_collaboration_mode.clone(), active_collaboration_mask: self.active_collaboration_mask.clone(), @@ -352,6 +357,10 @@ impl ChatWidget { self.input_queue.queued_user_messages.len(), UserMessageHistoryRecord::UserMessageText, ); + self.input_queue.server_queued_messages = input_state.server_queued_messages; + self.input_queue.server_queue_has_pending_turn = + input_state.server_queue_has_pending_turn; + self.input_queue.server_queue_barrier = input_state.server_queue_barrier; } else { self.turn_lifecycle .restore_running(/*running*/ false, Instant::now()); diff --git a/codex-rs/tui/src/chatwidget/input_submission.rs b/codex-rs/tui/src/chatwidget/input_submission.rs index 578a35728f..808f0e247f 100644 --- a/codex-rs/tui/src/chatwidget/input_submission.rs +++ b/codex-rs/tui/src/chatwidget/input_submission.rs @@ -2,6 +2,12 @@ use super::*; +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum SubmissionTarget { + Turn, + Queue, +} + impl ChatWidget { pub(super) fn user_message_from_submission( &mut self, @@ -69,6 +75,18 @@ impl ChatWidget { ); } + pub(super) fn submit_user_message_to_queue(&mut self, user_message: UserMessage) { + let (accepted, _) = self.submit_user_message_with_history_and_shell_escape_policy( + user_message, + UserMessageHistoryRecord::UserMessageText, + ShellEscapePolicy::Disallow, + SubmissionTarget::Queue, + ); + if !accepted { + self.note_server_queue_submission_failed(); + } + } + pub(super) fn submit_user_message_with_history_record( &mut self, user_message: UserMessage, @@ -78,6 +96,7 @@ impl ChatWidget { user_message, history_record, ShellEscapePolicy::Allow, + SubmissionTarget::Turn, ) .0 } @@ -91,6 +110,7 @@ impl ChatWidget { user_message, UserMessageHistoryRecord::UserMessageText, shell_escape_policy, + SubmissionTarget::Turn, ) .1 } @@ -100,6 +120,7 @@ impl ChatWidget { user_message: UserMessage, history_record: UserMessageHistoryRecord, shell_escape_policy: ShellEscapePolicy, + submission_target: SubmissionTarget, ) -> (bool, Option) { if !self.is_session_configured() { tracing::warn!("cannot submit user message before session is configured; queueing"); @@ -137,6 +158,7 @@ impl ChatWidget { ); return (false, None); } + let fallback_user_message = user_message.clone(); let UserMessage { text, local_images, @@ -145,7 +167,8 @@ impl ChatWidget { mention_bindings, } = user_message; - let render_in_history = !self.turn_lifecycle.agent_turn_running; + let render_in_history = + submission_target == SubmissionTarget::Turn && !self.turn_lifecycle.agent_turn_running; let mut items: Vec = Vec::new(); // Special-case: "!cmd" executes a local shell command instead of sending to the model. @@ -319,37 +342,49 @@ impl ChatWidget { } else { None }; - let pending_steer = (!render_in_history).then(|| PendingSteer { - user_message: UserMessage { - text: text.clone(), - local_images: local_images.clone(), - remote_image_urls: remote_image_urls.clone(), - text_elements: text_elements.clone(), - mention_bindings: mention_bindings.clone(), - }, - history_record: history_record.clone(), - compare_key: Self::pending_steer_compare_key_from_items(&items), - }); - let personality = self - .config - .personality - .filter(|_| self.config.features.enabled(Feature::Personality)) - .filter(|_| self.current_model_supports_personality()); - let service_tier = self.service_tier_update_for_core(); - let active_permission_profile = self.config.permissions.active_permission_profile(); - let op = AppCommand::user_turn( - items, - self.config.cwd.to_path_buf(), - AskForApproval::from(self.config.permissions.approval_policy.value()), - active_permission_profile, - effective_mode.model().to_string(), - effective_mode.reasoning_effort(), - /*summary*/ None, - service_tier, - /*final_output_json_schema*/ None, - collaboration_mode, - personality, - ); + let pending_steer = (submission_target == SubmissionTarget::Turn && !render_in_history) + .then(|| PendingSteer { + user_message: UserMessage { + text: text.clone(), + local_images: local_images.clone(), + remote_image_urls: remote_image_urls.clone(), + text_elements: text_elements.clone(), + mention_bindings: mention_bindings.clone(), + }, + history_record: history_record.clone(), + compare_key: Self::pending_steer_compare_key_from_items(&items), + }); + let op = match submission_target { + SubmissionTarget::Turn => { + let personality = self + .config + .personality + .filter(|_| self.config.features.enabled(Feature::Personality)) + .filter(|_| self.current_model_supports_personality()); + let service_tier = self.service_tier_update_for_core(); + let active_permission_profile = self.config.permissions.active_permission_profile(); + AppCommand::user_turn( + items, + self.config.cwd.to_path_buf(), + AskForApproval::from(self.config.permissions.approval_policy.value()), + active_permission_profile, + effective_mode.model().to_string(), + effective_mode.reasoning_effort(), + /*summary*/ None, + service_tier, + /*final_output_json_schema*/ None, + collaboration_mode, + personality, + ) + } + SubmissionTarget::Queue => AppCommand::queue_turn( + TurnSubmission { + input: items, + ..Default::default() + }, + fallback_user_message, + ), + }; if !self.submit_op(op.clone()) { return (false, None); diff --git a/codex-rs/tui/src/chatwidget/interaction.rs b/codex-rs/tui/src/chatwidget/interaction.rs index d9451fc9d5..654c660275 100644 --- a/codex-rs/tui/src/chatwidget/interaction.rs +++ b/codex-rs/tui/src/chatwidget/interaction.rs @@ -101,7 +101,7 @@ impl ChatWidget { if key_event.kind == KeyEventKind::Press && self.chat_keymap.edit_queued_message.is_pressed(key_event) - && self.has_queued_follow_up_messages() + && self.input_queue.has_editable_queued_message() && self.bottom_pane.no_modal_or_popup_active() { if let Some(user_message) = self.pop_latest_queued_user_message() { diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index 5215489621..17a53f9c64 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -54,7 +54,16 @@ impl ChatWidget { ServerNotification::ThreadSettingsUpdated(notification) => { self.on_thread_settings_updated(notification); } + ServerNotification::ThreadQueueChanged(notification) => { + self.input_queue.set_server_queued_turns( + notification.queued_turns, + notification.dispatching_queued_turn_id, + ); + self.refresh_pending_input_preview(); + self.maybe_send_next_queued_input(); + } ServerNotification::TurnStarted(notification) => { + self.input_queue.note_server_queue_turn_started(); self.turn_lifecycle.last_turn_id = Some(notification.turn.id); self.last_non_retry_error = None; if !matches!(replay_kind, Some(ReplayKind::ResumeInitialMessages)) { @@ -62,6 +71,7 @@ impl ChatWidget { } } ServerNotification::TurnCompleted(notification) => { + self.input_queue.note_server_queue_turn_completed(); self.handle_turn_completed_notification(notification, replay_kind); } ServerNotification::ItemStarted(notification) => { @@ -222,7 +232,6 @@ impl ChatWidget { | ServerNotification::ThreadStatusChanged(_) | ServerNotification::ThreadArchived(_) | ServerNotification::ThreadUnarchived(_) - | ServerNotification::ThreadQueueChanged(_) | ServerNotification::RawResponseItemCompleted(_) | ServerNotification::CommandExecOutputDelta(_) | ServerNotification::ProcessOutputDelta(_) 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 777439a1cd..b4b7c9a6ad 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 @@ -16,6 +16,7 @@ expression: normalize_snapshot_paths(term.backend().vt100().screen().contents()) • Messages to be submitted at end of turn ↳ Steer submitted while /compact was running. + ⌥ + ↑ edit last queued message › Ask Codex 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 d90ccd9c5f..707772c6df 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 @@ -16,6 +16,7 @@ expression: normalize_snapshot_paths(term.backend().vt100().screen().contents()) • Messages to be submitted at end of turn ↳ Steer submitted while /review was running. + ⌥ + ↑ edit last queued message › Ask Codex to do anything diff --git a/codex-rs/tui/src/chatwidget/tests/app_server.rs b/codex-rs/tui/src/chatwidget/tests/app_server.rs index f4fcc85e43..9e9384914a 100644 --- a/codex-rs/tui/src/chatwidget/tests/app_server.rs +++ b/codex-rs/tui/src/chatwidget/tests/app_server.rs @@ -153,6 +153,65 @@ async fn thread_settings_updated_updates_visible_state_without_transcript() { assert_eq!(chat.current_model(), "gpt-5.4"); } +#[tokio::test] +async fn thread_queue_changed_updates_server_queue_preview() { + let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let thread_id = ThreadId::new(); + + chat.handle_server_notification( + ServerNotification::ThreadQueueChanged( + codex_app_server_protocol::ThreadQueueChangedNotification { + thread_id: thread_id.to_string(), + dispatching_queued_turn_id: None, + queued_turns: vec![codex_app_server_protocol::QueuedTurn { + id: "queued-turn".to_string(), + submission: codex_app_server_protocol::TurnSubmission { + input: vec![UserInput::Text { + text: "queued follow-up".to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }, + status: codex_app_server_protocol::QueuedTurnStatus::Pending, + }], + }, + ), + /*replay_kind*/ None, + ); + + assert_eq!( + chat.input_queue.server_queued_messages, + vec!["queued follow-up".to_string()] + ); +} + +#[tokio::test] +async fn empty_thread_queue_snapshot_releases_local_queue() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let thread_id = ThreadId::new(); + chat.thread_id = Some(thread_id); + chat.input_queue.note_server_queue_submission(); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("local fallback").into()); + chat.input_queue + .queued_user_message_history_records + .push_back(UserMessageHistoryRecord::UserMessageText); + + chat.handle_server_notification( + ServerNotification::ThreadQueueChanged( + codex_app_server_protocol::ThreadQueueChangedNotification { + thread_id: thread_id.to_string(), + dispatching_queued_turn_id: None, + queued_turns: Vec::new(), + }, + ), + /*replay_kind*/ None, + ); + + assert_matches!(next_submit_op(&mut op_rx), Op::UserTurn { .. }); +} + #[tokio::test] async fn thread_settings_updated_preserves_default_settings_for_plan_mode() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(Some("gpt-5.3-codex")).await; diff --git a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs index 27ae4daff2..95df6c718a 100644 --- a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs +++ b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs @@ -926,6 +926,195 @@ async fn empty_enter_during_task_does_not_queue() { assert!(chat.input_queue.queued_user_messages.is_empty()); } +#[tokio::test] +async fn app_server_queue_gate_routes_plain_follow_up_to_durable_queue() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_feature_enabled(Feature::AppServerQueue, /*enabled*/ true); + chat.thread_id = Some(ThreadId::new()); + chat.bottom_pane.set_task_running(/*running*/ true); + + chat.queue_user_message("queued submission".into()); + + assert!(chat.input_queue.queued_user_messages.is_empty()); + assert!(chat.input_queue.blocks_local_queue_autosend()); + match op_rx.try_recv() { + Ok(Op::QueueTurn { + submission, + fallback_user_message, + }) => { + assert_eq!( + submission.input, + vec![UserInput::Text { + text: "queued submission".to_string(), + text_elements: Vec::new(), + }] + ); + assert_eq!(fallback_user_message.text, "queued submission"); + } + other => panic!("expected Op::QueueTurn, got {other:?}"), + } +} + +#[tokio::test] +async fn app_server_queue_gate_keeps_ephemeral_follow_up_local() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_feature_enabled(Feature::AppServerQueue, /*enabled*/ true); + chat.config.ephemeral = true; + chat.thread_id = Some(ThreadId::new()); + chat.bottom_pane.set_task_running(/*running*/ true); + + chat.queue_user_message("side conversation follow-up".into()); + + assert_eq!(chat.input_queue.queued_user_messages.len(), 1); + assert!(op_rx.try_recv().is_err()); +} + +#[tokio::test] +async fn app_server_queue_gate_keeps_shell_follow_up_local() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_feature_enabled(Feature::AppServerQueue, /*enabled*/ true); + chat.thread_id = Some(ThreadId::new()); + chat.bottom_pane.set_task_running(/*running*/ true); + + chat.queue_user_message("!echo hello".into()); + + assert_eq!(chat.input_queue.queued_user_messages.len(), 1); + assert!(op_rx.try_recv().is_err()); +} + +#[tokio::test] +async fn app_server_queue_gate_keeps_plain_follow_up_local_behind_older_shell_input() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_feature_enabled(Feature::AppServerQueue, /*enabled*/ true); + chat.thread_id = Some(ThreadId::new()); + chat.bottom_pane.set_task_running(/*running*/ true); + + chat.queue_user_message("!echo hello".into()); + chat.queue_user_message("plain follow-up".into()); + + assert_eq!(chat.input_queue.queued_user_messages.len(), 2); + assert_eq!(chat.input_queue.queued_user_messages[0].text, "!echo hello"); + assert_eq!( + chat.input_queue.queued_user_messages[1].text, + "plain follow-up" + ); + assert!(op_rx.try_recv().is_err()); +} + +#[tokio::test] +async fn app_server_queue_gate_keeps_plain_follow_up_local_behind_pending_steer() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_feature_enabled(Feature::AppServerQueue, /*enabled*/ true); + chat.thread_id = Some(ThreadId::new()); + chat.bottom_pane.set_task_running(/*running*/ true); + chat.input_queue + .pending_steers + .push_back(pending_steer("pending steer")); + + chat.queue_user_message("plain follow-up".into()); + + assert_eq!(chat.input_queue.queued_user_messages.len(), 1); + assert_eq!( + chat.input_queue.queued_user_messages[0].text, + "plain follow-up" + ); + assert!(op_rx.try_recv().is_err()); +} + +#[tokio::test] +async fn failed_app_server_queue_submission_falls_back_to_local_queue() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_feature_enabled(Feature::AppServerQueue, /*enabled*/ true); + chat.thread_id = Some(ThreadId::new()); + chat.bottom_pane.set_task_running(/*running*/ true); + + chat.queue_user_message("queued submission".into()); + let Op::QueueTurn { + fallback_user_message, + .. + } = op_rx.try_recv().expect("queued turn op") + else { + panic!("expected queued turn op"); + }; + + chat.requeue_failed_server_submission(fallback_user_message); + + assert!(!chat.input_queue.blocks_local_queue_autosend()); + assert_eq!(chat.input_queue.queued_user_messages.len(), 1); + assert_eq!( + chat.input_queue.queued_user_messages[0].text, + "queued submission" + ); + assert!(matches!( + chat.input_queue + .queued_user_message_history_records + .front(), + Some(UserMessageHistoryRecord::Override(history)) if history.text.is_empty() + )); +} + +#[tokio::test] +async fn failed_app_server_queue_submission_starts_local_fallback_when_idle() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + chat.input_queue.note_server_queue_submission(); + chat.requeue_failed_server_submission("queued submission".into()); + + assert!(chat.maybe_send_next_queued_input()); + assert_matches!(next_submit_op(&mut op_rx), Op::UserTurn { .. }); +} + +#[tokio::test] +async fn failed_app_server_queue_submission_returns_to_front_of_local_queue() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_feature_enabled(Feature::AppServerQueue, /*enabled*/ true); + chat.thread_id = Some(ThreadId::new()); + chat.bottom_pane.set_task_running(/*running*/ true); + + chat.queue_user_message("durable first".into()); + let Op::QueueTurn { + fallback_user_message, + .. + } = op_rx.try_recv().expect("queued turn op") + else { + panic!("expected queued turn op"); + }; + chat.queue_user_message("!echo local second".into()); + + chat.requeue_failed_server_submission(fallback_user_message); + + assert_eq!(chat.input_queue.queued_user_messages.len(), 2); + assert_eq!( + chat.input_queue.queued_user_messages[0].text, + "durable first" + ); + assert_eq!( + chat.input_queue.queued_user_messages[1].text, + "!echo local second" + ); +} + +#[tokio::test] +async fn interrupted_turn_keeps_server_owned_follow_up_scheduled() { + let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.on_task_started(); + chat.input_queue.server_queued_messages = vec!["durable follow-up".to_string()]; + chat.input_queue.server_queue_has_pending_turn = true; + chat.input_queue.server_queue_barrier = ServerQueueBarrier::WaitingForTurn; + + chat.on_interrupted_turn(TurnAbortReason::Interrupted); + + assert_eq!( + chat.input_queue.server_queued_messages, + vec!["durable follow-up".to_string()] + ); + assert_eq!( + chat.input_queue.server_queue_barrier, + ServerQueueBarrier::WaitingForTurn + ); + assert!(chat.bottom_pane.composer_text().is_empty()); +} + #[tokio::test] async fn pending_steer_esc_does_not_steal_vim_insert_escape() { let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; @@ -995,6 +1184,9 @@ async fn restore_thread_input_state_syncs_sleep_inhibitor_state() { rejected_steer_history_records: VecDeque::new(), queued_user_messages: VecDeque::new(), queued_user_message_history_records: VecDeque::new(), + server_queued_messages: Vec::new(), + server_queue_has_pending_turn: false, + server_queue_barrier: ServerQueueBarrier::Inactive, user_turn_pending_start: false, current_collaboration_mode: chat.current_collaboration_mode.clone(), active_collaboration_mask: chat.active_collaboration_mask.clone(), @@ -1013,6 +1205,30 @@ async fn restore_thread_input_state_syncs_sleep_inhibitor_state() { assert!(!chat.bottom_pane.is_task_running()); } +#[tokio::test] +async fn restore_thread_input_state_preserves_server_queue_state() { + let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.input_queue.server_queued_messages = vec!["durable follow-up".to_string()]; + chat.input_queue.server_queue_has_pending_turn = true; + chat.input_queue.server_queue_barrier = ServerQueueBarrier::WaitingForTurn; + let input_state = chat + .capture_thread_input_state() + .expect("captured thread input state"); + chat.input_queue.clear(); + + chat.restore_thread_input_state(Some(input_state)); + + assert_eq!( + chat.input_queue.server_queued_messages, + vec!["durable follow-up".to_string()] + ); + assert_eq!( + chat.input_queue.server_queue_barrier, + ServerQueueBarrier::WaitingForTurn + ); + assert!(chat.input_queue.server_queue_has_pending_turn); +} + #[tokio::test] async fn alt_up_edits_most_recent_queued_message() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; diff --git a/codex-rs/tui/src/chatwidget/tests/exec_flow.rs b/codex-rs/tui/src/chatwidget/tests/exec_flow.rs index 8dbace4378..1b38ac94b2 100644 --- a/codex-rs/tui/src/chatwidget/tests/exec_flow.rs +++ b/codex-rs/tui/src/chatwidget/tests/exec_flow.rs @@ -1005,6 +1005,7 @@ async fn bang_shell_enter_while_task_running_submits_run_user_shell_command() { #[tokio::test] async fn user_message_during_user_shell_command_is_queued_not_steered() { let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_feature_enabled(Feature::AppServerQueue, /*enabled*/ true); chat.thread_id = Some(ThreadId::new()); handle_turn_started(&mut chat, "turn-1"); let begin = begin_exec_with_source( diff --git a/codex-rs/tui/src/chatwidget/tests/review_mode.rs b/codex-rs/tui/src/chatwidget/tests/review_mode.rs index faf8636ad8..a9c55f6e10 100644 --- a/codex-rs/tui/src/chatwidget/tests/review_mode.rs +++ b/codex-rs/tui/src/chatwidget/tests/review_mode.rs @@ -359,6 +359,9 @@ async fn restore_thread_input_state_restores_pending_steers_without_downgrading_ rejected_steer_history_records: VecDeque::new(), queued_user_messages, queued_user_message_history_records: VecDeque::new(), + server_queued_messages: Vec::new(), + server_queue_has_pending_turn: false, + server_queue_barrier: ServerQueueBarrier::Inactive, user_turn_pending_start: false, current_collaboration_mode: chat.current_collaboration_mode.clone(), active_collaboration_mask: chat.active_collaboration_mask.clone(), diff --git a/codex-rs/tui/src/chatwidget/tests/slash_commands.rs b/codex-rs/tui/src/chatwidget/tests/slash_commands.rs index 4f425c25c9..7582dc0f10 100644 --- a/codex-rs/tui/src/chatwidget/tests/slash_commands.rs +++ b/codex-rs/tui/src/chatwidget/tests/slash_commands.rs @@ -1568,6 +1568,28 @@ async fn queued_follow_up_suppresses_agent_turn_complete_notification() { assert_matches!(next_submit_op(&mut op_rx), Op::UserTurn { .. }); } +#[tokio::test] +async fn durable_queued_follow_up_suppresses_agent_turn_complete_notification() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.set_feature_enabled(Feature::AppServerQueue, /*enabled*/ true); + chat.thread_id = Some(ThreadId::new()); + handle_turn_started(&mut chat, "turn-1"); + chat.queue_user_message("Continue durably".into()); + assert_matches!(op_rx.try_recv(), Ok(Op::QueueTurn { .. })); + chat.input_queue.set_server_queued_turns( + vec![codex_app_server_protocol::QueuedTurn { + id: "queued".to_string(), + submission: Default::default(), + status: codex_app_server_protocol::QueuedTurnStatus::Pending, + }], + /*dispatching_queued_turn_id*/ None, + ); + + complete_turn_with_message(&mut chat, "turn-1", Some("Still working")); + + assert_matches!(chat.pending_notification, None); +} + #[tokio::test] async fn queued_menu_slash_keeps_agent_turn_complete_notification() { let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(Some("gpt-5.2")).await; diff --git a/codex-rs/tui/src/chatwidget/turn_runtime.rs b/codex-rs/tui/src/chatwidget/turn_runtime.rs index 4868d6a25c..d7a05965e4 100644 --- a/codex-rs/tui/src/chatwidget/turn_runtime.rs +++ b/codex-rs/tui/src/chatwidget/turn_runtime.rs @@ -191,11 +191,12 @@ impl ChatWidget { .current_goal_status .as_ref() .is_some_and(GoalStatusState::is_active); + let durable_follow_up_scheduled = self.input_queue.blocks_local_queue_autosend(); // Emit a notification when the agent is truly waiting for the user. - // Queued follow-up input and active goal continuation both start the - // next turn immediately, so notifying at that boundary would feel like - // a false "needs attention". - if !follow_up_started && !active_goal_continuing { + // Queued follow-up input and active goal continuation both start or + // schedule the next turn immediately, so notifying at that boundary + // would feel like a false "needs attention". + if !follow_up_started && !durable_follow_up_scheduled && !active_goal_continuing { self.notify(Notification::AgentTurnComplete { response: notification_response, }); diff --git a/codex-rs/tui/src/chatwidget/user_messages.rs b/codex-rs/tui/src/chatwidget/user_messages.rs index 80a36b5815..c2ac38f95b 100644 --- a/codex-rs/tui/src/chatwidget/user_messages.rs +++ b/codex-rs/tui/src/chatwidget/user_messages.rs @@ -22,6 +22,7 @@ use codex_protocol::user_input::ByteRange; use codex_protocol::user_input::TextElement; use super::ChatWidget; +use super::input_queue::ServerQueueBarrier; #[derive(Debug, Clone, PartialEq)] pub(crate) struct UserMessage { @@ -125,6 +126,9 @@ pub(crate) struct ThreadInputState { pub(super) rejected_steer_history_records: VecDeque, pub(super) queued_user_messages: VecDeque, pub(super) queued_user_message_history_records: VecDeque, + pub(super) server_queued_messages: Vec, + pub(super) server_queue_has_pending_turn: bool, + pub(super) server_queue_barrier: ServerQueueBarrier, pub(super) user_turn_pending_start: bool, pub(super) current_collaboration_mode: CollaborationMode, pub(super) active_collaboration_mask: Option,