app-server: refresh installed plugins periodically

codex-rs/app-server/src/background_refresh/mod.rs

@@ -0,0 +1,5 @@
+mod periodic;
+mod plugins;
+
+pub(crate) use periodic::PeriodicRefreshWorker;
+pub(crate) use plugins::spawn_plugins_refresh_worker;

codex-rs/app-server/src/background_refresh/periodic.rs

@@ -0,0 +1,62 @@
+use std::future::Future;
+use std::time::Duration;
+
+use tokio::task::JoinHandle;
+use tokio_util::sync::CancellationToken;
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(super) enum RefreshControl {
+    Continue,
+    Stop,
+}
+
+#[derive(Debug)]
+pub(crate) struct PeriodicRefreshWorker {
+    shutdown: CancellationToken,
+    _task: JoinHandle<()>,
+}
+
+impl PeriodicRefreshWorker {
+    pub(crate) fn shutdown(&self) {
+        self.shutdown.cancel();
+    }
+}
+
+impl Drop for PeriodicRefreshWorker {
+    fn drop(&mut self) {
+        self.shutdown();
+    }
+}
+
+pub(super) fn spawn<F, Fut>(refresh_interval: Duration, mut refresh: F) -> PeriodicRefreshWorker
+where
+    F: FnMut() -> Fut + Send + 'static,
+    Fut: Future<Output = RefreshControl> + Send + 'static,
+{
+    let shutdown = CancellationToken::new();
+    let worker_shutdown = shutdown.clone();
+    let task = tokio::spawn(async move {
+        loop {
+            let refresh_control = tokio::select! {
+                _ = worker_shutdown.cancelled() => break,
+                refresh_control = refresh() => refresh_control,
+            };
+            if refresh_control == RefreshControl::Stop {
+                break;
+            }
+
+            tokio::select! {
+                _ = worker_shutdown.cancelled() => break,
+                _ = tokio::time::sleep(refresh_interval) => {}
+            }
+        }
+    });
+    PeriodicRefreshWorker {
+        shutdown,
+        _task: task,
+    }
+}
+
+#[cfg(test)]
+#[path = "periodic_tests.rs"]
+mod tests;

codex-rs/app-server/src/background_refresh/periodic_tests.rs

@@ -0,0 +1,73 @@
+use std::sync::Arc;
+use std::sync::atomic::AtomicUsize;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use pretty_assertions::assert_eq;
+use tokio::sync::Notify;
+
+use super::*;
+
+#[tokio::test]
+async fn refreshes_immediately_periodically_and_stops_when_dropped() {
+    let refresh_count = Arc::new(AtomicUsize::new(0));
+    let refreshed = Arc::new(Notify::new());
+    let hold_second_refresh = Arc::new(Notify::new());
+    let worker = spawn(Duration::from_millis(10), {
+        let refresh_count = Arc::clone(&refresh_count);
+        let refreshed = Arc::clone(&refreshed);
+        let hold_second_refresh = Arc::clone(&hold_second_refresh);
+        move || {
+            let refresh_count = Arc::clone(&refresh_count);
+            let refreshed = Arc::clone(&refreshed);
+            let hold_second_refresh = Arc::clone(&hold_second_refresh);
+            async move {
+                let refresh_index = refresh_count.fetch_add(1, Ordering::SeqCst);
+                refreshed.notify_one();
+                if refresh_index == 1 {
+                    hold_second_refresh.notified().await;
+                }
+                RefreshControl::Continue
+            }
+        }
+    });
+
+    tokio::time::timeout(Duration::from_secs(1), async {
+        while refresh_count.load(Ordering::SeqCst) < 2 {
+            refreshed.notified().await;
+        }
+    })
+    .await
+    .expect("expected two refreshes");
+    drop(worker);
+    hold_second_refresh.notify_one();
+    tokio::time::sleep(Duration::from_millis(30)).await;
+
+    assert_eq!(refresh_count.load(Ordering::SeqCst), 2);
+}
+
+#[tokio::test]
+async fn stops_when_refresh_requests_it() {
+    let refresh_count = Arc::new(AtomicUsize::new(0));
+    let refreshed = Arc::new(Notify::new());
+    let worker = spawn(Duration::from_millis(10), {
+        let refresh_count = Arc::clone(&refresh_count);
+        let refreshed = Arc::clone(&refreshed);
+        move || {
+            let refresh_count = Arc::clone(&refresh_count);
+            let refreshed = Arc::clone(&refreshed);
+            async move {
+                refresh_count.fetch_add(1, Ordering::SeqCst);
+                refreshed.notify_one();
+                RefreshControl::Stop
+            }
+        }
+    });
+
+    tokio::time::timeout(Duration::from_secs(1), refreshed.notified())
+        .await
+        .expect("expected refresh");
+    drop(worker);
+
+    assert_eq!(refresh_count.load(Ordering::SeqCst), 1);
+}

