From feabb3a9853e867bfbbf3b9a36696cfaa1eb9c52 Mon Sep 17 00:00:00 2001 From: Zanie Blue Date: Tue, 9 Jun 2026 11:03:38 -0500 Subject: [PATCH] test app-server sender shutdown behavior --- codex-rs/app-server/src/extensions.rs | 48 +++++++++++++++++++++++++ codex-rs/app-server/src/mcp_refresh.rs | 49 +------------------------- 2 files changed, 49 insertions(+), 48 deletions(-) diff --git a/codex-rs/app-server/src/extensions.rs b/codex-rs/app-server/src/extensions.rs index fc2b1d60bf..ec190bb382 100644 --- a/codex-rs/app-server/src/extensions.rs +++ b/codex-rs/app-server/src/extensions.rs @@ -162,6 +162,8 @@ pub(crate) fn guardian_agent_spawner( mod tests { use std::time::Duration; + use crate::outgoing_message::OutgoingEnvelope; + use crate::outgoing_message::OutgoingMessage; use codex_protocol::protocol::ThreadGoal as CoreThreadGoal; use codex_protocol::protocol::ThreadGoalStatus; use codex_protocol::protocol::ThreadGoalUpdatedEvent; @@ -218,6 +220,52 @@ mod tests { ); } + #[tokio::test] + async fn app_server_event_sink_sends_fallback_while_sender_is_alive() { + let (outgoing_tx, mut outgoing_rx) = mpsc::channel(/*buffer*/ 4); + let outgoing = Arc::new(OutgoingMessageSender::new( + outgoing_tx, + AnalyticsEventsClient::disabled(), + )); + let sink = + app_server_extension_event_sink(Arc::clone(&outgoing), ThreadStateManager::new()); + let thread_id = ThreadId::default(); + let event = thread_goal_updated_event(thread_id, "turn-1"); + + sink.emit(event); + + let envelope = timeout(Duration::from_secs(/*secs*/ 1), outgoing_rx.recv()) + .await + .expect("timed out waiting for fallback notification") + .expect("outgoing channel closed unexpectedly"); + let OutgoingEnvelope::Broadcast { + message: + OutgoingMessage::AppServerNotification(ServerNotification::ThreadGoalUpdated( + notification, + )), + } = envelope + else { + panic!("expected a global thread goal update notification"); + }; + assert_eq!( + notification, + ThreadGoalUpdatedNotification { + thread_id: thread_id.to_string(), + turn_id: Some("turn-1".to_string()), + goal: ThreadGoal { + thread_id: thread_id.to_string(), + objective: "wire extension events".to_string(), + status: codex_app_server_protocol::ThreadGoalStatus::Active, + token_budget: Some(123), + tokens_used: 45, + time_used_seconds: 6, + created_at: 7, + updated_at: 8, + }, + } + ); + } + fn thread_goal_updated_event(thread_id: ThreadId, turn_id: &str) -> Event { Event { id: turn_id.to_string(), diff --git a/codex-rs/app-server/src/mcp_refresh.rs b/codex-rs/app-server/src/mcp_refresh.rs index 6bcc06e654..d11678ff5a 100644 --- a/codex-rs/app-server/src/mcp_refresh.rs +++ b/codex-rs/app-server/src/mcp_refresh.rs @@ -97,13 +97,9 @@ async fn queue_refresh( mod tests { use super::*; use crate::extensions::ThreadExtensionDependencies; - use crate::extensions::app_server_extension_event_sink; use crate::extensions::guardian_agent_spawner; use crate::extensions::thread_extensions; - use crate::outgoing_message::OutgoingMessageSender; - use crate::thread_state::ThreadStateManager; use async_trait::async_trait; - use codex_analytics::AnalyticsEventsClient; use codex_arg0::Arg0DispatchPaths; use codex_config::CloudConfigBundleLoader; use codex_config::LoaderOverrides; @@ -116,7 +112,6 @@ mod tests { use codex_core::init_state_db; use codex_core::thread_store_from_config; use codex_exec_server::EnvironmentManager; - use codex_extension_api::ExtensionEventSink; use codex_extension_api::NoopExtensionEventSink; use codex_login::AuthManager; use codex_login::CodexAuth; @@ -125,10 +120,7 @@ mod tests { use pretty_assertions::assert_eq; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; - use std::time::Duration; use tempfile::TempDir; - use tokio::sync::mpsc; - use tokio::time::timeout; #[tokio::test] async fn strict_refresh_reports_thread_planning_failures() -> anyhow::Result<()> { @@ -153,50 +145,11 @@ mod tests { Ok(()) } - #[tokio::test] - async fn retained_thread_manager_does_not_retain_app_server_sender() -> anyhow::Result<()> { - let (outgoing_tx, mut outgoing_rx) = mpsc::channel(/*buffer*/ 4); - let outgoing = Arc::new(OutgoingMessageSender::new( - outgoing_tx, - AnalyticsEventsClient::disabled(), - )); - let event_sink = - app_server_extension_event_sink(Arc::clone(&outgoing), ThreadStateManager::new()); - let (_temp_dir, thread_manager, _config_manager, _loader) = - refresh_test_state_with_event_sink(event_sink).await?; - - // Plugin refresh tasks can outlive MessageProcessor while retaining ThreadManager. - let detached_refresh_thread_manager = Arc::clone(&thread_manager); - drop(thread_manager); - drop(outgoing); - - assert!(matches!( - timeout(Duration::from_millis(/*millis*/ 100), outgoing_rx.recv()).await, - Ok(None) - )); - - detached_refresh_thread_manager - .shutdown_all_threads_bounded(Duration::from_secs(/*secs*/ 10)) - .await; - Ok(()) - } - async fn refresh_test_state() -> anyhow::Result<( TempDir, Arc, ConfigManager, Arc, - )> { - refresh_test_state_with_event_sink(Arc::new(NoopExtensionEventSink)).await - } - - async fn refresh_test_state_with_event_sink( - event_sink: Arc, - ) -> anyhow::Result<( - TempDir, - Arc, - ConfigManager, - Arc, )> { let temp_dir = TempDir::new()?; let good_cwd = temp_dir.path().join("good"); @@ -242,7 +195,7 @@ mod tests { thread_extensions( guardian_agent_spawner(thread_manager.clone()), ThreadExtensionDependencies { - event_sink, + event_sink: Arc::new(NoopExtensionEventSink), auth_manager: auth_manager.clone(), state_db: Some(state_db.clone()), analytics_events_client: codex_analytics::AnalyticsEventsClient::disabled(),