openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-25 18:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(core-plugins): fetch installed plugins once

Browse Source
This commit is contained in:
Eric Ning
2026-06-07 22:42:12 -07:00
parent 5a440c03f2
commit acdaa0272b
5 changed files with 489 additions and 284 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/plugins.rs
+6 -18
View File
@@ -617,14 +617,9 @@ impl PluginRequestProcessor {
// TODO(remote plugins): Remove this once remote plugins are ready and vertical plugins are
// served directly from the normal remote catalog.
if include_vertical && !config.features.enabled(Feature::RemotePlugin) {
let remote_plugin_service_config = RemotePluginServiceConfig {
chatgpt_base_url: config.chatgpt_base_url.clone(),
};
match codex_core_plugins::remote::fetch_openai_curated_remote_collection_marketplace(
&remote_plugin_service_config,
auth.as_ref(),
)
.await
match plugins_manager
.fetch_openai_curated_remote_collection_marketplace(&plugins_input, auth.as_ref())
.await
{
Ok(Some(remote_marketplace)) => {
data.push(remote_marketplace_to_info(remote_marketplace));
@@ -656,16 +651,9 @@ impl PluginRequestProcessor {
remote_sources.push(RemoteMarketplaceSource::SharedWithMe);
}
if !remote_sources.is_empty() {
let remote_plugin_service_config = RemotePluginServiceConfig {
chatgpt_base_url: config.chatgpt_base_url.clone(),
};
match codex_core_plugins::remote::fetch_remote_marketplaces(
&remote_plugin_service_config,
auth.as_ref(),
&remote_sources,
/*global_catalog_cache_path*/ Some(config.codex_home.as_path()),
)
.await
match plugins_manager
.fetch_remote_marketplaces(&plugins_input, auth.as_ref(), &remote_sources)
.await
{
Ok(remote_marketplaces) => {
for remote_marketplace in remote_marketplaces
codex-rs/app-server/tests/suite/v2/plugin_list.rs
+62 -77
View File
@@ -35,6 +35,7 @@ use wiremock::matchers::header;
use wiremock::matchers::method;
use wiremock::matchers::path;
use wiremock::matchers::query_param;
use wiremock::matchers::query_param_is_missing;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(30);
const TEST_CURATED_PLUGIN_SHA: &str = "0123456789abcdef0123456789abcdef01234567";
@@ -225,10 +226,16 @@ enabled = true
.as_array_mut()
.expect("installed plugins should be an array")
.push(remote_only);
let mut personal = global_installed_body["plugins"][0].clone();
personal["id"] = serde_json::json!("plugins~Plugin_22222222222222222222222222222222");
personal["name"] = serde_json::json!("personal-only");
personal["scope"] = serde_json::json!("USER");
global_installed_body["plugins"]
.as_array_mut()
.expect("installed plugins should be an array")
.push(personal);
let global_installed_body = serde_json::to_string(&global_installed_body)?;
mount_remote_installed_plugins(&server, "GLOBAL", &global_installed_body).await;
mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body())
.await;
mount_aggregate_remote_installed_plugins(&server, &global_installed_body).await;
let mut app_server = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, app_server.initialize()).await??;
@@ -1495,9 +1502,7 @@ async fn app_server_startup_sync_downloads_remote_installed_plugin_bundles() ->
/*enabled*/ true,
remote_app_manifest.clone(),
);
mount_remote_installed_plugins(&server, "GLOBAL", &global_installed_body).await;
mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body())
.await;
mount_aggregate_remote_installed_plugins(&server, &global_installed_body).await;
let installed_path = codex_home
.path()
@@ -1566,9 +1571,7 @@ async fn plugin_list_sync_upgrades_and_removes_remote_installed_plugin_bundles()
);
mount_remote_plugin_list(&server, "GLOBAL", &global_installed_body).await;
mount_remote_plugin_list(&server, "WORKSPACE", empty_remote_installed_plugins_body()).await;
mount_remote_installed_plugins(&server, "GLOBAL", &global_installed_body).await;
mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body())
.await;
mount_aggregate_remote_installed_plugins(&server, &global_installed_body).await;
let old_path = codex_home
.path()
@@ -1739,20 +1742,12 @@ async fn plugin_list_includes_remote_marketplaces_when_remote_plugin_enabled() -
.await;
Mock::given(method("GET"))
.and(path("/backend-api/ps/plugins/installed"))
.and(query_param("scope", "GLOBAL"))
.and(query_param_is_missing("scope"))
.and(header("authorization", "Bearer chatgpt-token"))
.and(header("chatgpt-account-id", "account-123"))
.and(header("oai-product-sku", "codex"))
.respond_with(ResponseTemplate::new(200).set_body_string(global_installed_body))
.mount(&server)
.await;
Mock::given(method("GET"))
.and(path("/backend-api/ps/plugins/installed"))
.and(query_param("scope", "WORKSPACE"))
.and(header("authorization", "Bearer chatgpt-token"))
.and(header("chatgpt-account-id", "account-123"))
.and(header("oai-product-sku", "codex"))
.respond_with(ResponseTemplate::new(200).set_body_string(empty_page_body))
.expect(1)
.mount(&server)
.await;
@@ -1907,9 +1902,7 @@ async fn plugin_list_includes_openai_curated_remote_collection_when_requested()
}
}"#;
mount_openai_curated_remote_collection_plugin_list(&server, collection_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_aggregate_remote_installed_plugins(&server, empty_remote_installed_plugins_body()).await;
let mut mcp = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
@@ -1994,9 +1987,7 @@ async fn plugin_list_fail_opens_openai_curated_remote_collection_errors() -> Res
.respond_with(ResponseTemplate::new(500).set_body_string("temporary failure"))
.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_aggregate_remote_installed_plugins(&server, empty_remote_installed_plugins_body()).await;
let mut mcp = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
@@ -2221,8 +2212,11 @@ plugin_sharing = true
.push(unlisted_installed_body["plugins"][0].clone());
let workspace_installed_body = serde_json::to_string(&workspace_installed_body)?;
let global_installed_body = remote_installed_plugin_body("", "1.2.3", /*enabled*/ true);
mount_remote_installed_plugins(&server, "GLOBAL", &global_installed_body).await;
mount_remote_installed_plugins(&server, "WORKSPACE", &workspace_installed_body).await;
let aggregate_installed_body = aggregate_remote_installed_plugin_bodies(&[
&global_installed_body,
&workspace_installed_body,
])?;
mount_aggregate_remote_installed_plugins(&server, &aggregate_installed_body).await;
let mut mcp = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
@@ -2270,8 +2264,12 @@ plugin_sharing = true
)
]
);
wait_for_remote_installed_scope_request(&server, "WORKSPACE").await?;
wait_for_remote_installed_scope_request(&server, "GLOBAL").await?;
wait_for_remote_plugin_request_count(
&server,
"/ps/plugins/installed",
/*expected_count*/ 1,
)
.await?;
Ok(())
}
@@ -2309,9 +2307,7 @@ plugin_sharing = false
.await;
let global_installed_body =
remote_installed_plugin_body(&bundle_url, "1.2.3", /*enabled*/ true);
mount_remote_installed_plugins(&server, "GLOBAL", &global_installed_body).await;
mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body())
.await;
mount_aggregate_remote_installed_plugins(&server, &global_installed_body).await;
let mut mcp = TestAppServer::new_with_env(
codex_home.path(),
@@ -2348,8 +2344,12 @@ plugin_sharing = false
.path()
.join("plugins/cache/openai-curated-remote/linear/1.2.3/.codex-plugin/plugin.json");
wait_for_path_exists(&installed_path).await?;
wait_for_remote_installed_scope_request(&server, "GLOBAL").await?;
wait_for_remote_installed_scope_request(&server, "WORKSPACE").await?;
wait_for_remote_plugin_request_count(
&server,
"/ps/plugins/installed",
/*expected_count*/ 1,
)
.await?;
Ok(())
}
@@ -2385,7 +2385,7 @@ async fn plugin_list_fetches_workspace_directory_kind_without_remote_plugin_flag
/*enabled*/ Some(false),
);
mount_remote_plugin_list(&server, "WORKSPACE", &workspace_plugin_body).await;
mount_remote_installed_plugins(&server, "WORKSPACE", &workspace_installed_body).await;
mount_aggregate_remote_installed_plugins(&server, &workspace_installed_body).await;
let mut mcp = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
@@ -2501,8 +2501,7 @@ async fn plugin_list_fetches_shared_with_me_kind() -> Result<()> {
.push(unlisted_installed_body["plugins"][0].clone());
let workspace_installed_body = serde_json::to_string(&workspace_installed_body)?;
mount_shared_workspace_plugins(&server, &shared_plugin_body).await;
mount_remote_installed_plugins(&server, "GLOBAL", empty_remote_installed_plugins_body()).await;
mount_remote_installed_plugins(&server, "WORKSPACE", &workspace_installed_body).await;
mount_aggregate_remote_installed_plugins(&server, &workspace_installed_body).await;
let mut mcp = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
@@ -2630,8 +2629,12 @@ async fn plugin_list_fetches_shared_with_me_kind() -> Result<()> {
share_context.discoverability,
Some(PluginShareDiscoverability::Unlisted)
);
wait_for_remote_installed_scope_request(&server, "WORKSPACE").await?;
wait_for_remote_installed_scope_request(&server, "GLOBAL").await?;
wait_for_remote_plugin_request_count(
&server,
"/ps/plugins/installed",
/*expected_count*/ 1,
)
.await?;
wait_for_remote_plugin_request_count(&server, "/ps/plugins/list", /*expected_count*/ 0).await?;
Ok(())
}
@@ -2782,19 +2785,7 @@ async fn plugin_list_marks_remote_plugin_disabled_by_admin() -> Result<()> {
.mount(&server)
.await;
}
for (scope, body) in [
("GLOBAL", global_installed_body),
("WORKSPACE", empty_page_body),
] {
Mock::given(method("GET"))
.and(path("/backend-api/ps/plugins/installed"))
.and(query_param("scope", scope))
.and(header("authorization", "Bearer chatgpt-token"))
.and(header("chatgpt-account-id", "account-123"))
.respond_with(ResponseTemplate::new(200).set_body_string(body))
.mount(&server)
.await;
}
mount_aggregate_remote_installed_plugins(&server, global_installed_body).await;
let mut mcp = TestAppServer::new(codex_home.path()).await?;
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
@@ -3002,29 +2993,6 @@ async fn wait_for_remote_plugin_request_count(
Ok(())
}
async fn wait_for_remote_installed_scope_request(server: &MockServer, scope: &str) -> Result<()> {
timeout(DEFAULT_TIMEOUT, async {
loop {
let Some(requests) = server.received_requests().await else {
bail!("wiremock did not record requests");
};
if requests.iter().any(|request| {
request.method == "GET"
&& request.url.path().ends_with("/ps/plugins/installed")
&& request
.url
.query_pairs()
.any(|(name, value)| name == "scope" && value == scope)
}) {
return Ok::<(), anyhow::Error>(());
}
tokio::time::sleep(Duration::from_millis(10)).await;
}
})
.await??;
Ok(())
}
async fn wait_for_path_exists(path: &std::path::Path) -> Result<()> {
timeout(DEFAULT_TIMEOUT, async {
loop {
@@ -3087,17 +3055,34 @@ async fn mount_shared_workspace_plugins(server: &MockServer, body: &str) {
.await;
}
async fn mount_remote_installed_plugins(server: &MockServer, scope: &str, body: &str) {
async fn mount_aggregate_remote_installed_plugins(server: &MockServer, body: &str) {
Mock::given(method("GET"))
.and(path("/backend-api/ps/plugins/installed"))
.and(query_param("scope", scope))
.and(header("authorization", "Bearer chatgpt-token"))
.and(header("chatgpt-account-id", "account-123"))
.and(query_param_is_missing("scope"))
.respond_with(ResponseTemplate::new(200).set_body_string(body))
.expect(1)
.mount(server)
.await;
}
fn aggregate_remote_installed_plugin_bodies(bodies: &[&str]) -> Result<String> {
let mut aggregate: serde_json::Value =
serde_json::from_str(empty_remote_installed_plugins_body())?;
let aggregate_plugins = aggregate["plugins"]
.as_array_mut()
.ok_or_else(|| anyhow::anyhow!("installed plugins should be an array"))?;
for body in bodies {
let mut page: serde_json::Value = serde_json::from_str(body)?;
let page_plugins = page["plugins"]
.as_array_mut()
.ok_or_else(|| anyhow::anyhow!("installed plugins should be an array"))?;
aggregate_plugins.append(page_plugins);
}
Ok(serde_json::to_string(&aggregate)?)
}
fn empty_remote_installed_plugins_body() -> &'static str {
r#"{
"plugins": [],
codex-rs/core-plugins/src/manager.rs
+216 -38
View File
@@ -37,6 +37,7 @@ use crate::marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome;
use crate::marketplace_upgrade::configured_git_marketplace_names;
use crate::marketplace_upgrade::upgrade_configured_git_marketplaces;
use crate::remote::RemoteInstalledPlugin;
use crate::remote::RemoteInstalledPluginsSnapshot;
use crate::remote::RemotePluginCatalogError;
use crate::remote::RemotePluginScope;
use crate::remote::RemotePluginServiceConfig;
@@ -82,6 +83,8 @@ use tracing::warn;
static CURATED_REPO_SYNC_STARTED: AtomicBool = AtomicBool::new(false);
const FEATURED_PLUGIN_IDS_CACHE_TTL: std::time::Duration =
std::time::Duration::from_secs(60 * 60 * 3);
const REMOTE_INSTALLED_PLUGINS_SNAPSHOT_CACHE_TTL: std::time::Duration =
std::time::Duration::from_secs(60 * 5);
#[derive(Debug, Clone)]
pub struct PluginsConfigInput {
@@ -122,6 +125,21 @@ struct CachedFeaturedPluginIds {
featured_plugin_ids: Vec<String>,
}
#[derive(Clone, PartialEq, Eq)]
struct RemoteInstalledPluginsSnapshotCacheKey {
chatgpt_base_url: String,
account_id: Option<String>,
chatgpt_user_id: Option<String>,
is_workspace_account: bool,
}
#[derive(Clone)]
struct CachedRemoteInstalledPluginsSnapshot {
key: RemoteInstalledPluginsSnapshotCacheKey,
expires_at: Instant,
snapshot: RemoteInstalledPluginsSnapshot,
}
struct RemoteInstalledPluginsCacheRefreshRequest {
service_config: RemotePluginServiceConfig,
auth: Option<CodexAuth>,
@@ -188,6 +206,18 @@ fn featured_plugin_ids_cache_key(
}
}
fn remote_installed_plugins_snapshot_cache_key(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
) -> RemoteInstalledPluginsSnapshotCacheKey {
RemoteInstalledPluginsSnapshotCacheKey {
chatgpt_base_url: config.chatgpt_base_url.clone(),
account_id: auth.and_then(CodexAuth::get_account_id),
chatgpt_user_id: auth.and_then(CodexAuth::get_chatgpt_user_id),
is_workspace_account: auth.is_some_and(CodexAuth::is_workspace_account),
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PluginInstallRequest {
pub plugin_name: String,
@@ -301,6 +331,8 @@ pub struct PluginsManager {
enabled_outcome_cache: RwLock<EnabledOutcomeCache>,
enabled_outcome_load_semaphore: Semaphore,
remote_installed_plugins_cache: RwLock<Option<Vec<RemoteInstalledPlugin>>>,
remote_installed_plugins_snapshot_cache: RwLock<Option<CachedRemoteInstalledPluginsSnapshot>>,
remote_installed_plugins_fetch_semaphore: Semaphore,
remote_installed_plugins_cache_refresh_state: RwLock<RemoteInstalledPluginsCacheRefreshState>,
restriction_product: Option<Product>,
analytics_events_client: RwLock<Option<AnalyticsEventsClient>>,
@@ -352,6 +384,8 @@ impl PluginsManager {
enabled_outcome_cache: RwLock::new(EnabledOutcomeCache::default()),
enabled_outcome_load_semaphore: Semaphore::new(/*permits*/ 1),
remote_installed_plugins_cache: RwLock::new(None),
remote_installed_plugins_snapshot_cache: RwLock::new(None),
remote_installed_plugins_fetch_semaphore: Semaphore::new(/*permits*/ 1),
remote_installed_plugins_cache_refresh_state: RwLock::new(
RemoteInstalledPluginsCacheRefreshState::default(),
),
@@ -582,20 +616,160 @@ impl PluginsManager {
visible_scopes: &[RemotePluginScope],
on_effective_plugins_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
) -> Result<Vec<crate::remote::RemoteMarketplace>, RemotePluginCatalogError> {
let plugins = crate::remote::fetch_remote_installed_plugins(
&remote_plugin_service_config(config),
auth,
)
.await?;
let service_config = remote_plugin_service_config(config);
let (snapshot, changed) = self
.load_remote_installed_plugins_snapshot(
&service_config,
auth,
/*force_refresh*/ false,
)
.await?;
let plugins = snapshot.effective_plugins()?;
let marketplaces =
crate::remote::group_remote_installed_plugins_by_marketplaces(&plugins, visible_scopes);
let changed = self.write_remote_installed_plugins_cache(plugins);
if changed && let Some(on_effective_plugins_changed) = on_effective_plugins_changed {
on_effective_plugins_changed();
}
Ok(marketplaces)
}
pub async fn fetch_remote_marketplaces(
&self,
config: &PluginsConfigInput,
auth: Option<&CodexAuth>,
sources: &[crate::remote::RemoteMarketplaceSource],
) -> Result<Vec<crate::remote::RemoteMarketplace>, RemotePluginCatalogError> {
let snapshot = self.remote_installed_plugins_snapshot(config, auth).await?;
crate::remote::fetch_remote_marketplaces_from_snapshot(
&remote_plugin_service_config(config),
auth,
sources,
Some(self.codex_home.as_path()),
&snapshot,
)
.await
}
pub async fn fetch_openai_curated_remote_collection_marketplace(
&self,
config: &PluginsConfigInput,
auth: Option<&CodexAuth>,
) -> Result<Option<crate::remote::RemoteMarketplace>, RemotePluginCatalogError> {
let snapshot = self.remote_installed_plugins_snapshot(config, auth).await?;
crate::remote::fetch_openai_curated_remote_collection_marketplace_from_snapshot(
&remote_plugin_service_config(config),
auth,
&snapshot,
)
.await
}
pub(crate) async fn remote_installed_plugins_snapshot(
&self,
config: &PluginsConfigInput,
auth: Option<&CodexAuth>,
) -> Result<RemoteInstalledPluginsSnapshot, RemotePluginCatalogError> {
let service_config = remote_plugin_service_config(config);
let (snapshot, _) = self
.load_remote_installed_plugins_snapshot(
&service_config,
auth,
/*force_refresh*/ false,
)
.await?;
Ok(snapshot)
}
async fn load_remote_installed_plugins_snapshot(
&self,
service_config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
force_refresh: bool,
) -> Result<(RemoteInstalledPluginsSnapshot, bool), RemotePluginCatalogError> {
let cache_key = remote_installed_plugins_snapshot_cache_key(service_config, auth);
if !force_refresh
&& let Some(snapshot) = self.cached_remote_installed_plugins_snapshot(&cache_key)
{
return Ok((snapshot, false));
}
let Ok(_fetch_permit) = self
.remote_installed_plugins_fetch_semaphore
.acquire()
.await
else {
return Err(RemotePluginCatalogError::UnexpectedResponse(
"remote installed plugins fetch semaphore closed".to_string(),
));
};
if !force_refresh
&& let Some(snapshot) = self.cached_remote_installed_plugins_snapshot(&cache_key)
{
return Ok((snapshot, false));
}
let snapshot = crate::remote::fetch_remote_installed_plugins(service_config, auth).await?;
let changed = self.write_remote_installed_plugins_snapshot(cache_key, snapshot.clone())?;
Ok((snapshot, changed))
}
fn cached_remote_installed_plugins_snapshot(
&self,
cache_key: &RemoteInstalledPluginsSnapshotCacheKey,
) -> Option<RemoteInstalledPluginsSnapshot> {
{
let cache = match self.remote_installed_plugins_snapshot_cache.read() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
let now = Instant::now();
if let Some(cached) = cache.as_ref()
&& now < cached.expires_at
&& cached.key == *cache_key
{
return Some(cached.snapshot.clone());
}
}
let mut cache = match self.remote_installed_plugins_snapshot_cache.write() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
let now = Instant::now();
if cache
.as_ref()
.is_some_and(|cached| now >= cached.expires_at || cached.key != *cache_key)
{
*cache = None;
}
None
}
fn write_remote_installed_plugins_snapshot(
&self,
cache_key: RemoteInstalledPluginsSnapshotCacheKey,
snapshot: RemoteInstalledPluginsSnapshot,
) -> Result<bool, RemotePluginCatalogError> {
let plugins = snapshot.effective_plugins()?;
match self.remote_installed_plugins_snapshot_cache.write() {
Ok(mut cache) => {
*cache = Some(CachedRemoteInstalledPluginsSnapshot {
key: cache_key,
expires_at: Instant::now() + REMOTE_INSTALLED_PLUGINS_SNAPSHOT_CACHE_TTL,
snapshot,
});
}
Err(err) => {
*err.into_inner() = Some(CachedRemoteInstalledPluginsSnapshot {
key: cache_key,
expires_at: Instant::now() + REMOTE_INSTALLED_PLUGINS_SNAPSHOT_CACHE_TTL,
snapshot,
});
}
}
Ok(self.write_remote_installed_plugins_cache(plugins))
}
fn write_remote_installed_plugins_cache(&self, plugins: Vec<RemoteInstalledPlugin>) -> bool {
let mut cache = match self.remote_installed_plugins_cache.write() {
Ok(cache) => cache,
@@ -611,6 +785,10 @@ impl PluginsManager {
}
pub fn clear_remote_installed_plugins_cache(&self) -> bool {
match self.remote_installed_plugins_snapshot_cache.write() {
Ok(mut cache) => *cache = None,
Err(err) => *err.into_inner() = None,
}
let mut cache = match self.remote_installed_plugins_cache.write() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
@@ -684,22 +862,31 @@ impl PluginsManager {
}
let manager = Arc::clone(self);
let config_for_refresh = config.clone();
let auth_for_refresh = auth.clone();
let on_local_cache_changed = Arc::new(move || {
manager.maybe_start_remote_installed_plugins_cache_refresh_after_mutation(
&config_for_refresh,
auth_for_refresh.clone(),
on_effective_plugins_changed.clone(),
);
let config = config.clone();
tokio::spawn(async move {
match manager
.remote_installed_plugins_snapshot(&config, auth.as_ref())
.await
{
Ok(snapshot) => {
crate::remote::maybe_start_remote_installed_plugin_bundle_sync(
manager.codex_home.clone(),
snapshot,
on_effective_plugins_changed,
);
}
Err(
RemotePluginCatalogError::AuthRequired
| RemotePluginCatalogError::UnsupportedAuthMode,
) => {}
Err(err) => {
warn!(
error = %err,
"failed to load installed plugins for remote bundle sync"
);
}
}
});
crate::remote::maybe_start_remote_installed_plugin_bundle_sync(
self.codex_home.clone(),
remote_plugin_service_config(config),
auth,
Some(on_local_cache_changed),
);
}
pub fn maybe_start_plugin_list_background_tasks_for_config(
@@ -710,11 +897,6 @@ impl PluginsManager {
on_effective_plugins_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
) {
self.maybe_start_non_curated_plugin_cache_refresh(roots);
self.maybe_start_remote_installed_plugins_cache_refresh(
config,
auth.clone(),
on_effective_plugins_changed.clone(),
);
self.maybe_start_remote_installed_plugin_bundle_sync(
config,
auth,
@@ -1299,11 +1481,6 @@ impl PluginsManager {
let on_effective_plugins_changed = on_effective_plugins_changed.clone();
tokio::spawn(async move {
let auth = auth_manager_for_remote_sync.auth().await;
manager.maybe_start_remote_installed_plugins_cache_refresh(
&config_for_remote_sync,
auth.clone(),
on_effective_plugins_changed.clone(),
);
manager.maybe_start_remote_installed_plugin_bundle_sync(
&config_for_remote_sync,
auth.clone(),
@@ -1569,16 +1746,17 @@ impl PluginsManager {
}
};
let installed_plugins = crate::remote::fetch_remote_installed_plugins(
&request.service_config,
request.auth.as_ref(),
)
.await;
let installed_plugins = self
.load_remote_installed_plugins_snapshot(
&request.service_config,
request.auth.as_ref(),
/*force_refresh*/ true,
)
.await;
match installed_plugins {
Ok(installed_plugins) => {
Ok((_, changed)) => {
// TODO(remote plugins): reconcile missing or stale local bundles before
// publishing remote installed state as effective local plugin config.
let changed = self.write_remote_installed_plugins_cache(installed_plugins);
let should_notify = changed
|| matches!(
request.notify,
codex-rs/core-plugins/src/remote.rs
+124 -56
View File
@@ -126,6 +126,11 @@ pub struct RemoteInstalledPlugin {
pub keywords: Vec<String>,
}
#[derive(Debug, Clone, PartialEq)]
pub(crate) struct RemoteInstalledPluginsSnapshot {
plugins: Vec<RemotePluginInstalledItem>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct RemotePluginSummary {
pub id: String,
@@ -336,6 +341,8 @@ pub enum RemotePluginCatalogError {
pub enum RemotePluginScope {
#[serde(rename = "GLOBAL")]
Global,
#[serde(rename = "USER")]
User,
#[serde(rename = "WORKSPACE")]
Workspace,
}
@@ -344,6 +351,7 @@ impl RemotePluginScope {
fn api_value(self) -> &'static str {
match self {
Self::Global => "GLOBAL",
Self::User => "USER",
Self::Workspace => "WORKSPACE",
}
}
@@ -351,6 +359,7 @@ impl RemotePluginScope {
fn marketplace_name(self) -> &'static str {
match self {
Self::Global => REMOTE_GLOBAL_MARKETPLACE_NAME,
Self::User => unreachable!("personal plugins do not map to a Codex marketplace"),
Self::Workspace => REMOTE_WORKSPACE_MARKETPLACE_NAME,
}
}
@@ -358,6 +367,7 @@ impl RemotePluginScope {
fn marketplace_display_name(self) -> &'static str {
match self {
Self::Global => REMOTE_GLOBAL_MARKETPLACE_DISPLAY_NAME,
Self::User => unreachable!("personal plugins do not map to a Codex marketplace"),
Self::Workspace => REMOTE_WORKSPACE_MARKETPLACE_DISPLAY_NAME,
}
}
@@ -374,6 +384,33 @@ impl RemotePluginScope {
}
}
impl RemoteInstalledPluginsSnapshot {
fn plugins_for_scope(&self, scope: RemotePluginScope) -> Vec<RemotePluginInstalledItem> {
self.plugins
.iter()
.filter(|plugin| plugin.plugin.scope == scope)
.cloned()
.collect()
}
pub(crate) fn effective_plugins(
&self,
) -> Result<Vec<RemoteInstalledPlugin>, RemotePluginCatalogError> {
let mut plugins = self
.plugins
.iter()
.filter(|plugin| plugin.plugin.scope != RemotePluginScope::User)
.map(remote_installed_plugin_to_cache_entry)
.collect::<Result<Vec<_>, _>>()?;
plugins.sort_by(|left, right| {
left.marketplace_name
.cmp(&right.marketplace_name)
.then_with(|| left.id.cmp(&right.id))
});
Ok(plugins)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Deserialize)]
struct RemotePluginPagination {
next_page_token: Option<String>,
@@ -496,6 +533,10 @@ fn remote_plugin_canonical_marketplace_name(
) -> Result<&'static str, RemotePluginCatalogError> {
match plugin.scope {
RemotePluginScope::Global => Ok(REMOTE_GLOBAL_MARKETPLACE_NAME),
RemotePluginScope::User => Err(RemotePluginCatalogError::UnexpectedResponse(format!(
"personal plugin `{}` does not map to a Codex marketplace",
plugin.id
))),
RemotePluginScope::Workspace => match workspace_plugin_discoverability(plugin)? {
RemotePluginShareDiscoverability::Listed => Ok(REMOTE_WORKSPACE_MARKETPLACE_NAME),
RemotePluginShareDiscoverability::Private
@@ -557,36 +598,43 @@ pub async fn fetch_remote_marketplaces(
auth: Option<&CodexAuth>,
sources: &[RemoteMarketplaceSource],
global_catalog_cache_path: Option<&Path>,
) -> Result<Vec<RemoteMarketplace>, RemotePluginCatalogError> {
let installed_plugins = fetch_remote_installed_plugins(config, auth).await?;
fetch_remote_marketplaces_from_snapshot(
config,
auth,
sources,
global_catalog_cache_path,
&installed_plugins,
)
.await
}
pub(crate) async fn fetch_remote_marketplaces_from_snapshot(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
sources: &[RemoteMarketplaceSource],
global_catalog_cache_path: Option<&Path>,
installed_plugins: &RemoteInstalledPluginsSnapshot,
) -> Result<Vec<RemoteMarketplace>, RemotePluginCatalogError> {
let auth = ensure_chatgpt_auth(auth)?;
let mut marketplaces = Vec::new();
let needs_workspace_installed = sources.iter().any(|source| {
matches!(
source,
RemoteMarketplaceSource::WorkspaceDirectory | RemoteMarketplaceSource::SharedWithMe
)
});
let workspace_installed_plugins = if needs_workspace_installed {
Some(fetch_installed_plugins_for_scope(config, auth, RemotePluginScope::Workspace).await?)
} else {
None
};
let workspace_installed_plugins =
installed_plugins.plugins_for_scope(RemotePluginScope::Workspace);
for source in sources {
match source {
RemoteMarketplaceSource::Global => {
let scope = RemotePluginScope::Global;
let (directory_plugins, installed_plugins) = tokio::try_join!(
fetch_directory_plugins_for_scope(config, auth, scope),
fetch_installed_plugins_for_scope(config, auth, scope),
)?;
let directory_plugins =
fetch_directory_plugins_for_scope(config, auth, scope).await?;
let directory_plugins_for_cache =
global_catalog_cache_path.map(|_| directory_plugins.clone());
if let Some(marketplace) = build_remote_marketplace(
scope.marketplace_name(),
scope.marketplace_display_name(),
directory_plugins,
installed_plugins,
installed_plugins.plugins_for_scope(scope),
/*include_installed_only*/ true,
)? {
marketplaces.push(marketplace);
@@ -610,7 +658,7 @@ pub async fn fetch_remote_marketplaces(
scope.marketplace_name(),
scope.marketplace_display_name(),
directory_plugins,
workspace_installed_plugins.clone().unwrap_or_default(),
workspace_installed_plugins.clone(),
/*include_installed_only*/ false,
)? {
marketplaces.push(marketplace);
@@ -640,7 +688,7 @@ pub async fn fetch_remote_marketplaces(
REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME,
REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_DISPLAY_NAME,
directly_shared_plugins,
workspace_installed_plugins.clone().unwrap_or_default(),
workspace_installed_plugins.clone(),
/*include_installed_only*/ false,
)? {
marketplaces.push(marketplace);
@@ -648,7 +696,6 @@ pub async fn fetch_remote_marketplaces(
let unlisted_installed_plugins = workspace_installed_plugins
.clone()
.unwrap_or_default()
.into_iter()
.filter_map(
|plugin| match workspace_plugin_discoverability(&plugin.plugin) {
@@ -713,24 +760,36 @@ pub fn cached_global_remote_discoverable_plugins(
pub async fn fetch_openai_curated_remote_collection_marketplace(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
) -> Result<Option<RemoteMarketplace>, RemotePluginCatalogError> {
let installed_plugins = fetch_remote_installed_plugins(config, auth).await?;
fetch_openai_curated_remote_collection_marketplace_from_snapshot(
config,
auth,
&installed_plugins,
)
.await
}
pub(crate) async fn fetch_openai_curated_remote_collection_marketplace_from_snapshot(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
installed_plugins: &RemoteInstalledPluginsSnapshot,
) -> Result<Option<RemoteMarketplace>, RemotePluginCatalogError> {
let auth = ensure_chatgpt_auth(auth)?;
let scope = RemotePluginScope::Global;
let (directory_plugins, installed_plugins) = tokio::try_join!(
fetch_directory_plugins_for_scope_with_collection(
config,
auth,
scope,
OPENAI_CURATED_REMOTE_COLLECTION_KEY,
),
fetch_installed_plugins_for_scope(config, auth, scope),
)?;
let directory_plugins = fetch_directory_plugins_for_scope_with_collection(
config,
auth,
scope,
OPENAI_CURATED_REMOTE_COLLECTION_KEY,
)
.await?;
build_remote_marketplace(
REMOTE_GLOBAL_MARKETPLACE_NAME,
REMOTE_GLOBAL_MARKETPLACE_DISPLAY_NAME,
directory_plugins,
installed_plugins,
installed_plugins.plugins_for_scope(scope),
/*include_installed_only*/ false,
)
}
@@ -786,31 +845,9 @@ fn build_remote_marketplace(
pub(crate) async fn fetch_remote_installed_plugins(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
) -> Result<Vec<RemoteInstalledPlugin>, RemotePluginCatalogError> {
) -> Result<RemoteInstalledPluginsSnapshot, RemotePluginCatalogError> {
let auth = ensure_chatgpt_auth(auth)?;
let global = async {
let scope = RemotePluginScope::Global;
let installed_plugins = fetch_installed_plugins_for_scope(config, auth, scope).await?;
Ok::<_, RemotePluginCatalogError>((scope, installed_plugins))
};
let workspace = async {
let scope = RemotePluginScope::Workspace;
let installed_plugins = fetch_installed_plugins_for_scope(config, auth, scope).await?;
Ok::<_, RemotePluginCatalogError>((scope, installed_plugins))
};
let (global, workspace) = tokio::try_join!(global, workspace)?;
let mut installed_plugins = [global, workspace]
.into_iter()
.flat_map(|(_scope, plugins)| plugins)
.map(|plugin| remote_installed_plugin_to_cache_entry(&plugin))
.collect::<Result<Vec<_>, _>>()?;
installed_plugins.sort_by(|left, right| {
left.marketplace_name
.cmp(&right.marketplace_name)
.then_with(|| left.id.cmp(&right.id))
});
Ok(installed_plugins)
fetch_installed_plugins(config, auth, /*include_download_urls*/ true).await
}
pub fn group_remote_installed_plugins_by_marketplaces(
@@ -1220,6 +1257,10 @@ fn remote_plugin_share_context(
) -> Result<Option<RemotePluginShareContext>, RemotePluginCatalogError> {
match plugin.scope {
RemotePluginScope::Global => Ok(None),
RemotePluginScope::User => Err(RemotePluginCatalogError::UnexpectedResponse(format!(
"personal plugin `{}` does not expose Codex share context",
plugin.id
))),
RemotePluginScope::Workspace => {
let discoverability = workspace_plugin_discoverability(plugin)?;
Ok(Some(RemotePluginShareContext {
@@ -1455,6 +1496,31 @@ async fn fetch_installed_plugins_for_scope(
.await
}
async fn fetch_installed_plugins(
config: &RemotePluginServiceConfig,
auth: &CodexAuth,
include_download_urls: bool,
) -> Result<RemoteInstalledPluginsSnapshot, RemotePluginCatalogError> {
let mut plugins = Vec::new();
let mut page_token = None;
loop {
let response = get_remote_plugin_installed_page(
config,
auth,
/*scope*/ None,
page_token.as_deref(),
include_download_urls,
)
.await?;
plugins.extend(response.plugins);
let Some(next_page_token) = response.pagination.next_page_token else {
break;
};
page_token = Some(next_page_token);
}
Ok(RemoteInstalledPluginsSnapshot { plugins })
}
async fn fetch_installed_plugins_for_scope_with_download_url(
config: &RemotePluginServiceConfig,
auth: &CodexAuth,
@@ -1467,7 +1533,7 @@ async fn fetch_installed_plugins_for_scope_with_download_url(
let response = get_remote_plugin_installed_page(
config,
auth,
scope,
Some(scope),
page_token.as_deref(),
include_download_urls,
)
@@ -1522,7 +1588,7 @@ async fn get_remote_shared_workspace_plugins_page(
async fn get_remote_plugin_installed_page(
config: &RemotePluginServiceConfig,
auth: &CodexAuth,
scope: RemotePluginScope,
scope: Option<RemotePluginScope>,
page_token: Option<&str>,
include_download_urls: bool,
) -> Result<RemotePluginInstalledResponse, RemotePluginCatalogError> {
@@ -1530,7 +1596,9 @@ async fn get_remote_plugin_installed_page(
let url = format!("{base_url}/ps/plugins/installed");
let client = build_reqwest_client();
let mut request = authenticated_request(client.get(&url), auth)?;
request = request.query(&[("scope", scope.api_value())]);
if let Some(scope) = scope {
request = request.query(&[("scope", scope.api_value())]);
}
if include_download_urls {
request = request.query(&[("includeDownloadUrls", true)]);
}
codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs
+81 -95
View File
@@ -3,11 +3,11 @@ use super::REMOTE_WORKSPACE_MARKETPLACE_NAME;
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME;
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_PRIVATE_MARKETPLACE_NAME;
use super::REMOTE_WORKSPACE_SHARED_WITH_ME_UNLISTED_MARKETPLACE_NAME;
use super::RemoteInstalledPluginsSnapshot;
use super::RemotePluginCatalogError;
use super::RemotePluginScope;
use super::RemotePluginServiceConfig;
use super::ensure_chatgpt_auth;
use super::fetch_installed_plugins_for_scope_with_download_url;
use super::fetch_remote_installed_plugins;
use super::remote_plugin_canonical_marketplace_name;
use crate::store::PLUGINS_CACHE_DIR;
use crate::store::PluginStore;
@@ -80,13 +80,9 @@ pub struct RemotePluginCacheMutationGuard {
pub(crate) fn maybe_start_remote_installed_plugin_bundle_sync(
codex_home: PathBuf,
config: RemotePluginServiceConfig,
auth: Option<CodexAuth>,
installed_plugins: RemoteInstalledPluginsSnapshot,
on_local_cache_changed: Option<Arc<dyn Fn() + Send + Sync + 'static>>,
) {
let Some(auth) = auth else {
return;
};
let key = RemoteInstalledPluginBundleSyncKey {
plugin_cache_root: remote_plugin_cache_root(&codex_home),
};
@@ -96,7 +92,7 @@ pub(crate) fn maybe_start_remote_installed_plugin_bundle_sync(
tokio::spawn(async move {
let result =
sync_remote_installed_plugin_bundles_once(codex_home, &config, Some(&auth)).await;
sync_remote_installed_plugin_bundles_from_snapshot(codex_home, installed_plugins).await;
match result {
Ok(outcome) => {
if outcome.changed_local_cache()
@@ -127,25 +123,14 @@ pub async fn sync_remote_installed_plugin_bundles_once(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
) -> Result<RemoteInstalledPluginBundleSyncOutcome, RemoteInstalledPluginBundleSyncError> {
let auth = ensure_chatgpt_auth(auth)?;
let global = async {
let scope = RemotePluginScope::Global;
let installed_plugins = fetch_installed_plugins_for_scope_with_download_url(
config, auth, scope, /*include_download_urls*/ true,
)
.await?;
Ok::<_, RemotePluginCatalogError>((scope, installed_plugins))
};
let workspace = async {
let scope = RemotePluginScope::Workspace;
let installed_plugins = fetch_installed_plugins_for_scope_with_download_url(
config, auth, scope, /*include_download_urls*/ true,
)
.await?;
Ok::<_, RemotePluginCatalogError>((scope, installed_plugins))
};
let installed_plugins = fetch_remote_installed_plugins(config, auth).await?;
sync_remote_installed_plugin_bundles_from_snapshot(codex_home, installed_plugins).await
}
let (global, workspace) = tokio::try_join!(global, workspace)?;
async fn sync_remote_installed_plugin_bundles_from_snapshot(
codex_home: PathBuf,
installed_plugins: RemoteInstalledPluginsSnapshot,
) -> Result<RemoteInstalledPluginBundleSyncOutcome, RemoteInstalledPluginBundleSyncError> {
let store = PluginStore::try_new(codex_home.clone())?;
let mut installed_plugin_names_by_marketplace =
BTreeMap::<String, BTreeSet<String>>::from_iter([
@@ -170,79 +155,80 @@ pub async fn sync_remote_installed_plugin_bundles_once(
let mut installed_plugin_ids = BTreeSet::new();
let mut failed_remote_plugin_ids = BTreeSet::new();
for (_scope, installed_plugins) in [global, workspace] {
for installed_plugin in installed_plugins {
let plugin = installed_plugin.plugin;
let marketplace_name = remote_plugin_canonical_marketplace_name(&plugin)?.to_string();
installed_plugin_names_by_marketplace
.entry(marketplace_name.clone())
.or_default()
.insert(plugin.name.clone());
let plugin_id = match PluginId::new(plugin.name.clone(), marketplace_name.clone()) {
Ok(plugin_id) => plugin_id,
Err(err) => {
warn!(
remote_plugin_id = %plugin.id,
plugin = %plugin.name,
marketplace = %marketplace_name,
error = %err,
"skipping remote installed plugin with invalid local cache id"
);
failed_remote_plugin_ids.insert(plugin.id);
continue;
}
};
let release_version = plugin
.release
.version
.as_deref()
.map(str::trim)
.filter(|version| !version.is_empty());
if store.active_plugin_version(&plugin_id).as_deref() == release_version {
for installed_plugin in installed_plugins.plugins {
if installed_plugin.plugin.scope == RemotePluginScope::User {
continue;
}
let plugin = installed_plugin.plugin;
let marketplace_name = remote_plugin_canonical_marketplace_name(&plugin)?.to_string();
installed_plugin_names_by_marketplace
.entry(marketplace_name.clone())
.or_default()
.insert(plugin.name.clone());
let plugin_id = match PluginId::new(plugin.name.clone(), marketplace_name.clone()) {
Ok(plugin_id) => plugin_id,
Err(err) => {
warn!(
remote_plugin_id = %plugin.id,
plugin = %plugin.name,
marketplace = %marketplace_name,
error = %err,
"skipping remote installed plugin with invalid local cache id"
);
failed_remote_plugin_ids.insert(plugin.id);
continue;
}
};
let release_version = plugin
.release
.version
.as_deref()
.map(str::trim)
.filter(|version| !version.is_empty());
if store.active_plugin_version(&plugin_id).as_deref() == release_version {
continue;
}
let bundle = match crate::remote_bundle::validate_remote_plugin_bundle(
&plugin.id,
&marketplace_name,
&plugin.name,
release_version,
plugin.release.bundle_download_url.as_deref(),
plugin.release.app_manifest.clone(),
) {
Ok(bundle) => bundle,
Err(err) => {
warn!(
remote_plugin_id = %plugin.id,
plugin = %plugin.name,
marketplace = %marketplace_name,
error = %err,
"skipping remote installed plugin bundle download"
);
failed_remote_plugin_ids.insert(plugin.id);
continue;
}
};
let bundle = match crate::remote_bundle::validate_remote_plugin_bundle(
&plugin.id,
&marketplace_name,
&plugin.name,
release_version,
plugin.release.bundle_download_url.as_deref(),
plugin.release.app_manifest.clone(),
) {
Ok(bundle) => bundle,
Err(err) => {
warn!(
remote_plugin_id = %plugin.id,
plugin = %plugin.name,
marketplace = %marketplace_name,
error = %err,
"skipping remote installed plugin bundle download"
);
failed_remote_plugin_ids.insert(plugin.id);
continue;
}
};
match crate::remote_bundle::download_and_install_remote_plugin_bundle(
codex_home.clone(),
bundle,
)
.await
{
Ok(result) => {
installed_plugin_ids.insert(result.plugin_id.as_key());
}
Err(err) => {
warn!(
remote_plugin_id = %plugin.id,
plugin = %plugin.name,
marketplace = %marketplace_name,
error = %err,
"failed to download remote installed plugin bundle"
);
failed_remote_plugin_ids.insert(plugin.id);
}
match crate::remote_bundle::download_and_install_remote_plugin_bundle(
codex_home.clone(),
bundle,
)
.await
{
Ok(result) => {
installed_plugin_ids.insert(result.plugin_id.as_key());
}
Err(err) => {
warn!(
remote_plugin_id = %plugin.id,
plugin = %plugin.name,
marketplace = %marketplace_name,
error = %err,
"failed to download remote installed plugin bundle"
);
failed_remote_plugin_ids.insert(plugin.id);
}
}
}