Route TUI question answers through Cloud message admission (#60462)

## What changed

Submit asynchronous question answers through Cloud's start-or-steer admission path when a turn is running, and use start-if-idle submission otherwise. Resolve the submission intent against the session mode so local and remote sessions retain their existing behavior.

Apply normal message admission handling to Cloud answers, clearing accepted replies and restoring rejected replies to the composer.

## Testing

Add a regression test covering accepted and rejected Cloud question replies, including the submitted answer and client message ID, composer state, and absence of extra requests.

GitOrigin-RevId: 757556e8a11bc148a92cde8ca05fe06aa291ee2c

codex-rs/tui/src/app/tests/session_lifecycle_requests.rs

@@ -200,6 +200,120 @@ async fn cloud_enter_recovers_drafts_from_server_receipts() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn cloud_async_question_reply_uses_cloud_admission() -> Result<()> {
+    use codex_app_server_client::CloudOperation;
+    use codex_app_server_client::CloudUserMessageReceipt;
+    use codex_app_server_protocol::AsyncUserInputQuestion;
+    use codex_app_server_protocol::ItemCompletedNotification;
+    use codex_context_fragments::AnsweredQuestion;
+    use codex_context_fragments::ContextualUserFragment;
+    use codex_protocol_test_support::ContentItemTestExt;
+
+    let title = "Which region?";
+    let answer = "Europe";
+    let question_id = serde_json::json!(["request_user_input_async", "question", 0]).to_string();
+    let input = vec![UserInput::Text {
+        text: AnsweredQuestion::new(&question_id, title, answer)
+            .render()
+            .expect_text(),
+        text_elements: Vec::new(),
+    }];
+
+    for rejected in [false, true] {
+        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.ensure_thread_channel(id)
+            .store
+            .lock()
+            .await
+            .set_active_turn_id("active".into());
+        let chat = &mut app.chat_widget;
+        chat.handle_server_notification(
+            turn_started_notification(id, "active"),
+            /*replay_kind*/ None,
+        );
+        chat.handle_server_notification(
+            ServerNotification::ItemCompleted(ItemCompletedNotification::new(
+                id.to_string(),
+                "active".into(),
+                ThreadItemPayload::AgentMessage {
+                    moderation: None,
+                    id: "question".into(),
+                    text: title.into(),
+                    phase: None,
+                    memory_citation: None,
+                    delivery: None,
+                    questions: Some(vec![AsyncUserInputQuestion {
+                        title: title.into(),
+                        options: None,
+                    }]),
+                },
+                /*started_at_ms*/ None,
+                /*completed_at_ms*/ 1,
+            )),
+            /*replay_kind*/ None,
+        );
+        chat.handle_key_event(KeyEvent::new(KeyCode::Up, KeyModifiers::ALT));
+        chat.handle_paste(answer.into());
+        chat.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!();
+        };
+        let expected_id = client_user_message_id.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 (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!("a Cloud question reply should use native start-or-steer admission");
+            };
+            assert_eq!(params.thread_id, id.to_string());
+            assert_eq!(params.client_user_message_id, Some(expected_id));
+            assert_eq!(params.input, input);
+            let receipt = if rejected {
+                Err(JSONRPCErrorError {
+                    code: -32602,
+                    message: "reply stream closed before submission".into(),
+                    data: None,
+                })
+            } else {
+                let turn = test_turn("active", TurnStatus::InProgress, Vec::new());
+                Ok(serde_json::to_value(CloudUserMessageReceipt::Steered(turn)).unwrap())
+            };
+            command.response.send(Ok(receipt)).unwrap();
+        });
+        result?;
+        let composer = app.chat_widget.composer_text_with_pending();
+        if rejected {
+            insta::assert_snapshot!(composer, @r"
+            > Which region?
+
+            Europe
+            ");
+        } else {
+            assert_eq!(composer, "");
+        }
+        assert!(commands.try_recv().is_err());
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn cloud_resume_picker_reconciles_admitted_steers() -> Result<()> {
     use crate::app::agents_overview::tests::overview_thread;

codex-rs/tui/src/app/user_turn_submission.rs

@@ -25,6 +25,26 @@ pub(super) async fn try_submit_user_turn(
     else {
         return Ok(false);
     };
+    let intent = if *turn_trigger == "user" {
+        match (*cloud_submission, app_server.thread_params_mode()) {
+            (
+                CloudSubmissionIntent::QuestionAnswer { during_turn },
+                crate::ThreadParamsMode::Cloud,
+            ) => {
+                if during_turn {
+                    CloudSubmissionIntent::StartOrSteer
+                } else {
+                    CloudSubmissionIntent::StartIfIdle
+                }
+            }
+            (CloudSubmissionIntent::QuestionAnswer { .. }, _) => {
+                CloudSubmissionIntent::UseKnownTurn
+            }
+            (intent, _) => intent,
+        }
+    } else {
+        CloudSubmissionIntent::UseKnownTurn
+    };
     let original_active = app.active_turn_id_for_thread(thread_id).await;
     let mut active = original_active.clone();
     let preflight = || {
@@ -40,7 +60,14 @@ pub(super) async fn try_submit_user_turn(
     };
     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)
+        .submit_user_message(
+            thread_id,
+            op,
+            intent,
+            &mut active,
+            preflight,
+            start_settings,
+        )
         .await;
     if active != original_active
         && let Some(channel) = app.thread_event_channels.get(&thread_id)
@@ -55,8 +82,7 @@ pub(super) async fn try_submit_user_turn(
         }
     }
 
-    let ordinary =
-        *turn_trigger == "user" && *cloud_submission != CloudSubmissionIntent::UseKnownTurn;
+    let ordinary = intent != CloudSubmissionIntent::UseKnownTurn;
     if let Err(UserMessageSubmissionFailure::Unhandled(error)) = result {
         return Err(error);
     }

codex-rs/tui/src/app_command.rs

@@ -102,6 +102,8 @@ pub(crate) enum CloudSubmissionIntent {
     StartOrSteer,
     /// A message leaving the local queue starts a fresh turn.
     StartIfIdle,
+    /// Use Cloud admission for answers, with steering enabled during a running turn.
+    QuestionAnswer { during_turn: bool },
 }
 
 #[allow(clippy::large_enum_variant)]

codex-rs/tui/src/app_server_session/user_message_submission.rs

@@ -81,13 +81,13 @@ impl AppServerSession {
         &mut self,
         thread_id: ThreadId,
         op: &AppCommand,
+        intent: CloudSubmissionIntent,
         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,
@@ -102,11 +102,6 @@ impl AppServerSession {
         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" {

codex-rs/tui/src/chatwidget/input_submission.rs

@@ -308,9 +308,13 @@ impl ChatWidget {
                 UserMessageSource::Prompt,
             ),
             UserMessageSource::Prompt => (CloudSubmissionIntent::StartIfIdle, source),
-            UserMessageSource::QuestionAnswer | UserMessageSource::SafetyRecourse => {
-                (CloudSubmissionIntent::UseKnownTurn, source)
-            }
+            UserMessageSource::QuestionAnswer => (
+                CloudSubmissionIntent::QuestionAnswer {
+                    during_turn: !render_in_history,
+                },
+                source,
+            ),
+            UserMessageSource::SafetyRecourse => (CloudSubmissionIntent::UseKnownTurn, source),
         };
         if !text.is_empty() {
             items.push(UserInput::Text {