From 4fae1c904c841746bcc7bf31a013d8c2781f3685 Mon Sep 17 00:00:00 2001 From: Zanie Blue Date: Mon, 8 Jun 2026 13:05:47 -0500 Subject: [PATCH] test app-server sender retention through refresh --- codex-rs/app-server/src/mcp_refresh.rs | 49 +++++++++++++++++++++++++- 1 file changed, 48 insertions(+), 1 deletion(-) diff --git a/codex-rs/app-server/src/mcp_refresh.rs b/codex-rs/app-server/src/mcp_refresh.rs index d11678ff5a..a0bd9e15ee 100644 --- a/codex-rs/app-server/src/mcp_refresh.rs +++ b/codex-rs/app-server/src/mcp_refresh.rs @@ -97,9 +97,13 @@ 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; @@ -112,6 +116,7 @@ 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; @@ -120,7 +125,10 @@ 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<()> { @@ -145,11 +153,50 @@ 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(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(100), outgoing_rx.recv()).await, + Ok(None) + )); + + detached_refresh_thread_manager + .shutdown_all_threads_bounded(Duration::from_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"); @@ -195,7 +242,7 @@ mod tests { thread_extensions( guardian_agent_spawner(thread_manager.clone()), ThreadExtensionDependencies { - event_sink: Arc::new(NoopExtensionEventSink), + event_sink, auth_manager: auth_manager.clone(), state_db: Some(state_db.clone()), analytics_events_client: codex_analytics::AnalyticsEventsClient::disabled(),