app-server: refresh Codex Apps tools periodically

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

@@ -0,0 +1,79 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use codex_chatgpt::connectors;
+use codex_core::McpManager;
+use codex_exec_server::EnvironmentManager;
+use tracing::warn;
+
+use super::PeriodicRefreshWorker;
+use super::periodic;
+use super::periodic::InitialRefresh;
+use super::periodic::RefreshControl;
+use crate::config_manager::ConfigManager;
+
+const APPS_REFRESH_INTERVAL: Duration = Duration::from_secs(5 * 60);
+
+pub(crate) fn spawn_apps_refresh_worker(
+    config_manager: ConfigManager,
+    environment_manager: &Arc<EnvironmentManager>,
+    mcp_manager: &Arc<McpManager>,
+) -> PeriodicRefreshWorker {
+    spawn_with_interval(
+        config_manager,
+        environment_manager,
+        mcp_manager,
+        APPS_REFRESH_INTERVAL,
+    )
+}
+
+fn spawn_with_interval(
+    config_manager: ConfigManager,
+    environment_manager: &Arc<EnvironmentManager>,
+    mcp_manager: &Arc<McpManager>,
+    refresh_interval: Duration,
+) -> PeriodicRefreshWorker {
+    let environment_manager = Arc::downgrade(environment_manager);
+    let mcp_manager = Arc::downgrade(mcp_manager);
+    // app/list populates this cache on demand. Delay the background pass so startup RPCs do not
+    // race a second force-refresh of the same Codex Apps MCP server.
+    periodic::spawn(refresh_interval, InitialRefresh::AfterInterval, move || {
+        let config_manager = config_manager.clone();
+        let environment_manager = environment_manager.clone();
+        let mcp_manager = mcp_manager.clone();
+        async move {
+            let Some(environment_manager) = environment_manager.upgrade() else {
+                return RefreshControl::Stop;
+            };
+            let Some(mcp_manager) = mcp_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 apps refresh");
+                    return RefreshControl::Continue;
+                }
+            };
+            if let Err(err) =
+                connectors::list_accessible_connectors_from_mcp_tools_with_mcp_manager(
+                    &config,
+                    /*force_refetch*/ true,
+                    environment_manager,
+                    mcp_manager,
+                )
+                .await
+            {
+                warn!(error = %err, "failed to refresh Codex Apps tools cache");
+            }
+            RefreshControl::Continue
+        }
+    })
+}
+
+#[cfg(test)]
+#[path = "apps_tests.rs"]
+mod tests;

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