codex-rs/app-server/src/background_refresh/plugins.rs

@@ -0,0 +1,83 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use codex_core_plugins::PluginsManager;
+use codex_login::AuthManager;
+use tracing::warn;
+
+use super::PeriodicRefreshWorker;
+use super::periodic;
+use super::periodic::RefreshControl;
+use crate::config_manager::ConfigManager;
+
+const PLUGINS_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60);
+
+pub(crate) fn spawn_plugins_refresh_worker(
+    plugins_manager: &Arc<PluginsManager>,
+    auth_manager: &Arc<AuthManager>,
+    config_manager: ConfigManager,
+    on_effective_plugins_changed: Arc<dyn Fn() + Send + Sync>,
+) -> PeriodicRefreshWorker {
+    spawn_with_interval(
+        plugins_manager,
+        auth_manager,
+        config_manager,
+        on_effective_plugins_changed,
+        PLUGINS_REFRESH_INTERVAL,
+    )
+}
+
+fn spawn_with_interval(
+    plugins_manager: &Arc<PluginsManager>,
+    auth_manager: &Arc<AuthManager>,
+    config_manager: ConfigManager,
+    on_effective_plugins_changed: Arc<dyn Fn() + Send + Sync>,
+    refresh_interval: Duration,
+) -> PeriodicRefreshWorker {
+    let plugins_manager = Arc::downgrade(plugins_manager);
+    let auth_manager = Arc::downgrade(auth_manager);
+    periodic::spawn(refresh_interval, move || {
+        let plugins_manager = plugins_manager.clone();
+        let auth_manager = auth_manager.clone();
+        let config_manager = config_manager.clone();
+        let on_effective_plugins_changed = Arc::clone(&on_effective_plugins_changed);
+        async move {
+            let Some(plugins_manager) = plugins_manager.upgrade() else {
+                return RefreshControl::Stop;
+            };
+            let Some(auth_manager) = auth_manager.upgrade() else {
+                return RefreshControl::Stop;
+            };
+            let config = match config_manager
+                .load_latest_config(/*fallback_cwd*/ None)
+                .await
+            {
+                Ok(config) => config,
+                Err(err) => {
+                    warn!(
+                        error = %err,
+                        "failed to reload config for periodic plugin refresh"
+                    );
+                    return RefreshControl::Continue;
+                }
+            };
+            let auth = auth_manager.auth().await;
+            let plugins_config = config.plugins_config_input();
+            plugins_manager.maybe_start_remote_installed_plugins_cache_refresh(
+                &plugins_config,
+                auth.clone(),
+                Some(Arc::clone(&on_effective_plugins_changed)),
+            );
+            plugins_manager.maybe_start_remote_installed_plugin_bundle_sync(
+                &plugins_config,
+                auth,
+                Some(on_effective_plugins_changed),
+            );
+            RefreshControl::Continue
+        }
+    })
+}
+
+#[cfg(test)]
+#[path = "plugins_tests.rs"]
+mod tests;

codex-rs/app-server/src/background_refresh/plugins_tests.rs

