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

codex-rs/app-server-client/src/cloud_backend/mod.rs

@@ -26,6 +26,10 @@ type ReplyStream = BoxStream<'static, Result<cloud::Event, cloud::Error>>;
 #[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<AppServerEvent>,
-    ) -> 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::<PreflightRejection>()
-                    || matches!(
-                        error.downcast_ref::<cloud::MutationError>(),
-                        Some(cloud::MutationError::InvalidRequest)
-                    ))
+            } else if error.is::<PreflightRejection>()
+                || user_message_submission
+                    && (error.is::<MissingReplyAttachment>()
+                        || matches!(
+                            error.downcast_ref::<cloud::MutationError>(),
+                            Some(cloud::MutationError::InvalidRequest)
+                        ))
             {
                 -32602
             } else {

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

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<AppServerTurnError> {
-    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<ActiveTurnSteerRace> {
-    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<String> {
     let TypedRequestError::Server { method, source } = error else {
         return None;

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<codex_app_server_protocol::ThreadSettingsUpdatedNotification>,
     ) {
         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 {

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(),

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(

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;
 

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(())
 }
 

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(),

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::<Vec<_>>()
+            .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<Mutex<Vec<JSONRPCRequest>>>;
 pub(super) type RecordingAppServer = (AppServerSession, RecordedRequests, JoinHandle<Result<()>>);
 

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… (<elapsed>)
 
 • Queued follow-up inputs
+  ↳ only once
   ↳ queued before disconnect
     ⇧← edit last queued message
 

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

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;

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<Item = &'a ThreadBufferedEvent>,
+) -> Vec<String> {
+    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(&notification.item.payload)
+            } else if let ServerNotification::ItemCompleted(notification) = notification.as_ref() {
+                Some(&notification.item.payload)
+            } else {
+                None
+            }
+        }))
+        .filter_map(|item| {
+            if let ThreadItemPayload::UserMessage { client_id, .. } = item {
+                client_id.clone()
+            } else {
+                None
+            }
+        })
+        .collect::<Vec<_>>()
+}
+
 fn is_voice_handoff_item(item: &ThreadItemPayload) -> bool {
     matches!(item, ThreadItemPayload::UserMessage { content, .. }
         if crate::chatwidget::realtime_delegation_input(content).is_some())

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(&notification.item.payload)
-                } else if let ServerNotification::ItemCompleted(notification) =
-                    notification.as_ref()
-                {
-                    Some(&notification.item.payload)
-                } else {
-                    None
-                }
-            }))
-            .filter_map(|item| {
-                if let ThreadItemPayload::UserMessage { client_id, .. } = item {
-                    client_id.clone()
-                } else {
-                    None
-                }
-            })
-            .collect::<Vec<_>>();
+        let confirmed_message_ids = super::thread_events::confirmed_user_message_ids(
+            &snapshot.turns,
+            snapshot.events.iter(),
+        );
         let recovered_input = snapshot
             .input_state
             .as_ref()

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<bool> {
     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::<UnsupportedLegacyPermissionProfile>()
+                .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)
 }

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<UserInput>,
         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<UserInput>,
         cwd: PathBuf,
@@ -240,6 +252,7 @@ impl AppCommand {
     ) -> Self {
         Self::UserTurn {
             turn_trigger,
+            cloud_submission,
             client_user_message_id,
             items,
             cwd,

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)]

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<AskForApproval>,
+    pub(crate) approvals_reviewer: Option<ApprovalsReviewer>,
+    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::<UnsupportedLegacyPermissionProfile>()
+            .is_some()
+            || matches!(error.downcast_ref::<TypedRequestError>(),
+                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<String>,
+        preflight: impl FnOnce() -> Result<Option<CyberAccessProgram>, String>,
+        start_settings: impl FnOnce() -> TurnStartSettings<'a>,
+    ) -> Result<UserMessageAdmission, UserMessageSubmissionFailure> {
+        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<AppServerTurnError> {
+    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<ActiveTurnSteerRace> {
+    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 })
+}

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();

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())
     }

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<String>,
+    pub(super) pending_user_message_history_record: Option<UserMessageHistoryRecord>,
     /// 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.
@@ -57,6 +58,14 @@ pub(super) struct InputQueueState {
 }
 
 impl InputQueueState {
+    pub(super) fn take_pending_steer(&mut self, client_id: &str) -> Option<PendingSteer> {
+        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,

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();
+    }
+}

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

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

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<ReplayKind>,
     ) {
+        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, &notification.turn.id);
             }
             TurnStatus::Failed => {
                 if replay_kind.is_none() {

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();
     }
 

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;

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,

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<String>,
+    /// Cloud confirmed admission to a turn whose live start we have yet to process.
+    pub(super) admitted_turn_id: Option<String>,
     pub(super) budget_limited_turn_ids: HashSet<String>,
     /// Completion labels already inserted into this thread's visible history.
     pub(super) rendered_completion_turn_ids: HashSet<String>,
@@ -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();
     }

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(),
         );

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<String>,
+    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<QueuedUserMessage>,
+    history: &mut VecDeque<UserMessageHistoryRecord>,
+    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()