diff --git a/codex-rs/tui/src/app/agents_overview.rs b/codex-rs/tui/src/app/agents_overview.rs index 91baa6991e..ed8b8f0081 100644 --- a/codex-rs/tui/src/app/agents_overview.rs +++ b/codex-rs/tui/src/app/agents_overview.rs @@ -18,6 +18,7 @@ mod selection; use super::agents_overview_view::AgentsOverviewGroup; use super::agents_overview_view::AgentsOverviewRow; use super::agents_overview_view::AgentsOverviewView; +use super::app_server_requests::BufferedServerRequest; use super::session_lifecycle::ThreadAttachPresentation; use super::*; use crate::app_event::AgentsOverviewThreadRefresh; @@ -76,7 +77,7 @@ pub(super) struct AgentsOverviewState { pub(super) blank_sessions: HashMap, pub(super) input_states: HashMap, pub(super) new_session_draft: Option>, - pub(super) dispatched_requests: HashMap>, + pub(super) dispatched_requests: HashMap>, } impl Drop for AgentsOverviewState { @@ -501,7 +502,13 @@ impl App { .input_states .insert(*thread_id, input_state); } - let requests = store.pending_replay_requests(); + let requests: Vec<_> = store + .pending_replay_requests() + .into_iter() + .filter_map(|request| { + self.pending_app_server_requests.buffered_request(request) + }) + .collect(); if !requests.is_empty() { previous_pending_requests.push((*thread_id, requests)); } @@ -915,11 +922,7 @@ impl App { .flat_map(|(_, requests)| requests) .collect::>(); for request in requests { - self.handle_app_server_event( - app_server, - codex_app_server_client::AppServerEvent::ServerRequest(Box::new(request)), - ) - .await; + self.handle_server_request_event(app_server, request).await; } } diff --git a/codex-rs/tui/src/app/agents_overview_details.rs b/codex-rs/tui/src/app/agents_overview_details.rs index 03b3c9b14c..1a86a59098 100644 --- a/codex-rs/tui/src/app/agents_overview_details.rs +++ b/codex-rs/tui/src/app/agents_overview_details.rs @@ -230,7 +230,11 @@ impl App { self.agents_overview .dispatched_requests .get(&thread_id) - .and_then(|requests| requests.iter().find_map(request_preview)) + .and_then(|requests| { + requests + .iter() + .find_map(|request| request_preview(&request.request)) + }) .or_else(|| { let channel = self .thread_event_channels diff --git a/codex-rs/tui/src/app/agents_overview_tests.rs b/codex-rs/tui/src/app/agents_overview_tests.rs index f5313a25c3..78c018f14d 100644 --- a/codex-rs/tui/src/app/agents_overview_tests.rs +++ b/codex-rs/tui/src/app/agents_overview_tests.rs @@ -1,6 +1,7 @@ #[path = "agents_overview_discovery_tests.rs"] mod discovery; +use crate::app::app_server_requests::BufferedServerRequest; #[path = "agent_center_tests.rs"] mod command_center; @@ -837,7 +838,10 @@ async fn agents_overview_details_show_available_attention_without_expanding_rows is_blocking: true, auto_resolution_ms: None, }, - }], + }] + .into_iter() + .map(|request| BufferedServerRequest::new(request, &app_server.request_handle())) + .collect(), ); let view = app.agents_overview_view(threads.clone(), Some(root)); app.agents_overview.visible_thread_ids = view.thread_ids(); @@ -2547,7 +2551,10 @@ async fn command_center_attach_conflict_opens_read_only_and_retries() -> Result< is_blocking: true, auto_resolution_ms: None, }, - }], + }] + .into_iter() + .map(|request| BufferedServerRequest::new(request, &server.request_handle())) + .collect(), ); let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel(); app.app_event_tx = AppEventSender::new(tx); diff --git a/codex-rs/tui/src/app/app_server_events.rs b/codex-rs/tui/src/app/app_server_events.rs index 4ceb678804..2821d6d242 100644 --- a/codex-rs/tui/src/app/app_server_events.rs +++ b/codex-rs/tui/src/app/app_server_events.rs @@ -6,6 +6,7 @@ use super::ThreadBufferedEvent; use super::app_server_event_targets::ServerNotificationThreadTarget; use super::app_server_event_targets::server_notification_thread_target; use super::app_server_event_targets::server_request_thread_id; +use super::app_server_requests::BufferedServerRequest; use crate::app_command::AppCommand; use crate::app_event::AppEvent; use crate::app_event::RateLimitRefreshOrigin; @@ -18,6 +19,7 @@ use codex_app_server_client::AppServerEvent; use codex_app_server_client::TypedRequestError; use codex_app_server_protocol::AuthMode; use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::RateLimitReachedType; use codex_app_server_protocol::RequestId; use codex_app_server_protocol::ServerNotification; @@ -30,9 +32,37 @@ use codex_app_server_protocol::ThreadSource; use codex_app_server_protocol::ThreadStatus; use codex_protocol::ThreadId; use codex_protocol::protocol::SubAgentSource; +use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile; use std::time::Duration; impl App { + async fn reject_unsupported_app_server_request( + &mut self, + unsupported: super::app_server_requests::UnsupportedAppServerRequest, + request: &BufferedServerRequest, + ) { + tracing::warn!( + request_id = ?unsupported.request_id, + message = unsupported.message, + "rejecting unsupported app-server request" + ); + self.chat_widget + .add_error_message(unsupported.message.clone()); + let result = request + .reply + .clone() + .reject(JSONRPCErrorError { + code: -32000, + message: unsupported.message, + data: None, + }) + .await + .map_err(|err| format!("failed to reject app-server request: {err}")); + if let Err(err) = result { + tracing::warn!("{err}"); + } + } + pub(super) fn refresh_mcp_startup_expected_servers_from_config(&mut self) { if self .current_displayed_thread_id() @@ -110,7 +140,9 @@ impl App { } } AppServerEvent::ServerRequest(request) => { - self.handle_server_request_event(app_server_client, *request) + let request = + BufferedServerRequest::new(*request, &app_server_client.request_handle()); + self.handle_server_request_event(app_server_client, request) .await; self.repaint_agents_overview(); } @@ -273,13 +305,16 @@ impl App { && let Some(requests) = self.agents_overview.dispatched_requests.get_mut(&thread_id) { - requests.retain(|request| request.id() != ¬ification.request_id); + requests.retain(|request| request.request.id() != ¬ification.request_id); } - if let Some(request) = self + if let Some(resolved) = self .pending_app_server_requests .resolve_notification(¬ification.thread_id, ¬ification.request_id) + && self.chat_widget.thread_id().map(|id| id.to_string()) + == Some(resolved.thread_id) { - self.chat_widget.dismiss_app_server_request(&request); + self.chat_widget + .dismiss_app_server_request(&resolved.request); if self.startup_pending_protected_request { self.startup_pending_protected_request = self.chat_widget.has_active_modal() @@ -548,26 +583,26 @@ impl App { .handle_server_notification(notification, /*replay_kind*/ None); } - async fn handle_server_request_event( + pub(in crate::app) async fn handle_server_request_event( &mut self, app_server_client: &AppServerSession, - request: ServerRequest, + buffered_request: BufferedServerRequest, ) { - let thread_id = server_request_thread_id(&request); + let request = &buffered_request.request; + let thread_id = server_request_thread_id(request); if thread_id .is_some_and(|thread_id| self.temporary_structured_requests.contains_key(&thread_id)) { - if let Err(err) = self - .reject_app_server_request( - app_server_client, - request.id().clone(), - "temporary structured threads cannot request tools or user interaction" - .to_string(), - ) - .await - { - tracing::debug!("{err}"); - } + self.reject_unsupported_app_server_request( + super::app_server_requests::UnsupportedAppServerRequest { + request_id: request.id().clone(), + message: + "temporary structured threads cannot request tools or user interaction" + .to_string(), + }, + &buffered_request, + ) + .await; return; } if let ServerRequest::DynamicToolCall { request_id, params } = &request { @@ -666,16 +701,14 @@ impl App { && owner.thread_id() == thread_id }); if thread_id.is_some_and(|thread_id| self.abandoned_side_threads.contains(&thread_id)) { - if let Err(err) = self - .reject_app_server_request( - app_server_client, - request.id().clone(), - "side conversation was closed".to_string(), - ) - .await - { - tracing::warn!("{err}"); - } + self.reject_unsupported_app_server_request( + super::app_server_requests::UnsupportedAppServerRequest { + request_id: request.id().clone(), + message: "side conversation was closed".to_string(), + }, + &buffered_request, + ) + .await; return; } if thread_id.is_some() @@ -683,8 +716,18 @@ impl App { && !background_voice && self.pending_startup_thread_start { + if let Some(unsupported) = self + .pending_app_server_requests + .note_buffered_request(&buffered_request) + { + self.reject_unsupported_app_server_request(unsupported, &buffered_request) + .await; + return; + } self.pending_primary_events - .push_back(ThreadBufferedEvent::Request(Box::new(request))); + .push_back(ThreadBufferedEvent::Request(Box::new( + buffered_request.request, + ))); return; } let unsupported_request = matches!( @@ -695,9 +738,22 @@ impl App { | ServerRequest::ApplyPatchApproval { .. } | ServerRequest::ExecCommandApproval { .. } ); + if let ServerRequest::PermissionsRequestApproval { request_id, params } = &request + && let Err(err) = CoreRequestPermissionProfile::try_from(params.permissions.clone()) + { + self.reject_unsupported_app_server_request( + super::app_server_requests::UnsupportedAppServerRequest { + request_id: request_id.clone(), + message: format!("failed to localize requested filesystem paths: {err}"), + }, + &buffered_request, + ) + .await; + return; + } if self .pending_app_server_requests - .contains_server_request(&request) + .contains_server_request(request) { return; } @@ -708,7 +764,7 @@ impl App { && !background_voice && let Some(requests) = self.agents_overview.dispatched_requests.get_mut(&thread_id) { - requests.push(request); + requests.push(buffered_request); return; } if thread_id.is_some() @@ -756,33 +812,18 @@ impl App { .dispatched_requests .entry(thread_id) .or_default() - .push(request); + .push(buffered_request); } return; } } - if let Some(unsupported) = self + let unsupported = self .pending_app_server_requests - .note_server_request(&request) - { - tracing::warn!( - request_id = ?unsupported.request_id, - message = unsupported.message, - "rejecting unsupported app-server request" - ); - self.chat_widget - .add_error_message(unsupported.message.clone()); - if let Err(err) = self - .reject_app_server_request( - app_server_client, - unsupported.request_id, - unsupported.message, - ) - .await - { - tracing::warn!("{err}"); - } + .note_buffered_request(&buffered_request); + if let Some(unsupported) = unsupported { + self.reject_unsupported_app_server_request(unsupported, &buffered_request) + .await; return; } @@ -792,9 +833,11 @@ impl App { }; let result = if self.primary_thread_id.is_none() && !background_voice { - self.enqueue_primary_thread_request(request).await + self.enqueue_primary_thread_request(buffered_request.request) + .await } else { - self.enqueue_thread_request(thread_id, request).await + self.enqueue_thread_request(thread_id, buffered_request.request) + .await }; if let Err(err) = result { tracing::warn!("failed to enqueue app-server request: {err}"); diff --git a/codex-rs/tui/src/app/app_server_requests.rs b/codex-rs/tui/src/app/app_server_requests.rs index ee68f0c544..be5c776085 100644 --- a/codex-rs/tui/src/app/app_server_requests.rs +++ b/codex-rs/tui/src/app/app_server_requests.rs @@ -1,40 +1,21 @@ +//! Tracks pending approvals and retains their originating connection until resolution. + use std::collections::HashMap; use std::collections::VecDeque; -use super::App; use crate::app_command::AppCommand; use crate::app_server_approval_conversions::granted_permission_profile_from_request; -use crate::app_server_session::AppServerSession; +use codex_app_server_client::AppServerReplyHandle; +use codex_app_server_client::AppServerRequestHandle; use codex_app_server_protocol::CommandExecutionRequestApprovalResponse; use codex_app_server_protocol::FileChangeRequestApprovalResponse; -use codex_app_server_protocol::JSONRPCErrorError; use codex_app_server_protocol::McpServerElicitationRequestResponse; use codex_app_server_protocol::PermissionsRequestApprovalResponse; use codex_app_server_protocol::RequestId as AppServerRequestId; use codex_app_server_protocol::ServerRequest; use codex_protocol::request_permissions::RequestPermissionProfile as CoreRequestPermissionProfile; -impl App { - pub(super) async fn reject_app_server_request( - &self, - app_server_client: &AppServerSession, - request_id: AppServerRequestId, - reason: String, - ) -> std::result::Result<(), String> { - app_server_client - .reject_server_request( - request_id, - JSONRPCErrorError { - code: -32000, - message: reason, - data: None, - }, - ) - .await - .map_err(|err| format!("failed to reject app-server request: {err}")) - } -} - +#[cfg(test)] #[derive(Debug, Clone, PartialEq, Eq)] pub(super) struct AppServerRequestResolution { pub(super) request_id: AppServerRequestId, @@ -70,13 +51,33 @@ pub(crate) enum ResolvedAppServerRequest { }, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct ResolvedThreadAppServerRequest { + pub(crate) thread_id: String, + pub(crate) request: ResolvedAppServerRequest, +} + +pub(in crate::app) struct BufferedServerRequest { + pub(in crate::app) request: ServerRequest, + pub(in crate::app) reply: AppServerReplyHandle, +} + +impl BufferedServerRequest { + pub(in crate::app) fn new(request: ServerRequest, origin: &AppServerRequestHandle) -> Self { + Self { + reply: origin.server_request_reply_handle(request.id().clone()), + request, + } + } +} + #[derive(Debug, Default)] pub(super) struct PendingAppServerRequests { - exec_approvals: HashMap<(String, String), AppServerRequestId>, - file_change_approvals: HashMap<(String, String), AppServerRequestId>, - permissions_approvals: HashMap<(String, String), AppServerRequestId>, - user_inputs: HashMap>, - mcp_requests: HashMap, + exec_approvals: HashMap<(String, String), PendingServerRequest>, + file_change_approvals: HashMap<(String, String), PendingServerRequest>, + permissions_approvals: HashMap<(String, String), PendingServerRequest>, + user_inputs: HashMap<(String, String), VecDeque>, + mcp_requests: HashMap, pub(super) user_verification: super::user_verification_requests::UserVerificationRequests, } @@ -99,16 +100,30 @@ impl PendingAppServerRequests { pub(super) fn cancel_thread_verification(&mut self, thread_id: &str) { for (server_name, request_id) in self.user_verification.cancel_thread(thread_id) { self.mcp_requests.remove(&McpRequestKey { + thread_id: Self::canonical_thread_id(thread_id), server_name, request_id, }); } } + #[cfg(test)] pub(super) fn note_server_request( &mut self, request: &ServerRequest, ) -> Option { + self.note_server_request_owned(request, /*reply*/ None) + } + + fn note_server_request_owned( + &mut self, + request: &ServerRequest, + reply: Option, + ) -> Option { + let pending = |request_id: &AppServerRequestId| PendingServerRequest { + request_id: request_id.clone(), + reply: reply.clone(), + }; match request { ServerRequest::CommandExecutionRequestApproval { request_id, params } => { let approval_id = params @@ -117,7 +132,7 @@ impl PendingAppServerRequests { .unwrap_or_else(|| params.item_id.clone()); self.exec_approvals.insert( (Self::canonical_thread_id(¶ms.thread_id), approval_id), - request_id.clone(), + pending(request_id), ); None } @@ -127,7 +142,7 @@ impl PendingAppServerRequests { Self::canonical_thread_id(¶ms.thread_id), params.item_id.clone(), ), - request_id.clone(), + pending(request_id), ); None } @@ -149,28 +164,34 @@ impl PendingAppServerRequests { Self::canonical_thread_id(¶ms.thread_id), params.item_id.clone(), ), - request_id.clone(), + pending(request_id), ); None } ServerRequest::ToolRequestUserInput { request_id, params } => { self.user_inputs - .entry(params.turn_id.clone()) + .entry(( + Self::canonical_thread_id(¶ms.thread_id), + params.turn_id.clone(), + )) .or_default() .push_back(PendingUserInputRequest { item_id: params.item_id.clone(), - request_id: request_id.clone(), + request: pending(request_id), }); None } ServerRequest::McpServerElicitationRequest { request_id, params } => { - self.user_verification.note_request(request_id, params); + let thread_id = Self::canonical_thread_id(¶ms.thread_id); + self.user_verification + .note_request(&thread_id, request_id, params); self.mcp_requests.insert( McpRequestKey { + thread_id, server_name: params.server_name.clone(), request_id: request_id.clone(), }, - request_id.clone(), + pending(request_id), ); None } @@ -205,11 +226,35 @@ impl PendingAppServerRequests { } } + pub(in crate::app) fn note_buffered_request( + &mut self, + request: &BufferedServerRequest, + ) -> Option { + self.note_server_request_owned(&request.request, Some(request.reply.clone())) + } + + #[cfg(test)] pub(super) fn take_resolution( &mut self, thread_id: &str, op: T, ) -> Result, String> + where + T: Into, + { + Ok(self + .take_resolution_record(thread_id, op)? + .map(|resolution| AppServerRequestResolution { + request_id: resolution.request.request_id, + result: resolution.result, + })) + } + + pub(in crate::app) fn take_resolution_record( + &mut self, + thread_id: &str, + op: T, + ) -> Result, String> where T: Into, { @@ -263,15 +308,16 @@ impl PendingAppServerRequests { .. } = &op { - self.user_verification.remove(server_name, request_id); + self.user_verification + .remove(&thread_id, server_name, request_id); } let resolution = match &op { AppCommand::ExecApproval { id, decision, .. } => self .exec_approvals - .remove(&(thread_id, id.clone())) - .map(|request_id| { - Ok::(AppServerRequestResolution { - request_id, + .remove(&(thread_id.clone(), id.clone())) + .map(|request| { + Ok::(PendingResolution { + request, result: serde_json::to_value(CommandExecutionRequestApprovalResponse { decision: decision.clone(), }) @@ -285,10 +331,10 @@ impl PendingAppServerRequests { .transpose()?, AppCommand::PatchApproval { id, decision } => self .file_change_approvals - .remove(&(thread_id, id.clone())) - .map(|request_id| { - Ok::(AppServerRequestResolution { - request_id, + .remove(&(thread_id.clone(), id.clone())) + .map(|request| { + Ok::(PendingResolution { + request, result: serde_json::to_value(FileChangeRequestApprovalResponse { decision: decision.clone(), }) @@ -300,10 +346,10 @@ impl PendingAppServerRequests { .transpose()?, AppCommand::RequestPermissionsResponse { id, response } => self .permissions_approvals - .remove(&(thread_id, id.clone())) - .map(|request_id| { - Ok::(AppServerRequestResolution { - request_id, + .remove(&(thread_id.clone(), id.clone())) + .map(|request| { + Ok::(PendingResolution { + request, result: serde_json::to_value(PermissionsRequestApprovalResponse { permissions: granted_permission_profile_from_request( response.permissions.clone(), @@ -318,10 +364,10 @@ impl PendingAppServerRequests { }) .transpose()?, AppCommand::UserInputAnswer { id, response } => self - .pop_user_input_request_for_turn(id) + .pop_user_input_request_for_turn(&thread_id, id) .map(|pending| { - Ok::(AppServerRequestResolution { - request_id: pending.request_id, + Ok::(PendingResolution { + request: pending.request, result: serde_json::to_value(response).map_err(|err| { format!("failed to serialize request_user_input response: {err}") })?, @@ -337,12 +383,13 @@ impl PendingAppServerRequests { } => self .mcp_requests .remove(&McpRequestKey { + thread_id: thread_id.clone(), server_name: server_name.to_string(), request_id: request_id.clone(), }) - .map(|request_id| { - Ok::(AppServerRequestResolution { - request_id, + .map(|request| { + Ok::(PendingResolution { + request, result: serde_json::to_value(McpServerElicitationRequestResponse { action: *decision, content: content.clone(), @@ -363,55 +410,68 @@ impl PendingAppServerRequests { &mut self, thread_id: &str, request_id: &AppServerRequestId, - ) -> Option { + ) -> Option { let thread_id = Self::canonical_thread_id(thread_id); if let Some(key) = self.exec_approvals.iter().find_map(|(key, value)| { - (key.0 == thread_id && value == request_id).then(|| key.clone()) + (key.0 == thread_id && &value.request_id == request_id).then(|| key.clone()) }) { self.exec_approvals.remove(&key); - return Some(ResolvedAppServerRequest::ExecApproval { - thread_id: key.0, - id: key.1, + return Some(ResolvedThreadAppServerRequest { + thread_id: key.0.clone(), + request: ResolvedAppServerRequest::ExecApproval { + thread_id: key.0, + id: key.1, + }, }); } if let Some(key) = self.file_change_approvals.iter().find_map(|(key, value)| { - (key.0 == thread_id && value == request_id).then(|| key.clone()) + (key.0 == thread_id && &value.request_id == request_id).then(|| key.clone()) }) { self.file_change_approvals.remove(&key); - return Some(ResolvedAppServerRequest::FileChangeApproval { - thread_id: key.0, - id: key.1, + return Some(ResolvedThreadAppServerRequest { + thread_id: key.0.clone(), + request: ResolvedAppServerRequest::FileChangeApproval { + thread_id: key.0, + id: key.1, + }, }); } if let Some(key) = self.permissions_approvals.iter().find_map(|(key, value)| { - (key.0 == thread_id && value == request_id).then(|| key.clone()) + (key.0 == thread_id && &value.request_id == request_id).then(|| key.clone()) }) { self.permissions_approvals.remove(&key); - return Some(ResolvedAppServerRequest::PermissionsApproval { - thread_id: key.0, - id: key.1, + return Some(ResolvedThreadAppServerRequest { + thread_id: key.0.clone(), + request: ResolvedAppServerRequest::PermissionsApproval { + thread_id: key.0, + id: key.1, + }, }); } - if let Some(pending) = self.remove_user_input_request(request_id) { - return Some(ResolvedAppServerRequest::UserInput { - call_id: pending.item_id, + if let Some(pending) = self.remove_user_input_request(&thread_id, request_id) { + return Some(ResolvedThreadAppServerRequest { + thread_id, + request: ResolvedAppServerRequest::UserInput { + call_id: pending.item_id, + }, }); } - if let Some(key) = self - .mcp_requests - .iter() - .find_map(|(key, value)| (value == request_id).then(|| key.clone())) - { + if let Some(key) = self.mcp_requests.iter().find_map(|(key, value)| { + (key.thread_id == thread_id && &value.request_id == request_id).then(|| key.clone()) + }) { self.mcp_requests.remove(&key); self.user_verification - .remove(&key.server_name, &key.request_id); - return Some(ResolvedAppServerRequest::McpElicitation { - server_name: key.server_name, - request_id: key.request_id, + .remove(&key.thread_id, &key.server_name, &key.request_id); + return Some(ResolvedThreadAppServerRequest { + thread_id: key.thread_id, + request: ResolvedAppServerRequest::McpElicitation { + server_name: key.server_name, + request_id: key.request_id, + }, }); } @@ -419,71 +479,102 @@ impl PendingAppServerRequests { } pub(super) fn contains_server_request(&self, request: &ServerRequest) -> bool { + matches!(request, ServerRequest::ChatgptAuthTokensRefresh { .. }) + || self.pending_request(request).is_some() + } + + fn pending_request(&self, request: &ServerRequest) -> Option<&PendingServerRequest> { match request { - ServerRequest::CommandExecutionRequestApproval { request_id, .. } => self - .exec_approvals - .values() - .any(|pending_request_id| pending_request_id == request_id), - ServerRequest::FileChangeRequestApproval { request_id, .. } => self + ServerRequest::CommandExecutionRequestApproval { request_id, params } => { + let approval_id = params.approval_id.as_ref().unwrap_or(¶ms.item_id); + self.exec_approvals + .get(&( + Self::canonical_thread_id(¶ms.thread_id), + approval_id.clone(), + )) + .filter(|pending| &pending.request_id == request_id) + } + ServerRequest::FileChangeRequestApproval { request_id, params } => self .file_change_approvals - .values() - .any(|pending_request_id| pending_request_id == request_id), - ServerRequest::PermissionsRequestApproval { request_id, .. } => self + .get(&( + Self::canonical_thread_id(¶ms.thread_id), + params.item_id.clone(), + )) + .filter(|pending| &pending.request_id == request_id), + ServerRequest::PermissionsRequestApproval { request_id, params } => self .permissions_approvals - .values() - .any(|pending_request_id| pending_request_id == request_id), - ServerRequest::ToolRequestUserInput { request_id, .. } => { - self.user_inputs.values().any(|queue| { + .get(&( + Self::canonical_thread_id(¶ms.thread_id), + params.item_id.clone(), + )) + .filter(|pending| &pending.request_id == request_id), + ServerRequest::ToolRequestUserInput { request_id, params } => self + .user_inputs + .get(&( + Self::canonical_thread_id(¶ms.thread_id), + params.turn_id.clone(), + )) + .and_then(|queue| { queue .iter() - .any(|pending| &pending.request_id == request_id) + .find(|pending| &pending.request.request_id == request_id) + .map(|pending| &pending.request) + }), + ServerRequest::McpServerElicitationRequest { request_id, params } => { + self.mcp_requests.get(&McpRequestKey { + thread_id: Self::canonical_thread_id(¶ms.thread_id), + server_name: params.server_name.clone(), + request_id: request_id.clone(), }) } - ServerRequest::McpServerElicitationRequest { request_id, .. } => self - .mcp_requests - .values() - .any(|pending_request_id| pending_request_id == request_id), - ServerRequest::ChatgptAuthTokensRefresh { .. } => true, + ServerRequest::ChatgptAuthTokensRefresh { .. } => None, ServerRequest::DynamicToolCall { .. } | ServerRequest::AttestationGenerate { .. } | ServerRequest::CurrentTimeRead { .. } | ServerRequest::ApplyPatchApproval { .. } - | ServerRequest::ExecCommandApproval { .. } => false, + | ServerRequest::ExecCommandApproval { .. } => None, } } + pub(in crate::app) fn buffered_request( + &self, + request: ServerRequest, + ) -> Option { + let reply = self.pending_request(&request)?.reply.clone()?; + Some(BufferedServerRequest { request, reply }) + } + fn pop_user_input_request_for_turn( &mut self, + thread_id: &str, turn_id: &str, ) -> Option { - let pending = self - .user_inputs - .get_mut(turn_id) - .and_then(VecDeque::pop_front); - if self - .user_inputs - .get(turn_id) - .is_some_and(VecDeque::is_empty) - { - self.user_inputs.remove(turn_id); + let key = (thread_id.to_string(), turn_id.to_string()); + let pending = self.user_inputs.get_mut(&key).and_then(VecDeque::pop_front); + if self.user_inputs.get(&key).is_some_and(VecDeque::is_empty) { + self.user_inputs.remove(&key); } pending } fn remove_user_input_request( &mut self, + thread_id: &str, request_id: &AppServerRequestId, ) -> Option { - let (turn_id, index) = self.user_inputs.iter().find_map(|(turn_id, queue)| { + let (key, index) = self.user_inputs.iter().find_map(|(key, queue)| { + if key.0 != thread_id { + return None; + } queue .iter() - .position(|pending| &pending.request_id == request_id) - .map(|index| (turn_id.clone(), index)) + .position(|pending| &pending.request.request_id == request_id) + .map(|index| (key.clone(), index)) })?; - let queue = self.user_inputs.get_mut(&turn_id)?; + let queue = self.user_inputs.get_mut(&key)?; let removed = queue.remove(index); if queue.is_empty() { - self.user_inputs.remove(&turn_id); + self.user_inputs.remove(&key); } removed } @@ -492,11 +583,23 @@ impl PendingAppServerRequests { #[derive(Debug)] struct PendingUserInputRequest { item_id: String, - request_id: AppServerRequestId, + request: PendingServerRequest, +} + +#[derive(Debug)] +pub(in crate::app) struct PendingServerRequest { + pub(in crate::app) request_id: AppServerRequestId, + pub(in crate::app) reply: Option, +} + +pub(in crate::app) struct PendingResolution { + pub(in crate::app) request: PendingServerRequest, + pub(in crate::app) result: serde_json::Value, } #[derive(Debug, Clone, PartialEq, Eq, Hash)] struct McpRequestKey { + thread_id: String, server_name: String, request_id: AppServerRequestId, } @@ -505,6 +608,7 @@ struct McpRequestKey { mod tests { use super::PendingAppServerRequests; use super::ResolvedAppServerRequest; + use super::ResolvedThreadAppServerRequest; use super::UnsupportedAppServerRequest; use crate::app_command::AppCommand as Op; use codex_app_server_protocol::AdditionalFileSystemPermissions; @@ -979,9 +1083,12 @@ mod tests { &thread_id.replace('-', ""), &AppServerRequestId::Integer(41) ), - Some(ResolvedAppServerRequest::ExecApproval { + Some(ResolvedThreadAppServerRequest { thread_id: thread_id.clone(), - id: "approval-1".to_string(), + request: ResolvedAppServerRequest::ExecApproval { + thread_id: thread_id.clone(), + id: "approval-1".to_string(), + }, }) ); assert_eq!( @@ -1017,9 +1124,12 @@ mod tests { assert_eq!( pending.resolve_notification("thread-1", &AppServerRequestId::Integer(12)), - Some(ResolvedAppServerRequest::McpElicitation { - server_name: "example".to_string(), - request_id: AppServerRequestId::Integer(12), + Some(ResolvedThreadAppServerRequest { + thread_id: "thread-1".to_string(), + request: ResolvedAppServerRequest::McpElicitation { + server_name: "example".to_string(), + request_id: AppServerRequestId::Integer(12), + }, }) ); } @@ -1041,8 +1151,11 @@ mod tests { assert_eq!( pending.resolve_notification("thread-1", &AppServerRequestId::Integer(8)), - Some(ResolvedAppServerRequest::UserInput { - call_id: "tool-1".to_string(), + Some(ResolvedThreadAppServerRequest { + thread_id: "thread-1".to_string(), + request: ResolvedAppServerRequest::UserInput { + call_id: "tool-1".to_string(), + }, }) ); } @@ -1092,3 +1205,7 @@ mod tests { assert_eq!(second_response.request_id, AppServerRequestId::Integer(9)); } } + +#[cfg(test)] +#[path = "retained_reply_tests.rs"] +mod retained_reply_tests; diff --git a/codex-rs/tui/src/app/retained_reply_tests.rs b/codex-rs/tui/src/app/retained_reply_tests.rs new file mode 100644 index 0000000000..2a95b2a719 --- /dev/null +++ b/codex-rs/tui/src/app/retained_reply_tests.rs @@ -0,0 +1,210 @@ +//! Exercises origin-bound approval replies through real WebSocket workers. + +use super::*; +use crate::app::app_server_requests::BufferedServerRequest; +use crate::app_server_session::AppServerSession; +use crate::app_server_session::ThreadParamsMode; +use codex_app_server_client::AppServerEvent; +use codex_app_server_protocol::CommandExecutionApprovalDecision; +use codex_app_server_protocol::CommandExecutionRequestApprovalParams; +use codex_app_server_protocol::FileChangeApprovalDecision; +use codex_app_server_protocol::FileChangeRequestApprovalParams; +use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::JSONRPCResponse; +use codex_protocol::ThreadId; +use futures::SinkExt; +use futures::StreamExt; +use pretty_assertions::assert_eq; +use serde_json::json; +use tokio::sync::mpsc; +use tokio::task::JoinHandle; +use tokio::time::Duration; +use tokio::time::timeout; +use tokio_tungstenite::tungstenite::Message; + +async fn connection() -> color_eyre::Result<( + AppServerSession, + mpsc::UnboundedReceiver, + JoinHandle>, +)> { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let endpoint = crate::resolve_remote_addr(&format!("ws://{}", listener.local_addr()?))?; + let (observed, receiver) = mpsc::unbounded_channel(); + let task = tokio::spawn(async move { + let (stream, _) = listener.accept().await?; + let mut socket = tokio_tungstenite::accept_async(stream).await?; + while let Some(frame) = socket.next().await { + let Message::Text(text) = frame? else { + continue; + }; + let message: JSONRPCMessage = serde_json::from_str(&text)?; + match message { + JSONRPCMessage::Request(request) if request.method == "initialize" => { + socket + .send(Message::Text( + json!({"id":request.id,"result":{ + "userAgent":"retained-reply-test","codexHome":std::env::temp_dir(), + "platformFamily":std::env::consts::FAMILY, + "platformOs":std::env::consts::OS + }}) + .to_string() + .into(), + )) + .await?; + } + JSONRPCMessage::Notification(_) => {} + message => observed.send(message)?, + } + } + Ok(()) + }); + let client = crate::connect_remote_app_server(endpoint).await?; + Ok(( + AppServerSession::new(client, ThreadParamsMode::Remote), + receiver, + task, + )) +} + +fn patch_request(thread_id: ThreadId) -> ServerRequest { + ServerRequest::FileChangeRequestApproval { + request_id: AppServerRequestId::Integer(7), + params: FileChangeRequestApprovalParams { + thread_id: thread_id.to_string(), + turn_id: "turn".into(), + item_id: "patch".into(), + started_at_ms: 0, + reason: None, + grant_root: None, + }, + } +} + +fn command_request(thread_id: ThreadId) -> ServerRequest { + ServerRequest::CommandExecutionRequestApproval { + request_id: AppServerRequestId::Integer(7), + params: CommandExecutionRequestApprovalParams { + kind: Default::default(), + thread_id: thread_id.to_string(), + turn_id: "turn".into(), + item_id: "command".into(), + started_at_ms: 0, + approval_id: None, + environment_id: None, + reason: None, + network_approval_context: None, + command: Some("echo hello".into()), + cwd: None, + command_actions: None, + additional_permissions: None, + proposed_execpolicy_amendment: None, + proposed_network_policy_amendments: None, + available_decisions: None, + }, + } +} + +#[tokio::test] +async fn direct_and_buffered_colliding_ids_reply_on_their_origin() -> color_eyre::Result<()> { + let (mut app, _events, _ops) = crate::app::tests::make_test_app_with_channels().await; + let (command_server, mut command_replies, command_task) = connection().await?; + let (mut patch_server, mut patch_replies, patch_task) = connection().await?; + let command_thread = ThreadId::new(); + let patch_thread = ThreadId::new(); + app.primary_thread_id = Some(command_thread); + app.active_thread_id = Some(command_thread); + app.ensure_thread_channel(command_thread); + app.ensure_thread_channel(patch_thread); + app.agents_overview + .dispatched_requests + .entry(patch_thread) + .or_default(); + + app.handle_app_server_event( + &command_server, + AppServerEvent::ServerRequest(Box::new(command_request(command_thread))), + ) + .await; + app.handle_app_server_event( + &patch_server, + AppServerEvent::ServerRequest(Box::new(patch_request(patch_thread))), + ) + .await; + + app.submit_thread_op( + &mut patch_server, + command_thread, + AppCommand::ExecApproval { + id: "command".into(), + turn_id: None, + decision: CommandExecutionApprovalDecision::Decline, + }, + ) + .await?; + app.primary_thread_id = Some(patch_thread); + app.active_thread_id = Some(patch_thread); + app.replay_agents_overview_requests(&command_server, patch_thread) + .await; + app.submit_thread_op( + &mut patch_server, + patch_thread, + AppCommand::PatchApproval { + id: "patch".into(), + decision: FileChangeApprovalDecision::Accept, + }, + ) + .await?; + + assert_eq!( + timeout(Duration::from_secs(/*secs*/ 5), command_replies.recv()).await?, + Some(JSONRPCMessage::Response(JSONRPCResponse { + id: AppServerRequestId::Integer(7), + result: json!({"decision":"decline"}) + })) + ); + assert_eq!( + timeout(Duration::from_secs(/*secs*/ 5), patch_replies.recv()).await?, + Some(JSONRPCMessage::Response(JSONRPCResponse { + id: AppServerRequestId::Integer(7), + result: json!({"decision":"accept"}) + })) + ); + command_server.shutdown().await?; + patch_server.shutdown().await?; + command_task.await??; + patch_task.await??; + Ok(()) +} + +#[tokio::test] +async fn closed_origin_consumes_reply_without_falling_through() -> color_eyre::Result<()> { + let (mut app, _events, _ops) = crate::app::tests::make_test_app_with_channels().await; + let (original, _replies, task) = connection().await?; + let thread = ThreadId::new(); + let request = BufferedServerRequest::new(patch_request(thread), &original.request_handle()); + app.pending_app_server_requests + .note_buffered_request(&request); + app.enqueue_thread_request(thread, request.request).await?; + original.shutdown().await?; + task.await??; + + assert!( + app.try_resolve_app_server_request( + thread, + &AppCommand::PatchApproval { + id: "patch".into(), + decision: FileChangeApprovalDecision::Accept, + }, + ) + .await? + ); + assert!( + app.thread_event_channels[&thread] + .store + .lock() + .await + .pending_replay_requests() + .is_empty() + ); + Ok(()) +} diff --git a/codex-rs/tui/src/app/session_lifecycle.rs b/codex-rs/tui/src/app/session_lifecycle.rs index 0499b7b5d2..a4fff209ab 100644 --- a/codex-rs/tui/src/app/session_lifecycle.rs +++ b/codex-rs/tui/src/app/session_lifecycle.rs @@ -798,11 +798,16 @@ impl App { pub(super) async fn reset_thread_event_state(&mut self) { let voice_owner = self.voice_owner_thread_id(); + let mut previous_pending = std::mem::take(&mut self.pending_app_server_requests); // Move retained tasks' approvals to background routing before clearing request bookkeeping. for (thread_id, requests) in &mut self.agents_overview.dispatched_requests { if let Some(channel) = self.thread_event_channels.get(thread_id) { for request in channel.store.lock().await.pending_replay_requests() { - if !requests.iter().any(|pending| pending.id() == request.id()) { + if !requests + .iter() + .any(|pending| pending.request.id() == request.id()) + && let Some(request) = previous_pending.buffered_request(request) + { requests.push(request); } } @@ -812,8 +817,7 @@ impl App { for (thread_id, channel) in &self.thread_event_channels { if Some(*thread_id) != voice_owner && !self.side_threads.contains_key(thread_id) { for request in channel.store.lock().await.pending_replay_requests() { - self.pending_app_server_requests - .resolve_notification(&thread_id.to_string(), request.id()); + previous_pending.resolve_notification(&thread_id.to_string(), request.id()); } } } @@ -844,17 +848,20 @@ impl App { self.primary_session_configured = None; self.pending_primary_events.clear(); if voice_owner.is_none() { - self.pending_app_server_requests.clear(); let side_thread_ids: Vec<_> = self.side_threads.keys().copied().collect(); for thread_id in side_thread_ids { if let Some(channel) = self.thread_event_channels.get(&thread_id) { for request in channel.store.lock().await.pending_replay_requests() { - let _ = self - .pending_app_server_requests - .note_server_request(&request); + if let Some(request) = previous_pending.buffered_request(request) { + let _ = self + .pending_app_server_requests + .note_buffered_request(&request); + } } } } + } else { + self.pending_app_server_requests = previous_pending; } self.pending_startup_thread_start = false; self.pending_server_version_notice = None; @@ -904,31 +911,22 @@ impl App { ThreadBufferedEvent::HistoryEntryResponse(_) | ThreadBufferedEvent::FeedbackSubmission(_) => true, }); - self.pending_app_server_requests.clear(); - let mut unsupported_requests = Vec::new(); + let previous_pending = std::mem::take(&mut self.pending_app_server_requests); for event in &self.pending_primary_events { - if let ThreadBufferedEvent::Request(request) = event - && let Some(unsupported) = self - .pending_app_server_requests - .note_server_request(request) - { - unsupported_requests.push(unsupported); - } - } - for unsupported in unsupported_requests { - self.pending_primary_events.retain(|event| { - !matches!(event, ThreadBufferedEvent::Request(request) - if request.id() == &unsupported.request_id) - }); - if let Err(error) = self - .reject_app_server_request( - app_server, - unsupported.request_id, - unsupported.message, - ) - .await - { - tracing::warn!("{error}"); + if let ThreadBufferedEvent::Request(request) = event { + let retained = previous_pending.buffered_request((**request).clone()); + #[cfg(test)] + let retained = retained.or_else(|| { + Some(super::app_server_requests::BufferedServerRequest::new( + (**request).clone(), + &app_server.request_handle(), + )) + }); + if let Some(request) = retained { + let _ = self + .pending_app_server_requests + .note_buffered_request(&request); + } } } if started.blocks_direct_input { diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index f22d492847..40c30df923 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -556,26 +556,34 @@ async fn external_editor_writable_directory_rejected_snapshot() -> Result<()> { #[tokio::test] async fn enqueue_primary_thread_session_replays_buffered_approval_after_attach() -> Result<()> { let (mut app, mut app_event_rx, _op_rx) = make_test_app_with_channels().await; - let thread_id = ThreadId::new(); - let approval_request = - exec_approval_request(thread_id, "turn-1", "call-1", /*approval_id*/ None); - - assert_eq!( - app.pending_app_server_requests - .note_server_request(&approval_request), - None - ); - app.agents_overview - .dispatched_requests - .insert(thread_id, vec![approval_request.clone()]); - app.enqueue_primary_thread_request(approval_request.clone()) - .await?; let (mut app_server, requests, proxy) = session_lifecycle_requests::start_recording_app_server( &app.config, /*blocked_thread_list*/ None, /*failed_thread_name*/ None, ) .await?; + let thread_id = ThreadId::new(); + let approval_request = + exec_approval_request(thread_id, "turn-1", "call-1", /*approval_id*/ None); + + let buffered = super::app_server_requests::BufferedServerRequest::new( + approval_request.clone(), + &app_server.request_handle(), + ); + assert_eq!( + app.pending_app_server_requests + .note_buffered_request(&buffered), + None + ); + app.agents_overview.dispatched_requests.insert( + thread_id, + vec![super::app_server_requests::BufferedServerRequest::new( + approval_request.clone(), + &app_server.request_handle(), + )], + ); + app.enqueue_primary_thread_request(approval_request.clone()) + .await?; app.handle_app_server_event( &app_server, codex_app_server_client::AppServerEvent::ServerNotification(Box::new( @@ -624,8 +632,12 @@ async fn enqueue_primary_thread_session_replays_buffered_approval_after_attach() }; *request_id = AppServerRequestId::Integer(2); params.thread_id = background_thread_id.to_string(); + let buffered = super::app_server_requests::BufferedServerRequest::new( + background_request.clone(), + &app_server.request_handle(), + ); app.pending_app_server_requests - .note_server_request(&background_request); + .note_buffered_request(&buffered); app.enqueue_thread_request(background_thread_id, background_request.clone()) .await?; @@ -649,8 +661,12 @@ async fn enqueue_primary_thread_session_replays_buffered_approval_after_attach() !app.pending_app_server_requests .contains_server_request(&background_request) ); + let buffered = super::app_server_requests::BufferedServerRequest::new( + background_request.clone(), + &app_server.request_handle(), + ); app.pending_app_server_requests - .note_server_request(&background_request); + .note_buffered_request(&buffered); app.enqueue_thread_request(background_thread_id, background_request) .await?; @@ -711,7 +727,8 @@ async fn resolved_buffered_approval_does_not_become_actionable_after_drain() -> .pending_app_server_requests .resolve_notification(&thread_id.to_string(), &AppServerRequestId::Integer(1)) .expect("matching app-server request should resolve"); - app.chat_widget.dismiss_app_server_request(&resolved); + app.chat_widget + .dismiss_app_server_request(&resolved.request); while app_event_rx.try_recv().is_ok() {} let rx = app @@ -747,6 +764,43 @@ async fn resolved_buffered_approval_does_not_become_actionable_after_drain() -> Ok(()) } +#[tokio::test] +async fn background_resolution_keeps_matching_foreground_user_input_visible() -> Result<()> { + let (mut app, _app_event_rx, _op_rx) = make_test_app_with_channels().await; + let app_server = crate::start_embedded_app_server_for_picker(&app.config).await?; + let background_thread = ThreadId::new(); + let foreground_thread = ThreadId::new(); + let background_request = request_user_input_request(background_thread, "turn-1", "question"); + let foreground_request = request_user_input_request(foreground_thread, "turn-1", "question"); + app.enqueue_primary_thread_session( + test_thread_session(foreground_thread, test_path_buf("/tmp/project")), + Vec::new(), + ) + .await?; + app.pending_app_server_requests + .note_server_request(&background_request); + app.pending_app_server_requests + .note_server_request(&foreground_request); + app.handle_thread_event_now(ThreadBufferedEvent::Request(Box::new(foreground_request))); + assert!(app.chat_widget.has_active_view()); + + app.handle_app_server_event( + &app_server, + codex_app_server_client::AppServerEvent::ServerNotification(Box::new( + ServerNotification::ServerRequestResolved( + codex_app_server_protocol::ServerRequestResolvedNotification { + thread_id: background_thread.to_string(), + request_id: AppServerRequestId::Integer(2), + }, + ), + )), + ) + .await; + assert!(app.chat_widget.has_active_view()); + app_server.shutdown().await?; + Ok(()) +} + fn set_test_initial_prompt(app: &mut App, initial_prompt: String) { let config = app.config.clone(); let model = get_model_offline_for_tests(config.model.as_deref()); @@ -857,6 +911,9 @@ async fn reset_thread_event_state_aborts_listener_tasks() { } let mut app = make_test_app().await; + let app_server = crate::start_embedded_app_server_for_picker(&app.config) + .await + .expect("embedded app server"); let thread_id = ThreadId::new(); let side_thread_id = ThreadId::new(); let side_request = exec_approval_request( @@ -886,8 +943,12 @@ async fn reset_thread_event_state_aborts_listener_tasks() { .push_request(side_request.clone()); app.side_threads .insert(side_thread_id, SideThreadState::new(thread_id)); + let buffered = super::app_server_requests::BufferedServerRequest::new( + side_request.clone(), + &app_server.request_handle(), + ); app.pending_app_server_requests - .note_server_request(&side_request); + .note_buffered_request(&buffered); app.pending_server_profiles.insert( thread_id, PermissionProfileSelection { @@ -921,6 +982,7 @@ async fn reset_thread_event_state_aborts_listener_tasks() { .await .expect("timed out waiting for listener task abort") .expect("listener task drop notification should succeed"); + app_server.shutdown().await.expect("app server shutdown"); } #[tokio::test] @@ -6655,7 +6717,7 @@ async fn replace_goal_confirmation_snapshot() { ); } -fn test_thread_session(thread_id: ThreadId, cwd: PathBuf) -> ThreadSessionState { +pub(super) fn test_thread_session(thread_id: ThreadId, cwd: PathBuf) -> ThreadSessionState { ThreadSessionState { daybreak_enabled: false, windows_sandbox_host: crate::app::WindowsSandboxHost::Local, diff --git a/codex-rs/tui/src/app/tests/fork_workspace_roots_tests.rs b/codex-rs/tui/src/app/tests/fork_workspace_roots_tests.rs index b6e4a48dbb..762c26ac89 100644 --- a/codex-rs/tui/src/app/tests/fork_workspace_roots_tests.rs +++ b/codex-rs/tui/src/app/tests/fork_workspace_roots_tests.rs @@ -93,8 +93,11 @@ async fn check_fork_dispatch(mode: ThreadParamsMode) -> Result<()> { "command", /*approval_id*/ None, ); - app.enqueue_thread_request(source_thread_id, approval.clone()) - .await?; + app.handle_app_server_event( + &server, + AppServerEvent::ServerRequest(Box::new(approval.clone())), + ) + .await; app.drain_active_thread_events(&mut tui).await?; vec![approval] } else { @@ -121,8 +124,11 @@ async fn check_fork_dispatch(mode: ThreadParamsMode) -> Result<()> { assert!(recorded_params(&requests, "thread/unsubscribe").is_empty()); assert!(!app.thread_event_listener_tasks[&source_thread_id].is_finished()); assert_eq!( - app.agents_overview.dispatched_requests[&source_thread_id], - pending + app.agents_overview.dispatched_requests[&source_thread_id] + .iter() + .map(|pending| &pending.request) + .collect::>(), + pending.iter().collect::>() ); assert!(app.agents_overview.dispatched_requests[&child_id].is_empty()); let question = request_user_input_request(source_thread_id, "source-turn", "question"); @@ -132,7 +138,9 @@ async fn check_fork_dispatch(mode: ThreadParamsMode) -> Result<()> { ) .await; assert_eq!( - app.agents_overview.dispatched_requests[&source_thread_id].last(), + app.agents_overview.dispatched_requests[&source_thread_id] + .last() + .map(|pending| &pending.request), Some(&question) ); assert!(!app.chat_widget.has_active_view()); diff --git a/codex-rs/tui/src/app/tests/startup.rs b/codex-rs/tui/src/app/tests/startup.rs index da40cb7809..196ff1ab16 100644 --- a/codex-rs/tui/src/app/tests/startup.rs +++ b/codex-rs/tui/src/app/tests/startup.rs @@ -1500,22 +1500,6 @@ async fn startup_thread_started_discards_another_threads_buffered_events() { )) .await .expect("embedded app server"); - app.handle_app_server_event( - &app_server, - codex_app_server_client::AppServerEvent::ServerRequest(Box::new( - ServerRequest::CurrentTimeRead { - request_id: AppServerRequestId::Integer(2), - params: codex_app_server_protocol::CurrentTimeReadParams { - thread_id: other_thread_id.to_string(), - }, - }, - )), - ) - .await; - assert!(app.pending_primary_events.iter().any(|event| { - matches!(event, ThreadBufferedEvent::Request(request) - if matches!(request.as_ref(), ServerRequest::CurrentTimeRead { .. })) - })); let thread_id = ThreadId::new(); app.handle_startup_thread_started( &mut app_server, diff --git a/codex-rs/tui/src/app/tests/user_verification_routes_tests.rs b/codex-rs/tui/src/app/tests/user_verification_routes_tests.rs index c288b03a3e..d3d07eded4 100644 --- a/codex-rs/tui/src/app/tests/user_verification_routes_tests.rs +++ b/codex-rs/tui/src/app/tests/user_verification_routes_tests.rs @@ -165,7 +165,7 @@ async fn discarding_side_cancels_verification_and_ignores_late_proof() -> Result let attempt = app .pending_app_server_requests .user_verification - .begin("deployments", request.id()) + .begin(&side_thread_id.to_string(), "deployments", request.id()) .expect("pending side-thread verification"); app.discard_side_thread_in_background(&mut app_server, side_thread_id) diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index eedd0929a9..c8d16dc9c6 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -532,10 +532,7 @@ impl App { crate::session_log::log_outbound_op(&op); - if self - .try_resolve_app_server_request(app_server, thread_id, &op) - .await? - { + if self.try_resolve_app_server_request(thread_id, &op).await? { return Ok(()); } @@ -1090,37 +1087,33 @@ impl App { pub(super) async fn try_resolve_app_server_request( &mut self, - app_server: &AppServerSession, thread_id: ThreadId, op: &AppCommand, ) -> Result { let Some(resolution) = self .pending_app_server_requests - .take_resolution(&thread_id.to_string(), op) + .take_resolution_record(&thread_id.to_string(), op) .map_err(|err| color_eyre::eyre::eyre!(err))? else { return Ok(false); }; + let Some(reply) = resolution.request.reply else { + return Err(color_eyre::eyre::eyre!( + "Missing original app-server connection for pending request" + )); + }; - match app_server - .resolve_server_request(resolution.request_id, resolution.result) - .await - { - Ok(()) => { - if ThreadEventStore::op_can_change_pending_replay_state(op) { - self.note_thread_outbound_op(thread_id, op).await; - self.refresh_pending_thread_approvals().await; - self.refresh_side_parent_status_from_store(thread_id).await; - } - Ok(true) - } - Err(err) => { - self.chat_widget.add_error_message(format!( - "Failed to resolve app-server request for thread {thread_id}: {err}" - )); - Ok(false) - } + if let Err(err) = reply.resolve(resolution.result).await { + self.chat_widget.add_error_message(format!( + "Failed to resolve app-server request for thread {thread_id}: {err}" + )); } + if ThreadEventStore::op_can_change_pending_replay_state(op) { + self.note_thread_outbound_op(thread_id, op).await; + self.refresh_pending_thread_approvals().await; + self.refresh_side_parent_status_from_store(thread_id).await; + } + Ok(true) } pub(super) async fn refresh_pending_thread_approvals(&mut self) { diff --git a/codex-rs/tui/src/app/user_verification.rs b/codex-rs/tui/src/app/user_verification.rs index f7c7bc3800..23cd49af46 100644 --- a/codex-rs/tui/src/app/user_verification.rs +++ b/codex-rs/tui/src/app/user_verification.rs @@ -58,11 +58,11 @@ impl App { ); return Ok(()); } - let Some(attempt) = self - .pending_app_server_requests - .user_verification - .begin(&server_name, &request_id) - else { + let Some(attempt) = self.pending_app_server_requests.user_verification.begin( + &thread_id.to_string(), + &server_name, + &request_id, + ) else { return Ok(()); }; let handle = app_server.request_handle(); @@ -107,7 +107,12 @@ impl App { || !self .pending_app_server_requests .user_verification - .is_pending(&server_name, &request_id, attempt_id) + .is_pending( + &thread_id.to_string(), + &server_name, + &request_id, + attempt_id, + ) { return Ok(()); } diff --git a/codex-rs/tui/src/app/user_verification_requests.rs b/codex-rs/tui/src/app/user_verification_requests.rs index daf6ab0024..c9c81c54f1 100644 --- a/codex-rs/tui/src/app/user_verification_requests.rs +++ b/codex-rs/tui/src/app/user_verification_requests.rs @@ -11,12 +11,11 @@ use uuid::Uuid; #[derive(Debug, Default)] pub(super) struct UserVerificationRequests { - pending: HashMap<(String, RequestId), PendingVerification>, + pending: HashMap<(String, String, RequestId), PendingVerification>, } #[derive(Debug)] struct PendingVerification { - thread_id: String, params: UserVerificationVerifyParams, attempt: Option, cancelled: CancellationToken, @@ -37,6 +36,7 @@ impl Drop for PendingVerification { impl UserVerificationRequests { pub(super) fn note_request( &mut self, + thread_id: &str, request_id: &RequestId, params: &McpServerElicitationRequestParams, ) { @@ -48,9 +48,12 @@ impl UserVerificationRequests { } = ¶ms.request { self.pending.insert( - (params.server_name.clone(), request_id.clone()), + ( + thread_id.to_string(), + params.server_name.clone(), + request_id.clone(), + ), PendingVerification { - thread_id: params.thread_id.clone(), params: UserVerificationVerifyParams { title: title.clone(), description: description.clone(), @@ -65,12 +68,15 @@ impl UserVerificationRequests { pub(super) fn begin( &mut self, + thread_id: &str, server_name: &str, request_id: &RequestId, ) -> Option { - let pending = self - .pending - .get_mut(&(server_name.to_string(), request_id.clone()))?; + let pending = self.pending.get_mut(&( + thread_id.to_string(), + server_name.to_string(), + request_id.clone(), + ))?; if pending.attempt.is_some() { return None; } @@ -85,31 +91,40 @@ impl UserVerificationRequests { pub(super) fn is_pending( &self, + thread_id: &str, server_name: &str, request_id: &RequestId, attempt: Uuid, ) -> bool { self.pending - .get(&(server_name.to_string(), request_id.clone())) + .get(&( + thread_id.to_string(), + server_name.to_string(), + request_id.clone(), + )) .is_some_and(|pending| pending.attempt == Some(attempt)) } - pub(super) fn remove(&mut self, server_name: &str, request_id: &RequestId) { - self.pending - .remove(&(server_name.to_string(), request_id.clone())); + pub(super) fn remove(&mut self, thread_id: &str, server_name: &str, request_id: &RequestId) { + self.pending.remove(&( + thread_id.to_string(), + server_name.to_string(), + request_id.clone(), + )); } pub(super) fn cancel_thread(&mut self, thread_id: &str) -> Vec<(String, RequestId)> { - let discarded_keys = self - .pending - .iter() - .filter(|(_, pending)| pending.thread_id == thread_id) - .map(|(key, _)| key.clone()) - .collect::>(); - for (server_name, request_id) in &discarded_keys { - self.remove(server_name, request_id); - } - discarded_keys + let mut discarded = Vec::new(); + self.pending + .retain(|(pending_thread_id, server_name, request_id), _| { + if pending_thread_id == thread_id { + discarded.push((server_name.clone(), request_id.clone())); + false + } else { + true + } + }); + discarded } pub(super) fn clear(&mut self) { diff --git a/codex-rs/tui/src/app/user_verification_requests_tests.rs b/codex-rs/tui/src/app/user_verification_requests_tests.rs index 75d4cbda1c..654b5b5272 100644 --- a/codex-rs/tui/src/app/user_verification_requests_tests.rs +++ b/codex-rs/tui/src/app/user_verification_requests_tests.rs @@ -1,6 +1,7 @@ use crate::app::app_server_requests::AppServerRequestResolution; use crate::app::app_server_requests::PendingAppServerRequests; use crate::app::app_server_requests::ResolvedAppServerRequest; +use crate::app::app_server_requests::ResolvedThreadAppServerRequest; use crate::app_command::AppCommand; use crate::app_command::UserVerificationResponse; use codex_app_server_protocol::McpServerElicitationRequest; @@ -28,29 +29,44 @@ fn verification_request() -> ServerRequest { } } +fn thread_id(request: &ServerRequest) -> &str { + let ServerRequest::McpServerElicitationRequest { params, .. } = request else { + unreachable!() + }; + ¶ms.thread_id +} + #[test] fn user_verification_cancellation_invalidates_the_inflight_result() { let mut pending = PendingAppServerRequests::default(); - let request = verification_request(); + let mut request = verification_request(); + let ServerRequest::McpServerElicitationRequest { params, .. } = &mut request else { + unreachable!() + }; + params.thread_id.retain(|character| character != '-'); + let canonical_thread_id = ThreadId::from_string(¶ms.thread_id) + .expect("valid compact thread ID") + .to_string(); pending.note_server_request(&request); let attempt = pending .user_verification - .begin("deployments", request.id()) + .begin(&canonical_thread_id, "deployments", request.id()) .expect("pending request"); assert!( pending .user_verification - .begin("deployments", request.id()) + .begin(&canonical_thread_id, "deployments", request.id()) .is_none() ); - assert!( - pending - .user_verification - .is_pending("deployments", request.id(), attempt.id) - ); + assert!(pending.user_verification.is_pending( + &canonical_thread_id, + "deployments", + request.id(), + attempt.id + )); let resolution = pending .take_resolution( - "unused", + &canonical_thread_id, AppCommand::resolve_user_verification( "deployments".to_string(), request.id().clone(), @@ -65,11 +81,12 @@ fn user_verification_cancellation_invalidates_the_inflight_result() { result: serde_json::json!({ "action": "cancel", "content": null, "_meta": {"openai/userVerificationReason": "userCancelled"} }), }) ); - assert!( - !pending - .user_verification - .is_pending("deployments", request.id(), attempt.id) - ); + assert!(!pending.user_verification.is_pending( + &canonical_thread_id, + "deployments", + request.id(), + attempt.id + )); assert!(attempt.cancelled.is_cancelled()); } @@ -84,27 +101,27 @@ fn discarded_thread_cancels_only_its_verification_attempts() { pending.note_server_request(&request); let discarded_attempt = pending .user_verification - .begin("deployments", request.id()) + .begin(thread_id(&request), "deployments", request.id()) .expect("discarded thread attempt"); let mut other_request = verification_request(); - let ServerRequest::McpServerElicitationRequest { request_id, params } = &mut other_request - else { + let ServerRequest::McpServerElicitationRequest { params, .. } = &mut other_request else { unreachable!() }; - *request_id = RequestId::Integer(8); params.thread_id = ThreadId::new().to_string(); pending.note_server_request(&other_request); let other_attempt = pending .user_verification - .begin("deployments", other_request.id()) + .begin(thread_id(&other_request), "deployments", other_request.id()) .expect("other thread attempt"); + assert!(!discarded_attempt.cancelled.is_cancelled()); pending.cancel_thread_verification(&discarded_thread_id); assert!(discarded_attempt.cancelled.is_cancelled()); assert!(!pending.contains_server_request(&request)); assert!(!pending.user_verification.is_pending( + thread_id(&request), "deployments", request.id(), discarded_attempt.id @@ -112,6 +129,7 @@ fn discarded_thread_cancels_only_its_verification_attempts() { assert!(!other_attempt.cancelled.is_cancelled()); assert!(pending.contains_server_request(&other_request)); assert!(pending.user_verification.is_pending( + thread_id(&other_request), "deployments", other_request.id(), other_attempt.id @@ -125,25 +143,27 @@ fn user_verification_reconnect_cannot_accept_a_previous_connections_proof() { pending.note_server_request(&request); let previous = pending .user_verification - .begin("deployments", request.id()) + .begin(thread_id(&request), "deployments", request.id()) .expect("first request"); pending.clear(); assert!(previous.cancelled.is_cancelled()); pending.note_server_request(&request); let current = pending .user_verification - .begin("deployments", request.id()) + .begin(thread_id(&request), "deployments", request.id()) .expect("new request"); - assert!( - !pending - .user_verification - .is_pending("deployments", request.id(), previous.id) - ); - assert!( - pending - .user_verification - .is_pending("deployments", request.id(), current.id) - ); + assert!(!pending.user_verification.is_pending( + thread_id(&request), + "deployments", + request.id(), + previous.id + )); + assert!(pending.user_verification.is_pending( + thread_id(&request), + "deployments", + request.id(), + current.id + )); } #[test] @@ -153,20 +173,24 @@ fn user_verification_server_resolution_invalidates_the_inflight_result() { pending.note_server_request(&request); let attempt = pending .user_verification - .begin("deployments", request.id()) + .begin(thread_id(&request), "deployments", request.id()) .expect("pending request"); assert_eq!( - pending.resolve_notification("unused", request.id()), - Some(ResolvedAppServerRequest::McpElicitation { - server_name: "deployments".to_string(), - request_id: request.id().clone() + pending.resolve_notification(thread_id(&request), request.id()), + Some(ResolvedThreadAppServerRequest { + thread_id: thread_id(&request).to_string(), + request: ResolvedAppServerRequest::McpElicitation { + server_name: "deployments".to_string(), + request_id: request.id().clone(), + }, }) ); - assert!( - !pending - .user_verification - .is_pending("deployments", request.id(), attempt.id) - ); + assert!(!pending.user_verification.is_pending( + thread_id(&request), + "deployments", + request.id(), + attempt.id + )); assert!(attempt.cancelled.is_cancelled()); } @@ -197,7 +221,7 @@ fn user_verification_proof_is_sent_to_the_server_but_not_session_recording() { let mut pending = PendingAppServerRequests::default(); pending.note_server_request(&request); let resolution = pending - .take_resolution("unused", op) + .take_resolution(thread_id(&request), op) .expect("response should serialize") .expect("pending request"); assert_eq!( diff --git a/codex-rs/tui/src/app/user_verification_tests.rs b/codex-rs/tui/src/app/user_verification_tests.rs index de6a217be8..ed1cd6d0a0 100644 --- a/codex-rs/tui/src/app/user_verification_tests.rs +++ b/codex-rs/tui/src/app/user_verification_tests.rs @@ -95,8 +95,6 @@ async fn run_verification_rpc_scenario(scenario: RpcScenario) -> color_eyre::Res unreachable!() }; let thread_id = ThreadId::from_string(¶ms.thread_id)?; - app.pending_app_server_requests - .note_server_request(&request); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; let endpoint = crate::resolve_remote_addr(&format!("ws://{}", listener.local_addr()?))?; let expected_params = serde_json::json!({ "title": "Approve deployment?", "description": "Deploy the reviewed change.", "challenge": "AQID" }); @@ -219,6 +217,12 @@ async fn run_verification_rpc_scenario(scenario: RpcScenario) -> color_eyre::Res crate::app_server_session::ThreadParamsMode::Embedded }; let mut session = AppServerSession::new(client, mode); + let buffered = crate::app::app_server_requests::BufferedServerRequest::new( + request.clone(), + &session.request_handle(), + ); + app.pending_app_server_requests + .note_buffered_request(&buffered); let approval = if scenario == RpcScenario::RemoteWorkspace { event_rx.recv().await.expect("approval event") } else { diff --git a/codex-rs/tui/src/app_event_sender.rs b/codex-rs/tui/src/app_event_sender.rs index 7d064523c9..dac6b8cbe8 100644 --- a/codex-rs/tui/src/app_event_sender.rs +++ b/codex-rs/tui/src/app_event_sender.rs @@ -93,10 +93,16 @@ impl AppEventSender { ))); } - pub(crate) fn user_input_answer(&self, id: String, response: ToolRequestUserInputResponse) { - self.send(AppEvent::CodexOp(AppCommand::user_input_answer( - id, response, - ))); + pub(crate) fn user_input_answer( + &self, + thread_id: ThreadId, + id: String, + response: ToolRequestUserInputResponse, + ) { + self.send(AppEvent::SubmitThreadOp { + thread_id, + op: AppCommand::user_input_answer(id, response), + }); } pub(crate) fn exec_approval( diff --git a/codex-rs/tui/src/app_server_session.rs b/codex-rs/tui/src/app_server_session.rs index 6d15c852f7..fc78348507 100644 --- a/codex-rs/tui/src/app_server_session.rs +++ b/codex-rs/tui/src/app_server_session.rs @@ -1778,14 +1778,6 @@ impl AppServerSession { Ok(()) } - pub(crate) async fn reject_server_request( - &self, - request_id: RequestId, - error: JSONRPCErrorError, - ) -> std::io::Result<()> { - self.client.reject_server_request(request_id, error).await - } - pub(crate) async fn resolve_server_request( &self, request_id: RequestId, diff --git a/codex-rs/tui/src/bottom_pane/request_user_input/mod.rs b/codex-rs/tui/src/bottom_pane/request_user_input/mod.rs index 66c4a1c9dd..51dab27b64 100644 --- a/codex-rs/tui/src/bottom_pane/request_user_input/mod.rs +++ b/codex-rs/tui/src/bottom_pane/request_user_input/mod.rs @@ -51,6 +51,7 @@ use codex_app_server_protocol::ToolRequestUserInputOption; use codex_app_server_protocol::ToolRequestUserInputParams; use codex_app_server_protocol::ToolRequestUserInputQuestion; use codex_app_server_protocol::ToolRequestUserInputResponse; +use codex_protocol::ThreadId; use codex_protocol::user_input::TextElement; use unicode_width::UnicodeWidthStr; @@ -897,7 +898,12 @@ impl RequestUserInputOverlay { }, ); } + let Ok(thread_id) = ThreadId::from_string(&self.request.thread_id) else { + tracing::error!("user input request had an invalid thread ID"); + return; + }; self.app_event_tx.user_input_answer( + thread_id, self.request.turn_id.clone(), ToolRequestUserInputResponse { answers: answers.clone(), @@ -916,7 +922,12 @@ impl RequestUserInputOverlay { fn submit_empty_auto_resolution(&mut self, now: Instant) { self.confirm_unanswered = None; let answers: HashMap = HashMap::new(); + let Ok(thread_id) = ThreadId::from_string(&self.request.thread_id) else { + tracing::error!("user input request had an invalid thread ID"); + return; + }; self.app_event_tx.user_input_answer( + thread_id, self.request.turn_id.clone(), ToolRequestUserInputResponse { answers: answers.clone(), @@ -1705,7 +1716,7 @@ mod tests { questions: Vec, ) -> ToolRequestUserInputParams { ToolRequestUserInputParams { - thread_id: "thread-1".to_string(), + thread_id: "00000000-0000-0000-0000-000000000001".to_string(), item_id: "call-1".to_string(), turn_id: turn_id.to_string(), questions, @@ -1714,6 +1725,19 @@ mod tests { } } + fn expect_user_input_answer( + event: AppEvent, + ) -> (ThreadId, String, ToolRequestUserInputResponse) { + let AppEvent::SubmitThreadOp { + thread_id, + op: Op::UserInputAnswer { id, response }, + } = event + else { + panic!("expected thread-scoped UserInputAnswer event"); + }; + (thread_id, id, response) + } + fn request_event_with_auto_resolution( turn_id: &str, questions: Vec, @@ -1851,9 +1875,8 @@ mod tests { assert!(overlay.done); let event = rx.try_recv().expect("expected UserInputAnswer event"); - let AppEvent::CodexOp(Op::UserInputAnswer { id, response }) = event else { - panic!("expected UserInputAnswer event"); - }; + let (thread_id, id, response) = expect_user_input_answer(event); + assert_eq!(thread_id.to_string(), overlay.request.thread_id); assert_eq!(id, "turn-1"); assert_eq!(response.answers, HashMap::new()); } @@ -1976,9 +1999,8 @@ mod tests { assert!(overlay.done); let event = rx.try_recv().expect("expected UserInputAnswer event"); - let AppEvent::CodexOp(Op::UserInputAnswer { id, response }) = event else { - panic!("expected UserInputAnswer event"); - }; + let (thread_id, id, response) = expect_user_input_answer(event); + assert_eq!(thread_id.to_string(), overlay.request.thread_id); assert_eq!(id, "turn-1"); assert_eq!(response.answers, HashMap::new()); @@ -2155,7 +2177,7 @@ mod tests { let (tx, mut rx) = test_sender(); let mut overlay = RequestUserInputOverlay::new( ToolRequestUserInputParams { - thread_id: "thread-1".to_string(), + thread_id: "00000000-0000-0000-0000-000000000001".to_string(), item_id: "call-1".to_string(), turn_id: "turn-1".to_string(), questions: vec![question_with_options("q1", "First")], @@ -2222,9 +2244,8 @@ mod tests { overlay.submit_answers(); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { id, response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (thread_id, id, response) = expect_user_input_answer(event); + assert_eq!(thread_id.to_string(), overlay.request.thread_id); assert_eq!(id, "turn-1"); let answer = response.answers.get("q1").expect("answer missing"); assert_eq!(answer.answers, Vec::::new()); @@ -2244,9 +2265,7 @@ mod tests { overlay.handle_key_event(KeyEvent::from(KeyCode::Enter)); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (_, _, response) = expect_user_input_answer(event); let answer = response.answers.get("q1").expect("answer missing"); assert_eq!(answer.answers, vec!["Option 1".to_string()]); } @@ -2280,9 +2299,7 @@ mod tests { overlay.handle_key_event(KeyEvent::from(KeyCode::Enter)); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (_, _, response) = expect_user_input_answer(event); let mut expected = HashMap::new(); expected.insert( "q1".to_string(), @@ -2313,9 +2330,7 @@ mod tests { overlay.handle_key_event(KeyEvent::from(KeyCode::Char('2'))); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (_, _, response) = expect_user_input_answer(event); let answer = response.answers.get("q1").expect("answer missing"); assert_eq!(answer.answers, vec!["Option 2".to_string()]); } @@ -2683,9 +2698,7 @@ mod tests { overlay.handle_key_event(KeyEvent::from(KeyCode::Enter)); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (_, _, response) = expect_user_input_answer(event); let answer = response.answers.get("q1").expect("answer missing"); assert_eq!(answer.answers, Vec::::new()); let answer = response.answers.get("q2").expect("answer missing"); @@ -3074,9 +3087,7 @@ mod tests { overlay.submit_answers(); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (_, _, response) = expect_user_input_answer(event); let answer = response.answers.get("q1").expect("answer missing"); assert_eq!(answer.answers, Vec::::new()); } @@ -3099,9 +3110,7 @@ mod tests { overlay.submit_answers(); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (_, _, response) = expect_user_input_answer(event); let answer = response.answers.get("q1").expect("answer missing"); assert_eq!(answer.answers, Vec::::new()); } @@ -3142,9 +3151,7 @@ mod tests { overlay.submit_answers(); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (_, _, response) = expect_user_input_answer(event); let answer = response.answers.get("q1").expect("answer missing"); assert_eq!(answer.answers, Vec::::new()); } @@ -3178,9 +3185,7 @@ mod tests { overlay.submit_answers(); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (_, _, response) = expect_user_input_answer(event); let answer = response.answers.get("q1").expect("answer missing"); assert_eq!( answer.answers, @@ -3292,9 +3297,7 @@ mod tests { overlay.submit_answers(); let event = rx.try_recv().expect("expected AppEvent"); - let AppEvent::CodexOp(Op::UserInputAnswer { response, .. }) = event else { - panic!("expected UserInputAnswer"); - }; + let (_, _, response) = expect_user_input_answer(event); let answer = response.answers.get("q1").expect("answer missing"); assert_eq!( answer.answers,