@@ -0,0 +1,102 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use codex_core_plugins::PluginsManager;
+use codex_login::AuthManager;
+use codex_login::CodexAuth;
+use pretty_assertions::assert_eq;
+use tempfile::TempDir;
+use tokio::time::timeout;
+use wiremock::Mock;
+use wiremock::MockServer;
+use wiremock::ResponseTemplate;
+use wiremock::matchers::method;
+use wiremock::matchers::path;
+
+use super::*;
+
+const TEST_REFRESH_INTERVAL: Duration = Duration::from_millis(500);
+const TEST_TIMEOUT: Duration = Duration::from_secs(5);
+
+#[tokio::test]
+async fn refreshes_remote_installed_plugins_immediately_and_periodically() {
+    let codex_home = TempDir::new().expect("create Codex home");
+    let server = MockServer::start().await;
+    std::fs::write(
+        codex_home.path().join("config.toml"),
+        format!(
+            r#"chatgpt_base_url = "{}/backend-api/"
+
+[features]
+plugins = true
+"#,
+            server.uri()
+        ),
+    )
+    .expect("write config");
+    Mock::given(method("GET"))
+        .and(path("/backend-api/ps/plugins/installed"))
+        .respond_with(ResponseTemplate::new(200).set_body_string(empty_installed_plugins_body()))
+        .mount(&server)
+        .await;
+
+    let plugins_manager = Arc::new(PluginsManager::new(codex_home.path().to_path_buf()));
+    let auth_manager =
+        AuthManager::from_auth_for_testing(CodexAuth::create_dummy_chatgpt_auth_for_testing());
+    let config_manager =
+        ConfigManager::without_managed_config_for_tests(codex_home.path().to_path_buf());
+    let worker = spawn_with_interval(
+        &plugins_manager,
+        &auth_manager,
+        config_manager,
+        Arc::new(|| {}),
+        TEST_REFRESH_INTERVAL,
+    );
+
+    wait_for_bundle_sync_request_count(&server, /*expected*/ 6).await;
+    drop(worker);
+    tokio::time::sleep(TEST_REFRESH_INTERVAL * 2).await;
+    let request_count_after_shutdown = bundle_sync_request_count(&server).await;
+    tokio::time::sleep(TEST_REFRESH_INTERVAL * 2).await;
+
+    assert_eq!(
+        bundle_sync_request_count(&server).await,
+        request_count_after_shutdown
+    );
+}
+
+async fn wait_for_bundle_sync_request_count(server: &MockServer, expected: usize) {
+    timeout(TEST_TIMEOUT, async {
+        while bundle_sync_request_count(server).await < expected {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("expected {expected} remote installed plugin bundle requests"));
+}
+
+async fn bundle_sync_request_count(server: &MockServer) -> usize {
+    server
+        .received_requests()
+        .await
+        .unwrap_or_default()
+        .iter()
+        .filter(|request| {
+            request.url.path() == "/backend-api/ps/plugins/installed"
+                && request
+                    .url
+                    .query_pairs()
+                    .any(|(key, value)| key == "includeDownloadUrls" && value == "true")
+        })
+        .count()
+}
+
+fn empty_installed_plugins_body() -> &'static str {
+    r#"{
+  "plugins": [],
+  "pagination": {
+    "limit": 50,
+    "next_page_token": null
+  }
+}"#
+}

codex-rs/app-server/src/lib.rs

@@ -84,6 +84,7 @@ const SQLITE_RECOVERY_CONFIG_WARNING_SUMMARY: &str = "Codex rebuilt its local da
 mod analytics_utils;
 mod app_server_tracing;
 mod attestation;
+mod background_refresh;
 mod bespoke_event_handling;
 mod command_exec;
 mod config;

codex-rs/app-server/src/message_processor.rs

@@ -93,6 +93,7 @@ use tokio::time::timeout;
 use tokio_util::sync::CancellationToken;
 use tracing::Instrument;
 
+use crate::background_refresh::PeriodicRefreshWorker;
 use crate::models_refresh_worker::ModelsRefreshWorker;
 
 const EXTERNAL_AUTH_REFRESH_TIMEOUT: Duration = Duration::from_secs(10);
