openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-12 21:20:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: queue TUI follow-ups through app-server

Browse Source
This commit is contained in:
Edward Frazer
2026-06-16 07:02:12 +00:00
parent f371cb05a3
commit e096bc93f0
27 changed files with 978 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app/app_server_events.rs
+14
View File
@@ -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)
codex-rs/tui/src/app/tests.rs
+1
View File
@@ -3,6 +3,7 @@
mod model_catalog;
mod session_summary;
mod startup;
mod user_message_queue;
use super::*;
use crate::app_backtrack::BacktrackSelection;
codex-rs/tui/src/app/tests/user_message_queue.rs
+120
View File
@@ -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;
}
codex-rs/tui/src/app/thread_routing.rs
+86
View File
@@ -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
codex-rs/tui/src/app_command.rs
+9
View File
@@ -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<CollaborationMode>,
personality: Option<Personality>,
},
QueueUserMessage {
submission: TurnSubmissionParams,
#[serde(skip)]
fallback_user_message: UserMessage,
},
RefreshUserMessageQueue,
OverrideTurnContext {
cwd: Option<PathBuf>,
approval_policy: Option<AskForApproval>,
codex-rs/tui/src/app_server_session.rs
+73
View File
@@ -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::<ExperimentalFeatureListResponse>(
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::<ThreadQueueAddResponse>(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<ThreadQueueListResponse> {
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,
codex-rs/tui/src/bottom_pane/mod.rs
+5
View File
@@ -1194,10 +1194,12 @@ impl BottomPane {
queued: Vec<String>,
pending_steers: Vec<String>,
rejected_steers: Vec<String>,
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;
codex-rs/tui/src/bottom_pane/pending_input_preview.rs
+20
View File
@@ -24,6 +24,7 @@ pub(crate) struct PendingInputPreview {
pub pending_steers: Vec<String>,
pub rejected_steers: Vec<String>,
pub queued_messages: Vec<String>,
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<key_hint::KeyBinding>,
@@ -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();
codex-rs/tui/src/bottom_pane/snapshots/codex_tui__bottom_pane__pending_input_preview__tests__render_durable_message_without_edit_hint.snap
+19
View File
@@ -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,
]
}
codex-rs/tui/src/chatwidget/ide_context.rs
+4
View File
@@ -130,3 +130,7 @@ impl ChatWidget {
.set_ide_context_active(self.ide_context.is_enabled());
}
}
#[cfg(test)]
#[path = "ide_context_tests.rs"]
mod tests;
codex-rs/tui/src/chatwidget/ide_context_tests.rs
+21
View File
@@ -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"]
);
}
codex-rs/tui/src/chatwidget/input_flow.rs
+74 -1
View File
@@ -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,
);
}
codex-rs/tui/src/chatwidget/input_queue.rs
+88 -63
View File
@@ -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<QueuedItem>,
has_more: bool,
pub(crate) refresh_in_flight: bool,
}
#[derive(Debug, Default, PartialEq, Eq)]
pub(super) struct PendingInputPreview {
pub(super) queued_messages: Vec<String>,
pub(super) pending_steers: Vec<String>,
pub(super) rejected_steers: Vec<String>,
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<QueuedUserMessage>,
/// 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;
codex-rs/tui/src/chatwidget/input_queue_tests.rs
+134
View File
@@ -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());
}
codex-rs/tui/src/chatwidget/input_restore.rs
+9 -1
View File
@@ -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<ThreadInputState>) {
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());
codex-rs/tui/src/chatwidget/input_submission.rs
+37 -3
View File
@@ -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<AppCommand>) {
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<UserInput> = 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 {
codex-rs/tui/src/chatwidget/protocol.rs
+22 -1
View File
@@ -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(&notification.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(_)
codex-rs/tui/src/chatwidget/session_flow.rs
+13
View File
@@ -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) {
codex-rs/tui/src/chatwidget/slash_dispatch.rs
+45
View File
@@ -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 {
codex-rs/tui/src/chatwidget/tests.rs
+1
View File
@@ -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;
codex-rs/tui/src/chatwidget/tests/composer_submission.rs
+5 -1
View File
@@ -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]
codex-rs/tui/src/chatwidget/tests/review_mode.rs
+1
View File
@@ -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(),
codex-rs/tui/src/chatwidget/tests/side.rs
+24
View File
@@ -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;
codex-rs/tui/src/chatwidget/tests/slash_commands.rs
+41
View File
@@ -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;
codex-rs/tui/src/chatwidget/tests/user_message_queue.rs
+100
View File
@@ -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(),
}]
);
}
codex-rs/tui/src/chatwidget/turn_runtime.rs
+2 -1
View File
@@ -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,
});
codex-rs/tui/src/chatwidget/user_messages.rs
+10
View File
@@ -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<ThreadComposerState>,
pub(crate) user_message_queue: UserMessageQueueState,
pub(super) pending_steers: VecDeque<UserMessage>,
pub(super) pending_steer_history_records: VecDeque<UserMessageHistoryRecord>,
pub(super) pending_steer_compare_keys: VecDeque<PendingSteerCompareKey>,
@@ -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<String> for UserMessage {
fn from(text: String) -> Self {
Self {