From e1cc96712f9ad4f5404fe821b9eba7ff29598816 Mon Sep 17 00:00:00 2001 From: Benjamin Carlsson Date: Mon, 9 Nov 2026 21:34:54 +0000 Subject: [PATCH] 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 --- .../app/tests/session_lifecycle_requests.rs | 114 ++++++++++++++++++ codex-rs/tui/src/app/user_turn_submission.rs | 32 ++++- codex-rs/tui/src/app_command.rs | 2 + .../user_message_submission.rs | 7 +- .../tui/src/chatwidget/input_submission.rs | 10 +- 5 files changed, 153 insertions(+), 12 deletions(-) diff --git a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs b/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs index 3040e2d921..5d13cf3f41 100644 --- a/codex-rs/tui/src/app/tests/session_lifecycle_requests.rs +++ b/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; diff --git a/codex-rs/tui/src/app/user_turn_submission.rs b/codex-rs/tui/src/app/user_turn_submission.rs index d03ed28538..5b13691c0b 100644 --- a/codex-rs/tui/src/app/user_turn_submission.rs +++ b/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); } diff --git a/codex-rs/tui/src/app_command.rs b/codex-rs/tui/src/app_command.rs index 683e57be1a..82034eea3b 100644 --- a/codex-rs/tui/src/app_command.rs +++ b/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)] diff --git a/codex-rs/tui/src/app_server_session/user_message_submission.rs b/codex-rs/tui/src/app_server_session/user_message_submission.rs index 4e90d84843..d213a7104d 100644 --- a/codex-rs/tui/src/app_server_session/user_message_submission.rs +++ b/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, preflight: impl FnOnce() -> Result, String>, start_settings: impl FnOnce() -> TurnStartSettings<'a>, ) -> Result { let AppCommand::UserTurn { turn_trigger, - cloud_submission, client_user_message_id, items, final_output_json_schema, @@ -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" { diff --git a/codex-rs/tui/src/chatwidget/input_submission.rs b/codex-rs/tui/src/chatwidget/input_submission.rs index c616326e52..a7d2bbdf71 100644 --- a/codex-rs/tui/src/chatwidget/input_submission.rs +++ b/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 {