@@ -186,6 +187,7 @@ impl ExternalAuth for ExternalAuthRefreshBridge {
 pub(crate) struct MessageProcessor {
     outgoing: Arc<OutgoingMessageSender>,
     models_refresh_worker: ModelsRefreshWorker,
+    plugins_refresh_worker: Option<PeriodicRefreshWorker>,
     skills_watcher: Arc<SkillsWatcher>,
     account_processor: AccountRequestProcessor,
     apps_processor: AppsRequestProcessor,
@@ -510,18 +512,25 @@ impl MessageProcessor {
             thread_list_state_permit,
             Arc::clone(&skills_watcher),
         );
-        if matches!(plugin_startup_tasks, crate::PluginStartupTasks::Start) {
-            // Keep plugin startup warmups aligned at app-server startup.
-            let on_effective_plugins_changed =
-                plugin_processor.effective_plugins_changed_callback();
-            thread_manager
-                .plugins_manager()
-                .maybe_start_plugin_startup_tasks_for_config(
+        let plugins_refresh_worker =
+            if matches!(plugin_startup_tasks, crate::PluginStartupTasks::Start) {
+                // Keep plugin startup warmups aligned at app-server startup.
+                let on_effective_plugins_changed =
+                    plugin_processor.effective_plugins_changed_callback();
+                let plugins_manager = thread_manager.plugins_manager();
+                plugins_manager.maybe_start_plugin_startup_tasks_for_config(
                     &config.plugins_config_input(),
-                    auth_manager,
-                    Some(on_effective_plugins_changed),
+                    Arc::clone(&auth_manager),
                 );
-        }
+                Some(crate::background_refresh::spawn_plugins_refresh_worker(
+                    &plugins_manager,
+                    &auth_manager,
+                    config_manager.clone(),
+                    on_effective_plugins_changed,
+                ))
+            } else {
+                None
+            };
         let config_processor = ConfigRequestProcessor::new(
             outgoing.clone(),
             config_manager.clone(),
@@ -555,6 +564,7 @@ impl MessageProcessor {
         Self {
             outgoing,
             models_refresh_worker,
+            plugins_refresh_worker,
             skills_watcher,
             account_processor,
             apps_processor,
@@ -585,6 +595,9 @@ impl MessageProcessor {
         self.account_processor.clear_external_auth();
         self.apps_processor.shutdown();
         self.models_refresh_worker.shutdown();
+        if let Some(plugins_refresh_worker) = &self.plugins_refresh_worker {
+            plugins_refresh_worker.shutdown();
+        }
         self.skills_watcher.shutdown();
     }
 
@@ -762,6 +775,9 @@ impl MessageProcessor {
 
     pub(crate) async fn drain_background_tasks(&self) {
         self.models_refresh_worker.shutdown();
+        if let Some(plugins_refresh_worker) = &self.plugins_refresh_worker {
+            plugins_refresh_worker.shutdown();
+        }
         self.thread_processor.drain_background_tasks().await;
     }
 

codex-rs/core-plugins/src/manager.rs

@@ -808,10 +808,9 @@ impl PluginsManager {
         auth: Option<CodexAuth>,
         on_effective_plugins_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
     ) {
-        self.maybe_start_remote_installed_plugins_cache_refresh_with_notify(
+        self.maybe_start_remote_installed_plugins_cache_refresh(
             config,
             auth.clone(),
-            RemoteInstalledPluginsCacheRefreshNotify::IfCacheChanged,
             on_effective_plugins_changed,
         );
 
@@ -824,6 +823,20 @@ impl PluginsManager {
         });
     }
 
+    pub fn maybe_start_remote_installed_plugins_cache_refresh(
+        self: &Arc<Self>,
+        config: &PluginsConfigInput,
+        auth: Option<CodexAuth>,
+        on_effective_plugins_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
+    ) {
+        self.maybe_start_remote_installed_plugins_cache_refresh_with_notify(
+            config,
+            auth,
+            RemoteInstalledPluginsCacheRefreshNotify::IfCacheChanged,
+            on_effective_plugins_changed,
+        );
+    }
+
     pub fn maybe_start_remote_installed_plugins_cache_refresh_after_mutation(
         self: &Arc<Self>,
         config: &PluginsConfigInput,
@@ -1776,7 +1789,6 @@ impl PluginsManager {
         self: &Arc<Self>,
         config: &PluginsConfigInput,
         auth_manager: Arc<AuthManager>,
-        on_effective_plugins_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
     ) {
         if config.plugins_enabled {
             let use_remote_global_catalog =
@@ -1835,26 +1847,15 @@ impl PluginsManager {
                     warn!("failed to start configured marketplace auto-upgrade task: {err}");
                 }
             }
-            let config_for_remote_sync = config.clone();
-            let manager = Arc::clone(self);
-            let auth_manager_for_remote_sync = auth_manager.clone();
-            let on_effective_plugins_changed = on_effective_plugins_changed.clone();
-            tokio::spawn(async move {
-                let auth = auth_manager_for_remote_sync.auth().await;
-                manager.maybe_start_remote_plugin_caches_refresh(
-                    &config_for_remote_sync,
-                    auth.clone(),
-                    on_effective_plugins_changed.clone(),
-                );
-                manager.maybe_start_remote_installed_plugin_bundle_sync(
-                    &config_for_remote_sync,
-                    auth.clone(),
-                    on_effective_plugins_changed,
-                );
-                if config_for_remote_sync.remote_plugin_enabled {
+            if config.remote_plugin_enabled {
+                let config_for_remote_catalog = config.clone();
+                let manager = Arc::clone(self);
+                let auth_manager_for_remote_catalog = auth_manager.clone();
+                tokio::spawn(async move {
+                    let auth = auth_manager_for_remote_catalog.auth().await;
                     match crate::remote::fetch_and_cache_global_remote_plugin_catalog(
                         manager.codex_home.as_path(),
-                        &remote_plugin_service_config(&config_for_remote_sync),
+                        &remote_plugin_service_config(&config_for_remote_catalog),
                         auth.as_ref(),
                     )
                     .await
@@ -1871,8 +1872,8 @@ impl PluginsManager {
                             );
                         }
                     }
-                }
-            });
+                });
+            }
 
             let config_for_featured_plugins = config.clone();
             let manager = Arc::clone(self);