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