From 83ff1c2f809d810482be0bfda197f4f3f7abd697 Mon Sep 17 00:00:00 2001 From: brisebois-oai Date: Thu, 23 Jul 2026 00:13:19 +0000 Subject: [PATCH] Cache remote plugin catalogs by scope (#34849) ## What changed - Cache global, user, and workspace remote plugin catalogs on disk with a three-hour TTL and scope- and account-specific keys. - Serve cached catalogs from `plugin/list`, refresh stale entries in the background, and warm existing caches at app-server startup without blocking list requests. - Make `forceRefetch: true` bypass cached catalog data and replace an entry only after a successful fetch. - Invalidate user and workspace catalog caches after remote plugin sharing changes. ## Testing - Cover cache freshness, scope isolation, legacy global cache compatibility, identity requirements for private catalogs, forced refetches, and non-blocking startup refreshes. GitOrigin-RevId: 7844e9dfb72e43b26d749b4afd16fd9c1793caba --- codex-rs/app-server/README.md | 2 +- .../src/request_processors/plugins.rs | 46 +- .../app-server/tests/suite/v2/plugin_list.rs | 650 +++++++++++++++++- codex-rs/core-plugins/src/manager.rs | 175 +++-- codex-rs/core-plugins/src/remote.rs | 222 ++++-- .../core-plugins/src/remote/catalog_cache.rs | 120 +++- .../src/remote/catalog_cache_tests.rs | 127 ++++ codex-rs/core-plugins/src/remote_tests.rs | 24 + codex-rs/state/src/migrations_tests.rs | 3 +- 9 files changed, 1223 insertions(+), 146 deletions(-) create mode 100644 codex-rs/core-plugins/src/remote/catalog_cache_tests.rs diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index 1ef2eafbdc..79ef624186 100644 --- a/codex-rs/app-server/README.md +++ b/codex-rs/app-server/README.md @@ -216,7 +216,7 @@ Example with notification opt-out: - `marketplace/add` — add a remote plugin marketplace from an HTTP(S) Git URL, SSH Git URL, or GitHub `owner/repo` shorthand, then persist it into the user marketplace config. Returns the installed root path plus whether the marketplace was already present. - `marketplace/remove` — remove a configured marketplace by name from the user marketplace config, and delete its installed marketplace root when one exists. - `marketplace/upgrade` — upgrade all configured Git plugin marketplaces, or one named marketplace when `marketplaceName` is provided. Returns selected marketplace names, upgraded roots, and per-marketplace errors. -- `plugin/list` — list discovered plugin marketplaces and plugin state, including effective marketplace install/auth policy metadata, nullable remote install-policy provenance in `installPolicySource` (`WORKSPACE_SETTING` or `IMPLICIT_CANONICAL_APP`), the remote marketplace `version` and locally materialized `localVersion` when available, plugin `availability` (`AVAILABLE` by default or `DISABLED_BY_ADMIN` for remote plugins blocked upstream), fail-open `marketplaceLoadErrors` entries for marketplace files that could not be parsed or loaded, and best-effort `featuredPluginIds` for the official curated marketplace. Every `PluginSummary` returned by plugin list, installed, read, and share-list methods includes `mustShowInstallationInterstitial`: remote service values preserve `true` or `false`, while local plugins and remote responses that omit the policy return `null`. Clients should fail closed when the value is `null`. Clients can explicitly request the remote `workspace-directory`, `shared-with-me`, or `created-by-me-remote` marketplace kinds. The optional `forceRefetch` request field is currently accepted but has no effect. `interface.category` uses the marketplace category when present; otherwise it falls back to the plugin manifest category (**under development; do not call from production clients yet**). +- `plugin/list` — list discovered plugin marketplaces and plugin state, including effective marketplace install/auth policy metadata, nullable remote install-policy provenance in `installPolicySource` (`WORKSPACE_SETTING` or `IMPLICIT_CANONICAL_APP`), the remote marketplace `version` and locally materialized `localVersion` when available, plugin `availability` (`AVAILABLE` by default or `DISABLED_BY_ADMIN` for remote plugins blocked upstream), fail-open `marketplaceLoadErrors` entries for marketplace files that could not be parsed or loaded, and best-effort `featuredPluginIds` for the official curated marketplace. Every `PluginSummary` returned by plugin list, installed, read, and share-list methods includes `mustShowInstallationInterstitial`: remote service values preserve `true` or `false`, while local plugins and remote responses that omit the policy return `null`. Clients should fail closed when the value is `null`. Clients can explicitly request the remote `workspace-directory`, `shared-with-me`, or `created-by-me-remote` marketplace kinds. Set `forceRefetch: true` to bypass TTL-backed remote catalog caches for the requested marketplaces and wait for fresh data; cache entries are replaced only after a successful fetch. At app-server startup, existing cached catalogs remain available to `plugin/list` while they refresh in the background. `interface.category` uses the marketplace category when present; otherwise it falls back to the plugin manifest category (**under development; do not call from production clients yet**). - `plugin/installed` — list installed plugin rows plus any explicitly requested local install-suggestion plugin names, without fetching the broader remote catalog. Remote rows include nullable `installPolicySource`; local rows return `null`. Mention surfaces can use this narrower view when they need plugin mention payloads rather than plugin-page discovery data (**under development; do not call from production clients yet**). - `plugin/read` — read one plugin by `marketplacePath` plus `pluginName`, returning marketplace info, a list-style `summary`, manifest descriptions/interface metadata, and bundled skills/hooks/apps/MCP server names. Remote plugin details can include scheduled task summaries from the catalog; `scheduledTasks: null` means the metadata is unavailable, while an empty array means the catalog found no scheduled tasks. Remote plugin details expose the canonical `shareUrl` supplied by the remote catalog when available; it is `null` for local plugins or when the catalog omits it. This field is separate from `summary.shareContext`, which continues to describe user and workspace sharing state. Returned plugin skills include their current `enabled` state after local config filtering; bundled hooks are returned as lightweight declaration summaries keyed for correlation with `hooks/list`. Use `plugin/install`'s `appsNeedingAuth` to drive post-install authentication and `app/list`'s `isAccessible` to determine current connector accessibility (**under development; do not call from production clients yet**). - `plugin/skill/read` — read remote plugin skill markdown on demand by `remoteMarketplaceName`, `remotePluginId`, and `skillName`. This lets clients preview uninstalled remote plugin skills without downloading the plugin bundle. diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index c9e1b29ce7..b0e54e3935 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -17,6 +17,8 @@ use codex_core_plugins::remote::REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME use codex_core_plugins::remote::REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME; use codex_core_plugins::remote::REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME; use codex_core_plugins::remote::RemoteAppTemplateUnavailableReason; +use codex_core_plugins::remote::RemotePluginCatalogCacheMode; +use codex_core_plugins::remote::RemotePluginScope; use codex_core_plugins::remote::is_valid_remote_plugin_id; use codex_core_plugins::remote::validate_remote_plugin_id; use codex_core_plugins::remote_bundle::RemotePluginBundleInstallError; @@ -540,7 +542,7 @@ impl PluginRequestProcessor { let PluginListParams { cwds, marketplace_kinds, - force_refetch: _, + force_refetch, } = params; let roots = cwds.unwrap_or_default(); let explicit_marketplace_kinds = marketplace_kinds.is_some(); @@ -578,12 +580,12 @@ impl PluginRequestProcessor { let use_remote_global_catalog = include_global_remote && auth_mode.is_some_and(DomainAuthMode::uses_codex_backend); let remote_plugin_service_config = remote_plugin_service_config(&config); - let refresh_global_remote_catalog_cache = use_remote_global_catalog - && codex_core_plugins::remote::has_cached_global_remote_plugin_catalog( - config.codex_home.as_path(), - &remote_plugin_service_config, - auth.as_ref(), - ); + let remote_catalog_cache_mode = if force_refetch { + RemotePluginCatalogCacheMode::ForceRefetch + } else { + RemotePluginCatalogCacheMode::PreferCache + }; + let mut remote_catalog_cache_refresh_scopes = Default::default(); 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(); @@ -700,12 +702,15 @@ impl PluginRequestProcessor { &remote_plugin_service_config, auth.as_ref(), &remote_sources, - /*global_catalog_cache_path*/ Some(config.codex_home.as_path()), + /*catalog_cache_root*/ Some(config.codex_home.as_path()), + remote_catalog_cache_mode, ) .await { - Ok(remote_marketplaces) => { - for remote_marketplace in remote_marketplaces + Ok(outcome) => { + remote_catalog_cache_refresh_scopes = outcome.catalog_cache_refresh_scopes; + for remote_marketplace in outcome + .marketplaces .into_iter() .map(remote_marketplace_to_info) { @@ -743,13 +748,14 @@ impl PluginRequestProcessor { || include_created_by_me_remote || include_shared_with_me || include_global_remote + || !remote_catalog_cache_refresh_scopes.is_empty() { plugins_manager.maybe_start_plugin_list_background_tasks_for_config( &plugins_input, auth.clone(), &roots, PluginListBackgroundTaskOptions { - refresh_global_remote_catalog_cache, + remote_catalog_cache_refresh_scopes, }, Some(self.effective_plugins_changed_callback()), ); @@ -1254,6 +1260,12 @@ impl PluginRequestProcessor { ) .await .map_err(|err| remote_plugin_catalog_error_to_jsonrpc(err, "save remote plugin share"))?; + codex_core_plugins::remote::invalidate_cached_remote_plugin_catalog_scopes( + config.codex_home.as_path(), + &remote_plugin_service_config, + auth.as_ref(), + &[RemotePluginScope::User, RemotePluginScope::Workspace], + ); let remote_plugin_id = result.remote_plugin_id; self.clear_plugin_related_caches(); Ok(PluginShareSaveResponse { @@ -1292,6 +1304,12 @@ impl PluginRequestProcessor { .map_err(|err| { remote_plugin_catalog_error_to_jsonrpc(err, "update remote plugin share targets") })?; + codex_core_plugins::remote::invalidate_cached_remote_plugin_catalog_scopes( + config.codex_home.as_path(), + &remote_plugin_service_config, + auth.as_ref(), + &[RemotePluginScope::User, RemotePluginScope::Workspace], + ); self.clear_plugin_related_caches(); Ok(PluginShareUpdateTargetsResponse { principals: result @@ -1385,6 +1403,12 @@ impl PluginRequestProcessor { ) .await .map_err(|err| remote_plugin_catalog_error_to_jsonrpc(err, "delete remote plugin share"))?; + codex_core_plugins::remote::invalidate_cached_remote_plugin_catalog_scopes( + config.codex_home.as_path(), + &remote_plugin_service_config, + auth.as_ref(), + &[RemotePluginScope::User, RemotePluginScope::Workspace], + ); self.clear_plugin_related_caches(); Ok(PluginShareDeleteResponse {}) } 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 d5cf3d7786..29916e189d 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_list.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_list.rs @@ -4,7 +4,10 @@ use anyhow::Result; use anyhow::bail; use app_test_support::ChatGptAuthFixture; use app_test_support::TestAppServer; +use app_test_support::to_response; use app_test_support::write_chatgpt_auth; +use chrono::Duration as ChronoDuration; +use chrono::Utc; use codex_app_server_protocol::HookMetadata; use codex_app_server_protocol::HookTrustStatus; use codex_app_server_protocol::HooksListParams; @@ -1902,6 +1905,7 @@ async fn plugin_list_includes_remote_marketplaces_when_remote_plugin_enabled() - let cached_catalog: serde_json::Value = serde_json::from_slice(&std::fs::read(&cache_files[0])?)?; assert_eq!(cached_catalog["schema_version"], serde_json::json!(1)); + assert!(cached_catalog["fetched_at"].as_str().is_some()); assert_eq!( cached_catalog["plugins"][0]["installation_policy_source"], serde_json::json!("IMPLICIT_CANONICAL_APP") @@ -1943,7 +1947,7 @@ async fn plugin_list_includes_remote_marketplaces_when_remote_plugin_enabled() - } #[tokio::test] -async fn plugin_list_uses_cached_global_remote_catalog_and_refreshes_it() -> Result<()> { +async fn plugin_list_honors_global_remote_catalog_cache_ttl() -> Result<()> { let codex_home = TempDir::new()?; let server = MockServer::start().await; write_remote_plugin_catalog_config( @@ -2032,9 +2036,468 @@ async fn plugin_list_uses_cached_global_remote_catalog_and_refreshes_it() -> Res remote_marketplace.plugins[0].id, "linear@openai-curated-remote" ); + sleep(Duration::from_millis(100)).await; + wait_for_remote_plugin_request_count(&server, "/ps/plugins/list", /*expected_count*/ 0).await?; + wait_for_cached_remote_catalog_plugin_ids(codex_home.path(), &[cached_remote_plugin_id]) + .await?; + + rewrite_cached_remote_catalog_fetched_at( + codex_home.path(), + Utc::now() - ChronoDuration::hours(4), + )?; + server.reset().await; + mount_delayed_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; + mount_empty_user_installed_plugins(&server).await; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + force_refetch: false, + }) + .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 stale cached remote marketplace"); + assert_eq!( + remote_marketplace.plugins[0].id, + "linear@openai-curated-remote" + ); + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + force_refetch: false, + }) + .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 stale 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?; + sleep(Duration::from_millis(100)).await; + wait_for_remote_plugin_request_count(&server, "/ps/plugins/list", /*expected_count*/ 1).await?; + + Ok(()) +} + +#[tokio::test] +async fn app_server_startup_refreshes_cached_remote_catalog_without_blocking_plugin_list() +-> 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"; + mount_remote_plugin_list( + &server, + "GLOBAL", + &remote_plugin_list_body(cached_remote_plugin_id, "linear", "Linear", "Plan work"), + ) + .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; + mount_empty_user_installed_plugins(&server).await; + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + timeout(DEFAULT_TIMEOUT, app_server.initialize()).await??; + let request_id = app_server + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + force_refetch: false, + }) + .await?; + let _: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??, + )?; + wait_for_cached_remote_catalog_plugin_ids(codex_home.path(), &[cached_remote_plugin_id]) + .await?; + timeout(DEFAULT_TIMEOUT, app_server.shutdown_gracefully()).await??; + + server.reset().await; + let refreshed_body = remote_plugin_list_body( + refreshed_remote_plugin_id, + "notion", + "Notion", + "Capture notes", + ); + Mock::given(method("GET")) + .and(path("/backend-api/ps/plugins/list")) + .and(query_param("scope", "GLOBAL")) + .and(query_param("limit", "200")) + .and(header("authorization", "Bearer chatgpt-token")) + .and(header("chatgpt-account-id", "account-123")) + .respond_with( + ResponseTemplate::new(200) + .set_body_string(refreshed_body) + .set_delay(Duration::from_secs(/*secs*/ 2)), + ) + .mount(&server) + .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; + mount_empty_user_installed_plugins(&server).await; + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .with_plugin_startup_tasks() + .build() + .await?; + timeout(DEFAULT_TIMEOUT, app_server.initialize()).await??; + wait_for_remote_plugin_request_count(&server, "/ps/plugins/list", /*expected_count*/ 1).await?; + + let request_id = app_server + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + force_refetch: false, + }) + .await?; + let response: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + app_server.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_cached_remote_catalog_plugin_ids(codex_home.path(), &[refreshed_remote_plugin_id]) + .await?; + let request_id = app_server + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + force_refetch: false, + }) + .await?; + let response: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + app_server.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 refreshed remote marketplace"); + assert_eq!( + remote_marketplace.plugins[0].id, + "notion@openai-curated-remote" + ); + sleep(Duration::from_millis(100)).await; + wait_for_remote_plugin_request_count(&server, "/ps/plugins/list", /*expected_count*/ 1).await?; + + Ok(()) +} + +#[tokio::test] +async fn app_server_startup_skips_disabled_remote_plugin_catalog_scopes() -> Result<()> { + let codex_home = TempDir::new()?; + let server = MockServer::start().await; + let base_url = format!("{}/backend-api/", server.uri()); + write_remote_plugin_catalog_config(codex_home.path(), &base_url)?; + 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 global_plugin_id = "plugins~Plugin_00000000000000000000000000000000"; + let user_plugin_id = "plugins~Plugin_11111111111111111111111111111111"; + let workspace_plugin_id = "plugins~Plugin_22222222222222222222222222222222"; + let global_body = remote_plugin_list_body(global_plugin_id, "global-linear", "Linear", "Plan"); + let user_body = user_remote_plugin_page_body( + user_plugin_id, + "private-linear", + "Private Linear", + "PRIVATE", + /*enabled*/ None, + ); + let workspace_body = workspace_remote_plugin_page_body( + workspace_plugin_id, + "workspace-linear", + "Workspace Linear", + "LISTED", + /*enabled*/ None, + ); + mount_remote_plugin_list(&server, "GLOBAL", &global_body).await; + mount_remote_plugin_list(&server, "USER", &user_body).await; + mount_remote_plugin_list(&server, "WORKSPACE", &workspace_body).await; + for scope in ["GLOBAL", "USER", "WORKSPACE"] { + mount_remote_installed_plugins(&server, scope, empty_remote_installed_plugins_body()).await; + } + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build() + .await?; + timeout(DEFAULT_TIMEOUT, app_server.initialize()).await??; + for marketplace_kinds in [ + None, + Some(vec![ + PluginListMarketplaceKind::CreatedByMeRemote, + PluginListMarketplaceKind::WorkspaceDirectory, + ]), + ] { + let request_id = app_server + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds, + force_refetch: false, + }) + .await?; + let _: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??, + )?; + } + wait_for_cached_remote_catalog_plugin_ids( + codex_home.path(), + &[global_plugin_id, user_plugin_id, workspace_plugin_id], + ) + .await?; + timeout(DEFAULT_TIMEOUT, app_server.shutdown_gracefully()).await??; + + write_remote_plugins_disabled_config_with_base_url(codex_home.path(), &base_url)?; + server.reset().await; + mount_remote_plugin_list(&server, "GLOBAL", &global_body).await; + mount_remote_plugin_list(&server, "USER", &user_body).await; + mount_remote_plugin_list(&server, "WORKSPACE", &workspace_body).await; + for scope in ["GLOBAL", "USER", "WORKSPACE"] { + mount_remote_installed_plugins(&server, scope, empty_remote_installed_plugins_body()).await; + } + + let mut app_server = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .with_plugin_startup_tasks() + .build() + .await?; + timeout(DEFAULT_TIMEOUT, app_server.initialize()).await??; + wait_for_remote_plugin_list_scope_request_count( + &server, + "WORKSPACE", + /*expected_count*/ 1, + ) + .await?; + + let requested_scopes = server + .received_requests() + .await + .expect("wiremock should record requests") + .into_iter() + .filter(|request| { + request.method == "GET" && request.url.path().ends_with("/ps/plugins/list") + }) + .filter_map(|request| { + request + .url + .query_pairs() + .find(|(name, _)| name == "scope") + .map(|(_, scope)| scope.into_owned()) + }) + .collect::>(); + assert_eq!(requested_scopes, vec!["WORKSPACE".to_string()]); + + Ok(()) +} + +#[tokio::test] +async fn plugin_list_force_refetch_bypasses_fresh_global_remote_catalog_cache() -> 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"; + mount_remote_plugin_list( + &server, + "GLOBAL", + &remote_plugin_list_body(cached_remote_plugin_id, "linear", "Linear", "Plan work"), + ) + .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; + mount_empty_user_installed_plugins(&server).await; + + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .without_auto_env() + .build() + .await?; + timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + force_refetch: false, + }) + .await?; + let _: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??, + )?; + wait_for_cached_remote_catalog_plugin_ids(codex_home.path(), &[cached_remote_plugin_id]) + .await?; + + server.reset().await; + mount_delayed_remote_plugin_list( + &server, + "GLOBAL", + &remote_plugin_list_body( + refreshed_remote_plugin_id, + "notion", + "Notion", + "Capture notes", + ), + ) + .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; + mount_empty_user_installed_plugins(&server).await; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + force_refetch: true, + }) + .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 refreshed remote marketplace"); + assert_eq!( + remote_marketplace.plugins[0].id, + "notion@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?; + + server.reset().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; + mount_empty_user_installed_plugins(&server).await; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: None, + force_refetch: false, + }) + .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 refreshed remote marketplace"); + assert_eq!( + remote_marketplace.plugins[0].id, + "notion@openai-curated-remote" + ); + wait_for_remote_plugin_request_count(&server, "/ps/plugins/list", /*expected_count*/ 0).await?; Ok(()) } @@ -2948,6 +3411,13 @@ async fn plugin_list_fetches_workspace_directory_kind_when_remote_plugin_disable "LISTED", /*enabled*/ Some(false), ); + let refreshed_workspace_plugin_body = workspace_remote_plugin_page_body( + "plugins~Plugin_22222222222222222222222222222222", + "workspace-notion", + "Workspace Notion", + "LISTED", + /*enabled*/ None, + ); mount_remote_plugin_list(&server, "WORKSPACE", &workspace_plugin_body).await; mount_remote_installed_plugins(&server, "WORKSPACE", &workspace_installed_body).await; mount_empty_user_installed_plugins(&server).await; @@ -3002,6 +3472,100 @@ async fn plugin_list_fetches_workspace_directory_kind_when_remote_plugin_disable .query() .is_some_and(|query| query.contains("scope=GLOBAL"))) ); + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: Some(vec![PluginListMarketplaceKind::WorkspaceDirectory]), + force_refetch: false, + }) + .await?; + let response: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??, + )?; + assert_eq!( + response.marketplaces[0].plugins[0].id, + "workspace-linear@workspace-directory" + ); + sleep(Duration::from_millis(100)).await; + wait_for_remote_plugin_list_scope_request_count( + &server, + "WORKSPACE", + /*expected_count*/ 1, + ) + .await?; + + rewrite_cached_remote_catalog_fetched_at( + codex_home.path(), + Utc::now() - ChronoDuration::hours(4), + )?; + server.reset().await; + mount_delayed_remote_plugin_list(&server, "WORKSPACE", &refreshed_workspace_plugin_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; + mount_empty_user_installed_plugins(&server).await; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: Some(vec![PluginListMarketplaceKind::WorkspaceDirectory]), + force_refetch: false, + }) + .await?; + let response: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??, + )?; + assert_eq!( + response.marketplaces[0].plugins[0].id, + "workspace-linear@workspace-directory" + ); + + wait_for_remote_plugin_list_scope_request_count( + &server, + "WORKSPACE", + /*expected_count*/ 1, + ) + .await?; + wait_for_cached_remote_catalog_plugin_ids( + codex_home.path(), + &["plugins~Plugin_22222222222222222222222222222222"], + ) + .await?; + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: Some(vec![PluginListMarketplaceKind::WorkspaceDirectory]), + force_refetch: false, + }) + .await?; + let response: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??, + )?; + assert_eq!( + response.marketplaces[0].plugins[0].id, + "workspace-notion@workspace-directory" + ); + sleep(Duration::from_millis(100)).await; + wait_for_remote_plugin_list_scope_request_count( + &server, + "WORKSPACE", + /*expected_count*/ 1, + ) + .await?; Ok(()) } @@ -3137,6 +3701,24 @@ plugin_sharing = false .any(|(key, value)| key == "scope" && value != "USER") }) ); + + let request_id = mcp + .send_plugin_list_request(PluginListParams { + cwds: None, + marketplace_kinds: Some(vec![PluginListMarketplaceKind::CreatedByMeRemote]), + force_refetch: false, + }) + .await?; + let response: PluginListResponse = to_response( + timeout( + DEFAULT_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??, + )?; + assert_eq!(response.marketplaces[0].plugins.len(), 2); + sleep(Duration::from_millis(100)).await; + wait_for_remote_plugin_list_scope_request_count(&server, "USER", /*expected_count*/ 2).await?; Ok(()) } @@ -3759,6 +4341,42 @@ async fn wait_for_remote_plugin_request_count( Ok(()) } +async fn wait_for_remote_plugin_list_scope_request_count( + server: &MockServer, + scope: &str, + expected_count: usize, +) -> Result<()> { + timeout(DEFAULT_TIMEOUT, async { + loop { + let Some(requests) = server.received_requests().await else { + bail!("wiremock did not record requests"); + }; + let request_count = requests + .iter() + .filter(|request| { + request.method == "GET" + && request.url.path().ends_with("/ps/plugins/list") + && request + .url + .query_pairs() + .any(|(name, value)| name == "scope" && value == scope) + }) + .count(); + if request_count == expected_count { + return Ok::<(), anyhow::Error>(()); + } + if request_count > expected_count { + bail!( + "expected exactly {expected_count} /ps/plugins/list requests for scope {scope}, got {request_count}" + ); + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await??; + Ok(()) +} + async fn wait_for_remote_installed_scope_request(server: &MockServer, scope: &str) -> Result<()> { timeout(DEFAULT_TIMEOUT, async { loop { @@ -3828,6 +4446,20 @@ fn cached_remote_catalog_plugin_ids(codex_home: &std::path::Path) -> Result, +) -> Result<()> { + let cache_dir = codex_home.join("cache/remote_plugin_catalog"); + for entry in std::fs::read_dir(cache_dir)? { + let path = entry?.path(); + let mut cached_catalog: serde_json::Value = serde_json::from_slice(&std::fs::read(&path)?)?; + cached_catalog["fetched_at"] = serde_json::json!(fetched_at); + std::fs::write(path, serde_json::to_vec_pretty(&cached_catalog)?)?; + } + Ok(()) +} + async fn wait_for_path_exists(path: &std::path::Path) -> Result<()> { timeout(DEFAULT_TIMEOUT, async { loop { @@ -3914,6 +4546,22 @@ async fn mount_remote_plugin_list(server: &MockServer, scope: &str, body: &str) .await; } +async fn mount_delayed_remote_plugin_list(server: &MockServer, scope: &str, body: &str) { + Mock::given(method("GET")) + .and(path("/backend-api/ps/plugins/list")) + .and(query_param("scope", scope)) + .and(query_param("limit", "200")) + .and(header("authorization", "Bearer chatgpt-token")) + .and(header("chatgpt-account-id", "account-123")) + .respond_with( + ResponseTemplate::new(200) + .set_body_string(body) + .set_delay(Duration::from_millis(/*millis*/ 200)), + ) + .mount(server) + .await; +} + fn remote_plugin_list_body( remote_plugin_id: &str, plugin_name: &str, diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 8f9a470258..df8341284b 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -47,6 +47,7 @@ use crate::remote::RemoteInstalledPlugin; use crate::remote::RemoteInstalledPluginBundleSyncOutcome; use crate::remote::RemotePluginCatalogError; use crate::remote::RemotePluginMaterialization; +use crate::remote::RemotePluginScope; use crate::remote::RemotePluginServiceConfig; use crate::remote_legacy::RemotePluginFetchError; use crate::remote_legacy::RemotePluginMutationError; @@ -90,8 +91,10 @@ use codex_tools::DiscoverableTool; use codex_tools::filter_request_plugin_install_discoverable_tools_for_client; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_plugins::PluginSkillRoot; +use std::collections::BTreeSet; use std::collections::HashMap; use std::collections::HashSet; +use std::collections::VecDeque; use std::path::Path; use std::path::PathBuf; use std::sync::Arc; @@ -206,20 +209,40 @@ struct RemoteInstalledPluginsCacheRefreshState { in_flight: bool, } -struct GlobalRemoteCatalogCacheRefreshRequest { +struct RemoteCatalogCacheRefreshRequest { service_config: RemotePluginServiceConfig, auth: Option, + scopes: BTreeSet, + mode: RemoteCatalogCacheRefreshMode, +} + +impl RemoteCatalogCacheRefreshRequest { + fn has_same_cache_identity(&self, other: &Self) -> bool { + self.service_config == other.service_config + && self.auth.as_ref().and_then(CodexAuth::get_account_id) + == other.auth.as_ref().and_then(CodexAuth::get_account_id) + && self.auth.as_ref().and_then(CodexAuth::get_chatgpt_user_id) + == other.auth.as_ref().and_then(CodexAuth::get_chatgpt_user_id) + && self.auth.as_ref().map(CodexAuth::is_workspace_account) + == other.auth.as_ref().map(CodexAuth::is_workspace_account) + } +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum RemoteCatalogCacheRefreshMode { + OnlyIfStale, + Force, } #[derive(Default)] -struct GlobalRemoteCatalogCacheRefreshState { - requested: Option, +struct RemoteCatalogCacheRefreshState { + requests: VecDeque, in_flight: bool, } -#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +#[derive(Clone, Debug, Default, PartialEq, Eq)] pub struct PluginListBackgroundTaskOptions { - pub refresh_global_remote_catalog_cache: bool, + pub remote_catalog_cache_refresh_scopes: BTreeSet, } #[derive(Clone, PartialEq, Eq)] @@ -391,7 +414,7 @@ pub struct PluginsManager { tool_suggest_metadata_cache: ToolSuggestMetadataCache, remote_installed_plugins_cache: RwLock>>, remote_installed_plugins_cache_refresh_state: RwLock, - global_remote_catalog_cache_refresh_state: RwLock, + remote_catalog_cache_refresh_state: RwLock, restriction_product: Option, auth_mode: RwLock>, analytics_events_client: RwLock>, @@ -470,8 +493,8 @@ impl PluginsManager { remote_installed_plugins_cache_refresh_state: RwLock::new( RemoteInstalledPluginsCacheRefreshState::default(), ), - global_remote_catalog_cache_refresh_state: RwLock::new( - GlobalRemoteCatalogCacheRefreshState::default(), + remote_catalog_cache_refresh_state: RwLock::new( + RemoteCatalogCacheRefreshState::default(), ), restriction_product, auth_mode: RwLock::new(auth_mode), @@ -1047,18 +1070,22 @@ impl PluginsManager { ); } - fn maybe_start_global_remote_catalog_cache_refresh( + fn maybe_start_remote_catalog_cache_refresh( self: &Arc, config: &PluginsConfigInput, auth: Option, + scopes: BTreeSet, + mode: RemoteCatalogCacheRefreshMode, ) { - if !config.plugins_enabled || !config.remote_plugin_enabled { + if !config.plugins_enabled || scopes.is_empty() { return; } - self.schedule_global_remote_catalog_cache_refresh(GlobalRemoteCatalogCacheRefreshRequest { + self.schedule_remote_catalog_cache_refresh(RemoteCatalogCacheRefreshRequest { service_config: remote_plugin_service_config(config), auth, + scopes, + mode, }); } @@ -1071,9 +1098,12 @@ impl PluginsManager { on_effective_plugins_changed: Option, ) { self.maybe_start_non_curated_plugin_cache_refresh(config, roots); - if options.refresh_global_remote_catalog_cache { - self.maybe_start_global_remote_catalog_cache_refresh(config, auth.clone()); - } + self.maybe_start_remote_catalog_cache_refresh( + config, + auth.clone(), + options.remote_catalog_cache_refresh_scopes, + RemoteCatalogCacheRefreshMode::OnlyIfStale, + ); self.maybe_start_remote_plugin_caches_refresh( config, auth.clone(), @@ -2054,27 +2084,22 @@ impl PluginsManager { auth.clone(), on_effective_plugins_changed, ); + let mut scopes = crate::remote::cached_remote_plugin_catalog_scopes( + manager.codex_home.as_path(), + &remote_plugin_service_config(&config_for_remote_sync), + auth.as_ref(), + ); if config_for_remote_sync.remote_plugin_enabled { - match crate::remote::fetch_and_cache_global_remote_plugin_catalog( - manager.codex_home.as_path(), - &remote_plugin_service_config(&config_for_remote_sync), - auth.as_ref(), - ) - .await - { - Ok(()) => {} - Err( - RemotePluginCatalogError::AuthRequired - | RemotePluginCatalogError::UnsupportedAuthMode, - ) => {} - Err(err) => { - warn!( - error = %err, - "failed to warm remote plugin catalog cache" - ); - } - } + scopes.insert(RemotePluginScope::Global); + } else { + scopes.retain(|scope| *scope == RemotePluginScope::Workspace); } + manager.maybe_start_remote_catalog_cache_refresh( + &config_for_remote_sync, + auth, + scopes, + RemoteCatalogCacheRefreshMode::Force, + ); }); let config_for_featured_plugins = config.clone(); @@ -2232,16 +2257,35 @@ impl PluginsManager { }); } - fn schedule_global_remote_catalog_cache_refresh( + fn schedule_remote_catalog_cache_refresh( self: &Arc, - request: GlobalRemoteCatalogCacheRefreshRequest, + request: RemoteCatalogCacheRefreshRequest, ) { let should_spawn = { - let mut state = match self.global_remote_catalog_cache_refresh_state.write() { + let mut state = match self.remote_catalog_cache_refresh_state.write() { Ok(state) => state, Err(err) => err.into_inner(), }; - state.requested = Some(request); + if let Some(pending) = state + .requests + .iter_mut() + .find(|pending| pending.has_same_cache_identity(&request)) + { + pending.scopes.extend(request.scopes); + pending.auth = request.auth; + pending.mode = match (pending.mode, request.mode) { + (RemoteCatalogCacheRefreshMode::Force, _) + | (_, RemoteCatalogCacheRefreshMode::Force) => { + RemoteCatalogCacheRefreshMode::Force + } + ( + RemoteCatalogCacheRefreshMode::OnlyIfStale, + RemoteCatalogCacheRefreshMode::OnlyIfStale, + ) => RemoteCatalogCacheRefreshMode::OnlyIfStale, + }; + } else { + state.requests.push_back(request); + } if state.in_flight { false } else { @@ -2255,7 +2299,7 @@ impl PluginsManager { let manager = Arc::clone(self); tokio::spawn(async move { - manager.run_global_remote_catalog_cache_refresh_loop().await; + manager.run_remote_catalog_cache_refresh_loop().await; }); } @@ -2472,14 +2516,14 @@ impl PluginsManager { } } - async fn run_global_remote_catalog_cache_refresh_loop(self: Arc) { + async fn run_remote_catalog_cache_refresh_loop(self: Arc) { loop { let request = { - let mut state = match self.global_remote_catalog_cache_refresh_state.write() { + let mut state = match self.remote_catalog_cache_refresh_state.write() { Ok(state) => state, Err(err) => err.into_inner(), }; - match state.requested.take() { + match state.requests.pop_front() { Some(request) => request, None => { state.in_flight = false; @@ -2488,23 +2532,38 @@ impl PluginsManager { } }; - 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" - ); + for scope in request.scopes { + if request.mode == RemoteCatalogCacheRefreshMode::OnlyIfStale + && crate::remote::has_fresh_cached_remote_plugin_catalog( + self.codex_home.as_path(), + &request.service_config, + request.auth.as_ref(), + scope, + ) + { + continue; + } + + match crate::remote::fetch_and_cache_remote_plugin_catalog( + self.codex_home.as_path(), + &request.service_config, + request.auth.as_ref(), + scope, + ) + .await + { + Ok(()) => {} + Err( + RemotePluginCatalogError::AuthRequired + | RemotePluginCatalogError::UnsupportedAuthMode, + ) => {} + Err(err) => { + warn!( + error = %err, + scope = ?scope, + "failed to refresh cached remote plugin catalog" + ); + } } } } diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index f5d3fc7663..da5816510f 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -189,6 +189,18 @@ pub enum RemoteMarketplaceSource { SharedWithMe, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RemotePluginCatalogCacheMode { + PreferCache, + ForceRefetch, +} + +#[derive(Debug, Clone, PartialEq)] +pub struct RemoteMarketplacesFetchOutcome { + pub marketplaces: Vec, + pub catalog_cache_refresh_scopes: BTreeSet, +} + #[derive(Debug, Clone, PartialEq)] pub struct RemoteInstalledPlugin { pub marketplace_name: String, @@ -443,6 +455,8 @@ pub enum RemotePluginScope { } impl RemotePluginScope { + const CATALOG_CACHE_SCOPES: [Self; 3] = [Self::Global, Self::User, Self::Workspace]; + fn api_value(self) -> &'static str { match self { Self::Global => "GLOBAL", @@ -722,10 +736,12 @@ pub async fn fetch_remote_marketplaces( config: &RemotePluginServiceConfig, auth: Option<&CodexAuth>, sources: &[RemoteMarketplaceSource], - global_catalog_cache_path: Option<&Path>, -) -> Result, RemotePluginCatalogError> { + catalog_cache_root: Option<&Path>, + catalog_cache_mode: RemotePluginCatalogCacheMode, +) -> Result { let auth = ensure_chatgpt_auth(auth)?; let mut marketplaces = Vec::new(); + let mut catalog_cache_refresh_scopes = BTreeSet::new(); let needs_workspace_installed = sources.iter().any(|source| { matches!( source, @@ -742,61 +758,48 @@ 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_directory_plugins_for_scope_with_cache( + catalog_cache_root, + config, + auth, + scope, + catalog_cache_mode, + ), fetch_installed_plugins_for_scope(config, auth, scope), )?; - let directory_plugins_for_cache = - global_catalog_cache_path.map(|_| directory_plugins.clone()); + if directory_plugins.cache_refresh_needed { + catalog_cache_refresh_scopes.insert(scope); + } if let Some(marketplace) = build_remote_marketplace( scope.marketplace_name(), scope.marketplace_display_name(), - directory_plugins, + directory_plugins.plugins, installed_plugins, /*include_installed_only*/ true, )? { marketplaces.push(marketplace); } - if let (Some(codex_home), Some(directory_plugins)) = - (global_catalog_cache_path, directory_plugins_for_cache) - { - catalog_cache::write_cached_global_directory_plugins( - codex_home, - config, - auth, - &directory_plugins, - ); - } } RemoteMarketplaceSource::CreatedByMeRemote => { let scope = RemotePluginScope::User; let (directory_plugins, installed_plugins) = tokio::try_join!( - fetch_directory_plugins_for_scope(config, auth, scope), + fetch_directory_plugins_for_scope_with_cache( + catalog_cache_root, + config, + auth, + scope, + catalog_cache_mode, + ), fetch_installed_plugins_for_scope(config, auth, scope), )?; + if directory_plugins.cache_refresh_needed { + catalog_cache_refresh_scopes.insert(scope); + } if let Some(marketplace) = build_remote_marketplace( scope.marketplace_name(), scope.marketplace_display_name(), - directory_plugins, + directory_plugins.plugins, installed_plugins, /*include_installed_only*/ false, )? { @@ -805,12 +808,21 @@ pub async fn fetch_remote_marketplaces( } RemoteMarketplaceSource::WorkspaceDirectory => { let scope = RemotePluginScope::Workspace; - let directory_plugins = - fetch_directory_plugins_for_scope(config, auth, scope).await?; + let directory_plugins = fetch_directory_plugins_for_scope_with_cache( + catalog_cache_root, + config, + auth, + scope, + catalog_cache_mode, + ) + .await?; + if directory_plugins.cache_refresh_needed { + catalog_cache_refresh_scopes.insert(scope); + } if let Some(marketplace) = build_remote_marketplace( scope.marketplace_name(), scope.marketplace_display_name(), - directory_plugins, + directory_plugins.plugins, workspace_installed_plugins.clone().unwrap_or_default(), /*include_installed_only*/ false, )? { @@ -878,7 +890,22 @@ pub async fn fetch_remote_marketplaces( } } - Ok(marketplaces) + Ok(RemoteMarketplacesFetchOutcome { + marketplaces, + catalog_cache_refresh_scopes, + }) +} + +pub(crate) async fn fetch_and_cache_remote_plugin_catalog( + codex_home: &Path, + config: &RemotePluginServiceConfig, + auth: Option<&CodexAuth>, + scope: RemotePluginScope, +) -> Result<(), RemotePluginCatalogError> { + let auth = ensure_chatgpt_auth(auth)?; + let plugins = fetch_directory_plugins_for_scope(config, auth, scope).await?; + catalog_cache::write_cached_directory_plugins(codex_home, config, auth, scope, &plugins); + Ok(()) } pub async fn fetch_and_cache_global_remote_plugin_catalog( @@ -886,11 +913,21 @@ pub async fn fetch_and_cache_global_remote_plugin_catalog( config: &RemotePluginServiceConfig, auth: Option<&CodexAuth>, ) -> Result<(), RemotePluginCatalogError> { - let auth = ensure_chatgpt_auth(auth)?; - let plugins = - fetch_directory_plugins_for_scope(config, auth, RemotePluginScope::Global).await?; - catalog_cache::write_cached_global_directory_plugins(codex_home, config, auth, &plugins); - Ok(()) + fetch_and_cache_remote_plugin_catalog(codex_home, config, auth, RemotePluginScope::Global).await +} + +pub fn invalidate_cached_remote_plugin_catalog_scopes( + codex_home: &Path, + config: &RemotePluginServiceConfig, + auth: Option<&CodexAuth>, + scopes: &[RemotePluginScope], +) { + let Ok(auth) = ensure_chatgpt_auth(auth) else { + return; + }; + for scope in scopes { + catalog_cache::remove_cached_directory_plugins(codex_home, config, auth, *scope); + } } #[instrument(level = "trace", skip_all)] @@ -975,15 +1012,39 @@ fn recommended_plugins_mode(response: RecommendedPluginsResponse) -> Recommended } } -pub fn has_cached_global_remote_plugin_catalog( +pub(crate) fn has_fresh_cached_remote_plugin_catalog( codex_home: &Path, config: &RemotePluginServiceConfig, auth: Option<&CodexAuth>, + scope: RemotePluginScope, ) -> bool { let Ok(auth) = ensure_chatgpt_auth(auth) else { return false; }; - catalog_cache::load_cached_global_directory_plugins(codex_home, config, auth).is_some() + catalog_cache::load_cached_directory_plugins(codex_home, config, auth, scope).is_some_and( + |cached| { + matches!( + cached.freshness, + catalog_cache::RemotePluginCatalogCacheFreshness::Fresh + ) + }, + ) +} + +pub(crate) fn cached_remote_plugin_catalog_scopes( + codex_home: &Path, + config: &RemotePluginServiceConfig, + auth: Option<&CodexAuth>, +) -> BTreeSet { + let Ok(auth) = ensure_chatgpt_auth(auth) else { + return BTreeSet::new(); + }; + RemotePluginScope::CATALOG_CACHE_SCOPES + .into_iter() + .filter(|scope| { + catalog_cache::load_cached_directory_plugins(codex_home, config, auth, *scope).is_some() + }) + .collect() } pub fn cached_global_remote_discoverable_plugins( @@ -991,17 +1052,25 @@ pub fn cached_global_remote_discoverable_plugins( config: &RemotePluginServiceConfig, auth: &CodexAuth, ) -> Vec { - catalog_cache::load_cached_global_directory_plugins(codex_home, config, auth) - .unwrap_or_default() - .into_iter() - .filter_map(|plugin| match remote_discoverable_plugin_from_directory_item(&plugin) { + catalog_cache::load_cached_directory_plugins( + codex_home, + config, + auth, + RemotePluginScope::Global, + ) + .map(|cached| cached.plugins) + .unwrap_or_default() + .into_iter() + .filter_map( + |plugin| match remote_discoverable_plugin_from_directory_item(&plugin) { Ok(plugin) => Some(plugin), Err(err) => { tracing::warn!(error = %err, "ignoring cached remote plugin recommendation entry"); None } - }) - .collect() + }, + ) + .collect() } pub async fn fetch_openai_curated_remote_collection_marketplace( @@ -1768,6 +1837,42 @@ fn normalize_remote_default_prompt(prompt: &str) -> Option { Some(prompt.to_string()) } +struct DirectoryPluginsFetchOutcome { + plugins: Vec, + cache_refresh_needed: bool, +} + +async fn fetch_directory_plugins_for_scope_with_cache( + codex_home: Option<&Path>, + config: &RemotePluginServiceConfig, + auth: &CodexAuth, + scope: RemotePluginScope, + cache_mode: RemotePluginCatalogCacheMode, +) -> Result { + if cache_mode == RemotePluginCatalogCacheMode::PreferCache + && let Some(codex_home) = codex_home + && let Some(cached) = + catalog_cache::load_cached_directory_plugins(codex_home, config, auth, scope) + { + return Ok(DirectoryPluginsFetchOutcome { + plugins: cached.plugins, + cache_refresh_needed: matches!( + cached.freshness, + catalog_cache::RemotePluginCatalogCacheFreshness::Stale + ), + }); + } + + let plugins = fetch_directory_plugins_for_scope(config, auth, scope).await?; + if let Some(codex_home) = codex_home { + catalog_cache::write_cached_directory_plugins(codex_home, config, auth, scope, &plugins); + } + Ok(DirectoryPluginsFetchOutcome { + plugins, + cache_refresh_needed: false, + }) +} + async fn fetch_directory_plugins_for_scope( config: &RemotePluginServiceConfig, auth: &CodexAuth, @@ -1800,6 +1905,15 @@ async fn fetch_directory_plugins_for_scope_with_optional_collection( scope: RemotePluginScope, collection: Option<&str>, ) -> Result, RemotePluginCatalogError> { + tracing::info!( + operation = "plugins.remote_catalog.list", + http.method = "GET", + api.path = "ps/plugins/list", + plugin.scope = scope.api_value(), + plugin.collection = collection.unwrap_or_default(), + "fetching remote plugin catalog" + ); + let mut plugins = Vec::new(); let mut page_token = None; loop { diff --git a/codex-rs/core-plugins/src/remote/catalog_cache.rs b/codex-rs/core-plugins/src/remote/catalog_cache.rs index d49885dfbb..50999919c7 100644 --- a/codex-rs/core-plugins/src/remote/catalog_cache.rs +++ b/codex-rs/core-plugins/src/remote/catalog_cache.rs @@ -1,14 +1,21 @@ use super::RemotePluginDirectoryItem; +use super::RemotePluginScope; use super::RemotePluginServiceConfig; +use chrono::DateTime; +use chrono::Utc; use codex_login::CodexAuth; use serde::Deserialize; use serde::Serialize; use std::path::Path; use std::path::PathBuf; +use std::time::Duration; use tracing::warn; +// `plugin/list` and other callers that assume all plugins are available locally will be +// deprecated soon. Remove this catalog cache when those callers migrate to on-demand fetching. const REMOTE_PLUGIN_CATALOG_DISK_CACHE_SCHEMA_VERSION: u8 = 1; const REMOTE_PLUGIN_CATALOG_DISK_CACHE_DIR: &str = "cache/remote_plugin_catalog"; +const REMOTE_PLUGIN_CATALOG_DISK_CACHE_TTL: Duration = Duration::from_secs(60 * 60 * 3); #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] struct RemotePluginCatalogCacheKey { @@ -16,34 +23,65 @@ struct RemotePluginCatalogCacheKey { account_id: Option, chatgpt_user_id: Option, is_workspace_account: bool, + // Global catalogs predate scoped cache keys and must keep their existing filenames. + #[serde(skip_serializing_if = "Option::is_none")] + scope: Option, } impl RemotePluginCatalogCacheKey { - fn global(config: &RemotePluginServiceConfig, auth: &CodexAuth) -> Self { - Self { + fn new( + config: &RemotePluginServiceConfig, + auth: &CodexAuth, + scope: RemotePluginScope, + ) -> Option { + let cache_key = Self { chatgpt_base_url: config.chatgpt_base_url.clone(), account_id: auth.get_account_id(), chatgpt_user_id: auth.get_chatgpt_user_id(), is_workspace_account: auth.is_workspace_account(), + scope: (scope != RemotePluginScope::Global).then_some(scope), + }; + // Preserve global catalog caching for existing header-auth clients, but never share + // user or workspace catalogs when the auth mode cannot identify their owner. + if !matches!(scope, RemotePluginScope::Global) + && cache_key.account_id.is_none() + && cache_key.chatgpt_user_id.is_none() + { + return None; } + + Some(cache_key) } } #[derive(Debug, Clone, Serialize, Deserialize)] struct RemotePluginCatalogDiskCache { schema_version: u8, + #[serde(default, skip_serializing_if = "Option::is_none")] + fetched_at: Option>, plugins: Vec, } -pub(crate) fn load_cached_global_directory_plugins( +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(super) enum RemotePluginCatalogCacheFreshness { + Fresh, + Stale, +} + +#[derive(Debug, Clone)] +pub(super) struct CachedDirectoryPlugins { + pub plugins: Vec, + pub freshness: RemotePluginCatalogCacheFreshness, +} + +pub(crate) fn load_cached_directory_plugins( codex_home: &Path, config: &RemotePluginServiceConfig, auth: &CodexAuth, -) -> Option> { - let cache_path = cache_path( - codex_home, - &RemotePluginCatalogCacheKey::global(config, auth), - ); + scope: RemotePluginScope, +) -> Option { + let cache_key = RemotePluginCatalogCacheKey::new(config, auth, scope)?; + let cache_path = cache_path(codex_home, &cache_key); let bytes = match std::fs::read(&cache_path) { Ok(bytes) => bytes, Err(err) if err.kind() == std::io::ErrorKind::NotFound => return None, @@ -71,31 +109,58 @@ pub(crate) fn load_cached_global_directory_plugins( return None; } - Some(cache.plugins) + let freshness = if is_fresh(cache.fetched_at, Utc::now()) { + RemotePluginCatalogCacheFreshness::Fresh + } else { + RemotePluginCatalogCacheFreshness::Stale + }; + Some(CachedDirectoryPlugins { + plugins: cache.plugins, + freshness, + }) } -pub(crate) fn write_cached_global_directory_plugins( +pub(crate) fn write_cached_directory_plugins( codex_home: &Path, config: &RemotePluginServiceConfig, auth: &CodexAuth, + scope: RemotePluginScope, plugins: &[RemotePluginDirectoryItem], ) { - let cache_path = cache_path( - codex_home, - &RemotePluginCatalogCacheKey::global(config, auth), - ); - if let Some(parent) = cache_path.parent() - && std::fs::create_dir_all(parent).is_err() - { + let Some(cache_key) = RemotePluginCatalogCacheKey::new(config, auth, scope) else { return; - } - let Ok(bytes) = serde_json::to_vec_pretty(&RemotePluginCatalogDiskCache { + }; + let cache_path = cache_path(codex_home, &cache_key); + let Ok(contents) = serde_json::to_string_pretty(&RemotePluginCatalogDiskCache { schema_version: REMOTE_PLUGIN_CATALOG_DISK_CACHE_SCHEMA_VERSION, + fetched_at: Some(Utc::now()), plugins: plugins.to_vec(), }) else { return; }; - let _ = std::fs::write(cache_path, bytes); + let _ = codex_utils_path::write_atomically(&cache_path, &contents); +} + +pub(crate) fn remove_cached_directory_plugins( + codex_home: &Path, + config: &RemotePluginServiceConfig, + auth: &CodexAuth, + scope: RemotePluginScope, +) { + let Some(cache_key) = RemotePluginCatalogCacheKey::new(config, auth, scope) else { + return; + }; + let cache_path = cache_path(codex_home, &cache_key); + match std::fs::remove_file(&cache_path) { + Ok(()) => {} + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => { + warn!( + cache_path = %cache_path.display(), + "failed to remove remote plugin catalog disk cache: {err}" + ); + } + } } fn cache_path(codex_home: &Path, cache_key: &RemotePluginCatalogCacheKey) -> PathBuf { @@ -109,3 +174,18 @@ fn cache_path(codex_home: &Path, cache_key: &RemotePluginCatalogCacheKey) -> Pat .join(REMOTE_PLUGIN_CATALOG_DISK_CACHE_DIR) .join(format!("{cache_key_hash:016x}.json")) } + +fn is_fresh(fetched_at: Option>, now: DateTime) -> bool { + let Some(fetched_at) = fetched_at else { + return false; + }; + let Ok(ttl) = chrono::Duration::from_std(REMOTE_PLUGIN_CATALOG_DISK_CACHE_TTL) else { + return false; + }; + let age = now.signed_duration_since(fetched_at); + age >= chrono::Duration::zero() && age <= ttl +} + +#[cfg(test)] +#[path = "catalog_cache_tests.rs"] +mod tests; diff --git a/codex-rs/core-plugins/src/remote/catalog_cache_tests.rs b/codex-rs/core-plugins/src/remote/catalog_cache_tests.rs new file mode 100644 index 0000000000..7039398660 --- /dev/null +++ b/codex-rs/core-plugins/src/remote/catalog_cache_tests.rs @@ -0,0 +1,127 @@ +use super::*; +use chrono::TimeDelta; +use codex_login::AuthHeaders; +use pretty_assertions::assert_eq; + +#[test] +fn catalog_cache_freshness_honors_ttl() { + let now = Utc::now(); + + assert!(!is_fresh(/*fetched_at*/ None, now)); + assert!(is_fresh(Some(now), now)); + assert!(is_fresh(Some(now - TimeDelta::hours(3)), now,)); + assert!(!is_fresh( + Some(now - TimeDelta::hours(3) - TimeDelta::milliseconds(1)), + now, + )); + assert!(!is_fresh(Some(now + TimeDelta::milliseconds(1)), now)); +} + +#[test] +fn catalog_cache_paths_are_isolated_by_scope() { + let codex_home = Path::new("/tmp/codex-home"); + let cache_key_for_scope = |scope| RemotePluginCatalogCacheKey { + chatgpt_base_url: "https://chatgpt.com/backend-api".to_string(), + account_id: Some("account-id".to_string()), + chatgpt_user_id: Some("user-id".to_string()), + is_workspace_account: true, + scope: (scope != RemotePluginScope::Global).then_some(scope), + }; + + let paths = [ + RemotePluginScope::Global, + RemotePluginScope::User, + RemotePluginScope::Workspace, + ] + .map(|scope| cache_path(codex_home, &cache_key_for_scope(scope))); + + assert_ne!(paths[0], paths[1]); + assert_ne!(paths[0], paths[2]); + assert_ne!(paths[1], paths[2]); +} + +#[test] +fn global_catalog_cache_reuses_legacy_cache_file() { + let codex_home = tempfile::tempdir().expect("create codex home"); + let config = RemotePluginServiceConfig::new( + "https://chatgpt.com/backend-api".to_string(), + crate::test_support::test_http_client_factory(), + ); + let auth = CodexAuth::Headers(AuthHeaders::new(http::HeaderMap::new())); + let legacy_cache_path = codex_home + .path() + .join(REMOTE_PLUGIN_CATALOG_DISK_CACHE_DIR) + .join("f22564d6f8ca89f6.json"); + let legacy_cache = serde_json::json!({ + "schema_version": REMOTE_PLUGIN_CATALOG_DISK_CACHE_SCHEMA_VERSION, + "plugins": [], + }); + let contents = serde_json::to_string_pretty(&legacy_cache).expect("serialize legacy cache"); + codex_utils_path::write_atomically(&legacy_cache_path, &contents).expect("write legacy cache"); + + let cached = + load_cached_directory_plugins(codex_home.path(), &config, &auth, RemotePluginScope::Global) + .expect("load legacy global cache"); + assert!(cached.plugins.is_empty()); + assert_eq!(cached.freshness, RemotePluginCatalogCacheFreshness::Stale); + + write_cached_directory_plugins( + codex_home.path(), + &config, + &auth, + RemotePluginScope::Global, + &[], + ); + let refreshed_cache: RemotePluginCatalogDiskCache = serde_json::from_slice( + &std::fs::read(&legacy_cache_path).expect("read refreshed legacy cache"), + ) + .expect("parse refreshed legacy cache"); + assert!(refreshed_cache.fetched_at.is_some()); +} + +#[test] +fn header_auth_does_not_cache_private_catalogs_without_a_stable_identity() { + let codex_home = tempfile::tempdir().expect("create codex home"); + let config = RemotePluginServiceConfig::new( + "https://chatgpt.com/backend-api".to_string(), + crate::test_support::test_http_client_factory(), + ); + let auth = CodexAuth::Headers(AuthHeaders::new(http::HeaderMap::new())); + + write_cached_directory_plugins( + codex_home.path(), + &config, + &auth, + RemotePluginScope::Global, + &[], + ); + assert!( + load_cached_directory_plugins(codex_home.path(), &config, &auth, RemotePluginScope::Global) + .is_some() + ); + + for scope in [RemotePluginScope::User, RemotePluginScope::Workspace] { + let insecure_cache_key = RemotePluginCatalogCacheKey { + chatgpt_base_url: config.chatgpt_base_url.clone(), + account_id: None, + chatgpt_user_id: None, + is_workspace_account: false, + scope: Some(scope), + }; + let insecure_cache_path = cache_path(codex_home.path(), &insecure_cache_key); + + write_cached_directory_plugins(codex_home.path(), &config, &auth, scope, &[]); + assert!(!insecure_cache_path.exists()); + + let insecure_cache = RemotePluginCatalogDiskCache { + schema_version: REMOTE_PLUGIN_CATALOG_DISK_CACHE_SCHEMA_VERSION, + fetched_at: Some(Utc::now()), + plugins: Vec::new(), + }; + let contents = serde_json::to_string_pretty(&insecure_cache).expect("serialize cache"); + codex_utils_path::write_atomically(&insecure_cache_path, &contents) + .expect("write insecure cache"); + + assert!(load_cached_directory_plugins(codex_home.path(), &config, &auth, scope).is_none()); + } +} diff --git a/codex-rs/core-plugins/src/remote_tests.rs b/codex-rs/core-plugins/src/remote_tests.rs index 839e683342..4001ec8ca7 100644 --- a/codex-rs/core-plugins/src/remote_tests.rs +++ b/codex-rs/core-plugins/src/remote_tests.rs @@ -46,6 +46,30 @@ async fn remote_plugin_list_routes_the_complete_query_url() { ); } +#[test] +fn cached_remote_plugin_catalog_scopes_returns_existing_scopes() { + let codex_home = tempfile::tempdir().expect("create codex home"); + let config = RemotePluginServiceConfig::new( + "https://chatgpt.com/backend-api".to_string(), + crate::test_support::test_http_client_factory(), + ); + let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); + for scope in [RemotePluginScope::Global, RemotePluginScope::Workspace] { + catalog_cache::write_cached_directory_plugins( + codex_home.path(), + &config, + &auth, + scope, + &[], + ); + } + + assert_eq!( + cached_remote_plugin_catalog_scopes(codex_home.path(), &config, Some(&auth)), + BTreeSet::from([RemotePluginScope::Global, RemotePluginScope::Workspace]) + ); +} + #[test] fn build_remote_marketplace_preserves_directory_order_and_appends_installed_only_plugins() { let directory_plugins = vec![ diff --git a/codex-rs/state/src/migrations_tests.rs b/codex-rs/state/src/migrations_tests.rs index 3cd5c653f6..8ec2102d22 100644 --- a/codex-rs/state/src/migrations_tests.rs +++ b/codex-rs/state/src/migrations_tests.rs @@ -36,8 +36,9 @@ async fn pinned_threads_migration_defaults_existing_and_legacy_rows_to_unpinned( let _ = std::fs::remove_dir_all(sqlite_home); }); let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs()); + let state_path = sqlite.state_db_path(); let pool = sqlite - .open_read_write_pool(&state_db_path(&sqlite_home)) + .open_read_write_pool(&state_path) .await .expect("sqlite database should open"); migrator_through(/*version*/ 42)