From a142c90e2601d7b7bc9ab07e4d0552a3814b0501 Mon Sep 17 00:00:00 2001 From: Benjamin Carlsson Date: Fri, 6 Nov 2026 03:41:12 +0000 Subject: [PATCH] Handle TUI message submission receipts and preserve uncertain drafts (#59817) ## Why The TUI's cached active turn can be stale when a cloud message is submitted. A lost reply also leaves delivery uncertain, so automatically resending the draft risks submitting it twice. ## What changed - Let the cloud server choose whether an Enter submission starts or steers a turn, and apply its receipt to the matching client message ID. - Keep Tab-queued input waiting until the admitted turn finishes, including before its start notification arrives. - Restore rejected submissions to the composer. Hold duplicate or unconfirmed submissions out of automatic delivery until history confirms them. - Reconcile saved drafts and pending steers by client message ID when resuming sessions, and preserve submissions belonging to other turns during interruption recovery. - Classify cloud preflight failures as invalid parameters so rejected queued input can be restored safely. ## Testing Add regression coverage for cloud receipt outcomes, queued-input ordering, rejected drafts, lost replies, interruption recovery, and session resume with identical message text but distinct client IDs. GitOrigin-RevId: ec4eb581e450488434bc4614ec59273d73ffd0a1 --- .../src/cloud_backend/mod.rs | 23 +- .../src/cloud_backend/submission_tests.rs | 19 + codex-rs/tui/src/app.rs | 44 -- codex-rs/tui/src/app/agents_overview.rs | 4 +- codex-rs/tui/src/app/safety_buffering.rs | 2 + codex-rs/tui/src/app/session_picker.rs | 17 + codex-rs/tui/src/app/tests.rs | 3 + .../tui/src/app/tests/disconnect_tests.rs | 30 +- .../tui/src/app/tests/safety_buffering.rs | 8 +- .../app/tests/session_lifecycle_requests.rs | 502 ++++++++++++++++++ ...app__tests__disconnect__offline_draft.snap | 1 + ...cle_requests__cloud_composer_receipts.snap | 52 ++ codex-rs/tui/src/app/tests/turn_submission.rs | 64 +++ codex-rs/tui/src/app/thread_events.rs | 29 + codex-rs/tui/src/app/thread_routing.rs | 30 +- codex-rs/tui/src/app/user_turn_submission.rs | 218 ++++---- codex-rs/tui/src/app_command.rs | 13 + codex-rs/tui/src/app_server_session.rs | 1 + .../user_message_submission.rs | 269 ++++++++++ codex-rs/tui/src/chatwidget.rs | 61 ++- codex-rs/tui/src/chatwidget/input_flow.rs | 16 +- codex-rs/tui/src/chatwidget/input_queue.rs | 12 + codex-rs/tui/src/chatwidget/input_receipts.rs | 128 +++++ codex-rs/tui/src/chatwidget/input_restore.rs | 75 ++- .../tui/src/chatwidget/input_submission.rs | 31 +- codex-rs/tui/src/chatwidget/protocol.rs | 13 +- codex-rs/tui/src/chatwidget/reconnect.rs | 18 +- .../chatwidget/tests/composer_submission.rs | 45 ++ codex-rs/tui/src/chatwidget/tests/helpers.rs | 2 + codex-rs/tui/src/chatwidget/turn_lifecycle.rs | 8 + codex-rs/tui/src/chatwidget/turn_runtime.rs | 2 +- codex-rs/tui/src/chatwidget/user_messages.rs | 40 ++ 32 files changed, 1522 insertions(+), 258 deletions(-) create mode 100644 codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__session_lifecycle_requests__cloud_composer_receipts.snap create mode 100644 codex-rs/tui/src/app_server_session/user_message_submission.rs create mode 100644 codex-rs/tui/src/chatwidget/input_receipts.rs diff --git a/codex-rs/app-server-client/src/cloud_backend/mod.rs b/codex-rs/app-server-client/src/cloud_backend/mod.rs index 7b7c86ed79..8ac8c6ec2e 100644 --- a/codex-rs/app-server-client/src/cloud_backend/mod.rs +++ b/codex-rs/app-server-client/src/cloud_backend/mod.rs @@ -26,6 +26,10 @@ type ReplyStream = BoxStream<'static, Result>; #[error("{0}")] struct PreflightRejection(&'static str); +#[derive(Debug, thiserror::Error)] +#[error("Cloud stream is not connected. Reopen the conversation before sending.")] +struct MissingReplyAttachment; + struct Attachment { thread_id: String, stream: ReplyStream, @@ -124,7 +128,7 @@ impl Replies { &mut self, id: &str, output: &mpsc::UnboundedSender, - ) -> Result<(), PreflightRejection> { + ) -> Result<(), MissingReplyAttachment> { // Notice a ready disconnect before sending. A continuous stream of events // must still allow commands to make progress. let _ = tokio::task::unconstrained(self.forward_next(output)).now_or_never(); @@ -133,9 +137,7 @@ impl Replies { .as_ref() .is_none_or(|active| active.thread_id != id) { - return Err(PreflightRejection( - "Cloud stream is not connected. Reopen the conversation before sending.", - )); + return Err(MissingReplyAttachment); } Ok(()) } @@ -204,12 +206,13 @@ async fn run_worker( .is_some() { -32601 - } else if user_message_submission - && (error.is::() - || matches!( - error.downcast_ref::(), - Some(cloud::MutationError::InvalidRequest) - )) + } else if error.is::() + || user_message_submission + && (error.is::() + || matches!( + error.downcast_ref::(), + Some(cloud::MutationError::InvalidRequest) + )) { -32602 } else { diff --git a/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs b/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs index 0130c44896..f23cad2274 100644 --- a/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs +++ b/codex-rs/app-server-client/src/cloud_backend/submission_tests.rs @@ -118,6 +118,25 @@ async fn native_submission_uses_one_admission_and_keeps_preflight_guards() -> Re }; assert_eq!(source.code, -32602); } + let queued = handle + .request(protocol::ClientRequest::TurnStart { + request_id: protocol::RequestId::Integer(3), + params: protocol::TurnStartParams { + input: vec![protocol::UserInput::LocalImage { + path: "queued.png".into(), + detail: None, + }], + ..submission("queued-image") + }, + }) + .await? + .expect_err("a queued attachment must fail before admission"); + assert_eq!(queued.code, -32602); + assert_eq!( + queued.message, + "Cloud conversations currently accept plain text only." + ); + drop(reply_sender); let Err(crate::TypedRequestError::Server { source, .. }) = cloud_handle.submit_user_message(submission("closed")).await diff --git a/codex-rs/tui/src/app.rs b/codex-rs/tui/src/app.rs index d30d6c9739..3e393f2b54 100644 --- a/codex-rs/tui/src/app.rs +++ b/codex-rs/tui/src/app.rs @@ -136,7 +136,6 @@ use codex_app_server_protocol::ThreadMemoryMode; use codex_app_server_protocol::ThreadSettingsUpdateParams; use codex_app_server_protocol::ThreadStartSource; use codex_app_server_protocol::Turn; -use codex_app_server_protocol::TurnError as AppServerTurnError; use codex_app_server_protocol::TurnStatus; use codex_app_server_protocol::WriteStatus; use codex_config::CloudConfigBundleLoader; @@ -791,49 +790,6 @@ impl RuntimePermissionProfileOverride { } } -fn active_turn_not_steerable_turn_error(error: &TypedRequestError) -> Option { - let TypedRequestError::Server { source, .. } = error else { - return None; - }; - let turn_error: AppServerTurnError = serde_json::from_value(source.data.clone()?).ok()?; - matches!( - turn_error.codex_error_info, - Some(AppServerCodexErrorInfo::ActiveTurnNotSteerable { .. }) - ) - .then_some(turn_error) -} - -#[derive(Debug, Clone, PartialEq, Eq)] -enum ActiveTurnSteerRace { - Missing, - ExpectedTurnMismatch { actual_turn_id: String }, -} - -fn active_turn_steer_race(error: &TypedRequestError) -> Option { - let TypedRequestError::Server { method, source } = error else { - return None; - }; - if method != "turn/steer" { - return None; - } - if source.message == "no active turn to steer" { - return Some(ActiveTurnSteerRace::Missing); - } - - // App-server steer mismatches mean our cached active turn id is stale, but the response - // includes the server's current active turn so we can resynchronize and retry once. - let mismatch_prefix = "expected active turn id `"; - let mismatch_separator = "` but found `"; - let actual_turn_id = source - .message - .strip_prefix(mismatch_prefix)? - .split_once(mismatch_separator)? - .1 - .strip_suffix('`')? - .to_string(); - Some(ActiveTurnSteerRace::ExpectedTurnMismatch { actual_turn_id }) -} - fn active_turn_interrupt_race(error: &TypedRequestError) -> Option { let TypedRequestError::Server { method, source } = error else { return None; diff --git a/codex-rs/tui/src/app/agents_overview.rs b/codex-rs/tui/src/app/agents_overview.rs index 8559f1272f..c31484dc90 100644 --- a/codex-rs/tui/src/app/agents_overview.rs +++ b/codex-rs/tui/src/app/agents_overview.rs @@ -886,11 +886,13 @@ impl App { pub(super) async fn restore_session_draft( &mut self, thread_id: ThreadId, - input: ThreadInputState, + mut input: ThreadInputState, pending_settings: Option, ) { let preserve_in_flight_turn = !self.chat_widget.is_external_writer_view() && self.active_turn_id_for_thread(thread_id).await.is_some(); + self.reconcile_saved_session_input(thread_id, &mut input) + .await; self.chat_widget.restore_thread_input_state( Some(input), ThreadInputStateRestoreMode { diff --git a/codex-rs/tui/src/app/safety_buffering.rs b/codex-rs/tui/src/app/safety_buffering.rs index aae4a2f19d..46bc1a82d2 100644 --- a/codex-rs/tui/src/app/safety_buffering.rs +++ b/codex-rs/tui/src/app/safety_buffering.rs @@ -59,6 +59,7 @@ impl App { let AppCommand::UserTurn { items, + cloud_submission, cwd, active_permission_profile, approval_policy, @@ -73,6 +74,7 @@ impl App { ); return; }; + *cloud_submission = crate::app_command::CloudSubmissionIntent::UseKnownTurn; let permissions_override = Self::turn_permissions_override_from_config( &retry_config, active_permission_profile.as_ref(), diff --git a/codex-rs/tui/src/app/session_picker.rs b/codex-rs/tui/src/app/session_picker.rs index 37433d7233..cddb331ca3 100644 --- a/codex-rs/tui/src/app/session_picker.rs +++ b/codex-rs/tui/src/app/session_picker.rs @@ -19,6 +19,21 @@ impl App { } } + pub(super) async fn reconcile_saved_session_input( + &self, + thread_id: ThreadId, + input: &mut ThreadInputState, + ) { + if (!input.pending_steers.is_empty() || input.has_unconfirmed_messages()) + && let Some(channel) = self.thread_event_channels.get(&thread_id) + { + let store = channel.store.lock().await; + let confirmed = + super::thread_events::confirmed_user_message_ids(&store.turns, store.buffer.iter()); + input.reconcile_confirmed_messages(&confirmed); + } + } + pub(super) async fn open_resume_picker_for_backends( &mut self, tui: &mut tui::Tui, @@ -93,6 +108,8 @@ impl App { { // The resumed widget already reflects the latest server result. input_state.provider_error = None; + self.reconcile_saved_session_input(thread_id, &mut input_state) + .await; let preserve_in_flight_turn = self.active_turn_id_for_thread(thread_id).await.is_some(); self.chat_widget.restore_thread_input_state( diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 153207d984..9a2953f5a9 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -129,6 +129,9 @@ use crate::app_backtrack::BacktrackState; use crate::app_backtrack::nth_user_position; use crate::app_backtrack::user_count; use crate::app_event::HistoryBatchEntryResponse; +use crate::app_server_session::user_message_submission::ActiveTurnSteerRace; +use crate::app_server_session::user_message_submission::active_turn_not_steerable_turn_error; +use crate::app_server_session::user_message_submission::active_turn_steer_race; use crate::test_support::boxed_future; use codex_app_server_protocol::ThreadItem; diff --git a/codex-rs/tui/src/app/tests/disconnect_tests.rs b/codex-rs/tui/src/app/tests/disconnect_tests.rs index 41dd449acd..aa7743119c 100644 --- a/codex-rs/tui/src/app/tests/disconnect_tests.rs +++ b/codex-rs/tui/src/app/tests/disconnect_tests.rs @@ -31,6 +31,11 @@ async fn lost_mutation_reply_preserves_work_without_resubmitting() -> Result<()> app.primary_thread_id = Some(id); app.chat_widget .handle_thread_session(test_thread_session(id, app.config.cwd.to_path_buf())); + app.ensure_thread_channel(id) + .store + .lock() + .await + .set_active_turn_id("finished".into()); let listener = TcpListener::bind("127.0.0.1:0").await?; let endpoint = crate::resolve_remote_addr(&format!("ws://{}", listener.local_addr()?))?; app.app_server_target = AppServerTarget::Remote { @@ -38,8 +43,21 @@ async fn lost_mutation_reply_preserves_work_without_resubmitting() -> Result<()> }; let server = tokio::spawn(async move { let (stream, _) = listener.accept().await?; + let mut steered_message = None; serve_reconnect_requests(tokio_tungstenite::accept_async(stream).await?, |request| { + let params = request.params.unwrap(); + let message_id = params["clientUserMessageId"] + .as_str() + .expect("submission ID"); + if request.method == "turn/steer" { + assert_eq!(params["expectedTurnId"], "finished"); + steered_message = Some(message_id.to_owned()); + return std::future::ready(Some( + json!({"error": {"code": -32600, "message": "no active turn to steer"}}), + )); + } assert_eq!(request.method, "turn/start"); + assert_eq!(steered_message.as_deref(), Some(message_id)); std::future::ready(None) }) .await @@ -157,7 +175,12 @@ async fn lost_mutation_reply_preserves_work_without_resubmitting() -> Result<()> app.chat_widget.composer_text_with_pending(), "queued before disconnect" ); - assert!(!app.chat_widget.has_queued_follow_up_messages()); + assert_eq!(app.chat_widget.queued_user_message_texts(), ["only once"]); + assert!( + app.chat_widget + .capture_thread_input_state() + .is_some_and(|input| input.has_unconfirmed_messages()) + ); let paste = "copied text ".repeat(/*n*/ 92); app.handle_tui_event(&mut tui, &mut session, TuiEvent::Paste(paste.clone())) .await?; @@ -211,7 +234,10 @@ async fn lost_mutation_reply_preserves_work_without_resubmitting() -> Result<()> )); } session.shutdown().await?; - assert_eq!(server.await??, vec!["initialize", "turn/start"]); + assert_eq!( + server.await??, + vec!["initialize", "turn/steer", "turn/start"] + ); Ok(()) } diff --git a/codex-rs/tui/src/app/tests/safety_buffering.rs b/codex-rs/tui/src/app/tests/safety_buffering.rs index 4719506025..e13349f680 100644 --- a/codex-rs/tui/src/app/tests/safety_buffering.rs +++ b/codex-rs/tui/src/app/tests/safety_buffering.rs @@ -510,9 +510,15 @@ stream_max_retries = 0 unreachable!("user turn"); }; let expected_client_id = client_user_message_id.clone(); - let pending_input = app.chat_widget.capture_thread_input_state(); + // The server reports the current turn. The retried steer must keep this message ID. + app.ensure_thread_channel(thread_id) + .store + .lock() + .await + .set_active_turn_id("stale".into()); app.submit_thread_op(&mut app_server, thread_id, steer) .await?; + let pending_input = app.chat_widget.capture_thread_input_state(); let other_id = ThreadId::from_string( &app_test_support::create_fake_rollout( app.config.codex_home.as_path(), diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index 4407ec76ea..ecbb4d79e7 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs @@ -43,6 +43,508 @@ use tokio_tungstenite::tungstenite::Error as WebSocketError; use tokio_tungstenite::tungstenite::Message; use tokio_tungstenite::tungstenite::error::ProtocolError; +#[tokio::test] +async fn cloud_enter_recovers_drafts_from_server_receipts() -> Result<()> { + use codex_app_server_client::CloudOperation; + use codex_app_server_client::CloudUserMessageReceipt::Duplicate; + use codex_app_server_client::CloudUserMessageReceipt::Started; + use codex_app_server_client::CloudUserMessageReceipt::Steered; + + let turn = test_turn("actual", TurnStatus::InProgress, Vec::new()); + let mut rendered_receipts = Vec::new(); + for (label, receipt) in [ + ("Started", Some(Ok(Started(turn.clone())))), + ("Steered", Some(Ok(Steered(turn.clone())))), + ("Unconfirmed", None), + ( + "Rejected", + Some(Err(JSONRPCErrorError { + code: -32602, + message: "reply stream closed before submission".into(), + data: None, + })), + ), + ("Duplicate", Some(Ok(Duplicate { turn_id: None }))), + ] { + let (mut app, mut events, mut ops) = make_test_app_with_channels().await; + let id = ThreadId::new(); + app.enqueue_primary_thread_session( + test_thread_session(id, app.config.cwd.to_path_buf()), + Vec::new(), + ) + .await?; + app.ensure_thread_channel(id) + .store + .lock() + .await + .set_active_turn_id("stale".into()); + app.chat_widget.handle_server_notification( + turn_started_notification(id, "stale"), + /*replay_kind*/ None, + ); + while events.try_recv().is_ok() {} + app.chat_widget.apply_external_edit("follow up".into()); + app.chat_widget + .handle_key_event(KeyEvent::from(KeyCode::Enter)); + let op = next_user_turn_op(&mut ops); + let Op::UserTurn { + client_user_message_id, + items, + .. + } = &op + else { + unreachable!() + }; + let expected = (Some(client_user_message_id.clone()), items.clone()); + + let (client, mut commands, _events) = + codex_app_server_client::CloudClient::channel(/*capacity*/ 1); + let mut server = AppServerSession::new( + crate::backend_id::BackendId::new(), + codex_app_server_client::AppServerClient::Cloud(client), + crate::ThreadParamsMode::Cloud, + ); + let rejected = matches!(receipt, Some(Err(_))); + let admitted = matches!(receipt, Some(Ok(Started(_) | Steered(_)))); + let steered = matches!(receipt, Some(Ok(Steered(_)))); + let (result, ()) = tokio::join!(app.submit_thread_op(&mut server, id, op), async { + let command = commands.recv().await.unwrap(); + let CloudOperation::SubmitUserMessage(params) = command.operation else { + panic!("Enter must start or steer") + }; + assert_eq!(params.thread_id, id.to_string()); + assert_eq!((params.client_user_message_id, params.input), expected); + if let Some(receipt) = receipt { + command + .response + .send(Ok(receipt.map(|r| serde_json::to_value(r).unwrap()))) + .unwrap(); + } + }); + result?; + if !rejected { + app.chat_widget.handle_server_notification( + turn_completed_notification(id, "stale", TurnStatus::Interrupted), + /*replay_kind*/ None, + ); + assert_eq!(app.chat_widget.composer_text_with_pending(), ""); + } + if admitted { + app.chat_widget.handle_server_notification( + turn_started_notification(id, "actual"), + /*replay_kind*/ None, + ); + } + if steered { + app.chat_widget.handle_server_notification( + turn_completed_notification(id, "actual", TurnStatus::Interrupted), + /*replay_kind*/ None, + ); + } + let history = std::iter::from_fn(|| events.try_recv().ok()) + .filter_map(|event| match event { + AppEvent::InsertHistoryCell(cell) => { + Some(lines_to_single_string(&cell.display_lines(/*width*/ 80))) + } + _ => None, + }) + .collect::>() + .join("\n"); + let composer = + normalize_snapshot_paths(render_bottom_popup(&app.chat_widget, /*width*/ 80)); + rendered_receipts.push(format!("── {label} ──\n{history}\n{composer}")); + assert!( + !std::iter::from_fn(|| ops.try_recv().ok()).any(|op| matches!(op, Op::UserTurn { .. })) + ); + } + insta::assert_snapshot!("cloud_composer_receipts", rendered_receipts.join("\n\n")); + Ok(()) +} + +#[tokio::test] +async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> { + use crate::app::agents_overview::tests::overview_thread; + use crate::app_server_session::user_message_submission::UserMessageAdmission; + use codex_app_server_client::CloudOperation; + use codex_app_server_protocol::ItemCompletedNotification; + + let (mut app, mut events, mut ops) = make_test_app_with_channels().await; + let [original, other] = [ThreadId::new(), ThreadId::new()]; + app.app_server_target = AppServerTarget::Remote { + endpoint: crate::resolve_remote_addr("ws://127.0.0.1:1")?, + }; + app.enqueue_primary_thread_session( + test_thread_session(original, app.config.cwd.to_path_buf()), + Vec::new(), + ) + .await?; + + let turn_id = "finished-turn"; + let mut client_ids = Vec::new(); + for _ in 0..2 { + app.chat_widget.apply_external_edit("same text".into()); + app.chat_widget + .handle_key_event(KeyEvent::from(KeyCode::Enter)); + let op = next_user_turn_op(&mut ops); + let Op::UserTurn { + client_user_message_id, + .. + } = &op + else { + unreachable!(); + }; + app.chat_widget.user_message_admitted( + UserMessageAdmission::Steered(turn_id.into()), + &op, + client_user_message_id, + ); + client_ids.push(client_user_message_id.clone()); + } + let [confirmed_id, unmatched_id]: [String; 2] = client_ids.try_into().unwrap(); + let user_item = |client_id: String| ThreadItemPayload::UserMessage { + id: format!("item-{client_id}"), + client_id: Some(client_id), + content: vec![UserInput::Text { + text: "same text".into(), + text_elements: Vec::new(), + }], + }; + let mut original_thread = overview_thread( + original, + /*parent_thread_id*/ None, + "Original", + ThreadStatus::Idle, + ); + original_thread.cwd = app.config.cwd.clone(); + original_thread.turns = vec![test_turn( + turn_id, + TurnStatus::Completed, + vec![user_item(confirmed_id)], + )]; + let mut other_thread = overview_thread( + other, + /*parent_thread_id*/ None, + "Other", + ThreadStatus::Idle, + ); + other_thread.cwd = app.config.cwd.clone(); + + let (client, mut commands, _server_events) = + codex_app_server_client::CloudClient::channel(/*capacity*/ 1); + let mut server = AppServerSession::new( + crate::backend_id::BackendId::new(), + codex_app_server_client::AppServerClient::Cloud(client), + crate::ThreadParamsMode::Cloud, + ); + let backend_id = server.backend_id(); + let serve = tokio::spawn(async move { + let mut resumed_threads = Vec::new(); + while let Some(command) = commands.recv().await { + let reply = match command.operation { + CloudOperation::Request(request) => match *request { + ClientRequest::ThreadRead { params, .. } => { + let mut thread = if params.thread_id == original.to_string() { + original_thread.clone() + } else { + other_thread.clone() + }; + if !params.include_turns { + thread.turns.clear(); + } + Ok(serde_json::json!({"thread": thread})) + } + ClientRequest::ThreadResume { params, .. } => { + let thread = if params.thread_id == original.to_string() { + &original_thread + } else { + &other_thread + }; + resumed_threads.push(params.thread_id); + Ok(serde_json::json!({ + "thread": thread, + "model": "gpt-5.5", + "modelProvider": thread.model_provider, + "cwd": thread.cwd, + "approvalPolicy": "never", + "approvalsReviewer": "user", + "sandbox": {"type": "readOnly"}, + })) + } + ClientRequest::ThreadUnsubscribe { .. } => { + Ok(serde_json::json!({"status": "unsubscribed"})) + } + ClientRequest::ThreadLoadedList { .. } => { + Ok(serde_json::json!({"data": [], "nextCursor": null})) + } + ClientRequest::ThreadGoalGet { .. } => Ok(serde_json::json!({"goal": null})), + ClientRequest::TurnStart { .. } | ClientRequest::TurnSteer { .. } => { + panic!("resuming must not send a turn") + } + _ => Err(JSONRPCErrorError { + code: -32601, + message: "unsupported by the resume test".into(), + data: None, + }), + }, + CloudOperation::SubmitUserMessage(_) => panic!("resuming must not send a message"), + CloudOperation::Shutdown => { + command + .response + .send(Ok(Ok(serde_json::Value::Null))) + .unwrap(); + break; + } + _ => panic!("unexpected Cloud operation while resuming"), + }; + command.response.send(Ok(reply)).unwrap(); + } + resumed_threads + }); + + let mut tui = crate::tui::test_support::make_test_tui()?; + while events.try_recv().is_ok() {} + for thread_id in [other, original] { + app.apply_resume_picker_selection( + &mut tui, + &mut server, + SessionSelection::Resume(SessionTarget { + backend_id, + path: None, + thread_id, + cwd: None, + history_mode: None, + }), + ) + .await?; + assert_eq!(app.chat_widget.thread_id(), Some(thread_id)); + } + + assert_eq!(app.chat_widget.queued_user_message_texts(), ["same text"]); + assert!( + app.chat_widget + .capture_thread_input_state() + .is_some_and(|input| input.has_unconfirmed_messages()) + ); + assert!( + !std::iter::from_fn(|| events.try_recv().ok()) + .any(|event| { matches!(event, AppEvent::CodexOp(Op::UserTurn { .. })) }) + ); + app.chat_widget.handle_server_notification( + ServerNotification::ItemCompleted(ItemCompletedNotification::new( + original.to_string(), + turn_id.into(), + user_item(unmatched_id), + /*started_at_ms*/ None, + /*completed_at_ms*/ 1, + )), + /*replay_kind*/ None, + ); + assert!(app.chat_widget.queued_user_message_texts().is_empty()); + server.shutdown().await?; + assert_eq!(serve.await?, [other.to_string(), original.to_string()]); + Ok(()) +} + +#[tokio::test] +async fn conventional_steer_defers_review_and_holds_a_lost_reply() -> Result<()> { + use codex_app_server_client::CloudOperation; + use codex_app_server_protocol::ItemCompletedNotification; + use codex_app_server_protocol::NonSteerableTurnKind; + + for review_active in [true, false] { + let (mut app, _events, mut ops) = make_test_app_with_channels().await; + let id = ThreadId::new(); + app.enqueue_primary_thread_session( + test_thread_session(id, app.config.cwd.to_path_buf()), + Vec::new(), + ) + .await?; + let turn_id = "active"; + app.ensure_thread_channel(id) + .store + .lock() + .await + .set_active_turn_id(turn_id.into()); + app.chat_widget.handle_server_notification( + turn_started_notification(id, turn_id), + /*replay_kind*/ None, + ); + app.chat_widget.apply_external_edit("first".into()); + app.chat_widget.handle_key_event(KeyCode::Enter.into()); + let op = next_user_turn_op(&mut ops); + let Op::UserTurn { + client_user_message_id, + items, + .. + } = &op + else { + unreachable!(); + }; + let (client_id, input) = (Some(client_user_message_id.clone()), items.clone()); + if review_active { + app.chat_widget.apply_external_edit("second".into()); + app.chat_widget.handle_key_event(KeyCode::Tab.into()); + } + + let (client, mut commands, _server_events) = + codex_app_server_client::CloudClient::channel(/*capacity*/ 1); + let mut server = AppServerSession::new( + crate::backend_id::BackendId::new(), + codex_app_server_client::AppServerClient::Cloud(client), + crate::ThreadParamsMode::Embedded, + ); + let (result, ()) = tokio::join!(app.submit_thread_op(&mut server, id, op), async { + let command = commands.recv().await.unwrap(); + let CloudOperation::Request(request) = command.operation else { + panic!("expected conventional steer"); + }; + let ClientRequest::TurnSteer { params, .. } = *request else { + panic!("expected turn/steer"); + }; + assert_eq!(params.expected_turn_id, turn_id); + assert_eq!( + (params.client_user_message_id, params.input), + (client_id.clone(), input.clone()) + ); + if review_active { + let message = "cannot steer a review turn"; + let error = AppServerTurnError { + message: message.into(), + codex_error_info: Some(AppServerCodexErrorInfo::ActiveTurnNotSteerable { + turn_kind: NonSteerableTurnKind::Review, + }), + additional_details: None, + misalignment: None, + }; + command + .response + .send(Ok(Err(JSONRPCErrorError { + code: -32600, + message: message.into(), + data: Some(serde_json::to_value(error).unwrap()), + }))) + .unwrap(); + } + }); + result?; + if review_active { + assert_eq!( + app.chat_widget.queued_user_message_texts(), + ["first", "second"] + ); + assert_eq!( + app.active_turn_id_for_thread(id).await.as_deref(), + Some(turn_id) + ); + } else { + app.chat_widget.handle_server_notification( + turn_completed_notification(id, turn_id, TurnStatus::Interrupted), + /*replay_kind*/ None, + ); + assert_eq!(app.chat_widget.composer_text_with_pending(), ""); + for client_id in [None, client_id] { + let confirmed = client_id.is_some(); + assert_eq!(app.chat_widget.queued_user_message_texts(), ["first"]); + app.chat_widget.handle_server_notification( + ServerNotification::ItemCompleted(ItemCompletedNotification::new( + id.to_string(), + turn_id.into(), + ThreadItemPayload::UserMessage { + id: format!("user-item-{confirmed}"), + client_id, + content: input.clone(), + }, + /*started_at_ms*/ None, + /*completed_at_ms*/ 1, + )), + /*replay_kind*/ None, + ); + } + assert!(app.chat_widget.queued_user_message_texts().is_empty()); + } + assert!(commands.try_recv().is_err()); + assert!( + !std::iter::from_fn(|| ops.try_recv().ok()).any(|op| matches!(op, Op::UserTurn { .. })) + ); + } + Ok(()) +} + +#[tokio::test] +async fn queued_start_rejection_restores_the_draft() -> Result<()> { + use crate::ThreadParamsMode; + use codex_app_server_client::CloudOperation; + + for (mode, code, message) in [ + (ThreadParamsMode::Embedded, -32600, "input was rejected"), + (ThreadParamsMode::Embedded, -32001, "server overloaded"), + ( + ThreadParamsMode::Cloud, + -32602, + "Cloud conversations currently accept plain text only.", + ), + ] { + let (mut app, _events, mut ops) = make_test_app_with_channels().await; + let id = ThreadId::new(); + app.enqueue_primary_thread_session( + test_thread_session(id, app.config.cwd.to_path_buf()), + Vec::new(), + ) + .await?; + app.chat_widget.handle_server_notification( + turn_started_notification(id, "before-queue"), + /*replay_kind*/ None, + ); + app.chat_widget.apply_external_edit("queued draft".into()); + app.chat_widget.handle_key_event(KeyCode::Tab.into()); + app.chat_widget.handle_server_notification( + turn_completed_notification(id, "before-queue", TurnStatus::Completed), + /*replay_kind*/ None, + ); + let op = next_user_turn_op(&mut ops); + let Op::UserTurn { + cloud_submission: crate::app_command::CloudSubmissionIntent::StartIfIdle, + client_user_message_id, + items, + .. + } = &op + else { + panic!("expected the deferred draft"); + }; + let expected = (Some(client_user_message_id.clone()), items.clone()); + let (client, mut commands, _server_events) = + codex_app_server_client::CloudClient::channel(/*capacity*/ 1); + let mut server = AppServerSession::new( + crate::backend_id::BackendId::new(), + codex_app_server_client::AppServerClient::Cloud(client), + mode, + ); + let (result, ()) = tokio::join!(app.submit_thread_op(&mut server, id, op), async { + let command = commands.recv().await.unwrap(); + let CloudOperation::Request(request) = command.operation else { + panic!("queued draft must use typed turn/start"); + }; + let ClientRequest::TurnStart { params, .. } = *request else { + panic!("expected turn/start"); + }; + assert_eq!((params.client_user_message_id, params.input), expected); + command + .response + .send(Ok(Err(JSONRPCErrorError { + code, + message: message.into(), + data: None, + }))) + .unwrap(); + }); + result?; + assert_eq!(app.chat_widget.composer_text_with_pending(), "queued draft"); + assert!(app.chat_widget.queued_user_message_texts().is_empty()); + assert!(commands.try_recv().is_err()); + } + Ok(()) +} + pub(super) type RecordedRequests = Arc>>; pub(super) type RecordingAppServer = (AppServerSession, RecordedRequests, JoinHandle>); diff --git a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__disconnect__offline_draft.snap b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__disconnect__offline_draft.snap index f4e0fff53b..a3c7c90a91 100644 --- a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__disconnect__offline_draft.snap +++ b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__disconnect__offline_draft.snap @@ -8,6 +8,7 @@ expression: "format!(\"{}\\n{}\",\nlines_to_single_string(&app.transcript_cells. • Reconnecting to server… () • Queued follow-up inputs + ↳ only once ↳ queued before disconnect ⇧← edit last queued message diff --git a/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__session_lifecycle_requests__cloud_composer_receipts.snap b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__session_lifecycle_requests__cloud_composer_receipts.snap new file mode 100644 index 0000000000..b3bda716d2 --- /dev/null +++ b/codex-rs/tui/src/app/tests/snapshots/codex_tui__app__tests__session_lifecycle_requests__cloud_composer_receipts.snap @@ -0,0 +1,52 @@ +--- +source: tui/src/app/tests/session_lifecycle_requests.rs +expression: "rendered_receipts.join(\"\\n\\n\")" +--- +── Started ── + +› follow up + +■ Conversation interrupted - use /feedback if something went wrong +• Working (0s • esc to interrupt) + + +› Ask Codex to do anything + + gpt-test default · /tmp/project + +── Steered ── +■ Conversation interrupted - use /feedback if something went wrong +■ Conversation interrupted - use /feedback if something went wrong +› follow up + + gpt-test default · /tmp/project + +── Unconfirmed ── +■ Cloud submission could not be confirmed: cloud/submitUserMessage transport error: cloud backend dropped its response +■ Conversation interrupted - use /feedback if something went wrong +• Queued follow-up inputs + ↳ follow up + ⇧← edit last queued message + +› Ask Codex to do anything + + gpt-test default · /tmp/project + +── Rejected ── +■ reply stream closed before submission +• Working (0s • esc to interrupt) + + +› follow up + + tab to queue message 100% context left + +── Duplicate ── +■ Conversation interrupted - use /feedback if something went wrong +• Queued follow-up inputs + ↳ follow up + ⇧← edit last queued message + +› Ask Codex to do anything + + gpt-test default · /tmp/project diff --git a/codex-rs/tui/src/app/tests/turn_submission.rs b/codex-rs/tui/src/app/tests/turn_submission.rs index 3b19476064..8195cd0246 100644 --- a/codex-rs/tui/src/app/tests/turn_submission.rs +++ b/codex-rs/tui/src/app/tests/turn_submission.rs @@ -11,6 +11,70 @@ use codex_protocol::permissions::FileSystemSpecialPath; use codex_protocol::permissions::NetworkSandboxPolicy; use pretty_assertions::assert_eq; +#[tokio::test] +async fn cloud_admission_keeps_tab_input_queued_until_the_admitted_turn_finishes() { + use crate::app_command::CloudSubmissionIntent; + use crate::app_server_session::user_message_submission::UserMessageAdmission; + + for receipt in [ + UserMessageAdmission::Started("next".into()), + UserMessageAdmission::Steered("next".into()), + ] { + let (mut app, _events, mut ops) = make_test_app_with_channels().await; + let id = ThreadId::new(); + let session = test_thread_session(id, app.config.cwd.to_path_buf()); + let chat = &mut app.chat_widget; + chat.handle_thread_session_quiet(session); + chat.handle_server_notification( + turn_started_notification(id, "previous"), + /*replay_kind*/ None, + ); + chat.apply_external_edit("queued draft".into()); + chat.handle_key_event(KeyEvent::from(KeyCode::Tab)); + chat.apply_external_edit("enter input".into()); + chat.handle_key_event(KeyEvent::from(KeyCode::Enter)); + let mut turns = std::iter::from_fn(|| ops.try_recv().ok()) + .filter(|op| matches!(op, Op::UserTurn { .. })); + let op = turns.next().expect("expected Enter submission"); + let Op::UserTurn { + client_user_message_id, + .. + } = &op + else { + unreachable!(); + }; + chat.user_message_admitted(receipt, &op, client_user_message_id); + chat.handle_server_notification( + turn_completed_notification(id, "previous", TurnStatus::Completed), + /*replay_kind*/ None, + ); + assert_eq!(chat.queued_user_message_texts(), ["queued draft"]); + assert!(turns.next().is_none()); + chat.apply_external_edit("another draft".into()); + chat.handle_key_event(KeyEvent::from(KeyCode::Tab)); + assert_eq!( + chat.queued_user_message_texts(), + ["queued draft", "another draft"] + ); + chat.handle_server_notification( + turn_started_notification(id, "next"), + /*replay_kind*/ None, + ); + assert!(turns.next().is_none()); + chat.handle_server_notification( + turn_completed_notification(id, "next", TurnStatus::Completed), + /*replay_kind*/ None, + ); + assert!(matches!( + turns.next(), + Some(Op::UserTurn { cloud_submission: CloudSubmissionIntent::StartIfIdle, items, .. }) + if matches!(items.as_slice(), [UserInput::Text { text, .. }] if text == "queued draft") + )); + assert_eq!(chat.queued_user_message_texts(), ["another draft"]); + assert!(turns.next().is_none()); + } +} + #[tokio::test] async fn worktree_creation_rejects_untrusted_source_before_allocation() -> Result<()> { use crate::app_event::ManagedWorktreeMode; diff --git a/codex-rs/tui/src/app/thread_events.rs b/codex-rs/tui/src/app/thread_events.rs index 6b0193c78f..1a125bb0a2 100644 --- a/codex-rs/tui/src/app/thread_events.rs +++ b/codex-rs/tui/src/app/thread_events.rs @@ -26,6 +26,35 @@ pub(super) enum ThreadBufferedEvent { FeedbackSubmission(FeedbackThreadEvent), } +pub(super) fn confirmed_user_message_ids<'a>( + turns: &'a [Turn], + events: impl Iterator, +) -> Vec { + turns + .iter() + .flat_map(Turn::item_payloads) + .chain(events.filter_map(|event| { + let ThreadBufferedEvent::Notification(notification) = event else { + return None; + }; + if let ServerNotification::ItemStarted(notification) = notification.as_ref() { + Some(¬ification.item.payload) + } else if let ServerNotification::ItemCompleted(notification) = notification.as_ref() { + Some(¬ification.item.payload) + } else { + None + } + })) + .filter_map(|item| { + if let ThreadItemPayload::UserMessage { client_id, .. } = item { + client_id.clone() + } else { + None + } + }) + .collect::>() +} + fn is_voice_handoff_item(item: &ThreadItemPayload) -> bool { matches!(item, ThreadItemPayload::UserMessage { content, .. } if crate::chatwidget::realtime_delegation_input(content).is_some()) diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index 97e15e0f40..a614c9ce9e 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -1845,32 +1845,10 @@ impl App { self.chat_widget .remember_realtime_delegated_reasoning_turn(turn_id); } - let confirmed_message_ids = snapshot - .turns - .iter() - .flat_map(Turn::item_payloads) - .chain(snapshot.events.iter().filter_map(|event| { - let ThreadBufferedEvent::Notification(notification) = event else { - return None; - }; - if let ServerNotification::ItemStarted(notification) = notification.as_ref() { - Some(¬ification.item.payload) - } else if let ServerNotification::ItemCompleted(notification) = - notification.as_ref() - { - Some(¬ification.item.payload) - } else { - None - } - })) - .filter_map(|item| { - if let ThreadItemPayload::UserMessage { client_id, .. } = item { - client_id.clone() - } else { - None - } - }) - .collect::>(); + let confirmed_message_ids = super::thread_events::confirmed_user_message_ids( + &snapshot.turns, + snapshot.events.iter(), + ); let recovered_input = snapshot .input_state .as_ref() diff --git a/codex-rs/tui/src/app/user_turn_submission.rs b/codex-rs/tui/src/app/user_turn_submission.rs index cdbda63973..d03ed28538 100644 --- a/codex-rs/tui/src/app/user_turn_submission.rs +++ b/codex-rs/tui/src/app/user_turn_submission.rs @@ -3,6 +3,11 @@ //! This is the single app-server submission path for both new user turns and steering. use super::*; +use crate::app_command::CloudSubmissionIntent; +use crate::app_server_session::UnsupportedLegacyPermissionProfile; +use crate::app_server_session::user_message_submission::TurnStartSettings; +use crate::app_server_session::user_message_submission::UserMessageAdmission; +use crate::app_server_session::user_message_submission::UserMessageSubmissionFailure; pub(super) async fn try_submit_user_turn( app: &mut App, @@ -12,105 +17,116 @@ pub(super) async fn try_submit_user_turn( ) -> Result { let AppCommand::UserTurn { turn_trigger, + cloud_submission, client_user_message_id, - items, - cwd, - approval_policy, - approvals_reviewer, - active_permission_profile: _, model, - effort, - summary, - service_tier, - final_output_json_schema, - collaboration_mode, + .. } = op else { return Ok(false); }; - if let Some(turn_id) = app.active_turn_id_for_thread(thread_id).await { - if *turn_trigger != "user" { - color_eyre::eyre::bail!("the fork already has an active turn"); - } - let mut steer_turn_id = turn_id; - let mut retried_after_turn_mismatch = false; - loop { - match app_server - .turn_steer( - thread_id, - steer_turn_id.clone(), - client_user_message_id.clone(), - items.to_vec(), - ) - .await - { - Ok(_) => { - if app.active_thread_id == Some(thread_id) - && app.chat_widget.thread_id() == Some(thread_id) - { - crate::startup_recovery::acknowledged(client_user_message_id); - } - return Ok(true); - } - Err(error) => { - if let Some(turn_error) = active_turn_not_steerable_turn_error(&error) { - if !app.chat_widget.enqueue_rejected_steer() { - app.chat_widget.add_error_message(turn_error.message); - } - return Ok(true); - } - match active_turn_steer_race(&error) { - Some(ActiveTurnSteerRace::Missing) => { - if let Some(channel) = app.thread_event_channels.get(&thread_id) { - let mut store = channel.store.lock().await; - store.clear_active_turn_id(); - } - break; - } - Some(ActiveTurnSteerRace::ExpectedTurnMismatch { actual_turn_id }) - if !retried_after_turn_mismatch && actual_turn_id != steer_turn_id => - { - // Review flows can swap the active turn before the TUI - // processes the corresponding notification. Retry once with - // the server-reported turn id so non-steerable review turns - // still fall through to the existing queueing behavior. - if let Some(channel) = app.thread_event_channels.get(&thread_id) { - let mut store = channel.store.lock().await; - store.set_active_turn_id(actual_turn_id.clone()); - } - steer_turn_id = actual_turn_id; - retried_after_turn_mismatch = true; - } - Some(ActiveTurnSteerRace::ExpectedTurnMismatch { actual_turn_id }) => { - if let Some(channel) = app.thread_event_channels.get(&thread_id) { - let mut store = channel.store.lock().await; - store.set_active_turn_id(actual_turn_id); - } - return Err(error.into()); - } - None => return Err(error.into()), - } - } + let original_active = app.active_turn_id_for_thread(thread_id).await; + let mut active = original_active.clone(); + let preflight = || { + let enabled = app.chat_widget.daybreak_enabled + && !app.chat_widget.side_conversation_active() + && !app.side_threads.contains_key(&thread_id); + crate::daybreak::program_for_turn( + &app.chat_widget.model_catalog().models, + model, + app.chat_widget.daybreak_account_eligible(), + enabled, + ) + }; + let start_settings = || turn_start_settings(app, thread_id, op); + let result = app_server + .submit_user_message(thread_id, op, &mut active, preflight, start_settings) + .await; + if active != original_active + && let Some(channel) = app.thread_event_channels.get(&thread_id) + { + let mut store = channel.store.lock().await; + if store.active_turn_id() == original_active.as_deref() { + if let Some(active) = active { + store.set_active_turn_id(active); + } else { + store.clear_active_turn_id(); } } } - let eligible_account = app.chat_widget.daybreak_account_eligible(); - let enabled = app.chat_widget.daybreak_enabled - && !app.chat_widget.side_conversation_active() - && !app.side_threads.contains_key(&thread_id); - let cyber_access_program = match crate::daybreak::program_for_turn( - &app.chat_widget.model_catalog().models, - model, - eligible_account, - enabled, - ) { - Ok(program) => program, - Err(message) => { + + let ordinary = + *turn_trigger == "user" && *cloud_submission != CloudSubmissionIntent::UseKnownTurn; + if let Err(UserMessageSubmissionFailure::Unhandled(error)) = result { + return Err(error); + } + if app.active_thread_id != Some(thread_id) || app.chat_widget.thread_id() != Some(thread_id) { + if let Err(error) = result { + tracing::warn!(%thread_id, ?error, "User submission failed in a background thread"); + } + return Ok(true); + } + + match result { + Ok(admission) => { + if !matches!(&admission, UserMessageAdmission::Queued(_)) { + crate::startup_recovery::acknowledged(client_user_message_id); + } + if ordinary || !matches!(&admission, UserMessageAdmission::Steered(_)) { + app.chat_widget + .user_message_admitted(admission, op, client_user_message_id); + } + } + Err(UserMessageSubmissionFailure::Preflight(message)) if !ordinary => { if !app.chat_widget.handle_turn_start_rejection(message.clone()) { app.chat_widget.add_error_message(message); } - return Ok(true); } + Err(UserMessageSubmissionFailure::Preflight(message)) => { + app.chat_widget + .restore_rejected_user_message(client_user_message_id, message); + } + Err(UserMessageSubmissionFailure::Rejected(error)) => { + if error + .downcast_ref::() + .is_some() + { + app.chat_widget + .set_queue_autosend_suppressed(/*suppressed*/ true); + } + app.chat_widget + .restore_rejected_user_message(client_user_message_id, format!("{error:#}")); + } + Err(UserMessageSubmissionFailure::Unconfirmed { error, reconnect }) => { + app.chat_widget + .hold_unconfirmed_user_message(client_user_message_id); + if !reconnect || !app.recover_transport_error(&error) { + let label = if reconnect { + "Submission" + } else { + "Cloud submission" + }; + app.chat_widget + .add_error_message(format!("{label} could not be confirmed: {error}")); + } + } + Err(UserMessageSubmissionFailure::Unhandled(_)) => unreachable!("returned above"), + } + Ok(true) +} + +fn turn_start_settings<'a>( + app: &'a App, + thread_id: ThreadId, + op: &AppCommand, +) -> TurnStartSettings<'a> { + let AppCommand::UserTurn { + approval_policy, + approvals_reviewer, + .. + } = op + else { + unreachable!("turn/start requires a user turn"); }; let config = app.chat_widget.config_ref(); let selected_profile = app.pending_server_profiles.get(&thread_id).or_else(|| { @@ -150,30 +166,10 @@ pub(super) async fn try_submit_user_turn( }), explicit_profile.and_then(RuntimePermissionProfileOverride::turn_permission_profile), ); - let response = app_server - .turn_start( - thread_id, - turn_trigger, - client_user_message_id.clone(), - items.to_vec(), - cwd.clone(), - turn_approval_policy, - turn_approvals_reviewer, - permissions_override, - config.permissions.user_visible_workspace_roots(), - model.to_string(), - effort.clone(), - *summary, - service_tier.clone(), - collaboration_mode.clone(), - final_output_json_schema.clone(), - cyber_access_program.map(Into::into), - ) - .await?; - if app.active_thread_id == Some(thread_id) && app.chat_widget.thread_id() == Some(thread_id) { - crate::startup_recovery::acknowledged(client_user_message_id); - app.chat_widget - .record_safety_buffering_turn(response.turn.id, op); + TurnStartSettings { + approval_policy: turn_approval_policy, + approvals_reviewer: turn_approvals_reviewer, + permissions: permissions_override, + workspace_roots: config.permissions.user_visible_workspace_roots(), } - Ok(true) } diff --git a/codex-rs/tui/src/app_command.rs b/codex-rs/tui/src/app_command.rs index d67471c057..683e57be1a 100644 --- a/codex-rs/tui/src/app_command.rs +++ b/codex-rs/tui/src/app_command.rs @@ -94,6 +94,16 @@ impl Serialize for RealtimeSpeechText { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub(crate) enum CloudSubmissionIntent { + /// Use the TUI's known turn for explicit operations such as safety retries. + UseKnownTurn, + /// Enter lets the server choose whether to start or steer the current turn. + StartOrSteer, + /// A message leaving the local queue starts a fresh turn. + StartIfIdle, +} + #[allow(clippy::large_enum_variant)] #[derive(Debug, Clone, PartialEq, Serialize)] pub(crate) enum AppCommand { @@ -118,6 +128,7 @@ pub(crate) enum AppCommand { }, UserTurn { turn_trigger: &'static str, + cloud_submission: CloudSubmissionIntent, client_user_message_id: String, items: Vec, cwd: PathBuf, @@ -226,6 +237,7 @@ impl AppCommand { #[allow(clippy::too_many_arguments)] pub(crate) fn user_turn( turn_trigger: &'static str, + cloud_submission: CloudSubmissionIntent, client_user_message_id: String, items: Vec, cwd: PathBuf, @@ -240,6 +252,7 @@ impl AppCommand { ) -> Self { Self::UserTurn { turn_trigger, + cloud_submission, client_user_message_id, items, cwd, diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 0e698c17fa..49efbb5445 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -21,6 +21,7 @@ mod rollout_history; pub(crate) use read_state::unread_revision; mod startup_launch; mod thread_list; +pub(crate) mod user_message_submission; mod web_search; #[cfg(test)] diff --git a/codex-rs/tui/src/app_server_session/user_message_submission.rs b/codex-rs/tui/src/app_server_session/user_message_submission.rs new file mode 100644 index 0000000000..4e90d84843 --- /dev/null +++ b/codex-rs/tui/src/app_server_session/user_message_submission.rs @@ -0,0 +1,269 @@ +//! Route composer input and distinguish admission, rejection, and uncertainty. +//! A duplicate or an unconfirmed response must never trigger an automatic resend. + +use codex_app_server_client::AppServerRequestHandle; +use codex_app_server_client::CloudUserMessageReceipt; +use codex_app_server_client::TypedRequestError; +use codex_app_server_protocol::ApprovalsReviewer; +use codex_app_server_protocol::AskForApproval; +use codex_app_server_protocol::CodexErrorInfo as AppServerCodexErrorInfo; +use codex_app_server_protocol::TurnError as AppServerTurnError; +use codex_app_server_protocol::TurnStartParams; +use codex_protocol::ThreadId; +use codex_protocol::turn_input::CyberAccessProgram; +use codex_utils_absolute_path::AbsolutePathBuf; +use color_eyre::eyre::Report; + +use crate::app_command::AppCommand; +use crate::app_command::CloudSubmissionIntent; + +use super::AppServerSession; +use super::JSONRPC_INVALID_PARAMS; +use super::JSONRPC_INVALID_REQUEST; +use super::JSONRPC_METHOD_NOT_FOUND; +use super::ThreadParamsMode; +use super::TurnPermissionsOverride; +use super::UnsupportedLegacyPermissionProfile; + +// Both conventional transports return this before the request reaches its handler. +const JSONRPC_SERVER_OVERLOADED: i64 = -32001; + +pub(crate) struct TurnStartSettings<'a> { + pub(crate) approval_policy: Option, + pub(crate) approvals_reviewer: Option, + pub(crate) permissions: TurnPermissionsOverride, + pub(crate) workspace_roots: &'a [AbsolutePathBuf], +} + +pub(crate) enum UserMessageAdmission { + Started(String), + Steered(String), + /// The turn/start response confirms admission and does not distinguish start from steer. + StartResponse(String), + Duplicate, + Queued(String), +} + +#[derive(Debug)] +pub(crate) enum UserMessageSubmissionFailure { + Preflight(String), + Rejected(Report), + Unconfirmed { error: Report, reconnect: bool }, + Unhandled(Report), +} + +impl UserMessageSubmissionFailure { + fn conventional(error: Report, intent: CloudSubmissionIntent) -> Self { + if intent == CloudSubmissionIntent::UseKnownTurn + || super::model_provider_error(&error).is_some() + { + return Self::Unhandled(error); + } + if error + .downcast_ref::() + .is_some() + || matches!(error.downcast_ref::(), + Some(TypedRequestError::Server { source, .. }) + if matches!(source.code, JSONRPC_INVALID_REQUEST | JSONRPC_INVALID_PARAMS | JSONRPC_METHOD_NOT_FOUND | JSONRPC_SERVER_OVERLOADED)) + { + Self::Rejected(error) + } else { + Self::Unconfirmed { + error, + reconnect: true, + } + } + } +} + +impl AppServerSession { + pub(crate) async fn submit_user_message<'a>( + &mut self, + thread_id: ThreadId, + op: &AppCommand, + active_turn: &mut Option, + preflight: impl FnOnce() -> Result, String>, + start_settings: impl FnOnce() -> TurnStartSettings<'a>, + ) -> Result { + let AppCommand::UserTurn { + turn_trigger, + cloud_submission, + client_user_message_id, + items, + final_output_json_schema, + cwd, + model, + effort, + summary, + service_tier, + collaboration_mode, + .. + } = op + else { + unreachable!("user-message submission requires a user turn") + }; + let intent = if *turn_trigger == "user" { + *cloud_submission + } else { + CloudSubmissionIntent::UseKnownTurn + }; + let cloud = self.thread_params_mode() == ThreadParamsMode::Cloud; + if (!cloud || intent == CloudSubmissionIntent::UseKnownTurn) && active_turn.is_some() { + if *turn_trigger != "user" { + return Err(UserMessageSubmissionFailure::Unhandled( + color_eyre::eyre::eyre!("the fork already has an active turn"), + )); + } + let mut retried = false; + while let Some(turn_id) = active_turn.clone() { + let error = match self + .turn_steer( + thread_id, + turn_id.clone(), + client_user_message_id.clone(), + items.clone(), + ) + .await + { + Ok(response) => return Ok(UserMessageAdmission::Steered(response.turn_id)), + Err(error) => error, + }; + if let Some(error) = active_turn_not_steerable_turn_error(&error) { + return Ok(UserMessageAdmission::Queued(error.message)); + } + match active_turn_steer_race(&error) { + Some(ActiveTurnSteerRace::Missing) => { + *active_turn = None; + break; + } + Some(ActiveTurnSteerRace::ExpectedTurnMismatch { actual_turn_id }) => { + let retry = !retried && actual_turn_id != turn_id; + *active_turn = Some(actual_turn_id); + if retry { + retried = true; + continue; + } + } + None => {} + } + return Err(UserMessageSubmissionFailure::conventional( + error.into(), + intent, + )); + } + } + + let program = preflight().map_err(UserMessageSubmissionFailure::Preflight)?; + if cloud + && intent == CloudSubmissionIntent::StartOrSteer + && let AppServerRequestHandle::Cloud(handle) = &self.request_handle + { + let result = handle + .submit_user_message(TurnStartParams { + thread_id: thread_id.to_string(), + client_user_message_id: Some(client_user_message_id.clone()), + input: items.clone(), + output_schema: final_output_json_schema.clone(), + ..Default::default() + }) + .await; + return match result { + Ok(CloudUserMessageReceipt::Started(turn)) => { + Ok(UserMessageAdmission::Started(turn.id)) + } + Ok(CloudUserMessageReceipt::Steered(turn)) => { + Ok(UserMessageAdmission::Steered(turn.id)) + } + Ok(CloudUserMessageReceipt::Duplicate { .. }) => { + Ok(UserMessageAdmission::Duplicate) + } + // The adapter returns invalid params only for failures before admission. + Err(TypedRequestError::Server { source, .. }) + if source.code == JSONRPC_INVALID_PARAMS => + { + Err(UserMessageSubmissionFailure::Rejected(Report::msg( + source.message, + ))) + } + Err(error) => Err(UserMessageSubmissionFailure::Unconfirmed { + error: error.into(), + reconnect: false, + }), + }; + } + let settings = start_settings(); + self.turn_start( + thread_id, + turn_trigger, + client_user_message_id.clone(), + items.clone(), + cwd.clone(), + settings.approval_policy, + settings.approvals_reviewer, + settings.permissions, + settings.workspace_roots, + model.to_string(), + effort.clone(), + *summary, + service_tier.clone(), + collaboration_mode.clone(), + final_output_json_schema.clone(), + program.map(Into::into), + ) + .await + .map(|response| UserMessageAdmission::StartResponse(response.turn.id)) + .map_err( + |error| match UserMessageSubmissionFailure::conventional(error, intent) { + UserMessageSubmissionFailure::Rejected(error) => { + UserMessageSubmissionFailure::Rejected(error.wrap_err("Failed to start turn")) + } + failure => failure, + }, + ) + } +} + +pub(crate) fn active_turn_not_steerable_turn_error( + error: &TypedRequestError, +) -> Option { + let TypedRequestError::Server { source, .. } = error else { + return None; + }; + let turn_error: AppServerTurnError = serde_json::from_value(source.data.clone()?).ok()?; + matches!( + turn_error.codex_error_info, + Some(AppServerCodexErrorInfo::ActiveTurnNotSteerable { .. }) + ) + .then_some(turn_error) +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum ActiveTurnSteerRace { + Missing, + ExpectedTurnMismatch { actual_turn_id: String }, +} + +pub(crate) fn active_turn_steer_race(error: &TypedRequestError) -> Option { + let TypedRequestError::Server { method, source } = error else { + return None; + }; + if method != "turn/steer" { + return None; + } + if source.message == "no active turn to steer" { + return Some(ActiveTurnSteerRace::Missing); + } + + // App-server steer mismatches mean our cached active turn id is stale, but the response + // includes the server's current active turn so we can resynchronize and retry once. + let mismatch_prefix = "expected active turn id `"; + let mismatch_separator = "` but found `"; + let actual_turn_id = source + .message + .strip_prefix(mismatch_prefix)? + .split_once(mismatch_separator)? + .1 + .strip_suffix('`')? + .to_string(); + Some(ActiveTurnSteerRace::ExpectedTurnMismatch { actual_turn_id }) +} diff --git a/codex-rs/tui/src/chatwidget.rs b/codex-rs/tui/src/chatwidget.rs index cb44112c3f..01d781a7f8 100644 --- a/codex-rs/tui/src/chatwidget.rs +++ b/codex-rs/tui/src/chatwidget.rs @@ -286,6 +286,7 @@ mod reconnect; use self::input_queue::InputQueueState; mod image_submission; mod input_flow; +mod input_receipts; mod input_restore; mod input_submission; mod interrupts; @@ -1370,44 +1371,46 @@ impl ChatWidget { } // Servers may omit media from receipts, so prefer the submission identity. - if client_id.is_some() && self.last_rendered_user_message_client_id.as_deref() == client_id - { - return; - } - - if self - .input_queue - .pending_steers - .front() - .is_some_and(|pending| match client_id { - Some(client_id) => pending.client_id == client_id, - // Older app servers do not echo submission IDs. - None => pending.compare_key == Self::pending_steer_compare_key_from_items(items), - }) - { - if let Some(pending) = self.input_queue.pending_steers.pop_front() { - self.refresh_pending_input_preview(); - let pending_display = - user_message_display_for_history(pending.user_message, &pending.history_record); - self.on_user_message_display(pending_display); + let already_rendered = client_id.is_some() + && self.last_rendered_user_message_client_id.as_deref() == client_id; + let steers = &mut self.input_queue.pending_steers; + let matching = match client_id { + Some(id) => steers.iter().enumerate().position(|(index, pending)| { + pending.client_id == id + && (pending.admitted_turn_id.is_some() || (index == 0 && !already_rendered)) + }), + // Older app servers do not echo submission IDs. + None => steers.front().and_then(|pending| { + (pending.compare_key == Self::pending_steer_compare_key_from_items(items)) + .then_some(0) + }), + }; + if let Some(pending) = matching.and_then(|index| steers.remove(index)) { + self.refresh_pending_input_preview(); + if !already_rendered { + if !pending.optimistically_rendered { + let pending_display = user_message_display_for_history( + pending.user_message, + &pending.history_record, + ); + self.on_user_message_display(pending_display); + } // Later receipts carry the wire media, which may differ from the local attachment. self.last_rendered_user_message_display = Some(display); self.last_rendered_user_message_client_id = Some(pending.client_id); - } else if self.last_rendered_user_message_display.as_ref() != Some(&display) { - tracing::warn!( - "pending steer matched receipt but queue was empty when rendering committed user message" - ); - self.on_user_message_display(display); } - } else if !self.review.is_review_mode + } else if !already_rendered + && !self.review.is_review_mode && self.last_rendered_user_message_display.as_ref() != Some(&display) { 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 + .input_queue + .queued_user_messages + .iter() + .any(|message| message.unconfirmed_user_message_id() == Some(client_id)) { self.reconcile_recovered_messages(&[client_id.to_string()]); self.maybe_send_next_queued_input(); diff --git a/codex-rs/tui/src/chatwidget/input_flow.rs b/codex-rs/tui/src/chatwidget/input_flow.rs index fd7fb8f86c..1cbabcbf39 100644 --- a/codex-rs/tui/src/chatwidget/input_flow.rs +++ b/codex-rs/tui/src/chatwidget/input_flow.rs @@ -79,7 +79,12 @@ impl ChatWidget { self.reasoning_header = None; self.reasoning_summary_parts.clear(); self.set_status_header(String::from("Working")); - self.submit_user_message(user_message); + self.submit_user_message_with_history_and_shell_escape_policy( + user_message, + UserMessageHistoryRecord::UserMessageText, + ShellEscapePolicy::Allow, + UserMessageSource::DirectEnter, + ); } else { self.queue_user_message(user_message); } @@ -261,9 +266,14 @@ impl ChatWidget { } let mut submitted_follow_up = false; while !self.is_user_turn_pending_or_running() { - let Some((queued_message, history_record)) = self.pop_next_queued_user_message() else { + let Some((mut queued_message, history_record)) = self.pop_next_queued_user_message() + else { break; }; + queued_message.source = match queued_message.source { + UserMessageSource::DirectEnter => UserMessageSource::Prompt, + source => source, + }; match queued_message.action { QueuedInputAction::Plain => { let source = queued_message.source; @@ -348,7 +358,7 @@ impl ChatWidget { pub(crate) fn is_user_turn_pending_or_running(&self) -> bool { self.pending_image_submission.is_some() || self.input_queue.user_turn_pending_start - || self.turn_lifecycle.agent_turn_running + || self.turn_lifecycle.is_running_or_admitted() || self.review.is_review_mode || (self.bottom_pane.is_task_running() && self.mcp_startup_status.is_none()) } diff --git a/codex-rs/tui/src/chatwidget/input_queue.rs b/codex-rs/tui/src/chatwidget/input_queue.rs index bdaedab6ca..dc162e982e 100644 --- a/codex-rs/tui/src/chatwidget/input_queue.rs +++ b/codex-rs/tui/src/chatwidget/input_queue.rs @@ -34,6 +34,7 @@ pub(super) struct InputQueueState { /// 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, + pub(super) pending_user_message_history_record: Option, /// User messages that tried to steer a non-regular turn and must be retried first. pub(super) rejected_steers_queue: VecDeque, /// The origin of each rejected steer, kept in lockstep with its message. @@ -57,6 +58,14 @@ pub(super) struct InputQueueState { } impl InputQueueState { + pub(super) fn take_pending_steer(&mut self, client_id: &str) -> Option { + let index = self + .pending_steers + .iter() + .position(|pending| pending.client_id == client_id)?; + self.pending_steers.remove(index) + } + pub(super) fn submissions_paused(&self) -> bool { self.suppress_queue_autosend || self.transcript_copy.strong_count() > 0 } @@ -78,6 +87,7 @@ impl InputQueueState { self.queued_user_message_history_records.clear(); self.user_turn_pending_start = false; self.pending_user_message_client_id = None; + self.pending_user_message_history_record = None; self.rejected_steers_queue.clear(); self.rejected_steer_sources.clear(); self.rejected_steer_history_records.clear(); @@ -139,6 +149,8 @@ mod tests { .push_back(UserMessage::from("rejected")); state.pending_steers.push_back(PendingSteer { client_id: "test-submission".to_string(), + admitted_turn_id: None, + optimistically_rendered: false, user_message: UserMessage::from("pending"), history_record: UserMessageHistoryRecord::UserMessageText, source: UserMessageSource::Prompt, diff --git a/codex-rs/tui/src/chatwidget/input_receipts.rs b/codex-rs/tui/src/chatwidget/input_receipts.rs new file mode 100644 index 0000000000..34ffd72a29 --- /dev/null +++ b/codex-rs/tui/src/chatwidget/input_receipts.rs @@ -0,0 +1,128 @@ +//! Apply submission receipts and recover draft input by client message ID. +//! Ambiguous delivery stays out of automatic submission until history confirms it. + +use super::*; +use crate::app_server_session::user_message_submission::UserMessageAdmission; + +impl ChatWidget { + pub(crate) fn user_message_admitted( + &mut self, + receipt: UserMessageAdmission, + turn: &AppCommand, + client_id: &str, + ) { + if let UserMessageAdmission::Started(turn_id) | UserMessageAdmission::Steered(turn_id) = + &receipt + && (!self.turn_lifecycle.agent_turn_running + || self.last_turn_id() != Some(turn_id.as_str())) + { + self.turn_lifecycle.admitted_turn_id = Some(turn_id.clone()); + self.update_task_running_state(); + } + match receipt { + UserMessageAdmission::StartResponse(turn_id) => { + self.record_safety_buffering_turn(turn_id, turn) + } + UserMessageAdmission::Queued(message) => { + if !self.enqueue_rejected_steer_by_id(Some(client_id)) { + self.add_error_message(message); + } + } + UserMessageAdmission::Started(turn_id) => { + if let Some(pending) = self.input_queue.take_pending_steer(client_id) { + let display = user_message_display_for_history( + pending.user_message.clone(), + &pending.history_record, + ); + self.on_user_message_display(display); + self.last_rendered_user_message_client_id = Some(pending.client_id); + self.safety_buffering_prompt = Some(pending.user_message); + self.safety_buffering_source = pending.source; + self.refresh_pending_input_preview(); + } + self.record_safety_buffering_turn(turn_id, turn); + } + // A duplicate identifies the message, but only its history event confirms + // whether a steer committed. Keep it out of automatic delivery until then. + UserMessageAdmission::Duplicate => { + self.hold_unconfirmed_user_message(client_id); + } + UserMessageAdmission::Steered(turn_id) => { + if let Some(pending) = self + .input_queue + .pending_steers + .iter_mut() + .find(|pending| pending.client_id == client_id) + { + pending.admitted_turn_id = Some(turn_id); + } else if let AppCommand::UserTurn { items, .. } = turn + && let Some((message, history_record, source)) = + self.take_submitted_message(client_id) + { + self.input_queue.pending_steers.push_back(PendingSteer { + client_id: client_id.to_string(), + admitted_turn_id: Some(turn_id), + optimistically_rendered: true, + user_message: message, + history_record, + source, + compare_key: Self::pending_steer_compare_key_from_items(items), + }); + } + if self.input_queue.pending_user_message_client_id.as_deref() == Some(client_id) { + self.input_queue.user_turn_pending_start = false; + self.input_queue.pending_user_message_client_id = None; + } + self.refresh_pending_input_preview(); + } + } + } + + pub(super) fn take_submitted_message( + &mut self, + client_id: &str, + ) -> Option<(UserMessage, UserMessageHistoryRecord, UserMessageSource)> { + let queue = &mut self.input_queue; + if queue.pending_user_message_client_id.as_deref() == Some(client_id) { + queue.pending_user_message_client_id = None; + queue.user_turn_pending_start = false; + let history = queue + .pending_user_message_history_record + .take() + .unwrap_or(UserMessageHistoryRecord::UserMessageText); + self.safety_buffering_prompt + .take() + .map(|message| (message, history, self.safety_buffering_source)) + } else { + queue + .take_pending_steer(client_id) + .map(|p| (p.user_message, p.history_record, p.source)) + } + } + + pub(crate) fn hold_unconfirmed_user_message(&mut self, client_id: &str) { + if let Some((message, record, source)) = self.take_submitted_message(client_id) { + self.input_queue + .queued_user_messages + .push_front(QueuedUserMessage { + source, + delivery: MessageDelivery::Unconfirmed(Some(client_id.to_owned())), + ..message.into() + }); + self.input_queue + .queued_user_message_history_records + .push_front(record); + self.refresh_pending_input_preview(); + self.update_task_running_state(); + } + } + + pub(crate) fn restore_rejected_user_message(&mut self, client_id: &str, message: String) { + if let Some((draft, history, _)) = self.take_submitted_message(client_id) { + self.restore_user_message_to_composer(user_message_for_restore(draft, &history)); + } + self.add_error_message(message); + self.refresh_pending_input_preview(); + self.update_task_running_state(); + } +} diff --git a/codex-rs/tui/src/chatwidget/input_restore.rs b/codex-rs/tui/src/chatwidget/input_restore.rs index 4c00af8a49..6db9befb30 100644 --- a/codex-rs/tui/src/chatwidget/input_restore.rs +++ b/codex-rs/tui/src/chatwidget/input_restore.rs @@ -1,6 +1,7 @@ //! Input queue restore and thread-input snapshot behavior for `ChatWidget`. use std::collections::HashSet; +use std::collections::VecDeque; use codex_protocol::models::local_image_label_text; @@ -344,23 +345,32 @@ impl ChatWidget { } } - pub(crate) fn enqueue_rejected_steer(&mut self) -> bool { - let Some(pending_steer) = self.input_queue.pending_steers.pop_front() else { + pub(super) fn enqueue_rejected_steer(&mut self) -> bool { + self.enqueue_rejected_steer_by_id(/*client_id*/ None) + } + + pub(super) fn enqueue_rejected_steer_by_id(&mut self, client_id: Option<&str>) -> bool { + let pending = match client_id { + Some(id) => self.take_submitted_message(id), + None => self + .input_queue + .pending_steers + .pop_front() + .map(|p| (p.user_message, p.history_record, p.source)), + }; + let Some((message, history, source)) = pending else { tracing::warn!( "received active-turn-not-steerable error without a matching pending steer" ); return false; }; - self.input_queue - .rejected_steers_queue - .push_back(pending_steer.user_message); - self.input_queue - .rejected_steer_sources - .push_back(pending_steer.source); + self.input_queue.rejected_steers_queue.push_back(message); + self.input_queue.rejected_steer_sources.push_back(source); self.input_queue .rejected_steer_history_records - .push_back(pending_steer.history_record); + .push_back(history); self.refresh_pending_input_preview(); + self.update_task_running_state(); true } @@ -368,6 +378,49 @@ impl ChatWidget { /// or review completion. /// When there are queued user messages, restore them into the composer /// separated by newlines rather than auto-submitting the next one. + pub(super) fn on_interrupted_server_turn(&mut self, reason: TurnAbortReason, turn_id: &str) { + let queue = &mut self.input_queue; + // Recovery for the interrupted turn must leave other admitted steers and + // unresolved submissions alone until their own events arrive. + let (later, interrupted): (VecDeque<_>, _) = std::mem::take(&mut queue.pending_steers) + .into_iter() + .partition(|steer| { + steer + .admitted_turn_id + .as_deref() + .is_some_and(|id| id != turn_id) + }); + queue.pending_steers = interrupted; + let mut unconfirmed = Vec::new(); + if queue.has_unconfirmed_messages() { + let history = std::mem::take(&mut queue.queued_user_message_history_records); + for (message, record) in std::mem::take(&mut queue.queued_user_messages) + .into_iter() + .zip( + history + .into_iter() + .chain(std::iter::repeat(UserMessageHistoryRecord::UserMessageText)), + ) + { + if matches!(message.delivery, MessageDelivery::Unconfirmed(_)) { + unconfirmed.push((message, record)); + } else { + queue.queued_user_messages.push_back(message); + queue.queued_user_message_history_records.push_back(record); + } + } + } + self.on_interrupted_turn(reason); + self.input_queue.pending_steers.extend(later); + for (message, record) in unconfirmed { + self.input_queue.queued_user_messages.push_back(message); + self.input_queue + .queued_user_message_history_records + .push_back(record); + } + self.refresh_pending_input_preview(); + } + pub(super) fn on_interrupted_turn(&mut self, reason: TurnAbortReason) { self.requeue_image_submission(); // Finalize, log a gentle prompt, and clear running state. @@ -731,7 +784,9 @@ 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 { + delivery: if input_state.reconnect_pending + || pending.admitted_turn_id.is_some() + { MessageDelivery::Unconfirmed(Some(pending.client_id)) } else { MessageDelivery::Unsent diff --git a/codex-rs/tui/src/chatwidget/input_submission.rs b/codex-rs/tui/src/chatwidget/input_submission.rs index 7d0c7403b5..c616326e52 100644 --- a/codex-rs/tui/src/chatwidget/input_submission.rs +++ b/codex-rs/tui/src/chatwidget/input_submission.rs @@ -165,9 +165,11 @@ impl ChatWidget { || self.pending_image_submission.is_some() || self.safety_recourse.is_some() { - let model_prompt = source == UserMessageSource::Prompt - && (shell_escape_policy == ShellEscapePolicy::Disallow - || !user_message.text.starts_with('!')); + let model_prompt = matches!( + source, + UserMessageSource::Prompt | UserMessageSource::DirectEnter + ) && (shell_escape_policy == ShellEscapePolicy::Disallow + || !user_message.text.starts_with('!')); // A prepared submission was accepted before any messages queued behind it. let queue_index = if prepared_images.is_some() { 0 @@ -198,9 +200,11 @@ impl ChatWidget { return (true, None); } if !self.is_session_configured() { - let model_prompt = source == UserMessageSource::Prompt - && (shell_escape_policy == ShellEscapePolicy::Disallow - || !user_message.text.starts_with('!')); + let model_prompt = matches!( + source, + UserMessageSource::Prompt | UserMessageSource::DirectEnter + ) && (shell_escape_policy == ShellEscapePolicy::Disallow + || !user_message.text.starts_with('!')); tracing::warn!("cannot submit user message before session is configured; queueing"); self.input_queue .queued_user_messages @@ -297,6 +301,17 @@ impl ChatWidget { })); } + use crate::app_command::CloudSubmissionIntent; + let (cloud_submission, source) = match source { + UserMessageSource::DirectEnter => ( + CloudSubmissionIntent::StartOrSteer, + UserMessageSource::Prompt, + ), + UserMessageSource::Prompt => (CloudSubmissionIntent::StartIfIdle, source), + UserMessageSource::QuestionAnswer | UserMessageSource::SafetyRecourse => { + (CloudSubmissionIntent::UseKnownTurn, source) + } + }; if !text.is_empty() { items.push(UserInput::Text { text: text.clone(), @@ -462,6 +477,8 @@ impl ChatWidget { crate::startup_recovery::bind_submission(&text, &client_user_message_id); let pending_steer = (!render_in_history).then(|| PendingSteer { client_id: client_user_message_id.clone(), + admitted_turn_id: None, + optimistically_rendered: false, user_message: UserMessage { text: text.clone(), local_images: local_images.clone(), @@ -481,6 +498,7 @@ impl ChatWidget { } else { "user" }, + cloud_submission, client_user_message_id.clone(), items, self.config.cwd.to_path_buf(), @@ -530,6 +548,7 @@ impl ChatWidget { 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()); + self.input_queue.pending_user_message_history_record = Some(history_record.clone()); } // Persist the submitted text to cross-session message history. Mentions are encoded into diff --git a/codex-rs/tui/src/chatwidget/protocol.rs b/codex-rs/tui/src/chatwidget/protocol.rs index 08a4731803..4f1746eb3d 100644 --- a/codex-rs/tui/src/chatwidget/protocol.rs +++ b/codex-rs/tui/src/chatwidget/protocol.rs @@ -114,6 +114,10 @@ impl ChatWidget { self.on_replayed_task_started(notification.turn.started_at); } else { self.on_task_started(); + if self.turn_lifecycle.admitted_turn_id == self.turn_lifecycle.last_turn_id + { + self.turn_lifecycle.admitted_turn_id = None; + } } } } @@ -447,6 +451,13 @@ impl ChatWidget { notification: TurnCompletedNotification, replay_kind: Option, ) { + if replay_kind.is_none() + && notification.turn.status != TurnStatus::InProgress + && self.turn_lifecycle.admitted_turn_id.as_deref() + == Some(notification.turn.id.as_str()) + { + self.turn_lifecycle.admitted_turn_id = None; + } // User-message dedupe only suppresses the app-server echo of a prompt // this TUI already rendered locally. Once that turn ends, another // client can submit the same text and it still needs its own user cell. @@ -506,7 +517,7 @@ impl ChatWidget { } else { TurnAbortReason::Interrupted }; - self.on_interrupted_turn(reason); + self.on_interrupted_server_turn(reason, ¬ification.turn.id); } TurnStatus::Failed => { if replay_kind.is_none() { diff --git a/codex-rs/tui/src/chatwidget/reconnect.rs b/codex-rs/tui/src/chatwidget/reconnect.rs index d2608b1226..58d5683948 100644 --- a/codex-rs/tui/src/chatwidget/reconnect.rs +++ b/codex-rs/tui/src/chatwidget/reconnect.rs @@ -126,19 +126,11 @@ impl ChatWidget { } 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 - }); + super::user_messages::remove_confirmed_queued_messages( + &mut self.input_queue.queued_user_messages, + &mut self.input_queue.queued_user_message_history_records, + confirmed_message_ids, + ); self.refresh_pending_input_preview(); } diff --git a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs index 88222d0ec0..6084535a89 100644 --- a/codex-rs/tui/src/chatwidget/tests/composer_submission.rs +++ b/codex-rs/tui/src/chatwidget/tests/composer_submission.rs @@ -1894,6 +1894,51 @@ async fn pending_steer_interrupt_uses_remapped_binding() { assert!(chat.input_queue.submit_pending_steers_after_interrupt); } +#[tokio::test] +async fn resumed_steers_reconcile_by_client_id_and_hold_unmatched_admissions() { + let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; + chat.thread_id = Some(ThreadId::new()); + // Both messages have the same text; only one has a receipt in the resumed history. + chat.input_queue.pending_steers = VecDeque::from([ + PendingSteer { + client_id: "confirmed".into(), + admitted_turn_id: Some("finished".into()), + ..pending_steer("same text") + }, + PendingSteer { + client_id: "unmatched".into(), + admitted_turn_id: Some("finished".into()), + ..pending_steer("same text") + }, + ]); + chat.input_queue + .queued_user_messages + .push_back(UserMessage::from("next draft").into()); + chat.input_queue + .queued_user_message_history_records + .push_back(UserMessageHistoryRecord::UserMessageText); + let mut input = chat + .capture_thread_input_state() + .expect("saved session input"); + input.reconcile_confirmed_messages(&["confirmed".into()]); + chat.restore_thread_input_state( + Some(input), + ThreadInputStateRestoreMode { + preserve_in_flight_turn: false, + }, + ); + assert_eq!( + chat.queued_user_message_texts(), + vec!["same text", "next draft"] + ); + assert_eq!( + chat.input_queue.queued_user_messages[0].delivery, + MessageDelivery::Unconfirmed(Some("unmatched".into())) + ); + assert!(!chat.maybe_send_next_queued_input()); + assert_no_submit_op(&mut op_rx); +} + #[tokio::test] async fn restore_thread_input_state_applies_running_state_policy() { let (mut chat, _rx, mut op_rx) = make_chatwidget_manual(/*model_override*/ None).await; diff --git a/codex-rs/tui/src/chatwidget/tests/helpers.rs b/codex-rs/tui/src/chatwidget/tests/helpers.rs index 02615bad64..fd9e614e10 100644 --- a/codex-rs/tui/src/chatwidget/tests/helpers.rs +++ b/codex-rs/tui/src/chatwidget/tests/helpers.rs @@ -1099,6 +1099,8 @@ pub(super) fn complete_assistant_message( pub(super) fn pending_steer(text: &str) -> PendingSteer { PendingSteer { client_id: "test-submission".to_string(), + admitted_turn_id: None, + optimistically_rendered: false, user_message: UserMessage::from(text), history_record: UserMessageHistoryRecord::UserMessageText, source: UserMessageSource::Prompt, diff --git a/codex-rs/tui/src/chatwidget/turn_lifecycle.rs b/codex-rs/tui/src/chatwidget/turn_lifecycle.rs index 67ced849ad..0562da3a63 100644 --- a/codex-rs/tui/src/chatwidget/turn_lifecycle.rs +++ b/codex-rs/tui/src/chatwidget/turn_lifecycle.rs @@ -11,6 +11,8 @@ pub(super) struct TurnLifecycleState { /// Tracks whether codex-core currently considers an agent turn to be in progress. pub(super) agent_turn_running: bool, pub(super) last_turn_id: Option, + /// Cloud confirmed admission to a turn whose live start we have yet to process. + pub(super) admitted_turn_id: Option, pub(super) budget_limited_turn_ids: HashSet, /// Completion labels already inserted into this thread's visible history. pub(super) rendered_completion_turn_ids: HashSet, @@ -23,6 +25,7 @@ impl TurnLifecycleState { sleep_inhibitor: SleepInhibitor::new(prevent_idle_sleep), agent_turn_running: false, last_turn_id: None, + admitted_turn_id: None, budget_limited_turn_ids: HashSet::new(), rendered_completion_turn_ids: HashSet::new(), goal_status_active_turn_started_at: None, @@ -42,6 +45,10 @@ impl TurnLifecycleState { .set_turn_running(/*turn_running*/ false); } + pub(super) fn is_running_or_admitted(&self) -> bool { + self.agent_turn_running || self.admitted_turn_id.is_some() + } + pub(super) fn restore_running(&mut self, running: bool, now: Instant) { self.agent_turn_running = running; self.goal_status_active_turn_started_at = running.then_some(now); @@ -51,6 +58,7 @@ impl TurnLifecycleState { pub(super) fn reset_thread(&mut self) { self.finish(); self.last_turn_id = None; + self.admitted_turn_id = None; self.budget_limited_turn_ids.clear(); self.rendered_completion_turn_ids.clear(); } diff --git a/codex-rs/tui/src/chatwidget/turn_runtime.rs b/codex-rs/tui/src/chatwidget/turn_runtime.rs index ef5954a853..566668d8b4 100644 --- a/codex-rs/tui/src/chatwidget/turn_runtime.rs +++ b/codex-rs/tui/src/chatwidget/turn_runtime.rs @@ -44,7 +44,7 @@ impl ChatWidget { /// both the agent turn lifecycle and MCP startup lifecycle. pub(super) fn update_task_running_state(&mut self) { self.bottom_pane.set_task_running( - self.turn_lifecycle.agent_turn_running + self.turn_lifecycle.is_running_or_admitted() || self.review.is_review_mode || self.mcp_startup_status.is_some(), ); diff --git a/codex-rs/tui/src/chatwidget/user_messages.rs b/codex-rs/tui/src/chatwidget/user_messages.rs index 9fa654fb1f..d11c9bbd2c 100644 --- a/codex-rs/tui/src/chatwidget/user_messages.rs +++ b/codex-rs/tui/src/chatwidget/user_messages.rs @@ -64,6 +64,7 @@ pub(super) enum ShellEscapePolicy { #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(super) enum UserMessageSource { Prompt, + DirectEnter, QuestionAnswer, SafetyRecourse, } @@ -86,6 +87,13 @@ pub(super) struct QueuedUserMessage { } impl QueuedUserMessage { + pub(super) fn unconfirmed_user_message_id(&self) -> Option<&str> { + match &self.delivery { + MessageDelivery::Unconfirmed(id) => id.as_deref(), + MessageDelivery::Unsent => None, + } + } + pub(super) fn new(user_message: UserMessage, action: QueuedInputAction) -> Self { Self { user_message, @@ -209,6 +217,9 @@ impl From<&str> for UserMessage { pub(crate) struct PendingSteer { /// Preserved across request retries and thread switches until this submission commits. pub(crate) client_id: String, + /// The admission identifies the turn whose interruption can discard this input. + pub(super) admitted_turn_id: Option, + pub(super) optimistically_rendered: bool, pub(super) user_message: UserMessage, pub(super) history_record: UserMessageHistoryRecord, pub(super) source: UserMessageSource, @@ -852,7 +863,36 @@ impl ChatWidget { } } +pub(super) fn remove_confirmed_queued_messages( + messages: &mut VecDeque, + history: &mut VecDeque, + confirmed_ids: &[String], +) { + let mut index = 0; + messages.retain(|message| { + let confirmed = message + .unconfirmed_user_message_id() + .is_some_and(|id| confirmed_ids.iter().any(|confirmed_id| confirmed_id == id)); + if confirmed { + history.remove(index); + } else { + index += 1; + } + !confirmed + }); +} + impl ThreadInputState { + pub(crate) fn reconcile_confirmed_messages(&mut self, confirmed_ids: &[String]) { + self.pending_steers + .retain(|steer| !confirmed_ids.contains(&steer.client_id)); + remove_confirmed_queued_messages( + &mut self.queued_user_messages, + &mut self.queued_user_message_history_records, + confirmed_ids, + ); + } + pub(crate) fn has_unconfirmed_messages(&self) -> bool { self.queued_user_messages .iter()