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 {