From 368462abc3bd3e81199abfcb91c16e273853a8bb 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 --- codex-rs/app-server/src/lib.rs | 1 + codex-rs/app-server/src/message_processor.rs | 37 ++++-- .../app-server/src/plugins_refresh_worker.rs | 111 ++++++++++++++++++ .../src/plugins_refresh_worker_tests.rs | 110 +++++++++++++++++ 4 files changed, 249 insertions(+), 10 deletions(-) create mode 100644 codex-rs/app-server/src/plugins_refresh_worker.rs create mode 100644 codex-rs/app-server/src/plugins_refresh_worker_tests.rs diff --git a/codex-rs/app-server/src/lib.rs b/codex-rs/app-server/src/lib.rs index f4cd83d732..9ff9e4bc94 100644 --- a/codex-rs/app-server/src/lib.rs +++ b/codex-rs/app-server/src/lib.rs @@ -104,6 +104,7 @@ mod message_processor; mod models; mod models_refresh_worker; mod outgoing_message; +mod plugins_refresh_worker; mod request_processors; mod request_serialization; mod server_request_error; diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index ad29e05010..d012e13562 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -94,6 +94,7 @@ use tokio_util::sync::CancellationToken; use tracing::Instrument; use crate::models_refresh_worker::ModelsRefreshWorker; +use crate::plugins_refresh_worker::PluginsRefreshWorker; const EXTERNAL_AUTH_REFRESH_TIMEOUT: Duration = Duration::from_secs(10); const CONNECTION_RPC_DRAIN_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 30); @@ -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,26 @@ 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(Arc::clone(&on_effective_plugins_changed)), ); - } + Some(crate::plugins_refresh_worker::spawn( + &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 +565,7 @@ impl MessageProcessor { Self { outgoing, models_refresh_worker, + plugins_refresh_worker, skills_watcher, account_processor, apps_processor, @@ -585,6 +596,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 +776,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/app-server/src/plugins_refresh_worker.rs b/codex-rs/app-server/src/plugins_refresh_worker.rs new file mode 100644 index 0000000000..f94af0ea63 --- /dev/null +++ b/codex-rs/app-server/src/plugins_refresh_worker.rs @@ -0,0 +1,111 @@ +use std::sync::Arc; +use std::time::Duration; + +use codex_core_plugins::PluginsManager; +use codex_login::AuthManager; +use tokio::task::JoinHandle; +use tokio_util::sync::CancellationToken; +use tracing::warn; + +use crate::config_manager::ConfigManager; + +const PLUGINS_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60); + +#[derive(Debug)] +pub(crate) struct PluginsRefreshWorker { + shutdown: CancellationToken, + _task: JoinHandle<()>, +} + +impl PluginsRefreshWorker { + pub(crate) fn shutdown(&self) { + self.shutdown.cancel(); + } +} + +impl Drop for PluginsRefreshWorker { + fn drop(&mut self) { + self.shutdown(); + } +} + +pub(crate) fn spawn( + plugins_manager: &Arc, + auth_manager: &Arc, + config_manager: ConfigManager, + on_effective_plugins_changed: Arc, +) -> PluginsRefreshWorker { + 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, +) -> PluginsRefreshWorker { + let plugins_manager = Arc::downgrade(plugins_manager); + let auth_manager = Arc::downgrade(auth_manager); + let shutdown = CancellationToken::new(); + let worker_shutdown = shutdown.clone(); + let task = tokio::spawn(async move { + loop { + // Plugin startup tasks perform the initial refresh. Wait before the first periodic + // pass so app-server startup does not issue duplicate remote requests. + tokio::select! { + _ = worker_shutdown.cancelled() => break, + _ = tokio::time::sleep(refresh_interval) => {} + } + + let Some(plugins_manager) = plugins_manager.upgrade() else { + break; + }; + let Some(auth_manager) = auth_manager.upgrade() else { + break; + }; + 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" + ); + continue; + } + }; + let auth = auth_manager.auth().await; + if worker_shutdown.is_cancelled() { + break; + } + let plugins_config = config.plugins_config_input(); + plugins_manager.maybe_start_remote_plugin_caches_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(Arc::clone(&on_effective_plugins_changed)), + ); + } + }); + PluginsRefreshWorker { + shutdown, + _task: task, + } +} + +#[cfg(test)] +#[path = "plugins_refresh_worker_tests.rs"] +mod tests; diff --git a/codex-rs/app-server/src/plugins_refresh_worker_tests.rs b/codex-rs/app-server/src/plugins_refresh_worker_tests.rs new file mode 100644 index 0000000000..0c46be77f7 --- /dev/null +++ b/codex-rs/app-server/src/plugins_refresh_worker_tests.rs @@ -0,0 +1,110 @@ +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_periodically_and_stops_when_dropped() { + 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; + Mock::given(method("GET")) + .and(path("/backend-api/ps/plugins/suggested")) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "enabled": true, + "plugins": [] + }))) + .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 + } +}"# +}