From 6059e58307b7652cac54ec60321559525b4d7507 Mon Sep 17 00:00:00 2001 From: James Watterson-Tenvooren Date: Sat, 20 Jun 2026 17:33:07 +0000 Subject: [PATCH] app-server: refresh Codex Apps tools periodically --- .../app-server/src/background_refresh/apps.rs | 79 +++++++++++++ .../src/background_refresh/apps_tests.rs | 108 ++++++++++++++++++ .../app-server/src/background_refresh/mod.rs | 2 + .../src/background_refresh/periodic.rs | 25 +++- .../src/background_refresh/periodic_tests.rs | 27 ++++- .../src/background_refresh/plugins.rs | 3 +- codex-rs/app-server/src/message_processor.rs | 9 ++ 7 files changed, 244 insertions(+), 9 deletions(-) create mode 100644 codex-rs/app-server/src/background_refresh/apps.rs create mode 100644 codex-rs/app-server/src/background_refresh/apps_tests.rs diff --git a/codex-rs/app-server/src/background_refresh/apps.rs b/codex-rs/app-server/src/background_refresh/apps.rs new file mode 100644 index 0000000000..9e1f39955f --- /dev/null +++ b/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, + mcp_manager: &Arc, +) -> PeriodicRefreshWorker { + spawn_with_interval( + config_manager, + environment_manager, + mcp_manager, + APPS_REFRESH_INTERVAL, + ) +} + +fn spawn_with_interval( + config_manager: ConfigManager, + environment_manager: &Arc, + mcp_manager: &Arc, + 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; diff --git a/codex-rs/app-server/src/background_refresh/apps_tests.rs b/codex-rs/app-server/src/background_refresh/apps_tests.rs new file mode 100644 index 0000000000..a3ab8f379b --- /dev/null +++ b/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::(&request.body) + .ok() + .and_then(|body| body.get("method").cloned()) + .as_ref() + .and_then(Value::as_str) + == Some("tools/list") + }) + .count() +} diff --git a/codex-rs/app-server/src/background_refresh/mod.rs b/codex-rs/app-server/src/background_refresh/mod.rs index 9daf35d7a8..cbe8fbf79a 100644 --- a/codex-rs/app-server/src/background_refresh/mod.rs +++ b/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; diff --git a/codex-rs/app-server/src/background_refresh/periodic.rs b/codex-rs/app-server/src/background_refresh/periodic.rs index e7a7b34d85..7f0aa318aa 100644 --- a/codex-rs/app-server/src/background_refresh/periodic.rs +++ b/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(refresh_interval: Duration, mut refresh: F) -> PeriodicRefreshWorker +pub(super) fn spawn( + refresh_interval: Duration, + initial_refresh: InitialRefresh, + mut refresh: F, +) -> PeriodicRefreshWorker where F: FnMut() -> Fut + Send + 'static, Fut: Future + 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 { diff --git a/codex-rs/app-server/src/background_refresh/periodic_tests.rs b/codex-rs/app-server/src/background_refresh/periodic_tests.rs index dd4e43ae6d..f2bedcbab6 100644 --- a/codex-rs/app-server/src/background_refresh/periodic_tests.rs +++ b/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); +} diff --git a/codex-rs/app-server/src/background_refresh/plugins.rs b/codex-rs/app-server/src/background_refresh/plugins.rs index 521710c731..eae6bba70e 100644 --- a/codex-rs/app-server/src/background_refresh/plugins.rs +++ b/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(); diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 35d979c601..8a89e9550d 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -186,6 +186,7 @@ impl ExternalAuth for ExternalAuthRefreshBridge { pub(crate) struct MessageProcessor { outgoing: Arc, + apps_refresh_worker: PeriodicRefreshWorker, models_refresh_worker: ModelsRefreshWorker, plugins_refresh_worker: Option, skills_watcher: Arc, @@ -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();