diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 79791c142..77040905c 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -7,6 +7,7 @@ use codex_app_server_protocol::PluginSharePrincipalRole; use codex_app_server_protocol::PluginShareTargetRole; use codex_config::types::McpServerConfig; use codex_core_plugins::OPENAI_CURATED_MARKETPLACE_NAME; +use codex_core_plugins::PluginListBackgroundTaskOptions; use codex_core_plugins::remote::REMOTE_GLOBAL_MARKETPLACE_NAME; use codex_core_plugins::remote::RemoteAppTemplateUnavailableReason; use codex_core_plugins::remote::RemotePluginScope; @@ -543,21 +544,30 @@ impl PluginRequestProcessor { return Ok(empty_response()); } let plugins_input = config.plugins_config_input(); - if include_local || marketplace_kinds.contains(&PluginListMarketplaceKind::SharedWithMe) { - plugins_manager.maybe_start_plugin_list_background_tasks_for_config( - &plugins_input, - auth.clone(), - &roots, - Some(self.effective_plugins_changed_callback()), + let include_shared_with_me = + marketplace_kinds.contains(&PluginListMarketplaceKind::SharedWithMe); + let include_global_remote = + !explicit_marketplace_kinds && config.features.enabled(Feature::RemotePlugin); + let remote_plugin_service_config = RemotePluginServiceConfig { + chatgpt_base_url: config.chatgpt_base_url.clone(), + }; + let refresh_global_remote_catalog_cache = include_global_remote + && codex_core_plugins::remote::has_cached_global_remote_plugin_catalog( + config.codex_home.as_path(), + &remote_plugin_service_config, + auth.as_ref(), ); - } let (mut data, marketplace_load_errors) = if include_local { let config_for_marketplace_listing = plugins_input.clone(); let plugins_manager_for_marketplace_listing = plugins_manager.clone(); + let roots_for_marketplace_listing = roots.clone(); let shared_plugin_ids_by_local_path = load_shared_plugin_ids_by_local_path(&config)?; match tokio::task::spawn_blocking(move || { let outcome = plugins_manager_for_marketplace_listing - .list_marketplaces_for_config(&config_for_marketplace_listing, &roots)?; + .list_marketplaces_for_config( + &config_for_marketplace_listing, + &roots_for_marketplace_listing, + )?; Ok::< ( Vec, @@ -617,9 +627,6 @@ impl PluginRequestProcessor { // TODO(remote plugins): Remove this once remote plugins are ready and vertical plugins are // served directly from the normal remote catalog. if include_vertical && !config.features.enabled(Feature::RemotePlugin) { - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; match codex_core_plugins::remote::fetch_openai_curated_remote_collection_marketplace( &remote_plugin_service_config, auth.as_ref(), @@ -644,21 +651,16 @@ impl PluginRequestProcessor { } let mut remote_sources = Vec::new(); - if !explicit_marketplace_kinds && config.features.enabled(Feature::RemotePlugin) { + if include_global_remote { remote_sources.push(RemoteMarketplaceSource::Global); } if marketplace_kinds.contains(&PluginListMarketplaceKind::WorkspaceDirectory) { remote_sources.push(RemoteMarketplaceSource::WorkspaceDirectory); } - if marketplace_kinds.contains(&PluginListMarketplaceKind::SharedWithMe) - && config.features.enabled(Feature::PluginSharing) - { + if include_shared_with_me && config.features.enabled(Feature::PluginSharing) { remote_sources.push(RemoteMarketplaceSource::SharedWithMe); } if !remote_sources.is_empty() { - let remote_plugin_service_config = RemotePluginServiceConfig { - chatgpt_base_url: config.chatgpt_base_url.clone(), - }; match codex_core_plugins::remote::fetch_remote_marketplaces( &remote_plugin_service_config, auth.as_ref(), @@ -702,6 +704,17 @@ impl PluginRequestProcessor { } } } + if include_local || include_shared_with_me || include_global_remote { + plugins_manager.maybe_start_plugin_list_background_tasks_for_config( + &plugins_input, + auth.clone(), + &roots, + PluginListBackgroundTaskOptions { + refresh_global_remote_catalog_cache, + }, + Some(self.effective_plugins_changed_callback()), + ); + } let featured_plugin_ids = if data .iter() diff --git a/codex-rs/app-server/tests/suite/v2/plugin_list.rs b/codex-rs/app-server/tests/suite/v2/plugin_list.rs index d9a0a0049..418f071e9 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_list.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_list.rs @@ -1863,6 +1863,102 @@ async fn plugin_list_includes_remote_marketplaces_when_remote_plugin_enabled() - Ok(()) } +#[tokio::test] +async fn plugin_list_uses_cached_global_remote_catalog_and_refreshes_it() -> Result<()> { + let codex_home = TempDir::new()?; + let server = MockServer::start().await; + write_remote_plugin_catalog_config( + codex_home.path(), + &format!("{}/backend-api/", server.uri()), + )?; + write_chatgpt_auth( + codex_home.path(), + ChatGptAuthFixture::new("chatgpt-token") + .account_id("account-123") + .chatgpt_user_id("user-123") + .chatgpt_account_id("account-123"), + AuthCredentialsStoreMode::File, + )?; + + let cached_remote_plugin_id = "plugins~Plugin_00000000000000000000000000000000"; + let refreshed_remote_plugin_id = "plugins~Plugin_11111111111111111111111111111111"; + let cached_body = + remote_plugin_list_body(cached_remote_plugin_id, "linear", "Linear", "Plan work"); + let refreshed_body = remote_plugin_list_body( + refreshed_remote_plugin_id, + "notion", + "Notion", + "Capture notes", + ); + mount_remote_plugin_list(&server, "GLOBAL", &cached_body).await; + mount_remote_installed_plugins(&server, "GLOBAL", empty_remote_installed_plugins_body()).await; + mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body()) + .await; + + let mut mcp = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + }) + .await?; + let response: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??, + )?; + let remote_marketplace = response + .marketplaces + .iter() + .find(|marketplace| marketplace.name == "openai-curated-remote") + .expect("expected warmed remote marketplace"); + assert_eq!( + remote_marketplace.plugins[0].id, + "linear@openai-curated-remote" + ); + wait_for_remote_plugin_request_count(&server, "/ps/plugins/list", /*expected_count*/ 1).await?; + wait_for_cached_remote_catalog_plugin_ids(codex_home.path(), &[cached_remote_plugin_id]) + .await?; + + server.reset().await; + mount_remote_plugin_list(&server, "GLOBAL", &refreshed_body).await; + mount_remote_installed_plugins(&server, "GLOBAL", empty_remote_installed_plugins_body()).await; + mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body()) + .await; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + }) + .await?; + let response: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??, + )?; + let remote_marketplace = response + .marketplaces + .iter() + .find(|marketplace| marketplace.name == "openai-curated-remote") + .expect("expected cached remote marketplace"); + assert_eq!( + remote_marketplace.plugins[0].id, + "linear@openai-curated-remote" + ); + wait_for_remote_plugin_request_count(&server, "/ps/plugins/list", /*expected_count*/ 1).await?; + wait_for_cached_remote_catalog_plugin_ids(codex_home.path(), &[refreshed_remote_plugin_id]) + .await?; + + Ok(()) +} + #[tokio::test] async fn plugin_list_includes_openai_curated_remote_collection_when_requested() -> Result<()> { let codex_home = TempDir::new()?; @@ -3025,6 +3121,52 @@ async fn wait_for_remote_installed_scope_request(server: &MockServer, scope: &st Ok(()) } +async fn wait_for_cached_remote_catalog_plugin_ids( + codex_home: &std::path::Path, + expected_plugin_ids: &[&str], +) -> Result<()> { + let mut expected_plugin_ids = expected_plugin_ids + .iter() + .copied() + .map(str::to_string) + .collect::>(); + expected_plugin_ids.sort(); + timeout(DEFAULT_TIMEOUT, async { + loop { + let plugin_ids = cached_remote_catalog_plugin_ids(codex_home)?; + if plugin_ids == expected_plugin_ids { + return Ok::<(), anyhow::Error>(()); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await??; + Ok(()) +} + +fn cached_remote_catalog_plugin_ids(codex_home: &std::path::Path) -> Result> { + let cache_dir = codex_home.join("cache/remote_plugin_catalog"); + if !cache_dir.exists() { + return Ok(Vec::new()); + } + let mut plugin_ids = Vec::new(); + for entry in std::fs::read_dir(cache_dir)? { + let path = entry?.path(); + let cached_catalog: serde_json::Value = serde_json::from_slice(&std::fs::read(path)?)?; + let Some(plugins) = cached_catalog["plugins"].as_array() else { + continue; + }; + plugin_ids.extend( + plugins + .iter() + .filter_map(|plugin| plugin["id"].as_str()) + .map(str::to_string), + ); + } + plugin_ids.sort(); + Ok(plugin_ids) +} + async fn wait_for_path_exists(path: &std::path::Path) -> Result<()> { timeout(DEFAULT_TIMEOUT, async { loop { @@ -3063,6 +3205,43 @@ async fn mount_remote_plugin_list(server: &MockServer, scope: &str, body: &str) .await; } +fn remote_plugin_list_body( + remote_plugin_id: &str, + plugin_name: &str, + display_name: &str, + short_description: &str, +) -> String { + format!( + r#"{{ + "plugins": [ + {{ + "id": "{remote_plugin_id}", + "name": "{plugin_name}", + "scope": "GLOBAL", + "installation_policy": "AVAILABLE", + "authentication_policy": "ON_USE", + "status": "ENABLED", + "release": {{ + "version": "1.2.3", + "display_name": "{display_name}", + "description": "{display_name}", + "app_ids": [], + "interface": {{ + "short_description": "{short_description}", + "capabilities": ["Read"] + }}, + "skills": [] + }} + }} + ], + "pagination": {{ + "limit": 50, + "next_page_token": null + }} +}}"# + ) +} + async fn mount_openai_curated_remote_collection_plugin_list(server: &MockServer, body: &str) { Mock::given(method("GET")) .and(path("/backend-api/ps/plugins/list")) diff --git a/codex-rs/core-plugins/src/lib.rs b/codex-rs/core-plugins/src/lib.rs index 8cfb9092f..84eeea5dc 100644 --- a/codex-rs/core-plugins/src/lib.rs +++ b/codex-rs/core-plugins/src/lib.rs @@ -34,6 +34,7 @@ pub use manager::PluginDetailsUnavailableReason; pub use manager::PluginInstallError; pub use manager::PluginInstallOutcome; pub use manager::PluginInstallRequest; +pub use manager::PluginListBackgroundTaskOptions; pub use manager::PluginReadOutcome; pub use manager::PluginReadRequest; pub use manager::PluginUninstallError; diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 1839b06e0..59cfd135e 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -146,6 +146,22 @@ struct RemoteInstalledPluginsCacheRefreshState { in_flight: bool, } +struct GlobalRemoteCatalogCacheRefreshRequest { + service_config: RemotePluginServiceConfig, + auth: Option, +} + +#[derive(Default)] +struct GlobalRemoteCatalogCacheRefreshState { + requested: Option, + in_flight: bool, +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct PluginListBackgroundTaskOptions { + pub refresh_global_remote_catalog_cache: bool, +} + #[derive(Clone, PartialEq, Eq)] struct NonCuratedCacheRefreshRequest { roots: Vec, @@ -302,6 +318,7 @@ pub struct PluginsManager { enabled_outcome_load_semaphore: Semaphore, remote_installed_plugins_cache: RwLock>>, remote_installed_plugins_cache_refresh_state: RwLock, + global_remote_catalog_cache_refresh_state: RwLock, restriction_product: Option, analytics_events_client: RwLock>, } @@ -355,6 +372,9 @@ impl PluginsManager { remote_installed_plugins_cache_refresh_state: RwLock::new( RemoteInstalledPluginsCacheRefreshState::default(), ), + global_remote_catalog_cache_refresh_state: RwLock::new( + GlobalRemoteCatalogCacheRefreshState::default(), + ), restriction_product, analytics_events_client: RwLock::new(None), } @@ -702,14 +722,33 @@ impl PluginsManager { ); } + fn maybe_start_global_remote_catalog_cache_refresh( + self: &Arc, + config: &PluginsConfigInput, + auth: Option, + ) { + if !config.plugins_enabled || !config.remote_plugin_enabled { + return; + } + + self.schedule_global_remote_catalog_cache_refresh(GlobalRemoteCatalogCacheRefreshRequest { + service_config: remote_plugin_service_config(config), + auth, + }); + } + pub fn maybe_start_plugin_list_background_tasks_for_config( self: &Arc, config: &PluginsConfigInput, auth: Option, roots: &[AbsolutePathBuf], + options: PluginListBackgroundTaskOptions, on_effective_plugins_changed: Option>, ) { self.maybe_start_non_curated_plugin_cache_refresh(roots); + if options.refresh_global_remote_catalog_cache { + self.maybe_start_global_remote_catalog_cache_refresh(config, auth.clone()); + } self.maybe_start_remote_installed_plugins_cache_refresh( config, auth.clone(), @@ -1447,6 +1486,33 @@ impl PluginsManager { }); } + fn schedule_global_remote_catalog_cache_refresh( + self: &Arc, + request: GlobalRemoteCatalogCacheRefreshRequest, + ) { + let should_spawn = { + let mut state = match self.global_remote_catalog_cache_refresh_state.write() { + Ok(state) => state, + Err(err) => err.into_inner(), + }; + state.requested = Some(request); + if state.in_flight { + false + } else { + state.in_flight = true; + true + } + }; + if !should_spawn { + return; + } + + let manager = Arc::clone(self); + tokio::spawn(async move { + manager.run_global_remote_catalog_cache_refresh_loop().await; + }); + } + fn schedule_non_curated_plugin_cache_refresh( self: &Arc, roots: &[AbsolutePathBuf], @@ -1613,6 +1679,44 @@ impl PluginsManager { } } + async fn run_global_remote_catalog_cache_refresh_loop(self: Arc) { + loop { + let request = { + let mut state = match self.global_remote_catalog_cache_refresh_state.write() { + Ok(state) => state, + Err(err) => err.into_inner(), + }; + match state.requested.take() { + Some(request) => request, + None => { + state.in_flight = false; + return; + } + } + }; + + match crate::remote::fetch_and_cache_global_remote_plugin_catalog( + self.codex_home.as_path(), + &request.service_config, + request.auth.as_ref(), + ) + .await + { + Ok(()) => {} + Err( + RemotePluginCatalogError::AuthRequired + | RemotePluginCatalogError::UnsupportedAuthMode, + ) => {} + Err(err) => { + warn!( + error = %err, + "failed to refresh cached global remote plugin catalog" + ); + } + } + } + } + fn run_non_curated_plugin_cache_refresh_loop(self: Arc) { loop { let request = { diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index b96746b63..332546fdc 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -576,6 +576,25 @@ pub async fn fetch_remote_marketplaces( match source { RemoteMarketplaceSource::Global => { let scope = RemotePluginScope::Global; + if let Some(codex_home) = global_catalog_cache_path + && let Some(directory_plugins) = + catalog_cache::load_cached_global_directory_plugins( + codex_home, config, auth, + ) + { + let installed_plugins = + fetch_installed_plugins_for_scope(config, auth, scope).await?; + if let Some(marketplace) = build_remote_marketplace( + scope.marketplace_name(), + scope.marketplace_display_name(), + directory_plugins, + installed_plugins, + /*include_installed_only*/ true, + )? { + marketplaces.push(marketplace); + } + continue; + } let (directory_plugins, installed_plugins) = tokio::try_join!( fetch_directory_plugins_for_scope(config, auth, scope), fetch_installed_plugins_for_scope(config, auth, scope), @@ -692,6 +711,17 @@ pub async fn fetch_and_cache_global_remote_plugin_catalog( Ok(()) } +pub fn has_cached_global_remote_plugin_catalog( + codex_home: &Path, + config: &RemotePluginServiceConfig, + auth: Option<&CodexAuth>, +) -> bool { + let Ok(auth) = ensure_chatgpt_auth(auth) else { + return false; + }; + catalog_cache::load_cached_global_directory_plugins(codex_home, config, auth).is_some() +} + pub fn cached_global_remote_discoverable_plugins( codex_home: &Path, config: &RemotePluginServiceConfig,