app-server: refresh installed plugins periodically

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;

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<OutgoingMessageSender>,
     models_refresh_worker: ModelsRefreshWorker,
+    plugins_refresh_worker: Option<PluginsRefreshWorker>,
     skills_watcher: Arc<SkillsWatcher>,
     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;
     }
 

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<PluginsManager>,
+    auth_manager: &Arc<AuthManager>,
+    config_manager: ConfigManager,
+    on_effective_plugins_changed: Arc<dyn Fn() + Send + Sync>,
+) -> PluginsRefreshWorker {
+    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,
+) -> 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;

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
+  }
+}"#
+}