From e096bc93f045b1912a5b76fbf162eafe0ae303ff Mon Sep 17 00:00:00 2001 From: Edward Frazer Date: Mon, 15 Jun 2026 19:33:20 +0000 Subject: [PATCH] feat: queue TUI follow-ups through app-server --- codex-rs/tui/src/app/app_server_events.rs | 14 ++ codex-rs/tui/src/app/tests.rs | 1 + .../tui/src/app/tests/user_message_queue.rs | 120 ++++++++++++++ codex-rs/tui/src/app/thread_routing.rs | 86 ++++++++++ codex-rs/tui/src/app_command.rs | 9 ++ codex-rs/tui/src/app_server_session.rs | 73 +++++++++ codex-rs/tui/src/bottom_pane/mod.rs | 5 + .../src/bottom_pane/pending_input_preview.rs | 20 +++ ...der_durable_message_without_edit_hint.snap | 19 +++ codex-rs/tui/src/chatwidget/ide_context.rs | 4 + .../tui/src/chatwidget/ide_context_tests.rs | 21 +++ codex-rs/tui/src/chatwidget/input_flow.rs | 75 ++++++++- codex-rs/tui/src/chatwidget/input_queue.rs | 151 ++++++++++-------- .../tui/src/chatwidget/input_queue_tests.rs | 134 ++++++++++++++++ codex-rs/tui/src/chatwidget/input_restore.rs | 10 +- .../tui/src/chatwidget/input_submission.rs | 40 ++++- codex-rs/tui/src/chatwidget/protocol.rs | 23 ++- codex-rs/tui/src/chatwidget/session_flow.rs | 13 ++ codex-rs/tui/src/chatwidget/slash_dispatch.rs | 45 ++++++ codex-rs/tui/src/chatwidget/tests.rs | 1 + .../chatwidget/tests/composer_submission.rs | 6 +- .../tui/src/chatwidget/tests/review_mode.rs | 1 + codex-rs/tui/src/chatwidget/tests/side.rs | 24 +++ .../src/chatwidget/tests/slash_commands.rs | 41 +++++ .../chatwidget/tests/user_message_queue.rs | 100 ++++++++++++ codex-rs/tui/src/chatwidget/turn_runtime.rs | 3 +- codex-rs/tui/src/chatwidget/user_messages.rs | 10 ++ 27 files changed, 978 insertions(+), 71 deletions(-) create mode 100644 codex-rs/tui/src/app/tests/user_message_queue.rs create mode 100644 codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_durable_message_without_edit_hint.snap create mode 100644 codex-rs/tui/src/chatwidget/ide_context_tests.rs create mode 100644 codex-rs/tui/src/chatwidget/input_queue_tests.rs create mode 100644 codex-rs/tui/src/chatwidget/tests/user_message_queue.rs diff --git a/codex-rs/tui/src/app/app_server_events.rs b/codex-rs/tui/src/app/app_server_events.rs index 567a4b26a8..554a719ff6 100644 --- a/codex-rs/tui/src/app/app_server_events.rs +++ b/codex-rs/tui/src/app/app_server_events.rs @@ -40,6 +40,20 @@ impl App { ); self.refresh_mcp_startup_expected_servers_from_config(); self.chat_widget.finish_mcp_startup_after_lag(); + if self + .config + .features + .enabled(codex_features::Feature::UserMessageQueue) + && !self.config.ephemeral + && !self.chat_widget.side_conversation_active() + && let Some(thread_id) = self.chat_widget.thread_id() + && self.chat_widget.start_server_queue_refresh(thread_id) + { + self.app_event_tx.send(AppEvent::SubmitThreadOp { + thread_id, + op: AppCommand::RefreshUserMessageQueue, + }); + } } AppServerEvent::ServerNotification(notification) => { self.handle_server_notification_event(app_server_client, notification) diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index c3c83e6d42..ee22406400 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -3,6 +3,7 @@ mod model_catalog; mod session_summary; mod startup; +mod user_message_queue; use super::*; use crate::app_backtrack::BacktrackSelection; diff --git a/codex-rs/tui/src/app/tests/user_message_queue.rs b/codex-rs/tui/src/app/tests/user_message_queue.rs new file mode 100644 index 0000000000..ba4bbf470f --- /dev/null +++ b/codex-rs/tui/src/app/tests/user_message_queue.rs @@ -0,0 +1,120 @@ +use super::*; +use crate::chatwidget::UserMessage; + +#[tokio::test] +async fn lag_refresh_targets_the_widget_thread() { + let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await; + let 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(); + app.chat_widget + .set_feature_enabled(Feature::UserMessageQueue, /*enabled*/ false); + app.chat_widget + .handle_thread_session(test_thread_session(thread_id, test_path_buf("/tmp/origin"))); + let _ = app.config.features.enable(Feature::UserMessageQueue); + app.chat_widget + .set_feature_enabled(Feature::UserMessageQueue, /*enabled*/ true); + app.active_thread_id = Some(ThreadId::new()); + while app_event_rx.try_recv().is_ok() {} + + app.handle_app_server_event( + &app_server, + codex_app_server_client::AppServerEvent::Lagged { skipped: 1 }, + ) + .await; + + assert!(matches!( + app_event_rx.try_recv(), + Ok(AppEvent::SubmitThreadOp { + thread_id: refresh_thread_id, + op: AppCommand::RefreshUserMessageQueue, + }) if refresh_thread_id == thread_id + )); +} + +#[tokio::test] +async fn queue_snapshot_updates_its_origin_thread() { + let mut app = make_test_app().await; + let thread_id = ThreadId::new(); + app.chat_widget + .handle_thread_session(test_thread_session(thread_id, test_path_buf("/tmp/origin"))); + let mut input_state = app + .chat_widget + .capture_thread_input_state() + .expect("chat widget input state"); + input_state.user_message_queue.refresh_in_flight = true; + app.ensure_thread_channel(thread_id) + .store + .lock() + .await + .input_state = Some(input_state.clone()); + let (chat_widget, _app_event_tx, _rx, _op_rx) = make_chatwidget_manual_with_sender().await; + app.chat_widget = chat_widget; + let snapshot = codex_app_server_protocol::ThreadQueueListResponse { + data: vec![codex_app_server_protocol::QueuedItem { + id: "queued-1".to_string(), + submission: codex_app_server_protocol::TurnSubmission::default(), + provenance: codex_app_server_protocol::QueuedItemProvenance::User, + status: codex_app_server_protocol::QueuedItemStatus::Pending, + }], + next_cursor: None, + }; + let mut expected = input_state; + expected.user_message_queue.set_snapshot(snapshot.clone()); + + app.apply_server_queue_snapshot(thread_id, snapshot).await; + + let actual = app + .thread_event_channels + .get(&thread_id) + .expect("origin thread channel") + .store + .lock() + .await + .input_state + .clone(); + pretty_assertions::assert_eq!(actual, Some(expected)); +} + +#[tokio::test] +async fn unsupported_server_queue_requeues_the_message_locally() { + Box::pin(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 started = app_server + .start_thread(app.chat_widget.config_ref()) + .await + .expect("thread/start should succeed"); + let thread_id = started.session.thread_id; + assert!(!app_server.thread_queue_supported(thread_id).await); + app.enqueue_primary_thread_session(started.session, started.turns) + .await + .expect("primary thread should be registered"); + app.chat_widget.handle_server_notification( + turn_started_notification(thread_id, "active-turn"), + /*replay_kind*/ None, + ); + let op = AppCommand::QueueUserMessage { + submission: codex_app_server_protocol::TurnSubmissionParams::default(), + fallback_user_message: UserMessage::from("use the local fallback"), + }; + + assert!( + app.try_submit_active_thread_op_via_app_server(&mut app_server, thread_id, &op) + .await + .expect("queue fallback should not fail") + ); + pretty_assertions::assert_eq!( + app.chat_widget.queued_user_message_texts(), + vec!["use the local fallback"] + ); + }) + .await; +} diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index c83920556c..2aeb5dcb88 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -5,7 +5,9 @@ //! when the visible thread changes. use super::*; +use crate::chatwidget::UserMessage; use crate::session_resume::read_session_model; +use codex_app_server_protocol::ThreadQueueListResponse; impl App { pub(super) async fn shutdown_current_thread(&mut self, app_server: &mut AppServerSession) { @@ -451,6 +453,45 @@ impl App { Ok(()) } + async fn apply_failed_server_queue_submission( + &mut self, + thread_id: ThreadId, + fallback: UserMessage, + ) { + if self.chat_widget.thread_id() == Some(thread_id) { + self.chat_widget.requeue_failed_server_submission(fallback); + self.chat_widget.maybe_send_next_queued_input(); + } else if let Some(channel) = self.thread_event_channels.get(&thread_id) + && let Some(input_state) = channel.store.lock().await.input_state.as_mut() + { + input_state.requeue_failed_server_submission(fallback); + } + } + + pub(super) async fn apply_server_queue_snapshot( + &mut self, + thread_id: ThreadId, + response: ThreadQueueListResponse, + ) { + if self.chat_widget.thread_id() == Some(thread_id) { + self.chat_widget.set_server_queue_snapshot(response); + } else if let Some(channel) = self.thread_event_channels.get(&thread_id) + && let Some(input_state) = channel.store.lock().await.input_state.as_mut() + { + input_state.user_message_queue.set_snapshot(response); + } + } + + async fn finish_server_queue_refresh(&mut self, thread_id: ThreadId) { + if self.chat_widget.thread_id() == Some(thread_id) { + self.chat_widget.finish_server_queue_refresh(); + } else if let Some(channel) = self.thread_event_channels.get(&thread_id) + && let Some(input_state) = channel.store.lock().await.input_state.as_mut() + { + input_state.user_message_queue.refresh_in_flight = false; + } + } + /// Persist prompt text in the local cross-session message history. pub(super) fn append_message_history_entry(&self, thread_id: ThreadId, text: String) { let history_config = codex_message_history::HistoryConfig::new( @@ -660,6 +701,51 @@ impl App { } Ok(true) } + AppCommand::QueueUserMessage { + submission, + fallback_user_message, + } => { + if !app_server.thread_queue_supported(thread_id).await { + self.apply_failed_server_queue_submission( + thread_id, + fallback_user_message.clone(), + ) + .await; + return Ok(true); + } + match app_server + .thread_queue_add(thread_id, submission.clone()) + .await + { + Ok(()) => {} + Err(err) => { + tracing::warn!(%err, "failed to durably queue follow-up; using local queue"); + self.apply_failed_server_queue_submission( + thread_id, + fallback_user_message.clone(), + ) + .await; + } + } + Ok(true) + } + AppCommand::RefreshUserMessageQueue => { + if !self.config.features.enabled(Feature::UserMessageQueue) + || self.config.ephemeral + || !app_server.thread_queue_supported(thread_id).await + { + self.finish_server_queue_refresh(thread_id).await; + return Ok(true); + } + match app_server.thread_queue_list(thread_id).await { + Ok(response) => self.apply_server_queue_snapshot(thread_id, response).await, + Err(err) => { + tracing::warn!(%err, "failed to refresh durable user message queue"); + self.finish_server_queue_refresh(thread_id).await; + } + } + 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 17587231a5..6d1a2b48db 100644 --- a/codex-rs/tui/src/app_command.rs +++ b/codex-rs/tui/src/app_command.rs @@ -7,6 +7,7 @@ use codex_app_server_protocol::McpServerElicitationAction; use codex_app_server_protocol::RequestId as AppServerRequestId; use codex_app_server_protocol::ReviewTarget; use codex_app_server_protocol::ToolRequestUserInputResponse; +use codex_app_server_protocol::TurnSubmissionParams; use codex_app_server_protocol::UserInput; use codex_config::types::ApprovalsReviewer; use codex_protocol::approvals::GuardianAssessmentEvent; @@ -21,6 +22,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 { @@ -45,6 +48,12 @@ pub(crate) enum AppCommand { collaboration_mode: Option, personality: Option, }, + QueueUserMessage { + submission: TurnSubmissionParams, + #[serde(skip)] + fallback_user_message: UserMessage, + }, + RefreshUserMessageQueue, OverrideTurnContext { cwd: Option, approval_policy: Option, diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 846970bfc0..7820153be2 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -25,6 +25,8 @@ use codex_app_server_protocol::AuthMode; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ConfigBatchWriteParams; use codex_app_server_protocol::ConfigWriteResponse; +use codex_app_server_protocol::ExperimentalFeatureListParams; +use codex_app_server_protocol::ExperimentalFeatureListResponse; use codex_app_server_protocol::ExternalAgentConfigDetectParams; use codex_app_server_protocol::ExternalAgentConfigDetectResponse; use codex_app_server_protocol::ExternalAgentConfigImportParams; @@ -39,6 +41,7 @@ use codex_app_server_protocol::MemoryResetResponse; use codex_app_server_protocol::Model as ApiModel; use codex_app_server_protocol::ModelListParams; use codex_app_server_protocol::ModelListResponse; +use codex_app_server_protocol::QueuedItemProvenance; use codex_app_server_protocol::RateLimitSnapshot; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ReviewDelivery; @@ -79,6 +82,10 @@ 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::ThreadQueueListParams; +use codex_app_server_protocol::ThreadQueueListResponse; use codex_app_server_protocol::ThreadReadParams; use codex_app_server_protocol::ThreadReadResponse; use codex_app_server_protocol::ThreadResumeParams; @@ -106,6 +113,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::TurnSubmissionParams; use codex_app_server_protocol::UserInput; use codex_otel::TelemetryAuthMode; use codex_protocol::ThreadId; @@ -242,6 +250,71 @@ impl AppServerSession { matches!(&self.client, AppServerClient::InProcess(_)) } + pub(crate) async fn thread_queue_supported(&mut self, thread_id: ThreadId) -> bool { + let request_id = self.next_request_id(); + match self + .client + .request_typed::( + ClientRequest::ExperimentalFeatureList { + request_id, + params: ExperimentalFeatureListParams { + cursor: None, + limit: None, + thread_id: Some(thread_id.to_string()), + }, + }, + ) + .await + { + Ok(response) => response + .data + .iter() + .any(|feature| feature.name == "user_message_queue" && feature.enabled), + Err(err) => { + tracing::debug!(%err, "app-server queue capability unavailable"); + false + } + } + } + + pub(crate) async fn thread_queue_add( + &mut self, + thread_id: ThreadId, + submission: TurnSubmissionParams, + ) -> 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, + provenance: QueuedItemProvenance::User, + }, + }) + .await + .map(|_| ()) + .wrap_err("thread/queue/add failed in TUI") + } + + pub(crate) async fn thread_queue_list( + &mut self, + thread_id: ThreadId, + ) -> Result { + let request_id = self.next_request_id(); + self.client + .request_typed(ClientRequest::ThreadQueueList { + request_id, + params: ThreadQueueListParams { + thread_id: thread_id.to_string(), + cursor: None, + limit: None, + }, + }) + .await + .wrap_err("thread/queue/list failed in TUI") + } + pub(crate) fn codex_home_path( &self, local_codex_home: &AbsolutePathBuf, diff --git a/codex-rs/tui/src/bottom_pane/mod.rs b/codex-rs/tui/src/bottom_pane/mod.rs index f6446f5437..505d8dea84 100644 --- a/codex-rs/tui/src/bottom_pane/mod.rs +++ b/codex-rs/tui/src/bottom_pane/mod.rs @@ -1194,10 +1194,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(); } @@ -2421,6 +2423,7 @@ mod tests { vec!["Queued follow-up question".to_string()], Vec::new(), Vec::new(), + /*has_editable_queued_message*/ true, ); let width = 48; @@ -2452,6 +2455,7 @@ mod tests { vec!["Queued follow-up question".to_string()], Vec::new(), Vec::new(), + /*has_editable_queued_message*/ true, ); pane.hide_status_indicator(); @@ -2484,6 +2488,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..2224adfb01 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: true, edit_binding: Some(key_hint::alt(KeyCode::Up)), interrupt_binding: Some(key_hint::plain(KeyCode::Esc)), } @@ -152,6 +154,7 @@ impl PendingInputPreview { } if !self.queued_messages.is_empty() + && self.has_editable_queued_message && let Some(edit_binding) = self.edit_binding { lines.push( @@ -212,6 +215,23 @@ mod tests { assert_snapshot!("render_one_message", format!("{buf:?}")); } + #[test] + fn render_durable_message_without_edit_hint() { + let mut queue = PendingInputPreview::new(); + queue + .queued_messages + .push("[deploy-hook] run the release check".to_string()); + queue.has_editable_queued_message = false; + let width = 48; + let height = queue.desired_height(width); + let mut buf = Buffer::empty(Rect::new(0, 0, width, height)); + queue.render(Rect::new(0, 0, width, height), &mut buf); + assert_snapshot!( + "render_durable_message_without_edit_hint", + format!("{buf:?}") + ); + } + #[test] fn render_one_message_with_shift_left_binding() { let mut queue = PendingInputPreview::new(); diff --git a/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_durable_message_without_edit_hint.snap b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_durable_message_without_edit_hint.snap new file mode 100644 index 0000000000..de565d96ba --- /dev/null +++ b/codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_durable_message_without_edit_hint.snap @@ -0,0 +1,19 @@ +--- +source: tui/src/bottom_pane/pending_input_preview.rs +assertion_line: 228 +expression: "format!(\"{buf:?}\")" +--- +Buffer { + area: Rect { x: 0, y: 0, width: 48, height: 2 }, + content: [ + "• Queued follow-up inputs ", + " ↳ [deploy-hook] run the release check ", + ], + styles: [ + x: 0, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: DIM, + x: 2, y: 0, fg: Reset, bg: Reset, underline: Reset, modifier: NONE, + x: 0, y: 1, fg: Reset, bg: Reset, underline: Reset, modifier: DIM, + x: 4, y: 1, fg: Reset, bg: Reset, underline: Reset, modifier: DIM | ITALIC, + x: 39, y: 1, fg: Reset, bg: Reset, underline: Reset, modifier: NONE, + ] +} diff --git a/codex-rs/tui/src/chatwidget/ide_context.rs b/codex-rs/tui/src/chatwidget/ide_context.rs index cf89be6b75..31bf719914 100644 --- a/codex-rs/tui/src/chatwidget/ide_context.rs +++ b/codex-rs/tui/src/chatwidget/ide_context.rs @@ -130,3 +130,7 @@ impl ChatWidget { .set_ide_context_active(self.ide_context.is_enabled()); } } + +#[cfg(test)] +#[path = "ide_context_tests.rs"] +mod tests; diff --git a/codex-rs/tui/src/chatwidget/ide_context_tests.rs b/codex-rs/tui/src/chatwidget/ide_context_tests.rs new file mode 100644 index 0000000000..6450a97919 --- /dev/null +++ b/codex-rs/tui/src/chatwidget/ide_context_tests.rs @@ -0,0 +1,21 @@ +use crate::chatwidget::UserMessage; +use crate::chatwidget::tests::helpers::make_chatwidget_manual_with_sender; +use codex_features::Feature; +use codex_protocol::ThreadId; + +#[tokio::test] +async fn ide_context_keeps_follow_ups_in_the_local_queue() { + let (mut chat, _app_event_tx, _rx, mut op_rx) = make_chatwidget_manual_with_sender().await; + chat.thread_id = Some(ThreadId::new()); + chat.set_feature_enabled(Feature::UserMessageQueue, /*enabled*/ true); + chat.ide_context.enable(); + chat.input_queue.user_turn_pending_start = true; + + chat.queue_user_message(UserMessage::from("use fresh IDE context")); + + assert!(op_rx.try_recv().is_err()); + assert_eq!( + chat.queued_user_message_texts(), + vec!["use fresh IDE context"] + ); +} diff --git a/codex-rs/tui/src/chatwidget/input_flow.rs b/codex-rs/tui/src/chatwidget/input_flow.rs index b24a4fde27..83c9258f5f 100644 --- a/codex-rs/tui/src/chatwidget/input_flow.rs +++ b/codex-rs/tui/src/chatwidget/input_flow.rs @@ -5,6 +5,7 @@ //! follow-ups, and restoring draft state across interrupts or thread switches. use super::*; +use codex_app_server_protocol::ThreadQueueListResponse; impl ChatWidget { pub(super) fn handle_composer_input_result( @@ -33,6 +34,12 @@ impl ChatWidget { self.queue_user_message(user_message); return; } + if !self.is_user_turn_pending_or_running() + && self.input_queue.blocks_local_queue_autosend() + { + self.queue_user_message(user_message); + return; + } // Submitted is emitted when user submits. // Reset any reasoning header only when we are actually submitting a turn. self.reasoning_buffer.clear(); @@ -84,7 +91,12 @@ impl ChatWidget { action: QueuedInputAction, pending_pastes: Vec<(String, String)>, ) { - if !self.is_session_configured() || self.is_user_turn_pending_or_running() { + if self.should_use_server_queue(&user_message, action) { + self.submit_user_message_to_queue(user_message); + } else if !self.is_session_configured() + || self.is_user_turn_pending_or_running() + || self.input_queue.blocks_local_queue_autosend() + { self.input_queue .queued_user_messages .push_back(QueuedUserMessage { @@ -101,11 +113,34 @@ impl ChatWidget { } } + fn should_use_server_queue( + &self, + user_message: &UserMessage, + action: QueuedInputAction, + ) -> bool { + self.is_session_configured() + && (self.input_queue.user_turn_pending_start + || self.turn_lifecycle.agent_turn_running + || self.input_queue.blocks_local_queue_autosend()) + && action == QueuedInputAction::Plain + && !user_message.text.starts_with('!') + && self.config.features.enabled(Feature::UserMessageQueue) + && !self.config.ephemeral + && !self.active_side_conversation + && !self.ide_context.is_enabled() + && !self.only_user_shell_commands_running() + && !self.input_queue.has_local_follow_up_messages() + && self.input_queue.pending_steers.is_empty() + } + /// If idle and there are queued inputs, submit exactly one to start the next turn. pub(crate) fn maybe_send_next_queued_input(&mut self) -> bool { 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; } @@ -147,6 +182,43 @@ impl ChatWidget { self.input_queue.user_turn_pending_start || self.bottom_pane.is_task_running() } + pub(crate) fn start_server_queue_refresh(&mut self, thread_id: ThreadId) -> bool { + if self.thread_id != Some(thread_id) + || self.input_queue.user_message_queue.refresh_in_flight + { + return false; + } + self.input_queue.user_message_queue.refresh_in_flight = true; + self.refresh_pending_input_preview(); + true + } + + pub(crate) fn finish_server_queue_refresh(&mut self) { + self.input_queue.user_message_queue.refresh_in_flight = false; + self.refresh_pending_input_preview(); + self.maybe_send_next_queued_input(); + } + + pub(crate) fn requeue_failed_server_submission(&mut self, fallback: UserMessage) { + self.input_queue + .queued_user_messages + .push_front(QueuedUserMessage { + user_message: fallback, + action: QueuedInputAction::Plain, + pending_pastes: Vec::new(), + }); + self.input_queue + .queued_user_message_history_records + .push_front(UserMessageHistoryRecord::UserMessageText); + self.refresh_pending_input_preview(); + } + + pub(crate) fn set_server_queue_snapshot(&mut self, response: ThreadQueueListResponse) { + self.input_queue.user_message_queue.set_snapshot(response); + self.refresh_pending_input_preview(); + self.maybe_send_next_queued_input(); + } + pub(super) fn only_user_shell_commands_running(&self) -> bool { self.turn_lifecycle.agent_turn_running && !self.running_commands.is_empty() @@ -163,6 +235,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..8ff1663d54 100644 --- a/codex-rs/tui/src/chatwidget/input_queue.rs +++ b/codex-rs/tui/src/chatwidget/input_queue.rs @@ -5,21 +5,36 @@ use std::collections::VecDeque; +use codex_app_server_protocol::QueuedItem; +use codex_app_server_protocol::QueuedItemProvenance; +use codex_app_server_protocol::QueuedItemStatus; +use codex_app_server_protocol::ThreadQueueListResponse; + +use super::ChatWidget; use super::PendingSteer; use super::QueuedUserMessage; use super::UserMessage; use super::UserMessageHistoryRecord; use super::user_message_preview_text; +#[derive(Clone, Debug, Default, PartialEq)] +pub(crate) struct UserMessageQueueState { + items: Vec, + has_more: bool, + pub(crate) refresh_in_flight: bool, +} + #[derive(Debug, Default, PartialEq, Eq)] 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)] pub(super) struct InputQueueState { + pub(super) user_message_queue: UserMessageQueueState, /// User inputs queued while a turn is in progress. pub(super) queued_user_messages: VecDeque, /// History records for queued user messages. Slash commands such as `/goal` @@ -44,12 +59,45 @@ pub(super) struct InputQueueState { pub(super) suppress_queue_autosend: bool, } +impl UserMessageQueueState { + pub(crate) fn set_snapshot(&mut self, response: ThreadQueueListResponse) { + let ThreadQueueListResponse { data, next_cursor } = response; + self.items = data; + self.has_more = next_cursor.is_some(); + self.refresh_in_flight = false; + } + + fn clear(&mut self) { + self.items.clear(); + self.has_more = false; + self.refresh_in_flight = false; + } +} + impl InputQueueState { - pub(super) fn has_queued_follow_up_messages(&self) -> bool { + pub(super) fn has_local_follow_up_messages(&self) -> bool { !self.rejected_steers_queue.is_empty() || !self.queued_user_messages.is_empty() } + pub(super) fn has_queued_follow_up_messages(&self) -> bool { + self.has_server_follow_up_messages() || self.has_local_follow_up_messages() + } + + pub(super) fn blocks_local_queue_autosend(&self) -> bool { + self.user_message_queue.refresh_in_flight || self.has_server_follow_up_messages() + } + + fn has_server_follow_up_messages(&self) -> bool { + self.user_message_queue.has_more + || self + .user_message_queue + .items + .iter() + .any(|item| matches!(item.status, QueuedItemStatus::Pending)) + } + pub(super) fn clear(&mut self) { + self.user_message_queue.clear(); self.queued_user_messages.clear(); self.queued_user_message_history_records.clear(); self.user_turn_pending_start = false; @@ -61,15 +109,21 @@ impl InputQueueState { pub(super) fn preview(&self) -> PendingInputPreview { let queued_messages = self - .queued_user_messages + .user_message_queue + .items .iter() - .enumerate() - .map(|(idx, message)| { - user_message_preview_text( - message, - self.queued_user_message_history_records.get(idx), - ) - }) + .map(server_queued_item_preview) + .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,64 +145,35 @@ impl InputQueueState { queued_messages, pending_steers, rejected_steers, + has_editable_queued_message: !self.queued_user_messages.is_empty(), } } } -#[cfg(test)] -mod tests { - use pretty_assertions::assert_eq; - - use super::*; - - #[test] - fn preview_keeps_queue_categories_separate() { - let mut state = InputQueueState::default(); - state - .queued_user_messages - .push_back(UserMessage::from("queued").into()); - state - .rejected_steers_queue - .push_back(UserMessage::from("rejected")); - state.pending_steers.push_back(PendingSteer { - user_message: UserMessage::from("pending"), - history_record: UserMessageHistoryRecord::UserMessageText, - compare_key: crate::chatwidget::user_messages::PendingSteerCompareKey { - message: "pending".to_string(), - image_count: 0, - }, - }); - - assert_eq!( - state.preview(), - PendingInputPreview { - queued_messages: vec!["queued".to_string()], - pending_steers: vec!["pending".to_string()], - rejected_steers: vec!["rejected".to_string()], - } - ); +fn server_queued_item_preview(item: &QueuedItem) -> String { + let display = ChatWidget::user_message_display_from_inputs(&item.submission.input); + let mut parts = Vec::new(); + if !display.message.is_empty() { + parts.push(display.message); } - - #[test] - fn clear_resets_all_input_queues() { - let mut state = InputQueueState::default(); - state - .queued_user_messages - .push_back(UserMessage::from("queued").into()); - state - .rejected_steers_queue - .push_back(UserMessage::from("rejected")); - state.user_turn_pending_start = true; - state.submit_pending_steers_after_interrupt = true; - - state.clear(); - - assert!(state.queued_user_messages.is_empty()); - assert!(state.queued_user_message_history_records.is_empty()); - assert!(!state.user_turn_pending_start); - assert!(state.rejected_steers_queue.is_empty()); - assert!(state.rejected_steer_history_records.is_empty()); - assert!(state.pending_steers.is_empty()); - assert!(!state.submit_pending_steers_after_interrupt); + parts.extend( + std::iter::repeat_n("[image]".to_string(), display.local_images.len()).chain( + std::iter::repeat_n("[image]".to_string(), display.remote_image_urls.len()), + ), + ); + if parts.is_empty() { + parts.push("[queued input]".to_string()); } + let mut preview = parts.join("\n"); + if let QueuedItemProvenance::ExternalEvent { source, .. } = &item.provenance { + preview = format!("[{source}] {preview}"); + } + if let QueuedItemStatus::Failed { error } = &item.status { + preview = format!("[failed: {error}] {preview}"); + } + preview } + +#[cfg(test)] +#[path = "input_queue_tests.rs"] +mod tests; diff --git a/codex-rs/tui/src/chatwidget/input_queue_tests.rs b/codex-rs/tui/src/chatwidget/input_queue_tests.rs new file mode 100644 index 0000000000..539602e131 --- /dev/null +++ b/codex-rs/tui/src/chatwidget/input_queue_tests.rs @@ -0,0 +1,134 @@ +use codex_app_server_protocol::TurnSubmission; +use pretty_assertions::assert_eq; + +use super::*; + +fn queued_item(text: &str, status: QueuedItemStatus) -> QueuedItem { + QueuedItem { + id: text.to_string(), + submission: TurnSubmission { + input: vec![codex_app_server_protocol::UserInput::Text { + text: text.to_string(), + text_elements: Vec::new(), + }], + ..Default::default() + }, + provenance: QueuedItemProvenance::User, + status, + } +} + +#[test] +fn preview_keeps_queue_categories_separate() { + let mut state = InputQueueState::default(); + state + .user_message_queue + .items + .push(queued_item("server queued", QueuedItemStatus::Pending)); + state + .queued_user_messages + .push_back(UserMessage::from("local queued").into()); + state + .rejected_steers_queue + .push_back(UserMessage::from("rejected")); + state.pending_steers.push_back(PendingSteer { + user_message: UserMessage::from("pending"), + history_record: UserMessageHistoryRecord::UserMessageText, + compare_key: crate::chatwidget::user_messages::PendingSteerCompareKey { + message: "pending".to_string(), + image_count: 0, + }, + }); + + assert_eq!( + state.preview(), + PendingInputPreview { + queued_messages: vec!["server queued".to_string(), "local queued".to_string()], + pending_steers: vec!["pending".to_string()], + rejected_steers: vec!["rejected".to_string()], + has_editable_queued_message: true, + } + ); +} + +#[test] +fn snapshot_replaces_the_displayed_server_queue() { + let mut state = InputQueueState::default(); + state + .user_message_queue + .items + .push(queued_item("old", QueuedItemStatus::Pending)); + + state + .user_message_queue + .set_snapshot(ThreadQueueListResponse { + data: vec![queued_item("current", QueuedItemStatus::Pending)], + next_cursor: None, + }); + + assert_eq!(state.preview().queued_messages, vec!["current"]); +} + +#[test] +fn preview_includes_external_provenance_and_failure() { + let mut item = queued_item( + "run the release check", + QueuedItemStatus::Failed { + error: "thread unavailable".to_string(), + }, + ); + item.provenance = QueuedItemProvenance::ExternalEvent { + source: "deploy-hook".to_string(), + metadata: Default::default(), + }; + let mut state = InputQueueState::default(); + state.user_message_queue.items.push(item); + + assert_eq!( + state.preview().queued_messages, + vec!["[failed: thread unavailable] [deploy-hook] run the release check"] + ); + assert!(!state.blocks_local_queue_autosend()); +} + +#[test] +fn server_queue_state_distinguishes_follow_ups_from_refreshes() { + let mut state = InputQueueState::default(); + state + .user_message_queue + .items + .push(queued_item("pending", QueuedItemStatus::Pending)); + assert!(state.blocks_local_queue_autosend()); + assert!(state.has_queued_follow_up_messages()); + + state.user_message_queue.items.clear(); + state.user_message_queue.has_more = true; + assert!(state.blocks_local_queue_autosend()); + assert!(state.has_queued_follow_up_messages()); + + state.user_message_queue.has_more = false; + state.user_message_queue.refresh_in_flight = true; + assert!(state.blocks_local_queue_autosend()); + assert!(!state.has_queued_follow_up_messages()); +} + +#[test] +fn clear_resets_local_and_server_queue_state() { + let mut state = InputQueueState::default(); + state + .user_message_queue + .items + .push(queued_item("server queued", QueuedItemStatus::Pending)); + state.user_message_queue.has_more = true; + state.user_message_queue.refresh_in_flight = true; + state + .queued_user_messages + .push_back(UserMessage::from("local queued").into()); + state.user_turn_pending_start = true; + + state.clear(); + + assert_eq!(state.preview(), PendingInputPreview::default()); + assert!(!state.user_turn_pending_start); + assert!(!state.blocks_local_queue_autosend()); +} diff --git a/codex-rs/tui/src/chatwidget/input_restore.rs b/codex-rs/tui/src/chatwidget/input_restore.rs index bcc4e956a7..d915ed5da2 100644 --- a/codex-rs/tui/src/chatwidget/input_restore.rs +++ b/codex-rs/tui/src/chatwidget/input_restore.rs @@ -51,7 +51,11 @@ impl ChatWidget { return; } if let Some(user_message) = self.initial_user_message.take() { - self.submit_user_message(user_message); + if self.input_queue.blocks_local_queue_autosend() { + self.queue_user_message(user_message); + } else { + self.submit_user_message(user_message); + } } } @@ -351,6 +355,7 @@ impl ChatWidget { }; Some(ThreadInputState { composer: composer.has_content().then_some(composer), + user_message_queue: self.input_queue.user_message_queue.clone(), pending_steers: self .input_queue .pending_steers @@ -387,11 +392,14 @@ impl ChatWidget { pub(crate) fn restore_thread_input_state(&mut self, input_state: Option) { let restored_task_running = input_state.as_ref().is_some_and(|state| state.task_running); if let Some(input_state) = input_state { + let refresh_in_flight = self.input_queue.user_message_queue.refresh_in_flight; self.current_collaboration_mode = input_state.current_collaboration_mode; self.active_collaboration_mask = input_state.active_collaboration_mask; self.turn_lifecycle .restore_running(input_state.agent_turn_running, Instant::now()); self.input_queue.user_turn_pending_start = input_state.user_turn_pending_start; + self.input_queue.user_message_queue = input_state.user_message_queue; + self.input_queue.user_message_queue.refresh_in_flight |= refresh_in_flight; self.update_collaboration_mode_indicator(); self.refresh_model_dependent_surfaces(); self.restore_composer_state(input_state.composer.unwrap_or_default()); diff --git a/codex-rs/tui/src/chatwidget/input_submission.rs b/codex-rs/tui/src/chatwidget/input_submission.rs index da2207dae0..eb0da0b418 100644 --- a/codex-rs/tui/src/chatwidget/input_submission.rs +++ b/codex-rs/tui/src/chatwidget/input_submission.rs @@ -1,6 +1,13 @@ //! User-message and shell-prompt submission behavior for `ChatWidget`. use super::*; +use codex_app_server_protocol::TurnSubmissionParams; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum SubmissionTarget { + Turn, + Queue, +} impl ChatWidget { pub(super) fn user_message_from_submission( @@ -69,6 +76,15 @@ impl ChatWidget { ); } + pub(super) fn submit_user_message_to_queue(&mut self, user_message: UserMessage) { + let _ = self.submit_user_message_with_history_and_shell_escape_policy( + user_message, + UserMessageHistoryRecord::UserMessageText, + ShellEscapePolicy::Disallow, + SubmissionTarget::Queue, + ); + } + pub(super) fn submit_user_message_with_history_record( &mut self, user_message: UserMessage, @@ -78,6 +94,7 @@ impl ChatWidget { user_message, history_record, ShellEscapePolicy::Allow, + SubmissionTarget::Turn, ) .0 } @@ -91,6 +108,7 @@ impl ChatWidget { user_message, UserMessageHistoryRecord::UserMessageText, shell_escape_policy, + SubmissionTarget::Turn, ) .1 } @@ -100,6 +118,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 +156,7 @@ impl ChatWidget { ); return (false, None); } + let fallback_user_message = user_message.clone(); let UserMessage { text, local_images, @@ -145,7 +165,6 @@ impl ChatWidget { mention_bindings, } = user_message; - let render_in_history = !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. @@ -292,6 +311,20 @@ impl ChatWidget { } } + if submission_target == SubmissionTarget::Queue { + let op = AppCommand::QueueUserMessage { + submission: TurnSubmissionParams { + input: items, + ..Default::default() + }, + fallback_user_message, + }; + if !self.submit_op(op.clone()) { + return (false, None); + } + return (true, Some(op)); + } + let effective_mode = self.effective_collaboration_mode(); if effective_mode.model().trim().is_empty() { self.add_error_message( @@ -319,6 +352,7 @@ impl ChatWidget { } else { None }; + let render_in_history = !self.turn_lifecycle.agent_turn_running; let pending_steer = (!render_in_history).then(|| PendingSteer { user_message: UserMessage { text: text.clone(), @@ -358,8 +392,8 @@ impl ChatWidget { self.input_queue.user_turn_pending_start = true; } - // Persist the submitted text to cross-session message history. Mentions are encoded into - // placeholder syntax so recall can reconstruct the mention bindings in a future session. + // Persist the submitted text to cross-session message history. Mentions are encoded + // so recall can reconstruct their bindings in a future session. let encoded_mentions = mention_bindings .iter() .map(|binding| LinkedMention { diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index c453144fb7..b8ca6ea0c4 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -57,6 +57,28 @@ impl ChatWidget { ServerNotification::ThreadSettingsUpdated(notification) => { self.on_thread_settings_updated(notification); } + ServerNotification::ThreadQueueChanged(notification) => { + if self.config.features.enabled(Feature::UserMessageQueue) + && !self.config.ephemeral + && !self.active_side_conversation + { + match ThreadId::from_string(¬ification.thread_id) { + Ok(thread_id) => { + if self.start_server_queue_refresh(thread_id) { + self.app_event_tx.send(AppEvent::SubmitThreadOp { + thread_id, + op: AppCommand::RefreshUserMessageQueue, + }); + } + } + Err(err) => tracing::warn!( + thread_id = notification.thread_id, + error = %err, + "ignoring queue invalidation with invalid thread id" + ), + } + } + } ServerNotification::TurnStarted(notification) => { self.turn_lifecycle.last_turn_id = Some(notification.turn.id); self.last_non_retry_error = None; @@ -191,7 +213,6 @@ impl ChatWidget { ServerNotification::ServerRequestResolved(_) | ServerNotification::AccountUpdated(_) | ServerNotification::AccountRateLimitsUpdated(_) - | ServerNotification::ThreadQueueChanged(_) | ServerNotification::ThreadStarted(_) | ServerNotification::ThreadStatusChanged(_) | ServerNotification::ThreadArchived(_) diff --git a/codex-rs/tui/src/chatwidget/session_flow.rs b/codex-rs/tui/src/chatwidget/session_flow.rs index 004f64c726..2ad69084df 100644 --- a/codex-rs/tui/src/chatwidget/session_flow.rs +++ b/codex-rs/tui/src/chatwidget/session_flow.rs @@ -154,6 +154,7 @@ impl ChatWidget { SessionConfiguredDisplay::Normal, fork_parent_title, ); + self.refresh_user_message_queue_after_session_configured(); } pub(crate) fn handle_thread_session_quiet(&mut self, session: ThreadSessionState) { @@ -163,6 +164,18 @@ impl ChatWidget { SessionConfiguredDisplay::Quiet, /*fork_parent_title*/ None, ); + self.refresh_user_message_queue_after_session_configured(); + } + + fn refresh_user_message_queue_after_session_configured(&mut self) { + if self.config.features.enabled(Feature::UserMessageQueue) + && !self.config.ephemeral + && let Some(thread_id) = self.thread_id + && self.start_server_queue_refresh(thread_id) + && !self.submit_op(AppCommand::RefreshUserMessageQueue) + { + self.finish_server_queue_refresh(); + } } pub(crate) fn handle_side_thread_session(&mut self, session: ThreadSessionState) { diff --git a/codex-rs/tui/src/chatwidget/slash_dispatch.rs b/codex-rs/tui/src/chatwidget/slash_dispatch.rs index 32ab70ed33..96b340ed71 100644 --- a/codex-rs/tui/src/chatwidget/slash_dispatch.rs +++ b/codex-rs/tui/src/chatwidget/slash_dispatch.rs @@ -148,6 +148,20 @@ impl ChatWidget { self.request_redraw(); return; } + if !self.turn_lifecycle.agent_turn_running + && self.input_queue.blocks_local_queue_autosend() + && matches!( + cmd, + SlashCommand::Init | SlashCommand::Compact | SlashCommand::Review + ) + { + self.queue_user_message_with_options( + UserMessage::from(format!("/{}", cmd.command())), + QueuedInputAction::ParseSlash, + Vec::new(), + ); + return; + } match cmd { SlashCommand::Feedback => { @@ -582,6 +596,37 @@ impl ChatWidget { else { return; }; + if !self.turn_lifecycle.agent_turn_running + && self.input_queue.blocks_local_queue_autosend() + && matches!(cmd, SlashCommand::Plan | SlashCommand::Review) + { + let prefix = format!("/{} ", cmd.command()); + let mut user_message = self.prepared_inline_user_message( + prepared_args, + prepared_elements, + Vec::new(), + Vec::new(), + Vec::new(), + SlashCommandDispatchSource::Live, + ); + user_message.text.insert_str(0, &prefix); + user_message.text_elements = user_message + .text_elements + .into_iter() + .map(|element| { + element.map_range(|range| ByteRange { + start: range.start + prefix.len(), + end: range.end + prefix.len(), + }) + }) + .collect(); + self.queue_user_message_with_options( + user_message, + QueuedInputAction::ParseSlash, + Vec::new(), + ); + return; + } self.dispatch_prepared_command_with_args( cmd, PreparedSlashCommandArgs { diff --git a/codex-rs/tui/src/chatwidget/tests.rs b/codex-rs/tui/src/chatwidget/tests.rs index a34d393842..60a8716eb0 100644 --- a/codex-rs/tui/src/chatwidget/tests.rs +++ b/codex-rs/tui/src/chatwidget/tests.rs @@ -252,6 +252,7 @@ mod status_and_layout; mod status_command_tests; mod status_surface_previews; mod terminal_title; +mod user_message_queue; pub(crate) use helpers::make_chatwidget_manual_with_sender; pub(crate) use helpers::set_chatgpt_auth; diff --git a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs index 154add1d67..7920e515b9 100644 --- a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs +++ b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs @@ -1047,11 +1047,13 @@ async fn pending_steer_interrupt_uses_remapped_binding() { } #[tokio::test] -async fn restore_thread_input_state_syncs_sleep_inhibitor_state() { +async fn restore_thread_input_state_preserves_active_runtime_state() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; chat.set_feature_enabled(Feature::PreventIdleSleep, /*enabled*/ true); + chat.input_queue.user_message_queue.refresh_in_flight = true; chat.restore_thread_input_state(Some(ThreadInputState { + user_message_queue: Default::default(), composer: None, pending_steers: VecDeque::new(), pending_steer_history_records: VecDeque::new(), @@ -1070,12 +1072,14 @@ async fn restore_thread_input_state_syncs_sleep_inhibitor_state() { assert!(chat.turn_lifecycle.agent_turn_running); assert!(chat.turn_lifecycle.sleep_inhibitor.is_turn_running()); assert!(chat.bottom_pane.is_task_running()); + assert!(chat.input_queue.user_message_queue.refresh_in_flight); chat.restore_thread_input_state(/*input_state*/ None); assert!(!chat.turn_lifecycle.agent_turn_running); assert!(!chat.turn_lifecycle.sleep_inhibitor.is_turn_running()); assert!(!chat.bottom_pane.is_task_running()); + assert!(!chat.input_queue.user_message_queue.refresh_in_flight); } #[tokio::test] diff --git a/codex-rs/tui/src/chatwidget/tests/review_mode.rs b/codex-rs/tui/src/chatwidget/tests/review_mode.rs index e8fe8ed300..0848e438f6 100644 --- a/codex-rs/tui/src/chatwidget/tests/review_mode.rs +++ b/codex-rs/tui/src/chatwidget/tests/review_mode.rs @@ -374,6 +374,7 @@ async fn restore_thread_input_state_restores_pending_steers_without_downgrading_ queued_user_messages.push_back(UserMessage::from("queued draft").into()); chat.restore_thread_input_state(Some(ThreadInputState { + user_message_queue: Default::default(), composer: None, pending_steers, pending_steer_history_records: VecDeque::new(), diff --git a/codex-rs/tui/src/chatwidget/tests/side.rs b/codex-rs/tui/src/chatwidget/tests/side.rs index 39a0cb9d73..1ef16fc52d 100644 --- a/codex-rs/tui/src/chatwidget/tests/side.rs +++ b/codex-rs/tui/src/chatwidget/tests/side.rs @@ -3,6 +3,30 @@ use pretty_assertions::assert_eq; use ratatui::Terminal; use ratatui::backend::TestBackend; +#[tokio::test] +async fn side_conversation_keeps_follow_ups_out_of_the_durable_queue() { + let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let thread_id = ThreadId::new(); + chat.thread_id = Some(thread_id); + chat.set_feature_enabled(Feature::UserMessageQueue, /*enabled*/ true); + chat.set_side_conversation_active(/*active*/ true); + chat.handle_server_notification( + ServerNotification::ThreadQueueChanged( + codex_app_server_protocol::ThreadQueueChangedNotification { + thread_id: thread_id.to_string(), + }, + ), + /*replay_kind*/ None, + ); + chat.input_queue.user_turn_pending_start = true; + + chat.queue_user_message(UserMessage::from("side follow-up")); + + assert!(rx.try_recv().is_err()); + assert!(op_rx.try_recv().is_err()); + assert_eq!(chat.queued_user_message_texts(), vec!["side follow-up"]); +} + #[tokio::test] async fn forked_thread_history_line_without_name_shows_id_once_snapshot() { let (mut chat, mut rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; diff --git a/codex-rs/tui/src/chatwidget/tests/slash_commands.rs b/codex-rs/tui/src/chatwidget/tests/slash_commands.rs index 37f20ad970..ddff0ecc37 100644 --- a/codex-rs/tui/src/chatwidget/tests/slash_commands.rs +++ b/codex-rs/tui/src/chatwidget/tests/slash_commands.rs @@ -1,5 +1,6 @@ use super::*; use crate::bottom_pane::slash_commands::ServiceTierCommand; +use codex_app_server_protocol::QueuedItem; use pretty_assertions::assert_eq; use serial_test::serial; @@ -217,6 +218,46 @@ async fn queued_slash_review_with_args_dispatches_after_active_turn() { } } +#[tokio::test] +async fn turn_starting_slash_commands_wait_behind_durable_queue() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + chat.set_server_queue_snapshot(codex_app_server_protocol::ThreadQueueListResponse { + data: vec![QueuedItem { + id: "queued-1".to_string(), + submission: codex_app_server_protocol::TurnSubmission::default(), + provenance: codex_app_server_protocol::QueuedItemProvenance::User, + status: codex_app_server_protocol::QueuedItemStatus::Pending, + }], + next_cursor: None, + }); + + chat.bottom_pane + .set_composer_text("/init".to_string(), Vec::new(), Vec::new()); + chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)); + chat.bottom_pane.set_composer_text( + "/review check ordering".to_string(), + Vec::new(), + Vec::new(), + ); + chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)); + chat.bottom_pane + .set_composer_text("/status".to_string(), Vec::new(), Vec::new()); + chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE)); + + assert!(op_rx.try_recv().is_err()); + assert_eq!( + chat.queued_user_message_texts(), + vec!["/init", "/review check ordering"] + ); + assert!( + chat.input_queue + .queued_user_messages + .iter() + .all(|message| message.action == QueuedInputAction::ParseSlash) + ); +} + #[tokio::test] async fn queued_slash_review_with_args_restores_for_edit() { let (mut chat, _rx, _op_rx) = make_chatwidget_manual(/*model_override*/ None).await; diff --git a/codex-rs/tui/src/chatwidget/tests/user_message_queue.rs b/codex-rs/tui/src/chatwidget/tests/user_message_queue.rs new file mode 100644 index 0000000000..78ffeaee48 --- /dev/null +++ b/codex-rs/tui/src/chatwidget/tests/user_message_queue.rs @@ -0,0 +1,100 @@ +use super::*; +use codex_app_server_protocol::QueuedItem; + +#[tokio::test] +async fn plain_follow_up_uses_the_server_queue() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + chat.set_feature_enabled(Feature::UserMessageQueue, /*enabled*/ true); + handle_turn_started(&mut chat, "active-turn"); + + chat.queue_user_message(UserMessage::from("queued follow-up")); + + let Op::QueueUserMessage { + submission, + fallback_user_message, + } = op_rx.try_recv().expect("server queue command") + else { + panic!("expected QueueUserMessage"); + }; + assert_eq!( + submission.input, + vec![UserInput::Text { + text: "queued follow-up".to_string(), + text_elements: Vec::new(), + }] + ); + assert_eq!(fallback_user_message.text, "queued follow-up"); + + chat.set_server_queue_snapshot(codex_app_server_protocol::ThreadQueueListResponse { + data: vec![QueuedItem { + id: "queued-1".to_string(), + submission: codex_app_server_protocol::TurnSubmission::default(), + provenance: codex_app_server_protocol::QueuedItemProvenance::User, + status: codex_app_server_protocol::QueuedItemStatus::Pending, + }], + next_cursor: None, + }); + handle_turn_completed(&mut chat, "active-turn", /*duration_ms*/ None); + chat.queue_user_message(UserMessage::from("another follow-up")); + assert!(matches!(op_rx.try_recv(), Ok(Op::QueueUserMessage { .. }))); +} + +#[tokio::test] +async fn queue_invalidation_refreshes_only_the_active_thread() { + let (mut chat, mut rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + let active_thread_id = ThreadId::new(); + chat.thread_id = Some(active_thread_id); + chat.set_feature_enabled(Feature::UserMessageQueue, /*enabled*/ true); + + chat.handle_server_notification( + ServerNotification::ThreadQueueChanged( + codex_app_server_protocol::ThreadQueueChangedNotification { + thread_id: ThreadId::new().to_string(), + }, + ), + /*replay_kind*/ None, + ); + assert!(rx.try_recv().is_err()); + + chat.handle_server_notification( + ServerNotification::ThreadQueueChanged( + codex_app_server_protocol::ThreadQueueChangedNotification { + thread_id: active_thread_id.to_string(), + }, + ), + /*replay_kind*/ None, + ); + + assert!(op_rx.try_recv().is_err()); + assert!(matches!( + rx.try_recv(), + Ok(AppEvent::SubmitThreadOp { + thread_id, + op: Op::RefreshUserMessageQueue, + }) if thread_id == active_thread_id + )); +} + +#[tokio::test] +async fn queue_refresh_sends_new_input_through_the_server_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.set_feature_enabled(Feature::UserMessageQueue, /*enabled*/ true); + assert!(chat.start_server_queue_refresh(thread_id)); + + chat.queue_user_message(UserMessage::from("wait for refresh")); + + let Op::QueueUserMessage { submission, .. } = op_rx.try_recv().expect("server queue command") + else { + panic!("expected QueueUserMessage"); + }; + assert_eq!( + submission.input, + vec![UserInput::Text { + text: "wait for refresh".to_string(), + text_elements: Vec::new(), + }] + ); +} diff --git a/codex-rs/tui/src/chatwidget/turn_runtime.rs b/codex-rs/tui/src/chatwidget/turn_runtime.rs index badab1c856..fd43b3fde6 100644 --- a/codex-rs/tui/src/chatwidget/turn_runtime.rs +++ b/codex-rs/tui/src/chatwidget/turn_runtime.rs @@ -194,11 +194,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 { + 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 6cfb739da5..633866d3f5 100644 --- a/codex-rs/tui/src/chatwidget/user_messages.rs +++ b/codex-rs/tui/src/chatwidget/user_messages.rs @@ -23,6 +23,7 @@ use codex_protocol::user_input::ByteRange; use codex_protocol::user_input::TextElement; use super::ChatWidget; +use super::input_queue::UserMessageQueueState; #[derive(Debug, Clone, PartialEq)] pub(crate) struct UserMessage { @@ -121,6 +122,7 @@ impl ThreadComposerState { #[derive(Debug, Clone, PartialEq)] pub(crate) struct ThreadInputState { pub(super) composer: Option, + pub(crate) user_message_queue: UserMessageQueueState, pub(super) pending_steers: VecDeque, pub(super) pending_steer_history_records: VecDeque, pub(super) pending_steer_compare_keys: VecDeque, @@ -135,6 +137,14 @@ pub(crate) struct ThreadInputState { pub(super) agent_turn_running: bool, } +impl ThreadInputState { + pub(crate) fn requeue_failed_server_submission(&mut self, fallback: UserMessage) { + self.queued_user_messages.push_front(fallback.into()); + self.queued_user_message_history_records + .push_front(UserMessageHistoryRecord::UserMessageText); + } +} + impl From for UserMessage { fn from(text: String) -> Self { Self {