From 9032334b543669ef8d3f8e2eb9905892faae89e4 Mon Sep 17 00:00:00 2001 From: James Watterson-Tenvooren Date: Sat, 20 Jun 2026 04:31:59 +0000 Subject: [PATCH] app-server: refresh installed plugins periodically --- .../app-server/src/background_refresh/mod.rs | 5 + .../src/background_refresh/periodic.rs | 62 +++++++++++ .../src/background_refresh/periodic_tests.rs | 73 +++++++++++++ .../src/background_refresh/plugins.rs | 83 ++++++++++++++ .../src/background_refresh/plugins_tests.rs | 102 ++++++++++++++++++ codex-rs/app-server/src/lib.rs | 1 + codex-rs/app-server/src/message_processor.rs | 36 +++++-- codex-rs/core-plugins/src/manager.rs | 47 ++++---- 8 files changed, 376 insertions(+), 33 deletions(-) create mode 100644 codex-rs/app-server/src/background_refresh/mod.rs create mode 100644 codex-rs/app-server/src/background_refresh/periodic.rs create mode 100644 codex-rs/app-server/src/background_refresh/periodic_tests.rs create mode 100644 codex-rs/app-server/src/background_refresh/plugins.rs create mode 100644 codex-rs/app-server/src/background_refresh/plugins_tests.rs diff --git a/codex-rs/app-server/src/background_refresh/mod.rs b/codex-rs/app-server/src/background_refresh/mod.rs new file mode 100644 index 0000000000..9daf35d7a8 --- /dev/null +++ b/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; diff --git a/codex-rs/app-server/src/background_refresh/periodic.rs b/codex-rs/app-server/src/background_refresh/periodic.rs new file mode 100644 index 0000000000..e7a7b34d85 --- /dev/null +++ b/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(refresh_interval: Duration, mut refresh: F) -> PeriodicRefreshWorker +where + F: FnMut() -> Fut + Send + 'static, + Fut: Future + 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; diff --git a/codex-rs/app-server/src/background_refresh/periodic_tests.rs b/codex-rs/app-server/src/background_refresh/periodic_tests.rs new file mode 100644 index 0000000000..dd4e43ae6d --- /dev/null +++ b/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); +} diff --git a/codex-rs/app-server/src/background_refresh/plugins.rs b/codex-rs/app-server/src/background_refresh/plugins.rs new file mode 100644 index 0000000000..521710c731 --- /dev/null +++ b/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, + auth_manager: &Arc, + config_manager: ConfigManager, + on_effective_plugins_changed: Arc, +) -> PeriodicRefreshWorker { + spawn_with_interval( + plugins_manager, + auth_manager, + config_manager, + on_effective_plugins_changed, + PLUGINS_REFRESH_INTERVAL, + ) +} + +fn spawn_with_interval( + plugins_manager: &Arc, + auth_manager: &Arc, + config_manager: ConfigManager, + on_effective_plugins_changed: Arc, + 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; diff --git a/codex-rs/app-server/src/background_refresh/plugins_tests.rs b/codex-rs/app-server/src/background_refresh/plugins_tests.rs new file mode 100644 index 0000000000..23d627aaec --- /dev/null +++ b/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 + } +}"# +} diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index f4cd83d732..810036fdac 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/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; diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index ad29e05010..35d979c601 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/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, models_refresh_worker: ModelsRefreshWorker, + plugins_refresh_worker: Option, skills_watcher: Arc, 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; } diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index b502106ea9..ce8e8ee4a3 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -808,10 +808,9 @@ impl PluginsManager { auth: Option, on_effective_plugins_changed: Option>, ) { - 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, + config: &PluginsConfigInput, + auth: Option, + on_effective_plugins_changed: Option>, + ) { + 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, config: &PluginsConfigInput, @@ -1776,7 +1789,6 @@ impl PluginsManager { self: &Arc, config: &PluginsConfigInput, auth_manager: Arc, - on_effective_plugins_changed: Option>, ) { 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);