@@ -0,0 +1,108 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use app_test_support::ChatGptAuthFixture;
+use app_test_support::write_chatgpt_auth;
+use codex_config::types::AuthCredentialsStoreMode;
+use codex_core::McpManager;
+use codex_core_plugins::PluginsManager;
+use codex_exec_server::EnvironmentManager;
+use codex_exec_server::ExecServerRuntimePaths;
+use core_test_support::apps_test_server::AppsTestServer;
+use serde_json::Value;
+use tempfile::TempDir;
+use tokio::time::timeout;
+use wiremock::MockServer;
+
+use super::*;
+
+const TEST_REFRESH_INTERVAL: Duration = Duration::from_millis(500);
+const TEST_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[tokio::test]
+async fn refreshes_codex_apps_tools_after_first_interval_and_periodically() {
+    let codex_home = TempDir::new().expect("create Codex home");
+    let server = MockServer::start().await;
+    let apps_server = AppsTestServer::mount(&server)
+        .await
+        .expect("mount apps server");
+    std::fs::write(
+        codex_home.path().join("config.toml"),
+        format!(
+            r#"chatgpt_base_url = "{}"
+cli_auth_credentials_store = "file"
+mcp_oauth_credentials_store = "file"
+
+[features]
+apps = true
+"#,
+            apps_server.chatgpt_base_url
+        ),
+    )
+    .expect("write config");
+    write_chatgpt_auth(
+        codex_home.path(),
+        ChatGptAuthFixture::new("access-chatgpt"),
+        AuthCredentialsStoreMode::File,
+    )
+    .expect("write auth");
+
+    let config_manager =
+        ConfigManager::without_managed_config_for_tests(codex_home.path().to_path_buf());
+    let runtime_paths = ExecServerRuntimePaths::from_optional_paths(
+        Some(std::env::current_exe().expect("resolve test executable")),
+        /*codex_linux_sandbox_exe*/ None,
+    )
+    .expect("resolve exec server runtime paths");
+    let environment_manager = Arc::new(
+        EnvironmentManager::from_codex_home(codex_home.path().to_path_buf(), Some(runtime_paths))
+            .await
+            .expect("create environment manager"),
+    );
+    let plugins_manager = Arc::new(PluginsManager::new(codex_home.path().to_path_buf()));
+    let mcp_manager = Arc::new(McpManager::new(plugins_manager));
+    let worker = spawn_with_interval(
+        config_manager,
+        &environment_manager,
+        &mcp_manager,
+        TEST_REFRESH_INTERVAL,
+    );
+
+    wait_for_tools_list_request_count(&server, /*expected*/ 4).await;
+    drop(worker);
+    let request_count_after_shutdown = tools_list_request_count(&server).await;
+    tokio::time::sleep(TEST_REFRESH_INTERVAL * 2).await;
+
+    assert_eq!(
+        tools_list_request_count(&server).await,
+        request_count_after_shutdown
+    );
+}
+
+async fn wait_for_tools_list_request_count(server: &MockServer, expected: usize) {
+    timeout(TEST_TIMEOUT, async {
+        while tools_list_request_count(server).await < expected {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("expected {expected} Codex Apps tools/list requests"));
+}
+
+async fn tools_list_request_count(server: &MockServer) -> usize {
+    server
+        .received_requests()
+        .await
+        .unwrap_or_default()
+        .iter()
+        .filter(|request| {
+            request.url.path() == "/api/codex/apps"
+                && serde_json::from_slice::<Value>(&request.body)
+                    .ok()
+                    .and_then(|body| body.get("method").cloned())
+                    .as_ref()
+                    .and_then(Value::as_str)
+                    == Some("tools/list")
+        })
+        .count()
+}

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

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

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

@@ -10,6 +10,12 @@ pub(super) enum RefreshControl {
     Stop,
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub(super) enum InitialRefresh {
+    Immediate,
+    AfterInterval,
+}
+
 #[derive(Debug)]
 pub(crate) struct PeriodicRefreshWorker {
     shutdown: CancellationToken,
@@ -28,7 +34,11 @@ impl Drop for PeriodicRefreshWorker {
     }
 }
 
-pub(super) fn spawn<F, Fut>(refresh_interval: Duration, mut refresh: F) -> PeriodicRefreshWorker
+pub(super) fn spawn<F, Fut>(
+    refresh_interval: Duration,
+    initial_refresh: InitialRefresh,
+    mut refresh: F,
+) -> PeriodicRefreshWorker
 where
     F: FnMut() -> Fut + Send + 'static,
     Fut: Future<Output = RefreshControl> + Send + 'static,
@@ -36,7 +46,14 @@ where
     let shutdown = CancellationToken::new();
     let worker_shutdown = shutdown.clone();
     let task = tokio::spawn(async move {
+        let mut delay_before_refresh = initial_refresh == InitialRefresh::AfterInterval;
         loop {
+            if delay_before_refresh {
+                tokio::select! {
+                    _ = worker_shutdown.cancelled() => break,
+                    _ = tokio::time::sleep(refresh_interval) => {}
+                }
+            }
             let refresh_control = tokio::select! {
                 _ = worker_shutdown.cancelled() => break,
                 refresh_control = refresh() => refresh_control,
@@ -44,11 +61,7 @@ where
             if refresh_control == RefreshControl::Stop {
                 break;
             }
-
-            tokio::select! {
-                _ = worker_shutdown.cancelled() => break,
-                _ = tokio::time::sleep(refresh_interval) => {}
-            }
+            delay_before_refresh = true;
         }
     });
     PeriodicRefreshWorker {

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

@@ -13,7 +13,7 @@ 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 worker = spawn(Duration::from_millis(10), InitialRefresh::Immediate, {
         let refresh_count = Arc::clone(&refresh_count);
         let refreshed = Arc::clone(&refreshed);
         let hold_second_refresh = Arc::clone(&hold_second_refresh);
@@ -50,7 +50,7 @@ async fn refreshes_immediately_periodically_and_stops_when_dropped() {
 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 worker = spawn(Duration::from_millis(10), InitialRefresh::Immediate, {
         let refresh_count = Arc::clone(&refresh_count);
         let refreshed = Arc::clone(&refreshed);
         move || {
@@ -71,3 +71,26 @@ async fn stops_when_refresh_requests_it() {
 
     assert_eq!(refresh_count.load(Ordering::SeqCst), 1);
 }
+
+#[tokio::test(start_paused = true)]
+async fn can_delay_the_first_refresh() {
+    let refresh_count = Arc::new(AtomicUsize::new(0));
+    let worker = spawn(Duration::from_secs(60), InitialRefresh::AfterInterval, {
+        let refresh_count = Arc::clone(&refresh_count);
+        move || {
+            let refresh_count = Arc::clone(&refresh_count);
+            async move {
+                refresh_count.fetch_add(1, Ordering::SeqCst);
+                RefreshControl::Continue
+            }
+        }
+    });
+
+    tokio::task::yield_now().await;
+    assert_eq!(refresh_count.load(Ordering::SeqCst), 0);
+    tokio::time::advance(Duration::from_secs(60)).await;
+    tokio::task::yield_now().await;
+    drop(worker);
+
+    assert_eq!(refresh_count.load(Ordering::SeqCst), 1);
+}

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

@@ -7,6 +7,7 @@ use tracing::warn;
 
 use super::PeriodicRefreshWorker;
 use super::periodic;
+use super::periodic::InitialRefresh;
 use super::periodic::RefreshControl;
 use crate::config_manager::ConfigManager;
 
@@ -36,7 +37,7 @@ fn spawn_with_interval(
 ) -> PeriodicRefreshWorker {
     let plugins_manager = Arc::downgrade(plugins_manager);
     let auth_manager = Arc::downgrade(auth_manager);
-    periodic::spawn(refresh_interval, move || {
+    periodic::spawn(refresh_interval, InitialRefresh::Immediate, move || {
         let plugins_manager = plugins_manager.clone();
         let auth_manager = auth_manager.clone();
         let config_manager = config_manager.clone();

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

@@ -186,6 +186,7 @@ impl ExternalAuth for ExternalAuthRefreshBridge {
 
 pub(crate) struct MessageProcessor {
     outgoing: Arc<OutgoingMessageSender>,
+    apps_refresh_worker: PeriodicRefreshWorker,
     models_refresh_worker: ModelsRefreshWorker,
     plugins_refresh_worker: Option<PeriodicRefreshWorker>,
     skills_watcher: Arc<SkillsWatcher>,
@@ -389,6 +390,11 @@ impl MessageProcessor {
         });
         let models_manager = thread_manager.get_models_manager();
         let models_refresh_worker = crate::models_refresh_worker::spawn(&models_manager);
+        let apps_refresh_worker = crate::background_refresh::spawn_apps_refresh_worker(
+            config_manager.clone(),
+            &thread_manager.environment_manager(),
+            &thread_manager.mcp_manager(),
+        );
         thread_manager
             .plugins_manager()
             .set_analytics_events_client(analytics_events_client.clone());
@@ -563,6 +569,7 @@ impl MessageProcessor {
 
         Self {
             outgoing,
+            apps_refresh_worker,
             models_refresh_worker,
             plugins_refresh_worker,
             skills_watcher,
@@ -594,6 +601,7 @@ impl MessageProcessor {
     pub(crate) fn clear_runtime_references(&self) {
         self.account_processor.clear_external_auth();
         self.apps_processor.shutdown();
+        self.apps_refresh_worker.shutdown();
         self.models_refresh_worker.shutdown();
         if let Some(plugins_refresh_worker) = &self.plugins_refresh_worker {
             plugins_refresh_worker.shutdown();
@@ -774,6 +782,7 @@ impl MessageProcessor {
     }
 
     pub(crate) async fn drain_background_tasks(&self) {
+        self.apps_refresh_worker.shutdown();
         self.models_refresh_worker.shutdown();
         if let Some(plugins_refresh_worker) = &self.plugins_refresh_worker {
             plugins_refresh_worker.shutdown();