test app-server sender shutdown behavior

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(),

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<ThreadManager>,
         ConfigManager,
         Arc<CountingThreadConfigLoader>,
-    )> {
-        refresh_test_state_with_event_sink(Arc::new(NoopExtensionEventSink)).await
-    }
-
-    async fn refresh_test_state_with_event_sink(
-        event_sink: Arc<dyn ExtensionEventSink>,
-    ) -> anyhow::Result<(
-        TempDir,
-        Arc<ThreadManager>,
-        ConfigManager,
-        Arc<CountingThreadConfigLoader>,
     )> {
         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(),