github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-20 08:19:50 +00:00
Code Issues Packages Projects Releases Wiki Activity

Route TUI request replies to their originating connection and thread (#53436)

Browse Source 
## Why

Pending requests can outlive a session switch, and different threads or connections can reuse request IDs. Replies and resolution notifications must stay associated with the request that prompted them.

## What changed

- Retain the original app-server reply handle for pending and buffered requests, including requests replayed after thread attachment or session reset.
- Scope user input, MCP elicitation, and user verification bookkeeping by thread, and send user input answers to the requesting thread.
- Keep background resolution notifications from dismissing foreground prompts.
- Consume failed replies to closed connections and clear pending replay state without falling through to another request route.

## Testing

Add WebSocket regression tests for direct and buffered approvals with colliding request IDs and replies to closed connections. Extend coverage for foreground prompt preservation and thread-scoped user verification cancellation.

GitOrigin-RevId: 541acb9910e92f6f2e83c27d2a2384185d4f8fec
This commit is contained in:
Benjamin Carlsson
2026-10-12 17:39:17 +00:00
committed by copyberry
parent b3c3bad976
commit 6f47fcb0be
19 changed files with 881 additions and 403 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/tui/src/app/agents_overview.rs
+10 -7
View File
@@ -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<ThreadId, crate::app_server_session::AppServerStartedThread>,
pub(super) input_states: HashMap<ThreadId, ThreadInputState>,
pub(super) new_session_draft: Option<Box<StartupDraftPump>>,
pub(super) dispatched_requests: HashMap<ThreadId, Vec<ServerRequest>>,
pub(super) dispatched_requests: HashMap<ThreadId, Vec<BufferedServerRequest>>,
}
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::<Vec<_>>();
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;
}
}
codex-rs/tui/src/app/agents_overview_details.rs
+5 -1
View File
@@ -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
codex-rs/tui/src/app/agents_overview_tests.rs
+9 -2
View File
@@ -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);
codex-rs/tui/src/app/app_server_events.rs
+97 -54
View File
@@ -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() != &notification.request_id);
requests.retain(|request| request.request.id() != &notification.request_id);
}
if let Some(request) = self
if let Some(resolved) = self
.pending_app_server_requests
.resolve_notification(&notification.thread_id, &notification.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}");
codex-rs/tui/src/app/app_server_requests.rs
+239 -122
View File
@@ -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<String, VecDeque<PendingUserInputRequest>>,
mcp_requests: HashMap<McpRequestKey, AppServerRequestId>,
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<PendingUserInputRequest>>,
mcp_requests: HashMap<McpRequestKey, PendingServerRequest>,
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<UnsupportedAppServerRequest> {
self.note_server_request_owned(request, /*reply*/ None)
}
fn note_server_request_owned(
&mut self,
request: &ServerRequest,
reply: Option<AppServerReplyHandle>,
) -> Option<UnsupportedAppServerRequest> {
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(&params.thread_id), approval_id),
request_id.clone(),
pending(request_id),
);
None
}
@@ -127,7 +142,7 @@ impl PendingAppServerRequests {
Self::canonical_thread_id(&params.thread_id),
params.item_id.clone(),
),
request_id.clone(),
pending(request_id),
);
None
}
@@ -149,28 +164,34 @@ impl PendingAppServerRequests {
Self::canonical_thread_id(&params.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(&params.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(&params.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<UnsupportedAppServerRequest> {
self.note_server_request_owned(&request.request, Some(request.reply.clone()))
}
#[cfg(test)]
pub(super) fn take_resolution<T>(
&mut self,
thread_id: &str,
op: T,
) -> Result<Option<AppServerRequestResolution>, String>
where
T: Into<AppCommand>,
{
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<T>(
&mut self,
thread_id: &str,
op: T,
) -> Result<Option<PendingResolution>, String>
where
T: Into<AppCommand>,
{
@@ -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, String>(AppServerRequestResolution {
request_id,
.remove(&(thread_id.clone(), id.clone()))
.map(|request| {
Ok::<PendingResolution, String>(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, String>(AppServerRequestResolution {
request_id,
.remove(&(thread_id.clone(), id.clone()))
.map(|request| {
Ok::<PendingResolution, String>(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, String>(AppServerRequestResolution {
request_id,
.remove(&(thread_id.clone(), id.clone()))
.map(|request| {
Ok::<PendingResolution, String>(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, String>(AppServerRequestResolution {
request_id: pending.request_id,
Ok::<PendingResolution, String>(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, String>(AppServerRequestResolution {
request_id,
.map(|request| {
Ok::<PendingResolution, String>(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<ResolvedAppServerRequest> {
) -> Option<ResolvedThreadAppServerRequest> {
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(&params.item_id);
self.exec_approvals
.get(&(
Self::canonical_thread_id(&params.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(&params.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(&params.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(&params.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(&params.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<BufferedServerRequest> {
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<PendingUserInputRequest> {
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<PendingUserInputRequest> {
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<AppServerReplyHandle>,
}
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;
codex-rs/tui/src/app/retained_reply_tests.rs
+210
View File
@@ -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<JSONRPCMessage>,
JoinHandle<color_eyre::Result<()>>,
)> {
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(())
}
codex-rs/tui/src/app/session_lifecycle.rs
+29 -31
View File
@@ -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 {
codex-rs/tui/src/app/tests.rs
+81 -19
View File
@@ -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,
codex-rs/tui/src/app/tests/fork_workspace_roots_tests.rs
+13 -5
View File
@@ -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::<Vec<_>>(),
pending.iter().collect::<Vec<_>>()
);
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());
codex-rs/tui/src/app/tests/startup.rs
-16
View File
@@ -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,
codex-rs/tui/src/app/tests/user_verification_routes_tests.rs
+1 -1
View File
@@ -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)
codex-rs/tui/src/app/thread_routing.rs
+17 -24
View File
@@ -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<bool> {
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) {
codex-rs/tui/src/app/user_verification.rs
+11 -6
View File
@@ -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(());
}
codex-rs/tui/src/app/user_verification_requests.rs
+36 -21
View File
@@ -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<Uuid>,
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 {
} = &params.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<UserVerificationAttempt> {
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::<Vec<_>>();
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) {
codex-rs/tui/src/app/user_verification_requests_tests.rs
+66 -42
View File
@@ -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!()
};
&params.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(&params.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!(
codex-rs/tui/src/app/user_verification_tests.rs
+6 -2
View File
@@ -95,8 +95,6 @@ async fn run_verification_rpc_scenario(scenario: RpcScenario) -> color_eyre::Res
unreachable!()
};
let thread_id = ThreadId::from_string(&params.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 {
codex-rs/tui/src/app_event_sender.rs
+10 -4
View File
@@ -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(
codex-rs/tui/src/app_server_session.rs
-8
View File
@@ -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,
codex-rs/tui/src/bottom_pane/request_user_input/mod.rs
+41 -38
View File
@@ -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<String, ToolRequestUserInputAnswer> = 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<ToolRequestUserInputQuestion>,
) -> 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<ToolRequestUserInputQuestion>,
@@ -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::<String>::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::<String>::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::<String>::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::<String>::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::<String>::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,