diff --git a/codex-rs/app-server-client/src/lib.rs b/codex-rs/app-server-client/src/lib.rs index abc844167b..ae383762f3 100644 --- a/codex-rs/app-server-client/src/lib.rs +++ b/codex-rs/app-server-client/src/lib.rs @@ -97,8 +97,8 @@ pub type RequestResult = std::result::Result; #[derive(Debug, Clone)] pub enum AppServerEvent { Lagged { skipped: usize }, - ServerNotification(ServerNotification), - ServerRequest(ServerRequest), + ServerNotification(Box), + ServerRequest(Box), Disconnected { message: String }, } @@ -205,7 +205,7 @@ where *skipped_events = skipped_events.saturating_add(1); warn!("dropping in-process app-server event because consumer queue is full"); if let InProcessServerEvent::ServerRequest(request) = event { - reject_server_request(request); + reject_server_request(*request); } return ForwardEventResult::Continue; } @@ -231,7 +231,7 @@ where *skipped_events = skipped_events.saturating_add(1); warn!("dropping in-process app-server event because consumer queue is full"); if let InProcessServerEvent::ServerRequest(request) = event { - reject_server_request(request); + reject_server_request(*request); } ForwardEventResult::Continue } @@ -522,9 +522,9 @@ impl InProcessAppServerClient { let Some(event) = event else { break; }; - if let InProcessServerEvent::ServerRequest( - ServerRequest::ChatgptAuthTokensRefresh { request_id, .. } - ) = &event + if let InProcessServerEvent::ServerRequest(request) = &event + && let ServerRequest::ChatgptAuthTokensRefresh { request_id, .. } = + request.as_ref() { let send_result = request_sender.fail_server_request( request_id.clone(), @@ -1332,9 +1332,9 @@ mod tests { async fn forward_in_process_event_preserves_transcript_notifications_under_backpressure() { let (event_tx, mut event_rx) = mpsc::channel(1); event_tx - .send(InProcessServerEvent::ServerNotification( + .send(InProcessServerEvent::ServerNotification(Box::new( command_execution_output_delta_notification("stdout-1"), - )) + ))) .await .expect("initial event should enqueue"); @@ -1342,8 +1342,8 @@ mod tests { let result = forward_in_process_event( &event_tx, &mut skipped_events, - InProcessServerEvent::ServerNotification(command_execution_output_delta_notification( - "stdout-2", + InProcessServerEvent::ServerNotification(Box::new( + command_execution_output_delta_notification("stdout-2"), )), |_| {}, ) @@ -1372,7 +1372,7 @@ mod tests { let result = forward_in_process_event( &event_tx, &mut skipped_events, - InProcessServerEvent::ServerNotification(notification), + InProcessServerEvent::ServerNotification(Box::new(notification)), |_| {}, ) .await; @@ -1385,9 +1385,12 @@ mod tests { .expect("receiver task should join successfully"); assert!(matches!( &events[0], - InProcessServerEvent::ServerNotification( - ServerNotification::CommandExecutionOutputDelta(notification) - ) if notification.delta == "stdout-1" + InProcessServerEvent::ServerNotification(notification) + if matches!( + notification.as_ref(), + ServerNotification::CommandExecutionOutputDelta(notification) + if notification.delta == "stdout-1" + ) )); assert!(matches!( &events[1], @@ -1395,24 +1398,35 @@ mod tests { )); assert!(matches!( &events[2], - InProcessServerEvent::ServerNotification(ServerNotification::AgentMessageDelta( - notification - )) if notification.delta == "hello" + InProcessServerEvent::ServerNotification(notification) + if matches!( + notification.as_ref(), + ServerNotification::AgentMessageDelta(notification) + if notification.delta == "hello" + ) )); assert!(matches!( &events[3], - InProcessServerEvent::ServerNotification(ServerNotification::ItemCompleted( - notification - )) if matches!( - ¬ification.item, - codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } if text == "hello" - ) + InProcessServerEvent::ServerNotification(notification) + if matches!( + notification.as_ref(), + ServerNotification::ItemCompleted(notification) + if matches!( + ¬ification.item, + codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } + if text == "hello" + ) + ) )); assert!(matches!( &events[4], - InProcessServerEvent::ServerNotification(ServerNotification::TurnCompleted( - notification - )) if notification.turn.status == codex_app_server_protocol::TurnStatus::Completed + InProcessServerEvent::ServerNotification(notification) + if matches!( + notification.as_ref(), + ServerNotification::TurnCompleted(notification) + if notification.turn.status + == codex_app_server_protocol::TurnStatus::Completed + ) )); } @@ -1753,7 +1767,8 @@ mod tests { let event = client.next_event().await.expect("event should arrive"); assert!(matches!( event, - AppServerEvent::ServerNotification(ServerNotification::AccountUpdated(_)) + AppServerEvent::ServerNotification(notification) + if matches!(notification.as_ref(), ServerNotification::AccountUpdated(_)) )); client.shutdown().await.expect("shutdown should complete"); @@ -1799,9 +1814,12 @@ mod tests { .expect("event stream should stay open"); assert!(matches!( first_event, - AppServerEvent::ServerNotification(ServerNotification::CommandExecutionOutputDelta( - notification - )) if notification.delta == "stdout-1" + AppServerEvent::ServerNotification(notification) + if matches!( + notification.as_ref(), + ServerNotification::CommandExecutionOutputDelta(notification) + if notification.delta == "stdout-1" + ) )); let mut remaining_events = Vec::new(); @@ -1818,30 +1836,31 @@ mod tests { for event in &remaining_events { match event { AppServerEvent::Lagged { skipped: 1 } => {} - AppServerEvent::ServerNotification( - ServerNotification::CommandExecutionOutputDelta(notification), - ) if notification.delta == "stdout-2" => {} - AppServerEvent::ServerNotification(ServerNotification::AgentMessageDelta( - notification, - )) if notification.delta == "hello" => { - transcript_event_names.push("agent_message_delta"); - } - AppServerEvent::ServerNotification(ServerNotification::ItemCompleted( - notification, - )) if matches!( - ¬ification.item, - codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } if text == "hello" - ) => - { - transcript_event_names.push("item_completed"); - } - AppServerEvent::ServerNotification(ServerNotification::TurnCompleted( - notification, - )) if notification.turn.status - == codex_app_server_protocol::TurnStatus::Completed => - { - transcript_event_names.push("turn_completed"); - } + AppServerEvent::ServerNotification(notification) => match notification.as_ref() { + ServerNotification::CommandExecutionOutputDelta(notification) + if notification.delta == "stdout-2" => {} + ServerNotification::AgentMessageDelta(notification) + if notification.delta == "hello" => + { + transcript_event_names.push("agent_message_delta"); + } + ServerNotification::ItemCompleted(notification) + if matches!( + ¬ification.item, + codex_app_server_protocol::ThreadItem::AgentMessage { text, .. } + if text == "hello" + ) => + { + transcript_event_names.push("item_completed"); + } + ServerNotification::TurnCompleted(notification) + if notification.turn.status + == codex_app_server_protocol::TurnStatus::Completed => + { + transcript_event_names.push("turn_completed"); + } + _ => panic!("unexpected remaining event: {event:?}"), + }, _ => panic!("unexpected remaining event: {event:?}"), } } @@ -2103,7 +2122,7 @@ mod tests { #[test] fn event_requires_delivery_marks_transcript_and_terminal_events() { assert!(event_requires_delivery( - &InProcessServerEvent::ServerNotification( + &InProcessServerEvent::ServerNotification(Box::new( codex_app_server_protocol::ServerNotification::TurnCompleted( codex_app_server_protocol::TurnCompletedNotification { thread_id: "thread".to_string(), @@ -2119,10 +2138,10 @@ mod tests { }, } ) - ) + )) )); assert!(event_requires_delivery( - &InProcessServerEvent::ServerNotification( + &InProcessServerEvent::ServerNotification(Box::new( codex_app_server_protocol::ServerNotification::AgentMessageDelta( codex_app_server_protocol::AgentMessageDeltaNotification { thread_id: "thread".to_string(), @@ -2131,10 +2150,10 @@ mod tests { delta: "hello".to_string(), } ) - ) + )) )); assert!(event_requires_delivery( - &InProcessServerEvent::ServerNotification( + &InProcessServerEvent::ServerNotification(Box::new( codex_app_server_protocol::ServerNotification::ItemCompleted( codex_app_server_protocol::ItemCompletedNotification { thread_id: "thread".to_string(), @@ -2148,23 +2167,23 @@ mod tests { }, } ) - ) + )) )); assert!(event_requires_delivery( - &InProcessServerEvent::ServerNotification( + &InProcessServerEvent::ServerNotification(Box::new( codex_app_server_protocol::ServerNotification::ExternalAgentConfigImportCompleted( codex_app_server_protocol::ExternalAgentConfigImportCompletedNotification { import_id: "import".to_string(), item_type_results: Vec::new(), }, ) - ) + )) )); assert!(!event_requires_delivery(&InProcessServerEvent::Lagged { skipped: 1 })); assert!(!event_requires_delivery( - &InProcessServerEvent::ServerNotification( + &InProcessServerEvent::ServerNotification(Box::new( codex_app_server_protocol::ServerNotification::CommandExecutionOutputDelta( codex_app_server_protocol::CommandExecutionOutputDeltaNotification { thread_id: "thread".to_string(), @@ -2173,7 +2192,7 @@ mod tests { delta: "stdout".to_string(), } ) - ) + )) )); } diff --git a/codex-rs/app-server-client/src/remote.rs b/codex-rs/app-server-client/src/remote.rs index e8cb7122f6..28ad04ea89 100644 --- a/codex-rs/app-server-client/src/remote.rs +++ b/codex-rs/app-server-client/src/remote.rs @@ -348,7 +348,7 @@ impl RemoteAppServerClient { Ok(request) => { if let Err(err) = deliver_event( &event_tx, - AppServerEvent::ServerRequest(request), + AppServerEvent::ServerRequest(Box::new(request)), ) { warn!(%err, "failed to deliver remote app-server server request"); @@ -862,7 +862,8 @@ where let method = request.method.clone(); match ServerRequest::try_from(request) { Ok(request) => { - pending_events.push(AppServerEvent::ServerRequest(request)); + pending_events + .push(AppServerEvent::ServerRequest(Box::new(request))); } Err(err) => { warn!(%err, method, "rejecting unknown remote app-server request during initialize"); @@ -940,7 +941,7 @@ where fn app_server_event_from_notification(notification: JSONRPCNotification) -> Option { match ServerNotification::try_from(notification) { - Ok(notification) => Some(AppServerEvent::ServerNotification(notification)), + Ok(notification) => Some(AppServerEvent::ServerNotification(Box::new(notification))), Err(_) => None, } } diff --git a/codex-rs/app-server/src/in_process.rs b/codex-rs/app-server/src/in_process.rs index 4a81f652e6..07a6f83029 100644 --- a/codex-rs/app-server/src/in_process.rs +++ b/codex-rs/app-server/src/in_process.rs @@ -160,9 +160,9 @@ pub struct InProcessStartArgs { #[derive(Debug, Clone)] pub enum InProcessServerEvent { /// Server request that requires client response/rejection. - ServerRequest(ServerRequest), + ServerRequest(Box), /// App-server notification directed to the embedded client. - ServerNotification(ServerNotification), + ServerNotification(Box), /// Indicates one or more events were dropped due to backpressure. Lagged { skipped: usize }, } @@ -663,7 +663,7 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult ( @@ -695,14 +695,18 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult { diff --git a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs index 030131a3db..decf45e98d 100644 --- a/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs +++ b/codex-rs/app-server/tests/suite/v2/remote_thread_store.rs @@ -191,9 +191,8 @@ async fn thread_delete_with_non_local_thread_store_does_not_create_local_persist let Some(event) = client.next_event().await else { anyhow::bail!("in-process app-server stopped before turn/completed"); }; - if let InProcessServerEvent::ServerNotification(ServerNotification::TurnCompleted( - completed, - )) = event + if let InProcessServerEvent::ServerNotification(notification) = event + && let ServerNotification::TurnCompleted(completed) = notification.as_ref() && completed.thread_id == thread.id { return Ok::<(), anyhow::Error>(()); @@ -332,9 +331,8 @@ async fn cold_thread_resume_reuses_non_local_history_probe() -> Result<()> { let Some(event) = client.next_event().await else { anyhow::bail!("in-process app-server stopped before turn/completed"); }; - if let InProcessServerEvent::ServerNotification(ServerNotification::TurnCompleted( - completed, - )) = event + if let InProcessServerEvent::ServerNotification(notification) = event + && let ServerNotification::TurnCompleted(completed) = notification.as_ref() && completed.thread_id == thread.id { return Ok::<(), anyhow::Error>(()); diff --git a/codex-rs/exec/src/lib.rs b/codex-rs/exec/src/lib.rs index 6c031b5418..31a502da31 100644 --- a/codex-rs/exec/src/lib.rs +++ b/codex-rs/exec/src/lib.rs @@ -997,9 +997,10 @@ async fn run_exec_session(args: ExecRunArgs) -> anyhow::Result<()> { match server_event { InProcessServerEvent::ServerRequest(request) => { - handle_server_request(&client, request, &mut error_seen).await; + handle_server_request(&client, *request, &mut error_seen).await; } - InProcessServerEvent::ServerNotification(mut notification) => { + InProcessServerEvent::ServerNotification(notification) => { + let mut notification = *notification; if let ServerNotification::Error(payload) = ¬ification { if payload.thread_id == primary_thread_id_for_requests && payload.turn_id == task_id diff --git a/codex-rs/tui/src/app/agent_status_feed.rs b/codex-rs/tui/src/app/agent_status_feed.rs index efaa6c5ad8..893fd740bf 100644 --- a/codex-rs/tui/src/app/agent_status_feed.rs +++ b/codex-rs/tui/src/app/agent_status_feed.rs @@ -85,14 +85,12 @@ impl AgentStatusThreadPreview { let mut activity = Vec::new(); for event in events { let item = match event { - ThreadBufferedEvent::Notification(ServerNotification::ItemCompleted(event)) => { - &event.item - } - ThreadBufferedEvent::Notification(ServerNotification::ItemStarted(event)) => { - &event.item - } - ThreadBufferedEvent::Notification(_) - | ThreadBufferedEvent::Request(_) + ThreadBufferedEvent::Notification(notification) => match notification.as_ref() { + ServerNotification::ItemCompleted(event) => &event.item, + ServerNotification::ItemStarted(event) => &event.item, + _ => continue, + }, + ThreadBufferedEvent::Request(_) | ThreadBufferedEvent::HistoryEntryResponse(_) | ThreadBufferedEvent::FeedbackSubmission(_) => continue, }; diff --git a/codex-rs/tui/src/app/app_server_events.rs b/codex-rs/tui/src/app/app_server_events.rs index 188fbd8dc8..aae6b33871 100644 --- a/codex-rs/tui/src/app/app_server_events.rs +++ b/codex-rs/tui/src/app/app_server_events.rs @@ -44,11 +44,11 @@ impl App { self.chat_widget.finish_mcp_startup_after_lag(); } AppServerEvent::ServerNotification(notification) => { - self.handle_server_notification_event(app_server_client, notification) + self.handle_server_notification_event(app_server_client, *notification) .await; } AppServerEvent::ServerRequest(request) => { - self.handle_server_request_event(app_server_client, request) + self.handle_server_request_event(app_server_client, *request) .await; } AppServerEvent::Disconnected { message } => { diff --git a/codex-rs/tui/src/app/background_requests.rs b/codex-rs/tui/src/app/background_requests.rs index 69f0612f18..c27a5e8ec3 100644 --- a/codex-rs/tui/src/app/background_requests.rs +++ b/codex-rs/tui/src/app/background_requests.rs @@ -626,7 +626,7 @@ impl App { { guard .pending_interactive_replay - .note_evicted_server_request(request); + .note_evicted_server_request(request.as_ref()); } guard.active }; diff --git a/codex-rs/tui/src/app/pending_interactive_replay.rs b/codex-rs/tui/src/app/pending_interactive_replay.rs index cc47f51ae0..a45c0757c5 100644 --- a/codex-rs/tui/src/app/pending_interactive_replay.rs +++ b/codex-rs/tui/src/app/pending_interactive_replay.rs @@ -702,8 +702,12 @@ mod tests { assert_eq!(snapshot.events.len(), 1); assert!(matches!( snapshot.events.first(), - Some(ThreadBufferedEvent::Request(ServerRequest::ToolRequestUserInput { params, .. })) - if params.item_id == "call-1" + Some(ThreadBufferedEvent::Request(request)) + if matches!( + request.as_ref(), + ServerRequest::ToolRequestUserInput { params, .. } + if params.item_id == "call-1" + ) )); } @@ -738,7 +742,11 @@ mod tests { snapshot.events.iter().all(|event| { !matches!( event, - ThreadBufferedEvent::Request(ServerRequest::ToolRequestUserInput { .. }) + ThreadBufferedEvent::Request(request) + if matches!( + request.as_ref(), + ServerRequest::ToolRequestUserInput { .. } + ) ) }), "server-resolved request_user_input prompt should not replay on thread switch" @@ -783,9 +791,11 @@ mod tests { snapshot.events.iter().all(|event| { !matches!( event, - ThreadBufferedEvent::Request( - ServerRequest::CommandExecutionRequestApproval { .. } - ) + ThreadBufferedEvent::Request(request) + if matches!( + request.as_ref(), + ServerRequest::CommandExecutionRequestApproval { .. } + ) ) }), "server-resolved exec approval prompt should not replay on thread switch" @@ -810,8 +820,12 @@ mod tests { assert_eq!(snapshot.events.len(), 1); assert!(matches!( snapshot.events.first(), - Some(ThreadBufferedEvent::Request(ServerRequest::ToolRequestUserInput { params, .. })) - if params.item_id == "call-2" + Some(ThreadBufferedEvent::Request(request)) + if matches!( + request.as_ref(), + ServerRequest::ToolRequestUserInput { params, .. } + if params.item_id == "call-2" + ) )); } @@ -832,8 +846,12 @@ mod tests { assert_eq!(snapshot.events.len(), 1); assert!(matches!( snapshot.events.first(), - Some(ThreadBufferedEvent::Request(ServerRequest::ToolRequestUserInput { params, .. })) - if params.item_id == "call-2" + Some(ThreadBufferedEvent::Request(request)) + if matches!( + request.as_ref(), + ServerRequest::ToolRequestUserInput { params, .. } + if params.item_id == "call-2" + ) )); } @@ -869,8 +887,12 @@ mod tests { assert!(snapshot.events.iter().all(|event| { !matches!( event, - ThreadBufferedEvent::Request(ServerRequest::CommandExecutionRequestApproval { .. }) - | ThreadBufferedEvent::Request(ServerRequest::FileChangeRequestApproval { .. }) + ThreadBufferedEvent::Request(request) + if matches!( + request.as_ref(), + ServerRequest::CommandExecutionRequestApproval { .. } + | ServerRequest::FileChangeRequestApproval { .. } + ) ) })); } @@ -935,7 +957,11 @@ mod tests { assert!(store.snapshot().events.iter().all(|event| { !matches!( event, - ThreadBufferedEvent::Request(ServerRequest::CommandExecutionRequestApproval { .. }) + ThreadBufferedEvent::Request(request) + if matches!( + request.as_ref(), + ServerRequest::CommandExecutionRequestApproval { .. } + ) ) })); } diff --git a/codex-rs/tui/src/app/replay_filter.rs b/codex-rs/tui/src/app/replay_filter.rs index 273ca0dd5a..df5a90b8f2 100644 --- a/codex-rs/tui/src/app/replay_filter.rs +++ b/codex-rs/tui/src/app/replay_filter.rs @@ -10,13 +10,15 @@ pub(super) fn snapshot_has_pending_interactive_request(snapshot: &ThreadEventSna snapshot.events.iter().any(|event| { matches!( event, - ThreadBufferedEvent::Request( - ServerRequest::CommandExecutionRequestApproval { .. } - | ServerRequest::FileChangeRequestApproval { .. } - | ServerRequest::McpServerElicitationRequest { .. } - | ServerRequest::PermissionsRequestApproval { .. } - | ServerRequest::ToolRequestUserInput { .. } - ) + ThreadBufferedEvent::Request(request) + if matches!( + request.as_ref(), + ServerRequest::CommandExecutionRequestApproval { .. } + | ServerRequest::FileChangeRequestApproval { .. } + | ServerRequest::McpServerElicitationRequest { .. } + | ServerRequest::PermissionsRequestApproval { .. } + | ServerRequest::ToolRequestUserInput { .. } + ) ) }) } @@ -24,10 +26,12 @@ pub(super) fn snapshot_has_pending_interactive_request(snapshot: &ThreadEventSna pub(super) fn event_is_notice(event: &ThreadBufferedEvent) -> bool { matches!( event, - ThreadBufferedEvent::Notification( - ServerNotification::Warning(_) - | ServerNotification::GuardianWarning(_) - | ServerNotification::ConfigWarning(_) - ) + ThreadBufferedEvent::Notification(notification) + if matches!( + notification.as_ref(), + ServerNotification::Warning(_) + | ServerNotification::GuardianWarning(_) + | ServerNotification::ConfigWarning(_) + ) ) } diff --git a/codex-rs/tui/src/app/tests.rs b/codex-rs/tui/src/app/tests.rs index 1a96ce0fe3..c542b2248e 100644 --- a/codex-rs/tui/src/app/tests.rs +++ b/codex-rs/tui/src/app/tests.rs @@ -198,9 +198,8 @@ async fn next_thread_settings_updated( .await .expect("app-server should emit an event") .expect("app-server event stream should remain open"); - if let codex_app_server_client::AppServerEvent::ServerNotification( - ServerNotification::ThreadSettingsUpdated(notification), - ) = event + if let codex_app_server_client::AppServerEvent::ServerNotification(notification) = event + && let ServerNotification::ThreadSettingsUpdated(notification) = *notification && notification.thread_id == thread_id.to_string() { return notification; @@ -289,10 +288,12 @@ async fn enqueue_primary_thread_session_replays_buffered_approval_after_attach() assert!(matches!( &event, - ThreadBufferedEvent::Request(ServerRequest::CommandExecutionRequestApproval { - params, - .. - }) if params.turn_id == "turn-1" + ThreadBufferedEvent::Request(request) + if matches!( + request.as_ref(), + ServerRequest::CommandExecutionRequestApproval { params, .. } + if params.turn_id == "turn-1" + ) )); app.handle_thread_event_now(event); @@ -353,10 +354,12 @@ async fn resolved_buffered_approval_does_not_become_actionable_after_drain() -> assert!(matches!( &event, - ThreadBufferedEvent::Request(ServerRequest::CommandExecutionRequestApproval { - params, - .. - }) if params.turn_id == "turn-1" + ThreadBufferedEvent::Request(request) + if matches!( + request.as_ref(), + ServerRequest::CommandExecutionRequestApproval { params, .. } + if params.turn_id == "turn-1" + ) )); app.handle_thread_event_now(event); @@ -801,9 +804,9 @@ async fn replayed_turn_complete_submits_restored_queued_follow_up() { ThreadEventSnapshot { session: None, turns: Vec::new(), - events: vec![ThreadBufferedEvent::Notification( + events: vec![ThreadBufferedEvent::Notification(Box::new( turn_completed_notification(thread_id, "turn-1", TurnStatus::Completed), - )], + ))], input_state: Some(input_state), }, /*resume_restored_queue*/ true, @@ -854,9 +857,9 @@ async fn replay_only_thread_keeps_restored_queue_visible() { ThreadEventSnapshot { session: None, turns: Vec::new(), - events: vec![ThreadBufferedEvent::Notification( + events: vec![ThreadBufferedEvent::Notification(Box::new( turn_completed_notification(thread_id, "turn-1", TurnStatus::Completed), - )], + ))], input_state: Some(input_state), }, /*resume_restored_queue*/ false, @@ -1026,12 +1029,14 @@ async fn replay_thread_snapshot_does_not_submit_queue_before_replay_catches_up() session: None, turns: Vec::new(), events: vec![ - ThreadBufferedEvent::Notification(turn_completed_notification( + ThreadBufferedEvent::Notification(Box::new(turn_completed_notification( thread_id, "turn-0", TurnStatus::Completed, - )), - ThreadBufferedEvent::Notification(turn_started_notification(thread_id, "turn-1")), + ))), + ThreadBufferedEvent::Notification(Box::new(turn_started_notification( + thread_id, "turn-1", + ))), ], input_state: Some(input_state), }, @@ -1294,9 +1299,9 @@ async fn replayed_interrupted_turn_restores_queued_input_to_composer() { ThreadEventSnapshot { session: None, turns: Vec::new(), - events: vec![ThreadBufferedEvent::Notification( + events: vec![ThreadBufferedEvent::Notification(Box::new( turn_completed_notification(thread_id, "turn-1", TurnStatus::Interrupted), - )], + ))], input_state: Some(input_state), }, /*resume_restored_queue*/ true, @@ -1323,10 +1328,8 @@ async fn token_usage_update_refreshes_status_line_with_runtime_context_window() assert_eq!(app.chat_widget.status_line_text(), None); - app.handle_thread_event_now(ThreadBufferedEvent::Notification(token_usage_notification( - ThreadId::new(), - "turn-1", - Some(950_000), + app.handle_thread_event_now(ThreadBufferedEvent::Notification(Box::new( + token_usage_notification(ThreadId::new(), "turn-1", Some(950_000)), ))); assert_eq!( @@ -1341,7 +1344,7 @@ async fn collab_receiver_notification_caches_thread_without_app_server_read() { let receiver_thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000123").expect("valid thread id"); - app.handle_thread_event_now(ThreadBufferedEvent::Notification( + app.handle_thread_event_now(ThreadBufferedEvent::Notification(Box::new( ServerNotification::ItemStarted(ItemStartedNotification { thread_id: ThreadId::new().to_string(), turn_id: "turn-1".to_string(), @@ -1358,7 +1361,7 @@ async fn collab_receiver_notification_caches_thread_without_app_server_read() { agents_states: HashMap::new(), }, }), - )); + ))); assert_eq!( app.agent_navigation.get(&receiver_thread_id), @@ -1378,7 +1381,7 @@ async fn collab_receiver_notification_does_not_cache_not_found_thread() { let receiver_thread_id = ThreadId::from_string("00000000-0000-0000-0000-000000000124").expect("valid thread id"); - app.handle_thread_event_now(ThreadBufferedEvent::Notification( + app.handle_thread_event_now(ThreadBufferedEvent::Notification(Box::new( ServerNotification::ItemCompleted(codex_app_server_protocol::ItemCompletedNotification { thread_id: ThreadId::new().to_string(), turn_id: "turn-1".to_string(), @@ -1401,7 +1404,7 @@ async fn collab_receiver_notification_does_not_cache_not_found_thread() { )]), }, }), - )); + ))); assert_eq!(app.agent_navigation.get(&receiver_thread_id), None); } @@ -2121,9 +2124,8 @@ async fn handle_start_side_seeds_navigation_before_thread_started() -> Result<() .await .expect("app-server should emit an event") .expect("app-server event stream should remain open"); - if let codex_app_server_client::AppServerEvent::ServerNotification( - ServerNotification::ThreadStarted(notification), - ) = event + if let codex_app_server_client::AppServerEvent::ServerNotification(notification) = event + && let ServerNotification::ThreadStarted(notification) = notification.as_ref() && notification.thread.id == side_thread_id.to_string() { saw_thread_started = true; @@ -2917,18 +2919,18 @@ async fn replay_snapshot_with_pending_request_suppresses_replay_notices() { session: Some(test_thread_session(thread_id, test_path_buf("/tmp/main"))), turns: Vec::new(), events: vec![ - ThreadBufferedEvent::Notification(ServerNotification::Warning( + ThreadBufferedEvent::Notification(Box::new(ServerNotification::Warning( WarningNotification { thread_id: Some(thread_id.to_string()), message: stale_warning.to_string(), }, - )), - ThreadBufferedEvent::Request(exec_approval_request( + ))), + ThreadBufferedEvent::Request(Box::new(exec_approval_request( thread_id, "turn-approval", "call-approval", /*approval_id*/ None, - )), + ))), ], input_state: None, }, @@ -4082,7 +4084,7 @@ async fn primary_thread_ignores_child_mcp_startup_notifications() { app.handle_app_server_event( &app_server, - codex_app_server_client::AppServerEvent::ServerNotification( + codex_app_server_client::AppServerEvent::ServerNotification(Box::new( ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification { thread_id: Some(child_thread_id.to_string()), name: "sentry".to_string(), @@ -4090,7 +4092,7 @@ async fn primary_thread_ignores_child_mcp_startup_notifications() { error: Some("sentry is not logged in".to_string()), failure_reason: None, }), - ), + )), ) .await; @@ -4106,9 +4108,11 @@ async fn primary_thread_ignores_child_mcp_startup_notifications() { assert!( matches!( child_snapshot.events.as_slice(), - [ThreadBufferedEvent::Notification( - ServerNotification::McpServerStatusUpdated(_) - )] + [ThreadBufferedEvent::Notification(notification)] + if matches!( + notification.as_ref(), + ServerNotification::McpServerStatusUpdated(_) + ) ), "child MCP startup notification should be buffered for the child thread" ); @@ -4155,7 +4159,7 @@ async fn app_scoped_mcp_startup_notifications_do_not_render_in_active_thread() { app.handle_app_server_event( &app_server, - codex_app_server_client::AppServerEvent::ServerNotification( + codex_app_server_client::AppServerEvent::ServerNotification(Box::new( ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification { thread_id: None, name: "sentry".to_string(), @@ -4163,7 +4167,7 @@ async fn app_scoped_mcp_startup_notifications_do_not_render_in_active_thread() { error: Some("sentry is not logged in".to_string()), failure_reason: None, }), - ), + )), ) .await; @@ -4219,7 +4223,7 @@ async fn active_side_thread_renders_live_mcp_startup_notifications() { ] { app.handle_app_server_event( &app_server, - codex_app_server_client::AppServerEvent::ServerNotification( + codex_app_server_client::AppServerEvent::ServerNotification(Box::new( ServerNotification::McpServerStatusUpdated(McpServerStatusUpdatedNotification { thread_id: Some(side_thread_id.to_string()), name: "sentry".to_string(), @@ -4228,7 +4232,7 @@ async fn active_side_thread_renders_live_mcp_startup_notifications() { .then(|| "sentry is not logged in".to_string()), failure_reason: None, }), - ), + )), ) .await; } @@ -6706,7 +6710,7 @@ async fn replace_chat_widget_reseeds_collab_agent_metadata_for_replay() { ThreadEventSnapshot { session: None, turns: Vec::new(), - events: vec![ThreadBufferedEvent::Notification( + events: vec![ThreadBufferedEvent::Notification(Box::new( ServerNotification::ItemStarted( codex_app_server_protocol::ItemStartedNotification { thread_id: "thread-1".to_string(), @@ -6726,7 +6730,7 @@ async fn replace_chat_widget_reseeds_collab_agent_metadata_for_replay() { }, }, ), - )], + ))], input_state: None, }, /*resume_restored_queue*/ false, @@ -7065,9 +7069,9 @@ async fn override_turn_context_sends_thread_settings_update() { app.handle_app_server_event( &app_server, - codex_app_server_client::AppServerEvent::ServerNotification( + codex_app_server_client::AppServerEvent::ServerNotification(Box::new( ServerNotification::ThreadSettingsUpdated(notification), - ), + )), ) .await; let updated_session = app diff --git a/codex-rs/tui/src/app/tests/rate_limits.rs b/codex-rs/tui/src/app/tests/rate_limits.rs index 1844c7fea6..419a37ac3e 100644 --- a/codex-rs/tui/src/app/tests/rate_limits.rs +++ b/codex-rs/tui/src/app/tests/rate_limits.rs @@ -56,11 +56,11 @@ async fn deliver_rolling_rate_limit_snapshot( ) { app.handle_app_server_event( app_server, - codex_app_server_client::AppServerEvent::ServerNotification( + codex_app_server_client::AppServerEvent::ServerNotification(Box::new( ServerNotification::AccountRateLimitsUpdated(AccountRateLimitsUpdatedNotification { rate_limits: snapshot, }), - ), + )), ) .await; } diff --git a/codex-rs/tui/src/app/tests/safety_buffering.rs b/codex-rs/tui/src/app/tests/safety_buffering.rs index dd87907f23..6e0e710f21 100644 --- a/codex-rs/tui/src/app/tests/safety_buffering.rs +++ b/codex-rs/tui/src/app/tests/safety_buffering.rs @@ -107,13 +107,13 @@ async fn next_turn_started( .await .expect("app-server should emit a turn/start event") .expect("app-server event stream should remain open"); - let started_turn_id = match &event { - AppServerEvent::ServerNotification(ServerNotification::TurnStarted(notification)) - if notification.thread_id == thread_id.to_string() => - { - Some(notification.turn.id.clone()) - } - _ => None, + let started_turn_id = if let AppServerEvent::ServerNotification(notification) = &event + && let ServerNotification::TurnStarted(notification) = notification.as_ref() + && notification.thread_id == thread_id.to_string() + { + Some(notification.turn.id.clone()) + } else { + None }; app.handle_app_server_event(app_server, event).await; drain_active_thread_events(app); @@ -138,8 +138,12 @@ async fn wait_for_turn_completed( .expect("app-server event stream should remain open"); let completed = matches!( &event, - AppServerEvent::ServerNotification(ServerNotification::TurnCompleted(notification)) - if notification.thread_id == thread_id.to_string() + AppServerEvent::ServerNotification(notification) + if matches!( + notification.as_ref(), + ServerNotification::TurnCompleted(notification) + if notification.thread_id == thread_id.to_string() + ) ); app.handle_app_server_event(app_server, event).await; drain_active_thread_events(app); @@ -237,12 +241,12 @@ stream_max_retries = 0 } while app_event_rx.try_recv().is_ok() {} - app.handle_thread_event_now(ThreadBufferedEvent::Notification( + app.handle_thread_event_now(ThreadBufferedEvent::Notification(Box::new( ServerNotification::Warning(WarningNotification { thread_id: Some(thread_id.to_string()), message: "event handled while interrupt is pending".to_string(), }), - )); + ))); assert!(matches!( app_event_rx.try_recv(), Ok(AppEvent::InsertHistoryCell(_)) @@ -507,16 +511,18 @@ goals = true app.handle_app_server_event( &app_server, - AppServerEvent::ServerNotification(ServerNotification::ModelSafetyBufferingUpdated( - ModelSafetyBufferingUpdatedNotification { - thread_id: source_thread_id.to_string(), - turn_id: active_turn_id.clone(), - model: CURRENT_MODEL.to_string(), - use_cases: Vec::new(), - reasons: Vec::new(), - show_buffering_ui: true, - faster_model: Some(FASTER_MODEL.to_string()), - }, + AppServerEvent::ServerNotification(Box::new( + ServerNotification::ModelSafetyBufferingUpdated( + ModelSafetyBufferingUpdatedNotification { + thread_id: source_thread_id.to_string(), + turn_id: active_turn_id.clone(), + model: CURRENT_MODEL.to_string(), + use_cases: Vec::new(), + reasons: Vec::new(), + show_buffering_ui: true, + faster_model: Some(FASTER_MODEL.to_string()), + }, + ), )), ) .await; @@ -584,16 +590,18 @@ goals = true .await; app.handle_app_server_event( &app_server, - AppServerEvent::ServerNotification(ServerNotification::ModelSafetyBufferingUpdated( - ModelSafetyBufferingUpdatedNotification { - thread_id: first_retry_thread_id.to_string(), - turn_id: first_retry_turn_id.clone(), - model: FASTER_MODEL.to_string(), - use_cases: Vec::new(), - reasons: Vec::new(), - show_buffering_ui: true, - faster_model: Some(FASTER_MODEL.to_string()), - }, + AppServerEvent::ServerNotification(Box::new( + ServerNotification::ModelSafetyBufferingUpdated( + ModelSafetyBufferingUpdatedNotification { + thread_id: first_retry_thread_id.to_string(), + turn_id: first_retry_turn_id.clone(), + model: FASTER_MODEL.to_string(), + use_cases: Vec::new(), + reasons: Vec::new(), + show_buffering_ui: true, + faster_model: Some(FASTER_MODEL.to_string()), + }, + ), )), ) .await; diff --git a/codex-rs/tui/src/app/thread_events.rs b/codex-rs/tui/src/app/thread_events.rs index 27472880f3..660ffab7ee 100644 --- a/codex-rs/tui/src/app/thread_events.rs +++ b/codex-rs/tui/src/app/thread_events.rs @@ -18,8 +18,8 @@ pub(super) struct ThreadEventSnapshot { #[derive(Debug, Clone)] pub(super) enum ThreadBufferedEvent { - Notification(ServerNotification), - Request(ServerRequest), + Notification(Box), + Request(Box), HistoryEntryResponse(HistoryLookupResponse), FeedbackSubmission(FeedbackThreadEvent), } @@ -53,14 +53,16 @@ pub(super) struct ThreadEventStore { impl ThreadEventStore { pub(super) fn event_survives_session_refresh(event: &ThreadBufferedEvent) -> bool { - matches!( - event, - ThreadBufferedEvent::Request(_) - | ThreadBufferedEvent::Notification(ServerNotification::HookStarted(_)) - | ThreadBufferedEvent::Notification(ServerNotification::HookCompleted(_)) - | ThreadBufferedEvent::Notification(ServerNotification::McpServerStatusUpdated(_)) - | ThreadBufferedEvent::FeedbackSubmission(_) - ) + match event { + ThreadBufferedEvent::Request(_) | ThreadBufferedEvent::FeedbackSubmission(_) => true, + ThreadBufferedEvent::Notification(notification) => matches!( + notification.as_ref(), + ServerNotification::HookStarted(_) + | ServerNotification::HookCompleted(_) + | ServerNotification::McpServerStatusUpdated(_) + ), + ThreadBufferedEvent::HistoryEntryResponse(_) => false, + } } pub(super) fn new(capacity: usize) -> Self { @@ -159,26 +161,29 @@ impl ThreadEventStore { } self.buffer - .push_back(ThreadBufferedEvent::Notification(notification.into_owned())); + .push_back(ThreadBufferedEvent::Notification(Box::new( + notification.into_owned(), + ))); if self.buffer.len() > self.capacity && let Some(removed) = self.buffer.pop_front() && let ThreadBufferedEvent::Request(request) = &removed { self.pending_interactive_replay - .note_evicted_server_request(request); + .note_evicted_server_request(request.as_ref()); } } pub(super) fn push_request(&mut self, request: ServerRequest) { self.pending_interactive_replay .note_server_request(&request); - self.buffer.push_back(ThreadBufferedEvent::Request(request)); + self.buffer + .push_back(ThreadBufferedEvent::Request(Box::new(request))); if self.buffer.len() > self.capacity && let Some(removed) = self.buffer.pop_front() && let ThreadBufferedEvent::Request(request) = &removed { self.pending_interactive_replay - .note_evicted_server_request(request); + .note_evicted_server_request(request.as_ref()); } } @@ -189,9 +194,9 @@ impl ThreadEventStore { ThreadBufferedEvent::Request(request) if self .pending_interactive_replay - .should_replay_snapshot_request(request) => + .should_replay_snapshot_request(request.as_ref()) => { - Some(request.clone()) + Some(request.as_ref().clone()) } ThreadBufferedEvent::Request(_) | ThreadBufferedEvent::Notification(_) @@ -210,18 +215,20 @@ impl ThreadEventStore { .iter() .rev() .find_map(|event| match event { - ThreadBufferedEvent::Notification(ServerNotification::ItemStarted( - notification, - )) if turn_id_matches(turn_id, ¬ification.turn_id) => { - file_change_item_changes(¬ification.item, item_id) - } - ThreadBufferedEvent::Notification(ServerNotification::ItemCompleted( - notification, - )) if turn_id_matches(turn_id, ¬ification.turn_id) => { - file_change_item_changes(¬ification.item, item_id) - } + ThreadBufferedEvent::Notification(notification) => match notification.as_ref() { + ServerNotification::ItemStarted(notification) + if turn_id_matches(turn_id, ¬ification.turn_id) => + { + file_change_item_changes(¬ification.item, item_id) + } + ServerNotification::ItemCompleted(notification) + if turn_id_matches(turn_id, ¬ification.turn_id) => + { + file_change_item_changes(¬ification.item, item_id) + } + _ => None, + }, ThreadBufferedEvent::Request(_) - | ThreadBufferedEvent::Notification(_) | ThreadBufferedEvent::HistoryEntryResponse(_) | ThreadBufferedEvent::FeedbackSubmission(_) => None, }) @@ -247,7 +254,7 @@ impl ThreadEventStore { .filter(|event| match event { ThreadBufferedEvent::Request(request) => self .pending_interactive_replay - .should_replay_snapshot_request(request), + .should_replay_snapshot_request(request.as_ref()), ThreadBufferedEvent::Notification(_) | ThreadBufferedEvent::HistoryEntryResponse(_) | ThreadBufferedEvent::FeedbackSubmission(_) => true, diff --git a/codex-rs/tui/src/app/thread_routing.rs b/codex-rs/tui/src/app/thread_routing.rs index a87d2b5a0f..180abdb784 100644 --- a/codex-rs/tui/src/app/thread_routing.rs +++ b/codex-rs/tui/src/app/thread_routing.rs @@ -632,7 +632,9 @@ impl App { }; if should_send && let Err(error) = thread_event_tx - .send(ThreadBufferedEvent::Notification(notification)) + .send(ThreadBufferedEvent::Notification(Box::new( + notification, + ))) .await { tracing::warn!(error = %error, "thread event channel closed"); @@ -1006,7 +1008,7 @@ impl App { } if let Some(notification) = notification { - match sender.try_send(ThreadBufferedEvent::Notification(notification)) { + match sender.try_send(ThreadBufferedEvent::Notification(Box::new(notification))) { Ok(()) => {} Err(TrySendError::Full(event)) => { tokio::spawn(async move { @@ -1132,7 +1134,7 @@ impl App { let request_status = SideParentStatus::for_request(&request); if should_send { - match sender.try_send(ThreadBufferedEvent::Request(request)) { + match sender.try_send(ThreadBufferedEvent::Request(Box::new(request))) { Ok(()) => {} Err(TrySendError::Full(event)) => { tokio::spawn(async move { @@ -1183,7 +1185,7 @@ impl App { { guard .pending_interactive_replay - .note_evicted_server_request(request); + .note_evicted_server_request(request.as_ref()); } } should_send @@ -1267,11 +1269,11 @@ impl App { for pending_event in pending { match pending_event { ThreadBufferedEvent::Notification(notification) => { - self.enqueue_thread_notification(thread_id, notification) + self.enqueue_thread_notification(thread_id, *notification) .await?; } ThreadBufferedEvent::Request(request) => { - self.enqueue_thread_request(thread_id, request).await?; + self.enqueue_thread_request(thread_id, *request).await?; } ThreadBufferedEvent::HistoryEntryResponse(event) => { self.enqueue_thread_history_entry_response(thread_id, event) @@ -1298,7 +1300,7 @@ impl App { .await; } self.pending_primary_events - .push_back(ThreadBufferedEvent::Notification(notification)); + .push_back(ThreadBufferedEvent::Notification(Box::new(notification))); Ok(()) } @@ -1310,7 +1312,7 @@ impl App { return self.enqueue_thread_request(thread_id, request).await; } self.pending_primary_events - .push_back(ThreadBufferedEvent::Request(request)); + .push_back(ThreadBufferedEvent::Request(Box::new(request))); Ok(()) } @@ -1541,22 +1543,26 @@ impl App { pub(super) fn handle_thread_event_now(&mut self, event: ThreadBufferedEvent) { let needs_refresh = matches!( &event, - ThreadBufferedEvent::Notification(ServerNotification::TurnStarted(_)) - | ThreadBufferedEvent::Notification(ServerNotification::ThreadTokenUsageUpdated(_)) + ThreadBufferedEvent::Notification(notification) + if matches!( + notification.as_ref(), + ServerNotification::TurnStarted(_) + | ServerNotification::ThreadTokenUsageUpdated(_) + ) ); match event { ThreadBufferedEvent::Notification(notification) => { - self.cache_collab_receiver_threads_for_notification(¬ification); + self.cache_collab_receiver_threads_for_notification(notification.as_ref()); self.chat_widget - .handle_server_notification(notification, /*replay_kind*/ None); + .handle_server_notification(*notification, /*replay_kind*/ None); } ThreadBufferedEvent::Request(request) => { if self .pending_app_server_requests - .contains_server_request(&request) + .contains_server_request(request.as_ref()) { self.chat_widget - .handle_server_request(request, /*replay_kind*/ None); + .handle_server_request(*request, /*replay_kind*/ None); } } ThreadBufferedEvent::HistoryEntryResponse(event) => { @@ -1575,10 +1581,10 @@ impl App { match event { ThreadBufferedEvent::Notification(notification) => self .chat_widget - .handle_server_notification(notification, Some(ReplayKind::ThreadSnapshot)), + .handle_server_notification(*notification, Some(ReplayKind::ThreadSnapshot)), ThreadBufferedEvent::Request(request) => self .chat_widget - .handle_server_request(request, Some(ReplayKind::ThreadSnapshot)), + .handle_server_request(*request, Some(ReplayKind::ThreadSnapshot)), ThreadBufferedEvent::HistoryEntryResponse(event) => { self.chat_widget.handle_history_entry_response(event) } @@ -1603,7 +1609,8 @@ impl App { // the exit marker when the currently active thread acknowledges shutdown. let pending_shutdown_exit_completed = matches!( &event, - ThreadBufferedEvent::Notification(ServerNotification::ThreadClosed(_)) + ThreadBufferedEvent::Notification(notification) + if matches!(notification.as_ref(), ServerNotification::ThreadClosed(_)) ) && self.pending_shutdown_exit_thread_id == self.active_thread_id; @@ -1617,7 +1624,7 @@ impl App { // failover, while true sub-agent deaths still do. if let ThreadBufferedEvent::Notification(notification) = &event && let Some((closed_thread_id, primary_thread_id)) = - self.active_non_primary_shutdown_target(notification) + self.active_non_primary_shutdown_target(notification.as_ref()) { self.mark_agent_picker_thread_closed(closed_thread_id); if self.side_threads.contains_key(&closed_thread_id) { diff --git a/codex-rs/tui/src/onboarding/onboarding_screen.rs b/codex-rs/tui/src/onboarding/onboarding_screen.rs index 405b37743c..0236d06894 100644 --- a/codex-rs/tui/src/onboarding/onboarding_screen.rs +++ b/codex-rs/tui/src/onboarding/onboarding_screen.rs @@ -556,7 +556,7 @@ pub(crate) async fn run_onboarding_app( if let Some(event) = event { match event { AppServerEvent::ServerNotification(notification) => { - onboarding_screen.handle_app_server_notification(notification); + onboarding_screen.handle_app_server_notification(*notification); } AppServerEvent::Disconnected { message } => { return Err(color_eyre::eyre::eyre!(message));