openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-18 17:18:52 +00:00
Code Issues Packages Projects Releases Wiki Activity

Resume unsent TUI input after reconnecting (#49105)

Browse Source 
## Why

Reconnecting paused all recovered input, including queued messages that had never been sent. Those messages can resume automatically once any uncertain submissions are resolved.

## What changed

- Track unsent and unconfirmed messages separately. Reconcile submitted prompts and steers using exact client message IDs from restored history, buffered events, and later receipts.
- Remove confirmed submissions from the recovered queue and resume unsent input after replay. Keep uncertain submissions paused without resending them, and show a notice identifying the message whose delivery could not be confirmed.
- Preserve pauses for unavailable conversations, pending compact/review requests, and existing input recovery failures.
- Simplify reconnect status text and remove blanket disconnect/reconnect transcript notices.

## Testing

Update reconnect integration tests and snapshots for automatic queue resumption. Add coverage for matching and unrelated receipts, missing history, late confirmations across navigation, pending compaction, and preservation of image-failure recovery pauses.

GitOrigin-RevId: d0b9aa43c4019d472985d12bbce8ba9bbfb9253b
This commit is contained in:
Eric Traut
2026-09-29 00:17:56 +00:00
committed by copyberry
parent 7e049b3eaa
commit 136391a23e
20 changed files with 355 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app/event_dispatch.rs
+1 -1
View File
@@ -574,7 +574,7 @@ impl App {
{
Ok(()) => {
// Keep local input without replacing the fork's running state.
self.chat_widget.restore_reconnected_input(retained_input);
self.chat_widget.restore_reconnected_input(retained_input, &[]);
if let Some(err) = name_error {
self.chat_widget.add_error_message(err);
}
codex-rs/tui/src/app/reconnect.rs
+11 -11
View File
@@ -1,6 +1,6 @@
//! TUI-only out-of-process reconnection. Each attempt initializes a fresh client and rejoins existing
//! threads using ordinary resume/history semantics; no user operation is retried.
//! Offline input and old async completions are quarantined.
//! threads using ordinary resume/history semantics. Confirmed submissions are reconciled before
//! unsent queues resume; uncertain submissions and old async completions remain quarantined.
use super::*;
use crate::app_server_session::ResumeModelSettings;
@@ -118,7 +118,7 @@ pub(super) async fn reconnect(
}
// Transport errors can contain endpoint credentials. Do not render or log them.
}
color_eyre::eyre::bail!("app-server session could not be restored")
color_eyre::eyre::bail!("Server connection could not be restored")
}
impl App {
@@ -261,7 +261,7 @@ impl App {
let displayed = self.current_displayed_thread_id();
let mut input = self.chat_widget.capture_thread_input_state();
if let Some(input) = input.as_mut() {
input.recovered_queue = true;
input.reconnect_pending = true;
}
self.store_active_thread_receiver().await;
// Old request handles stay attached to the dead connection. Rotating the event channel
@@ -347,7 +347,7 @@ impl App {
self.agents_overview.request_id = None;
self.agents_overview.refresh_pending = false;
for input in self.agents_overview.input_states.values_mut() {
input.recovered_queue = true;
input.reconnect_pending = true;
}
self.pending_startup_thread_start = false;
// Move cached UI state into fresh channels. Old producers retain the old sender/store,
@@ -370,7 +370,7 @@ impl App {
.buffer
.retain(|event| matches!(event, ThreadBufferedEvent::Notification(_)));
if let Some(input) = store.input_state.as_mut() {
input.recovered_queue = true;
input.reconnect_pending = true;
}
*replacement.store.lock().await = store;
*channel = replacement;
@@ -378,7 +378,7 @@ impl App {
if let Some(id) = self.current_displayed_thread_id()
&& let Some(mut input) = input.clone()
{
input.recovered_queue = true;
input.reconnect_pending = true;
self.agents_overview.input_states.insert(id, input);
}
if let Some(mut started) = thread {
@@ -442,7 +442,7 @@ impl App {
/*initial_user_message*/ None,
);
self.replace_chat_widget(ChatWidget::new_with_app_event(init));
self.chat_widget.restore_reconnected_input(input);
self.chat_widget.restore_reconnected_input(input, &[]);
}
// Discover tasks whose notifications were missed, without clearing retained rows.
// A hidden overview performs this discovery when it is next opened.
@@ -485,9 +485,9 @@ impl App {
);
}
self.feedback_audience = bootstrap.feedback_audience;
self.chat_widget.add_info_message(
"Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused.".into(), /*hint*/ None,
);
if displayed.is_some_and(|id| !self.thread_unavailable(id)) {
self.chat_widget.maybe_send_next_queued_input();
}
let connected_notice_key = crate::status::remote_connection::server_version_notice_key(
&self.app_server_target,
app_server.server_codex_home(),
codex-rs/tui/src/app/tests.rs
+38 -2
View File
@@ -1432,6 +1432,15 @@ async fn replay_only_thread_keeps_restored_queue_visible() {
let (mut app, _app_event_rx, _op_rx) = make_test_app_with_channels().await;
let thread_id = ThreadId::new();
let session = test_thread_session(thread_id, test_path_buf("/tmp/project"));
app.thread_event_channels.insert(
thread_id,
ThreadEventChannel::new_with_session(
THREAD_EVENT_CHANNEL_CAPACITY,
session.clone(),
Vec::new(),
),
);
app.activate_thread_channel(thread_id).await;
app.chat_widget.handle_thread_session(session.clone());
app.chat_widget.handle_server_notification(
turn_started_notification(thread_id, "turn-1"),
@@ -1445,10 +1454,11 @@ async fn replay_only_thread_keeps_restored_queue_visible() {
.apply_external_edit("queued follow-up".to_string());
app.chat_widget
.handle_key_event(KeyEvent::new(KeyCode::Tab, KeyModifiers::NONE));
let input_state = app
let mut input_state = app
.chat_widget
.capture_thread_input_state()
.expect("expected queued follow-up state");
input_state.reconnect_pending = true;
let (chat_widget, _app_event_tx, _rx, mut new_op_rx) =
make_chatwidget_manual_with_sender().await;
@@ -1620,14 +1630,25 @@ async fn replay_thread_snapshot_does_not_submit_queue_before_replay_catches_up()
agent_message_delta_notification(thread_id, "turn-1", "agent-1", "streaming"),
/*replay_kind*/ None,
);
app.chat_widget
.apply_external_edit("accepted steer".to_string());
app.chat_widget
.handle_key_event(KeyEvent::from(KeyCode::Enter));
app.chat_widget
.apply_external_edit("queued follow-up".to_string());
app.chat_widget
.handle_key_event(KeyEvent::new(KeyCode::Tab, KeyModifiers::NONE));
let input_state = app
let mut input_state = app
.chat_widget
.capture_thread_input_state()
.expect("expected queued follow-up state");
let client_id = input_state
.pending_steers
.front()
.unwrap()
.client_id
.clone();
input_state.reconnect_pending = true;
let (chat_widget, _app_event_tx, _rx, mut new_op_rx) =
make_chatwidget_manual_with_sender().await;
@@ -1641,6 +1662,21 @@ async fn replay_thread_snapshot_does_not_submit_queue_before_replay_catches_up()
session: None,
turns: Vec::new(),
events: vec![
ThreadBufferedEvent::Notification(Box::new(ServerNotification::ItemCompleted(
codex_app_server_protocol::ItemCompletedNotification {
thread_id: thread_id.to_string(),
turn_id: "turn-1".into(),
completed_at_ms: 0,
item: ThreadItem::UserMessage {
id: "accepted".into(),
client_id: Some(client_id),
content: vec![UserInput::Text {
text: "accepted steer".into(),
text_elements: Vec::new(),
}],
},
},
))),
ThreadBufferedEvent::Notification(Box::new(turn_completed_notification(
thread_id,
"turn-0",
codex-rs/tui/src/app/tests/active_reconnect_tests.rs
+31 -32
View File
@@ -12,7 +12,7 @@ use tokio::net::TcpListener;
use super::disconnect::serve_reconnect_requests;
#[tokio::test]
async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() -> Result<()> {
async fn reconnect_restores_history_permissions_and_resumes_unsent_input() -> Result<()> {
for (recovered_queue, edit_offline, resume_error_code, deferred_notice, notice_enabled) in [
(true, false, -32603, false, false),
(true, false, -32603, false, true),
@@ -143,9 +143,9 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() ->
"thread/list" | "thread/loaded/list" => Some(json!({"result": {"data": [], "nextCursor": null}})),
"thread/goal/get" => Some(json!({"result": {"goal": null}})),
"turn/start" => {
assert!(!recovered_queue);
assert!(!edit_offline);
let params = request.params.as_ref().unwrap();
assert_eq!(params["input"][0]["text"], "fresh follow-up");
assert_eq!(params["input"][0]["text"], if recovered_queue { "old queued input" } else { "fresh follow-up" });
assert_eq!(params["approvalPolicy"], if pending_profile { "never" } else { "on-request" });
assert_eq!(params["sandboxPolicy"]["type"], if pending_profile { json!(null) } else { json!("readOnly") });
assert_eq!(params["collaborationMode"], json!(expected_submitted_mode));
@@ -379,42 +379,41 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() ->
let notices = history
.lines()
.filter(|line| {
line.contains("Reconnected.") || line.contains("background Codex service")
line.contains("Couldn't confirm whether")
|| line.contains("background Codex service")
})
.collect::<Vec<_>>()
.join("\n");
if deferred_notice {
assert_snapshot!(notices, @r###"
• Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused.
⚠ A background Codex service is running v2.0.0, older than your Codex CLI
"###);
} else if edit_offline {
assert!(notices.contains("unacknowledged prompt"));
} else {
insta::allow_duplicates! {
assert_snapshot!(notices, @"• Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused.");
}
assert_eq!(notices, "");
}
assert_eq!(history.matches("only once").count(), 1);
if recovered_queue {
if edit_offline {
assert!(app.chat_widget.has_queued_follow_up_messages());
assert!(!app.chat_widget.maybe_send_next_queued_input());
if edit_offline {
app.handle_tui_event(
&mut tui,
&mut session,
TuiEvent::Key(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT)),
)
.await?;
assert_eq!(
app.chat_widget.composer_text_with_pending(),
"unacknowledged prompt"
);
assert!(ops.try_recv().is_err());
} else {
assert_snapshot!(
"restored_conversation",
render_bottom_popup(&app.chat_widget, /*width*/ 80)
);
}
app.handle_tui_event(
&mut tui,
&mut session,
TuiEvent::Key(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT)),
)
.await?;
assert_eq!(
app.chat_widget.composer_text_with_pending(),
"unacknowledged prompt"
);
assert!(ops.try_recv().is_err());
} else if recovered_queue {
assert!(!app.chat_widget.has_queued_follow_up_messages());
assert_snapshot!(
"restored_conversation",
render_bottom_popup(&app.chat_widget, /*width*/ 80)
);
} else {
app.chat_widget
.handle_key_event(KeyEvent::new(KeyCode::Char('u'), KeyModifiers::CONTROL));
@@ -426,6 +425,9 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() ->
app.handle_event(&mut tui, &mut session, event).await?;
}
}
while let Ok(event) = events.try_recv() {
app.handle_event(&mut tui, &mut session, event).await?;
}
session.shutdown().await?;
let methods = server.await??;
assert_eq!(
@@ -440,7 +442,7 @@ async fn reconnect_restores_history_permissions_and_keeps_old_input_paused() ->
.iter()
.filter(|method| *method == "turn/start")
.count(),
usize::from(!recovered_queue)
usize::from(!edit_offline)
);
}
Ok(())
@@ -742,13 +744,10 @@ pub(super) async fn drain_history(
events: &mut mpsc::UnboundedReceiver<AppEvent>,
) -> Result<String> {
while let Ok(event) = events.try_recv() {
assert!(!matches!(
event,
AppEvent::CodexOp(AppCommand::UserTurn { .. })
));
if matches!(
event,
AppEvent::InsertHistoryCell(_)
| AppEvent::CodexOp(AppCommand::UserTurn { .. })
| AppEvent::BeginThreadSwitchHistoryReplayBuffer
| AppEvent::EndInitialHistoryReplayBuffer
) {
codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__restored_conversation.snap
-4
View File
@@ -2,10 +2,6 @@
source: tui/src/app/tests/active_reconnect_tests.rs
expression: "render_bottom_popup(&app.chat_widget, 80)"
---
• Queued follow-up inputs
↳ old queued input
⌥+↑ edit last queued message
› kept draft
gpt-test default Plan mode (shift+tab to cycle)
codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__active_reconnect__unavailable_conversation.snap
-1
View File
@@ -3,7 +3,6 @@ source: tui/src/app/tests/active_reconnect_tests.rs
expression: "format!(\"{history}\\n{}\",\nrender_bottom_popup(&app.chat_widget,\n80)).replace(&test_path_buf(\"/tmp/project\").display().to_string(),\n\"/tmp/project\")"
---
• This conversation is unavailable. Its cached transcript and draft remain here; input is paused. Open the agent picker or return to the parent to continue.
• Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused.
⚠ A background Codex service is running v2.0.0, different from your Codex CLI
v0.0.0.
› unavailable draft
codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__disconnect__offline_draft.snap
+4 -2
View File
@@ -2,8 +2,10 @@
source: tui/src/app/tests/disconnect_tests.rs
expression: "format!(\"{}\\n{}\",\nlines_to_single_string(&app.transcript_cells.last().unwrap().display_lines(80)),\nrender_bottom_popup(&app.chat_widget, 80))"
---
■ Connection lost. Attempting to reconnect…
• Reconnecting to app-server… (0s)
› only once
• Reconnecting to server… (0s)
• Queued follow-up inputs
↳ queued before disconnect
codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__navigation_reconnect__reconnected_unavailable_conversation.snap
-1
View File
@@ -5,7 +5,6 @@ expression: "format!(\"{content}\\n{}\", render_bottom_popup(&app.chat_widget, 8
Cached previous conversation
• This conversation is unavailable. Its cached transcript and draft remain here; input is paused. Open the agent picker or return to the parent to continue.
• Reconnected. No input was resent. Review uncertain submissions before retrying; recovered queues remain paused.
• The rename target disappeared. Unsubmitted title: Keep this task draft!
› latest draft
codex-rs/tui/src/app/thread_routing.rs
+33 -3
View File
@@ -1878,10 +1878,38 @@ impl App {
self.chat_widget
.remember_realtime_delegated_reasoning_turn(turn_id);
}
let confirmed_message_ids = snapshot
.turns
.iter()
.flat_map(|turn| &turn.items)
.chain(snapshot.events.iter().filter_map(|event| {
let ThreadBufferedEvent::Notification(notification) = event else {
return None;
};
if let ServerNotification::ItemStarted(notification) = notification.as_ref() {
Some(&notification.item)
} else if let ServerNotification::ItemCompleted(notification) =
notification.as_ref()
{
Some(&notification.item)
} else {
None
}
}))
.filter_map(|item| {
if let ThreadItem::UserMessage { client_id, .. } = item {
client_id.clone()
} else {
None
}
})
.collect::<Vec<_>>();
let recovered_input = snapshot
.input_state
.as_ref()
.is_some_and(|input| input.recovered_queue)
.is_some_and(|input| {
input.recovered_queue || input.reconnect_pending || input.has_unconfirmed_messages()
})
.then(|| snapshot.input_state.take())
.flatten();
self.chat_widget.restore_thread_input_state(
@@ -1908,10 +1936,12 @@ impl App {
self.app_event_tx
.send(AppEvent::EndInitialHistoryReplayBuffer);
}
if recovered_input.is_some() {
let recovered = recovered_input.is_some();
if recovered {
let mode = has_resumed_collaboration_mode
.then(|| self.chat_widget.effective_collaboration_mode());
self.chat_widget.restore_reconnected_input(recovered_input);
self.chat_widget
.restore_reconnected_input(recovered_input, &confirmed_message_ids);
if let Some(mode) = mode {
self.chat_widget.set_effective_collaboration_mode(mode);
}
codex-rs/tui/src/chatwidget.rs
+9
View File
@@ -402,6 +402,7 @@ use self::turn_lifecycle::TurnLifecycleState;
mod usage;
mod user_messages;
mod working_directory;
use self::user_messages::MessageDelivery;
use self::user_messages::PendingSteer;
#[cfg(test)]
use self::user_messages::PendingSteerCompareKey;
@@ -1350,6 +1351,14 @@ impl ChatWidget {
{
self.on_user_message_display(display);
}
if let Some(client_id) = client_id
&& self.input_queue.queued_user_messages.iter().any(|message| {
matches!(&message.delivery, MessageDelivery::Unconfirmed(Some(id)) if id == client_id)
})
{
self.reconcile_recovered_messages(&[client_id.to_string()]);
self.maybe_send_next_queued_input();
}
}
fn on_user_message_display(&mut self, display: UserMessageDisplay) {
codex-rs/tui/src/chatwidget/input_flow.rs
+2
View File
@@ -206,6 +206,7 @@ impl ChatWidget {
.push_back(QueuedUserMessage {
user_message,
action,
delivery: MessageDelivery::Unsent,
pending_pastes,
source,
});
@@ -238,6 +239,7 @@ impl ChatWidget {
|| self.input_queue.suppress_queue_autosend
|| self.input_queue.rate_limit_recovery_pending
|| self.input_queue.recovered_queue
|| self.input_queue.has_unconfirmed_messages()
{
return false;
}
codex-rs/tui/src/chatwidget/input_queue.rs
+10
View File
@@ -5,6 +5,7 @@
use std::collections::VecDeque;
use super::MessageDelivery;
use super::PendingSteer;
use super::QueuedUserMessage;
use super::UserMessage;
@@ -32,6 +33,7 @@ pub(super) struct InputQueueState {
pub(super) queued_user_message_history_records: VecDeque<UserMessageHistoryRecord>,
/// A user turn has been submitted to core, but `TurnStarted` has not arrived yet.
pub(super) user_turn_pending_start: bool,
pub(super) pending_user_message_client_id: Option<String>,
/// User messages that tried to steer a non-regular turn and must be retried first.
pub(super) rejected_steers_queue: VecDeque<UserMessage>,
/// The origin of each rejected steer, kept in lockstep with its message.
@@ -49,10 +51,17 @@ pub(super) struct InputQueueState {
pub(super) suppress_queue_autosend: bool,
/// Hold submissions while a usage failure or backend-directed model fallback is resolved.
pub(super) rate_limit_recovery_pending: bool,
/// Pause for user recovery (for example an image or permission failure), independent of delivery.
pub(super) recovered_queue: bool,
}
impl InputQueueState {
pub(super) fn has_unconfirmed_messages(&self) -> bool {
self.queued_user_messages
.iter()
.any(|message| matches!(message.delivery, MessageDelivery::Unconfirmed(_)))
}
pub(super) fn has_queued_follow_up_messages(&self) -> bool {
!self.rejected_steers_queue.is_empty() || !self.queued_user_messages.is_empty()
}
@@ -63,6 +72,7 @@ impl InputQueueState {
self.queued_user_messages.clear();
self.queued_user_message_history_records.clear();
self.user_turn_pending_start = false;
self.pending_user_message_client_id = None;
self.rejected_steers_queue.clear();
self.rejected_steer_sources.clear();
self.rejected_steer_history_records.clear();
codex-rs/tui/src/chatwidget/input_restore.rs
+9
View File
@@ -584,7 +584,9 @@ impl ChatWidget {
.queued_user_message_history_records
.clone(),
recovered_queue: self.input_queue.recovered_queue,
reconnect_pending: false,
user_turn_pending_start: self.input_queue.user_turn_pending_start,
pending_user_message_client_id: self.input_queue.pending_user_message_client_id.clone(),
submit_pending_steers_after_interrupt: self
.input_queue
.submit_pending_steers_after_interrupt,
@@ -618,6 +620,8 @@ impl ChatWidget {
);
self.input_queue.user_turn_pending_start =
preserve_in_flight_turn && input_state.user_turn_pending_start;
self.input_queue.pending_user_message_client_id =
input_state.pending_user_message_client_id;
self.input_queue.submit_pending_steers_after_interrupt =
preserve_in_flight_turn && input_state.submit_pending_steers_after_interrupt;
self.update_collaboration_mode_indicator();
@@ -634,6 +638,11 @@ impl ChatWidget {
for pending in pending_steers.into_iter().rev() {
queued_user_messages.push_front(QueuedUserMessage {
source: pending.source,
delivery: if input_state.reconnect_pending {
MessageDelivery::Unconfirmed(Some(pending.client_id))
} else {
MessageDelivery::Unsent
},
..QueuedUserMessage::from(pending.user_message)
});
queued_user_message_history_records.push_front(pending.history_record);
codex-rs/tui/src/chatwidget/input_submission.rs
+1
View File
@@ -489,6 +489,7 @@ impl ChatWidget {
self.note_realtime_typed_input(&submitted_message.text);
if render_in_history {
self.input_queue.user_turn_pending_start = true;
self.input_queue.pending_user_message_client_id = Some(client_user_message_id.clone());
}
// Persist the submitted text to cross-session message history. Mentions are encoded into
codex-rs/tui/src/chatwidget/reconnect.rs
+61 -8
View File
@@ -1,4 +1,4 @@
//! Disconnects preserve editable input but never automatically retry queued submissions.
//! Disconnects preserve input and reconcile confirmed submissions before resuming unsent queues.
//! Unavailable threads retain uncertain prompts in recovered queues before clearing activity.
use super::realtime::RealtimeConversationPhase;
@@ -22,7 +22,6 @@ impl ChatWidget {
if let Some(questions) = &mut self.bottom_pane.questions {
questions.delivery_enabled = false;
}
self.input_queue.recovered_queue = true;
self.input_queue.suppress_queue_autosend = true;
self.set_initial_user_message_submit_suppressed(/*suppressed*/ true);
if let Some(message) = self.initial_user_message.take() {
@@ -31,28 +30,42 @@ impl ChatWidget {
self.bottom_pane.ensure_status_indicator();
self.bottom_pane
.set_interrupt_hint_visible(/*visible*/ false);
self.set_status_header("Reconnecting to app-server…".to_string());
self.set_status_header("Reconnecting to server…".to_string());
self.set_footer_hint_override(Some(vec![("ctrl+c".into(), "quit".into())]));
self.add_error_message("Connection lost. Attempting to reconnect…".into());
}
/// Restore local input only after replay, which can otherwise move interrupted queues into the draft.
pub(crate) fn restore_reconnected_input(&mut self, input: Option<ThreadInputState>) {
pub(crate) fn restore_reconnected_input(
&mut self,
input: Option<ThreadInputState>,
confirmed_message_ids: &[String],
) {
let running = self.turn_lifecycle.agent_turn_running;
let reconnect_pending = input.as_ref().is_some_and(|input| input.reconnect_pending);
if let Some(mut input) = input {
// Its acceptance is unknown. Keep a local copy for manual recovery without
// comparing against partial history or automatically submitting it again.
// Navigation can continue an unresolved recovery while new steers are in flight.
input.reconnect_pending |= input.has_unconfirmed_messages();
// Pending compact/review requests have no message receipt to reconcile.
input.recovered_queue |=
input.user_turn_pending_start && input.safety_buffering_prompt.is_none();
// Only an exact submission ID confirms receipt; missing or legacy history
// must never cause us to retry an already submitted message automatically.
if input.user_turn_pending_start
&& let Some(prompt) = input.safety_buffering_prompt.take()
{
input.queued_user_messages.push_front(QueuedUserMessage {
source: input.safety_buffering_source,
delivery: if input.reconnect_pending {
MessageDelivery::Unconfirmed(input.pending_user_message_client_id.take())
} else {
MessageDelivery::Unsent
},
..QueuedUserMessage::from(prompt)
});
input
.queued_user_message_history_records
.push_front(UserMessageHistoryRecord::UserMessageText);
input.recovered_queue = true;
input.recovered_queue |= !input.reconnect_pending;
}
input.current_collaboration_mode = self.current_collaboration_mode.clone();
// Keep the local selection for older servers. Replay reapplies a supplied mode.
@@ -67,10 +80,47 @@ impl ChatWidget {
},
);
}
if reconnect_pending || self.input_queue.has_unconfirmed_messages() {
self.reconcile_recovered_messages(confirmed_message_ids);
}
if reconnect_pending
&& let Some(message) = self
.input_queue
.queued_user_messages
.iter()
.find(|message| matches!(message.delivery, MessageDelivery::Unconfirmed(_)))
{
let preview = user_message_preview_text(message, /*history_record*/ None);
let preview = if preview.trim().is_empty() {
"message with attachments".into()
} else {
crate::text_formatting::truncate_text(&preview, /*max_graphemes*/ 80)
};
let notice =
format!("Couldn't confirm whether “{preview}” was sent. It hasn't been resent.");
self.add_info_message(notice, /*hint*/ None);
}
self.turn_lifecycle.restore_running(running, Instant::now());
self.update_task_running_state();
}
pub(super) fn reconcile_recovered_messages(&mut self, confirmed_message_ids: &[String]) {
let mut index = 0;
self.input_queue.queued_user_messages.retain(|message| {
let confirmed = matches!(&message.delivery,
MessageDelivery::Unconfirmed(Some(id)) if confirmed_message_ids.contains(id));
if confirmed {
self.input_queue
.queued_user_message_history_records
.remove(index);
} else {
index += 1;
}
!confirmed
});
self.refresh_pending_input_preview();
}
pub(crate) fn pause_unavailable_thread(&mut self) {
if let Some(questions) = &mut self.bottom_pane.questions {
questions.delivery_enabled = false;
@@ -87,6 +137,9 @@ impl ChatWidget {
.queued_user_messages
.push_front(QueuedUserMessage {
source: self.safety_buffering_source,
delivery: MessageDelivery::Unconfirmed(
self.input_queue.pending_user_message_client_id.clone(),
),
..QueuedUserMessage::from(prompt)
});
self.input_queue
codex-rs/tui/src/chatwidget/startup_submission.rs
+1
View File
@@ -54,6 +54,7 @@ impl ChatWidget {
|| self.input_queue.suppress_queue_autosend
|| self.input_queue.rate_limit_recovery_pending
|| self.input_queue.recovered_queue
|| self.input_queue.has_unconfirmed_messages()
{
return;
}
codex-rs/tui/src/chatwidget/tests/composer_submission.rs
+120 -33
View File
@@ -389,6 +389,7 @@ async fn parent_owned_thread_preserves_queued_input_before_draining() {
let queued_message = QueuedUserMessage {
user_message: UserMessage::from("keep this queued prompt"),
action: QueuedInputAction::Plain,
delivery: MessageDelivery::Unsent,
pending_pastes: vec![("[Image 1]".to_string(), "pasted contents".to_string())],
source: UserMessageSource::Prompt,
};
@@ -1826,7 +1827,9 @@ async fn restore_thread_input_state_applies_running_state_policy() {
queued_user_messages: VecDeque::from([UserMessage::from("already queued").into()]),
queued_user_message_history_records: VecDeque::from([queued_history.clone()]),
recovered_queue: false,
reconnect_pending: false,
user_turn_pending_start: true,
pending_user_message_client_id: None,
submit_pending_steers_after_interrupt: true,
current_collaboration_mode: chat.current_collaboration_mode.clone(),
active_collaboration_mask: chat.active_collaboration_mask.clone(),
@@ -1867,7 +1870,8 @@ async fn restore_thread_input_state_applies_running_state_policy() {
);
assert!(!chat.has_queued_follow_up_messages());
// Editing the last queued draft must not release the uncertain steer for replay.
assert!(chat.capture_thread_input_state().unwrap().recovered_queue);
assert_eq!(chat.input_queue.pending_steers.len(), 1);
assert!(chat.input_queue.suppress_queue_autosend);
chat.handle_restricted_key(
KeyEvent::new(KeyCode::Up, KeyModifiers::ALT),
RestrictedInputMode::Disconnected,
@@ -2359,46 +2363,106 @@ async fn interrupt_prepends_queued_messages_before_existing_composer_text() {
}
#[tokio::test]
async fn reconnect_holds_only_recovered_input_until_manually_edited() {
for (recovered, pending_start) in [
(None, false),
(Some("review this old input"), false),
(Some("unacknowledged prompt"), true),
] {
async fn reconnect_resumes_unsent_input_and_reconciles_confirmed_submissions() {
for pending_start in [false, true] {
let (mut chat, _rx, _ops) = make_chatwidget_manual(/*model_override*/ None).await;
if let Some(text) = recovered {
if pending_start {
chat.input_queue.user_turn_pending_start = true;
chat.safety_buffering_prompt = Some(UserMessage::from(text));
} else {
chat.input_queue
.queued_user_messages
.push_back(UserMessage::from(text).into());
}
}
chat.pause_for_disconnect();
let input = chat.capture_thread_input_state();
let (mut chat, _rx, mut ops) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
chat.restore_reconnected_input(input);
chat.set_queue_autosend_suppressed(/*suppressed*/ false);
if let Some(text) = recovered {
assert!(!chat.maybe_send_next_queued_input());
chat.handle_key_event(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT));
assert_eq!(chat.bottom_pane.composer_text(), text);
assert_no_submit_op(&mut ops);
chat.handle_key_event(KeyEvent::from(KeyCode::Enter));
assert_matches!(next_submit_op(&mut ops), Op::UserTurn { .. });
chat.input_queue.user_turn_pending_start = false;
if !pending_start {
handle_turn_started(&mut chat, "turn-1");
}
chat.submit_user_message("first message".into());
let client_id = if pending_start {
chat.input_queue
.pending_user_message_client_id
.clone()
.unwrap()
} else {
chat.input_queue.pending_steers[0].client_id.clone()
};
chat.input_queue
.queued_user_messages
.push_back(UserMessage::from("new follow-up").into());
assert!(chat.maybe_send_next_queued_input());
assert_matches!(next_submit_op(&mut ops), Op::UserTurn { .. });
.push_back(UserMessage::from("follow-up").into());
chat.pause_for_disconnect();
let mut input = chat.capture_thread_input_state();
// Missing history and unrelated receipts never authorize resubmitting uncertain input.
for ids in [
vec![],
vec!["different-submission".into()],
vec![client_id.clone()],
] {
let confirmed = ids.contains(&client_id);
input.as_mut().unwrap().reconnect_pending = true;
let (mut restored, mut rx, mut ops) =
make_chatwidget_manual(/*model_override*/ None).await;
restored.thread_id = Some(ThreadId::new());
if confirmed && pending_start {
// A fresh steer submitted during recovery must remain uncertain after navigation.
let steer = pending_steer("later steer");
let later_id = steer.client_id.clone();
input.as_mut().unwrap().pending_steers.push_back(steer);
input.as_mut().unwrap().reconnect_pending = false;
restored.restore_reconnected_input(input, &[]);
for (text, id) in [("first message", &client_id), ("later steer", &later_id)] {
assert_no_submit_op(&mut ops);
restored.on_committed_user_message(
&[UserInput::Text {
text: text.into(),
text_elements: Vec::new(),
}],
Some(id),
/*from_replay*/ false,
"turn-1",
);
}
} else {
restored.restore_reconnected_input(input, &ids);
assert_eq!(restored.maybe_send_next_queued_input(), confirmed);
}
if confirmed {
assert_matches!(next_submit_op(&mut ops), Op::UserTurn { items, .. }
if items == vec![UserInput::Text {
text: "follow-up".into(),
text_elements: Vec::new(),
}]);
assert!(!restored.has_queued_follow_up_messages());
} else {
assert_eq!(
restored.queued_user_message_texts(),
vec!["first message", "follow-up"]
);
assert_no_submit_op(&mut ops);
let notice = drain_insert_history(&mut rx)
.into_iter()
.flatten()
.map(|line| line.to_string())
.collect::<Vec<_>>()
.join("\n");
insta::allow_duplicates! {
assert_snapshot!(notice, @"• Couldn't confirm whether “first message” was sent. It hasn't been resent.");
}
}
input = restored.capture_thread_input_state();
}
}
}
#[tokio::test]
async fn reconnect_keeps_queue_paused_after_pending_compact() {
let (mut chat, _rx, mut ops) = make_chatwidget_manual(/*model_override*/ None).await;
chat.thread_id = Some(ThreadId::new());
chat.prepare_local_op_submission(&AppCommand::Compact);
chat.queue_user_message("follow-up".into());
chat.pause_for_disconnect();
let mut input = chat.capture_thread_input_state();
input.as_mut().unwrap().reconnect_pending = true;
chat.restore_reconnected_input(input, &[]);
chat.set_queue_autosend_suppressed(/*suppressed*/ false);
assert!(!chat.maybe_send_next_queued_input());
assert_eq!(chat.queued_user_message_texts(), vec!["follow-up"]);
assert_no_submit_op(&mut ops);
}
#[tokio::test(flavor = "multi_thread")]
async fn image_submission_is_portable_for_new_turns_and_steers() {
let dir = tempfile::tempdir().unwrap();
@@ -2522,6 +2586,9 @@ async fn image_preparation_failure_restores_full_input_without_submitting() {
chat.input_queue
.pending_steers
.push_back(pending_steer("already sent"));
chat.input_queue
.queued_user_messages
.push_back(UserMessage::from("follow-up").into());
}
let pending_steers = chat.input_queue.pending_steers.clone();
let dir = tempfile::tempdir().unwrap();
@@ -2596,6 +2663,26 @@ async fn image_preparation_failure_restores_full_input_without_submitting() {
normalize_snapshot_paths(render_bottom_popup(&chat, /*width*/ 80))
);
}
if running {
// A transport recovery must preserve the pause for the failed image draft.
chat.pause_for_disconnect();
let mut input = chat.capture_thread_input_state();
input.as_mut().unwrap().reconnect_pending = true;
chat.restore_reconnected_input(input, &[pending_steers[0].client_id.clone()]);
chat.on_committed_user_message(
&[UserInput::Text {
text: "already sent".into(),
text_elements: Vec::new(),
}],
Some(&pending_steers[0].client_id),
/*from_replay*/ false,
"turn",
);
chat.input_queue.suppress_queue_autosend = false;
handle_turn_completed(&mut chat, "turn", /*duration_ms*/ None);
assert_eq!(chat.queued_user_message_texts(), vec!["follow-up"]);
assert_no_submit_op(&mut op_rx);
}
pixels.save(&path).unwrap();
chat.handle_key_event(KeyEvent::new(KeyCode::Enter, KeyModifiers::NONE));
while let Some(event) = rx.recv().await {
codex-rs/tui/src/chatwidget/tests/questions_tests.rs
+2 -2
View File
@@ -363,7 +363,7 @@ async fn reconnect_preserves_buffered_question_answer_source() {
let (mut restored, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await;
restored.thread_id = Some(ThreadId::new());
restored.restore_reconnected_input(state);
restored.restore_reconnected_input(state, &[]);
assert_eq!(
restored
.input_queue
@@ -801,7 +801,7 @@ async fn question_drafts_survive_navigation_and_snapshot_replay() {
Some(ReplayKind::ThreadSnapshot),
);
assert_eq!(question_count(&chat), 1);
chat.restore_reconnected_input(saved);
chat.restore_reconnected_input(saved, &[]);
assert!(render_bottom_popup(&chat, /*width*/ 80).contains("second draft"));
chat.pause_unavailable_thread();
chat.handle_question_key(KeyEvent::new(KeyCode::Char(']'), KeyModifiers::CONTROL));
codex-rs/tui/src/chatwidget/tests/review_mode.rs
+2
View File
@@ -390,7 +390,9 @@ async fn restore_thread_input_state_restores_pending_steers_without_downgrading_
queued_user_messages,
queued_user_message_history_records: VecDeque::new(),
recovered_queue: false,
reconnect_pending: false,
user_turn_pending_start: false,
pending_user_message_client_id: None,
submit_pending_steers_after_interrupt: false,
current_collaboration_mode: chat.current_collaboration_mode.clone(),
active_collaboration_mask: chat.active_collaboration_mask.clone(),
codex-rs/tui/src/chatwidget/user_messages.rs
+20
View File
@@ -66,10 +66,18 @@ pub(super) enum UserMessageSource {
QuestionAnswer,
}
/// Whether a recovered message can be submitted without risking a duplicate.
#[derive(Debug, Clone, PartialEq)]
pub(super) enum MessageDelivery {
Unsent,
Unconfirmed(Option<String>),
}
#[derive(Debug, Clone, PartialEq)]
pub(super) struct QueuedUserMessage {
pub(super) user_message: UserMessage,
pub(super) action: QueuedInputAction,
pub(super) delivery: MessageDelivery,
pub(super) pending_pastes: Vec<(String, String)>,
pub(super) source: UserMessageSource,
}
@@ -79,6 +87,7 @@ impl QueuedUserMessage {
Self {
user_message,
action,
delivery: MessageDelivery::Unsent,
pending_pastes: Vec::new(),
source: UserMessageSource::Prompt,
}
@@ -145,7 +154,10 @@ pub(crate) struct ThreadInputState {
pub(super) queued_user_messages: VecDeque<QueuedUserMessage>,
pub(super) queued_user_message_history_records: VecDeque<UserMessageHistoryRecord>,
pub(crate) recovered_queue: bool,
/// Reconcile delivery after transport recovery, not after a fork or ordinary thread switch.
pub(crate) reconnect_pending: bool,
pub(super) user_turn_pending_start: bool,
pub(super) pending_user_message_client_id: Option<String>,
pub(super) submit_pending_steers_after_interrupt: bool,
pub(super) current_collaboration_mode: CollaborationMode,
pub(super) active_collaboration_mask: Option<CollaborationModeMask>,
@@ -838,3 +850,11 @@ impl ChatWidget {
)
}
}
impl ThreadInputState {
pub(crate) fn has_unconfirmed_messages(&self) -> bool {
self.queued_user_messages
.iter()
.any(|message| matches!(message.delivery, MessageDelivery::Unconfirmed(_)))
}
}