github/codex
4
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 16:17:49 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cache remote plugin catalogs by scope (#34849)

Browse Source 
## 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
This commit is contained in:
brisebois-oai
2026-07-23 00:52:55 +00:00
committed by copyberry
parent 9d82334302
commit 83ff1c2f80
9 changed files with 1223 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/README.md
+1 -1
View File
@@ -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.
codex-rs/app-server/src/request_processors/plugins.rs
+35 -11
View File
@@ -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 {})
}
codex-rs/app-server/tests/suite/v2/plugin_list.rs
+649 -1
View File
@@ -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::<Vec<_>>();
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<Vec<
Ok(plugin_ids)
}
fn rewrite_cached_remote_catalog_fetched_at(
codex_home: &std::path::Path,
fetched_at: chrono::DateTime<Utc>,
) -> 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,
codex-rs/core-plugins/src/manager.rs
+117 -58
View File
@@ -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<CodexAuth>,
scopes: BTreeSet<RemotePluginScope>,
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<GlobalRemoteCatalogCacheRefreshRequest>,
struct RemoteCatalogCacheRefreshState {
requests: VecDeque<RemoteCatalogCacheRefreshRequest>,
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<RemotePluginScope>,
}
#[derive(Clone, PartialEq, Eq)]
@@ -391,7 +414,7 @@ pub struct PluginsManager {
tool_suggest_metadata_cache: ToolSuggestMetadataCache,
remote_installed_plugins_cache: RwLock<Option<Vec<RemoteInstalledPlugin>>>,
remote_installed_plugins_cache_refresh_state: RwLock<RemoteInstalledPluginsCacheRefreshState>,
global_remote_catalog_cache_refresh_state: RwLock<GlobalRemoteCatalogCacheRefreshState>,
remote_catalog_cache_refresh_state: RwLock<RemoteCatalogCacheRefreshState>,
restriction_product: Option<Product>,
auth_mode: RwLock<Option<AuthMode>>,
analytics_events_client: RwLock<Option<AnalyticsEventsClient>>,
@@ -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<Self>,
config: &PluginsConfigInput,
auth: Option<CodexAuth>,
scopes: BTreeSet<RemotePluginScope>,
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<EffectivePluginsChangedCallback>,
) {
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<Self>,
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<Self>) {
async fn run_remote_catalog_cache_refresh_loop(self: Arc<Self>) {
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"
);
}
}
}
}
codex-rs/core-plugins/src/remote.rs
+168 -54
View File
@@ -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<RemoteMarketplace>,
pub catalog_cache_refresh_scopes: BTreeSet<RemotePluginScope>,
}
#[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<Vec<RemoteMarketplace>, RemotePluginCatalogError> {
catalog_cache_root: Option<&Path>,
catalog_cache_mode: RemotePluginCatalogCacheMode,
) -> Result<RemoteMarketplacesFetchOutcome, RemotePluginCatalogError> {
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<RemotePluginScope> {
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<RemoteDiscoverablePlugin> {
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<String> {
Some(prompt.to_string())
}
struct DirectoryPluginsFetchOutcome {
plugins: Vec<RemotePluginDirectoryItem>,
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<DirectoryPluginsFetchOutcome, RemotePluginCatalogError> {
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<Vec<RemotePluginDirectoryItem>, 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 {
codex-rs/core-plugins/src/remote/catalog_cache.rs
+100 -20
View File
@@ -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<String>,
chatgpt_user_id: Option<String>,
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<RemotePluginScope>,
}
impl RemotePluginCatalogCacheKey {
fn global(config: &RemotePluginServiceConfig, auth: &CodexAuth) -> Self {
Self {
fn new(
config: &RemotePluginServiceConfig,
auth: &CodexAuth,
scope: RemotePluginScope,
) -> Option<Self> {
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<DateTime<Utc>>,
plugins: Vec<RemotePluginDirectoryItem>,
}
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<RemotePluginDirectoryItem>,
pub freshness: RemotePluginCatalogCacheFreshness,
}
pub(crate) fn load_cached_directory_plugins(
codex_home: &Path,
config: &RemotePluginServiceConfig,
auth: &CodexAuth,
) -> Option<Vec<RemotePluginDirectoryItem>> {
let cache_path = cache_path(
codex_home,
&RemotePluginCatalogCacheKey::global(config, auth),
);
scope: RemotePluginScope,
) -> Option<CachedDirectoryPlugins> {
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<DateTime<Utc>>, now: DateTime<Utc>) -> 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;
codex-rs/core-plugins/src/remote/catalog_cache_tests.rs
+127
View File
@@ -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());
}
}
codex-rs/core-plugins/src/remote_tests.rs
+24
View File
@@ -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![
codex-rs/state/src/migrations_tests.rs
+2 -1
View File
@@ -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)