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

Harden remote installed plugin cache reconciliation (#40015)

Browse Source 
## What changed

- Scope remote installed-plugin and loaded-plugin snapshots to the active account, and discard in-flight loads when the account changes.
- Serialize bundle reconciliation with direct installs and uninstalls, and use cache generations to prevent stale refreshes from overwriting newer state.
- Validate the complete installed-plugin snapshot before downloading bundles or removing stale cache entries, while retaining valid installed metadata when materialization fails.

## Testing

- Cover account changes during plugin loading, incomplete snapshots, refresh/reconciliation races, and reconciliation cancellation recovery.

GitOrigin-RevId: 694c25577b420b93dacc6c65e3c0ecb7b8cc64c3
This commit is contained in:
Matthew Zeng
2026-08-21 21:28:16 +00:00
committed by copyberry
parent 9949c9eafa
commit e6a3877e95
8 changed files with 986 additions and 161 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/request_processors/plugins.rs
+45 -18
View File
@@ -1608,14 +1608,6 @@ impl PluginRequestProcessor {
);
return Err(invalid_request(error_message));
}
// Direct install writes the same cache tree that installed-plugin sync
// prunes before the backend installed snapshot can include this plugin.
let _remote_plugin_cache_mutation =
codex_core_plugins::remote::mark_remote_plugin_cache_mutation_in_flight(
config.codex_home.as_path(),
&actual_remote_marketplace_name,
&remote_plugin_name,
);
let validated_bundle = codex_core_plugins::remote_bundle::validate_remote_plugin_bundle(
&remote_plugin_id,
&actual_remote_marketplace_name,
@@ -1638,6 +1630,24 @@ impl PluginRequestProcessor {
remote_plugin_bundle_install_error_to_jsonrpc(err)
})?;
// Direct install writes the same cache tree that installed-plugin sync prunes. Hold the
// shared cache-root gate through the backend mutation so a full sync cannot commit a
// snapshot fetched before this plugin became installed.
let plugins_manager = self.thread_manager.plugins_manager();
let remote_plugin_sync_guard = plugins_manager
.acquire_remote_installed_plugin_sync_guard()
.await
.map_err(|err| {
internal_error(format!("failed to coordinate remote plugin install: {err}"))
})?;
// Keep this marker through downstream setup after releasing the shared gate. A new sync
// may start then, but it must not prune the bundle that this install just materialized.
let _remote_plugin_cache_mutation =
codex_core_plugins::remote::mark_remote_plugin_cache_mutation_in_flight(
config.codex_home.as_path(),
&actual_remote_marketplace_name,
&remote_plugin_name,
);
let result = codex_core_plugins::remote_bundle::download_and_install_remote_plugin_bundle(
&remote_plugin_service_config,
config.codex_home.to_path_buf(),
@@ -1693,13 +1703,12 @@ impl PluginRequestProcessor {
remote_plugin_catalog_error_to_jsonrpc(err, "install remote plugin")
})?;
self.thread_manager
.plugins_manager()
.maybe_start_remote_installed_plugins_cache_refresh_after_mutation(
&config.plugins_config_input(),
auth.clone(),
Some(self.effective_plugins_changed_callback()),
);
plugins_manager.maybe_start_remote_installed_plugins_cache_refresh_after_mutation(
&config.plugins_config_input(),
auth.clone(),
Some(self.effective_plugins_changed_callback()),
);
drop(remote_plugin_sync_guard);
let plugin_metadata = self
.thread_manager
@@ -2128,6 +2137,20 @@ impl PluginRequestProcessor {
plugin_telemetry.capability_summary =
Some(uninstall_target.fallback_capability_summary.clone());
}
let remote_plugin_sync_guard = plugins_manager
.acquire_remote_installed_plugin_sync_guard()
.await
.map_err(|err| {
internal_error(format!(
"failed to coordinate remote plugin uninstall: {err}"
))
})?;
let remote_plugin_cache_mutation =
codex_core_plugins::remote::mark_remote_plugin_cache_mutation_in_flight(
config.codex_home.as_path(),
&uninstall_target.plugin_id.marketplace_name,
&uninstall_target.plugin_id.plugin_name,
);
let uninstall_result = codex_core_plugins::remote::uninstall_remote_plugin(
&remote_plugin_service_config,
auth.as_ref(),
@@ -2136,21 +2159,25 @@ impl PluginRequestProcessor {
)
.await;
let mut refresh_effective_plugins = false;
if matches!(
&uninstall_result,
Ok(()) | Err(RemotePluginCatalogError::CacheRemove(_))
) {
self.analytics_events_client
.track_plugin_uninstalled(plugin_telemetry);
if plugins_manager.clear_remote_installed_plugins_cache() {
self.on_effective_plugins_changed().await;
}
refresh_effective_plugins = plugins_manager.clear_remote_installed_plugins_cache();
plugins_manager.maybe_start_remote_installed_plugins_cache_refresh_after_mutation(
&config.plugins_config_input(),
auth.clone(),
Some(self.effective_plugins_changed_callback()),
);
}
drop(remote_plugin_cache_mutation);
drop(remote_plugin_sync_guard);
if refresh_effective_plugins {
self.on_effective_plugins_changed().await;
}
uninstall_result.map_err(|err| {
remote_plugin_catalog_error_to_jsonrpc(err, "uninstall remote plugin")
codex-rs/core-plugins/src/discoverable_tests.rs
+5 -1
View File
@@ -946,7 +946,11 @@ plugins = true
let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing();
let mut plugins = load_plugins_config(codex_home.path(), codex_home.path()).await;
plugins.chatgpt_base_url = format!("{}/backend-api", server.uri());
let plugins_manager = test_plugins_manager(codex_home.path().to_path_buf());
let plugins_manager = test_plugins_manager_with_options(
codex_home.path().to_path_buf(),
Some(Product::Codex),
Some(AuthMode::Chatgpt),
);
fetch_and_cache_global_remote_plugin_catalog(
codex_home.path(),
&RemotePluginServiceConfig::new(
codex-rs/core-plugins/src/lib.rs
+1
View File
@@ -80,6 +80,7 @@ pub use manager::PluginUninstallError;
pub use manager::PluginsConfigInput;
pub use manager::PluginsManager;
pub use manager::RecommendedPluginCandidatesInput;
pub use manager::RemoteInstalledPluginSyncGuard;
pub use marketplace_policy::allowed_configured_marketplace_names;
pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeError as PluginMarketplaceUpgradeError;
pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome as PluginMarketplaceUpgradeOutcome;
codex-rs/core-plugins/src/manager.rs
+517 -83
View File
@@ -48,6 +48,8 @@ use crate::marketplace_upgrade::upgrade_configured_git_marketplaces_with_mode;
use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME;
use crate::remote::RecommendedPluginsMode;
use crate::remote::RemoteInstalledPlugin;
use crate::remote::RemoteInstalledPluginBundleSyncError;
use crate::remote::RemoteInstalledPluginBundleSyncOutcome;
use crate::remote::RemotePluginCatalogError;
use crate::remote::RemotePluginMaterialization;
use crate::remote::RemotePluginScope;
@@ -110,8 +112,10 @@ use std::sync::Arc;
use std::sync::RwLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
use tokio::sync::OnceCell;
use tokio::sync::OwnedSemaphorePermit;
use tokio::sync::Semaphore;
use tokio::sync::watch;
use tracing::instrument;
@@ -120,6 +124,7 @@ 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_PLUGIN_SYNC_WAIT_TIMEOUT: Duration = Duration::from_secs(120);
type EffectivePluginsChangedCallback = Arc<dyn Fn(EffectivePluginsChange) + Send + Sync + 'static>;
@@ -197,7 +202,66 @@ struct CachedFeaturedPluginIds {
featured_plugin_ids: Vec<String>,
}
#[derive(Clone, PartialEq, Eq)]
struct RemoteInstalledPluginsAuthIdentity {
auth_mode: Option<AuthMode>,
account_id: Option<String>,
chatgpt_user_id: Option<String>,
is_workspace_account: Option<bool>,
}
impl RemoteInstalledPluginsAuthIdentity {
fn from_auth(auth: Option<&CodexAuth>) -> Self {
Self {
auth_mode: auth.map(CodexAuth::api_auth_mode),
account_id: auth.and_then(CodexAuth::get_account_id),
chatgpt_user_id: auth.and_then(CodexAuth::get_chatgpt_user_id),
is_workspace_account: auth.map(CodexAuth::is_workspace_account),
}
}
}
#[derive(Default)]
struct RemoteInstalledPluginsCache {
generation: u64,
auth_identity: Option<RemoteInstalledPluginsAuthIdentity>,
reconciliation_generation: Option<u64>,
// A cancelled pass may mutate bundles before publication; retry must notify consumers.
needs_effective_plugins_refresh: bool,
plugins: Option<Vec<RemoteInstalledPlugin>>,
}
enum RemoteInstalledPluginsCachePublication {
Refresh,
Reconcile,
}
/// Holds the cache-root gate shared by full installed-bundle sync, reconciliation, and direct
/// remote plugin mutations.
///
/// Callers should keep this guard only while mutating the remote plugin cache and backend
/// installed state, then release it before unrelated analytics, OAuth, or runtime refresh work.
pub struct RemoteInstalledPluginSyncGuard {
_permit: OwnedSemaphorePermit,
}
struct RemoteInstalledPluginsReconciliationGuard<'a> {
manager: &'a PluginsManager,
generation: u64,
committed: bool,
}
impl Drop for RemoteInstalledPluginsReconciliationGuard<'_> {
fn drop(&mut self) {
if !self.committed {
self.manager
.abandon_remote_installed_plugins_reconcile(self.generation);
}
}
}
struct RemoteInstalledPluginsCacheRefreshRequest {
generation: u64,
service_config: RemotePluginServiceConfig,
auth: Option<CodexAuth>,
notify: RemoteInstalledPluginsCacheRefreshNotify,
@@ -439,12 +503,12 @@ pub struct PluginsManager {
non_curated_cache_refresh_lock: Semaphore,
non_curated_cache_refresh_state: RwLock<NonCuratedCacheRefreshState>,
non_curated_cache_refresh_completion: watch::Sender<NonCuratedCacheRefreshCompletion>,
// Keep the cache auth-independent so auth changes only need to resolve capabilities again.
// Remote installed plugins vary by account, so loaded capabilities are cached per auth.
loaded_plugins_cache: RwLock<LoadedPluginsCache>,
loaded_plugins_load_semaphore: Semaphore,
skill_root_loader: Arc<dyn SkillRootLoader<PluginSkillRoot>>,
tool_suggest_metadata_cache: ToolSuggestMetadataCache,
remote_installed_plugins_cache: RwLock<Option<Vec<RemoteInstalledPlugin>>>,
remote_installed_plugins_cache: RwLock<RemoteInstalledPluginsCache>,
remote_installed_plugin_bundle_sync_gate: Arc<Semaphore>,
remote_installed_plugins_cache_refresh_state: RwLock<RemoteInstalledPluginsCacheRefreshState>,
remote_catalog_cache_refresh_state: RwLock<RemoteCatalogCacheRefreshState>,
@@ -472,6 +536,7 @@ struct PluginLoadCacheKey {
configured_plugins: HashMap<String, PluginConfig>,
skill_config_rules: SkillConfigRules,
remote_global_catalog_active: bool,
auth_identity: Option<RemoteInstalledPluginsAuthIdentity>,
}
impl PluginLoadCacheKey {
@@ -479,6 +544,7 @@ impl PluginLoadCacheKey {
config: &PluginsConfigInput,
codex_home: &Path,
remote_global_catalog_active: bool,
auth_identity: RemoteInstalledPluginsAuthIdentity,
) -> Self {
Self {
configured_plugins: configured_plugins_from_stack(
@@ -487,6 +553,8 @@ impl PluginLoadCacheKey {
),
skill_config_rules: skill_config_rules_from_stack(&config.config_layer_stack),
remote_global_catalog_active,
// Local curated loads are auth-independent; only remote snapshots vary by account.
auth_identity: remote_global_catalog_active.then_some(auth_identity),
}
}
}
@@ -547,7 +615,7 @@ impl PluginsManager {
loaded_plugins_load_semaphore: Semaphore::new(/*permits*/ 1),
skill_root_loader,
tool_suggest_metadata_cache: ToolSuggestMetadataCache::new(),
remote_installed_plugins_cache: RwLock::new(None),
remote_installed_plugins_cache: RwLock::new(RemoteInstalledPluginsCache::default()),
remote_installed_plugin_bundle_sync_gate,
remote_installed_plugins_cache_refresh_state: RwLock::new(
RemoteInstalledPluginsCacheRefreshState::default(),
@@ -620,10 +688,12 @@ impl PluginsManager {
if !config.plugins_enabled {
return None;
}
let auth = self.auth_manager.auth_cached();
let key = PluginLoadCacheKey::from_config(
config,
self.codex_home.as_path(),
self.remote_global_catalog_active(config),
RemoteInstalledPluginsAuthIdentity::from_auth(auth.as_ref()),
);
self.loaded_plugins_cache
.read()
@@ -653,47 +723,95 @@ impl PluginsManager {
return PluginLoadOutcome::default();
}
let remote_global_catalog_active = self.remote_global_catalog_active(config);
let cache_key = PluginLoadCacheKey::from_config(
config,
self.codex_home.as_path(),
remote_global_catalog_active,
);
if !force_reload && let Some(plugins) = self.cached_loaded_plugins(&cache_key) {
return self.resolve_loaded_plugins_for_auth(plugins);
}
// Both the semaphore wait and skill loading can outlive the auth state captured at the
// start of this load. Account switches require a restart, so discard those passes;
// same-account revisions also cover ordinary token refreshes, where returning no plugins
// is wrong, so retry those instead.
let auth_change_receiver = self.auth_manager.auth_change_receiver();
let Ok(_load_permit) = self.loaded_plugins_load_semaphore.acquire().await else {
warn!("plugin load semaphore closed");
return PluginLoadOutcome::default();
};
if !force_reload && let Some(plugins) = self.cached_loaded_plugins(&cache_key) {
return self.resolve_loaded_plugins_for_auth(plugins);
loop {
let auth_revision = *auth_change_receiver.borrow();
let remote_global_catalog_active = self.remote_global_catalog_active(config);
let auth = self.auth_manager.auth_cached();
let auth_mode = auth.as_ref().map(CodexAuth::api_auth_mode);
let auth_identity = RemoteInstalledPluginsAuthIdentity::from_auth(auth.as_ref());
let cache_key = PluginLoadCacheKey::from_config(
config,
self.codex_home.as_path(),
remote_global_catalog_active,
auth_identity.clone(),
);
if !force_reload && let Some(plugins) = self.cached_loaded_plugins(&cache_key) {
let outcome = self.resolve_loaded_plugins_for_auth(plugins, auth_mode);
if *auth_change_receiver.borrow() == auth_revision {
return outcome;
}
if !self.remote_installed_plugins_auth_is_current(&auth_identity) {
return PluginLoadOutcome::default();
}
continue;
}
let Ok(_load_permit) = self.loaded_plugins_load_semaphore.acquire().await else {
warn!("plugin load semaphore closed");
return PluginLoadOutcome::default();
};
if *auth_change_receiver.borrow() != auth_revision {
if !self.remote_installed_plugins_auth_is_current(&auth_identity) {
return PluginLoadOutcome::default();
}
continue;
}
if !force_reload && let Some(plugins) = self.cached_loaded_plugins(&cache_key) {
let outcome = self.resolve_loaded_plugins_for_auth(plugins, auth_mode);
if *auth_change_receiver.borrow() == auth_revision {
return outcome;
}
if !self.remote_installed_plugins_auth_is_current(&auth_identity) {
return PluginLoadOutcome::default();
}
continue;
}
let cache_generation = self.loaded_plugins_cache_generation();
let plugin_skill_snapshots = new_plugin_skill_snapshots();
let plugins = load_plugins_from_layer_stack(
&config.config_layer_stack,
self.remote_installed_plugins_snapshot(),
&self.store,
Some(&plugin_skill_snapshots),
self.restriction_product,
remote_global_catalog_active,
self.skill_root_loader.as_ref(),
)
.await;
if *auth_change_receiver.borrow() != auth_revision {
if !self.remote_installed_plugins_auth_is_current(&auth_identity) {
return PluginLoadOutcome::default();
}
continue;
}
log_plugin_load_errors(&plugins);
self.cache_loaded_plugins_if_current(
cache_generation,
cache_key,
plugins.clone(),
plugin_skill_snapshots,
);
let outcome = self.resolve_loaded_plugins_for_auth(plugins, auth_mode);
if *auth_change_receiver.borrow() == auth_revision {
return outcome;
}
if !self.remote_installed_plugins_auth_is_current(&auth_identity) {
return PluginLoadOutcome::default();
}
}
let cache_generation = self.loaded_plugins_cache_generation();
let plugin_skill_snapshots = new_plugin_skill_snapshots();
let plugins = load_plugins_from_layer_stack(
&config.config_layer_stack,
self.remote_installed_plugins_snapshot(),
&self.store,
Some(&plugin_skill_snapshots),
self.restriction_product,
remote_global_catalog_active,
self.skill_root_loader.as_ref(),
)
.await;
log_plugin_load_errors(&plugins);
self.cache_loaded_plugins_if_current(
cache_generation,
cache_key,
plugins.clone(),
plugin_skill_snapshots,
);
self.resolve_loaded_plugins_for_auth(plugins)
}
fn resolve_loaded_plugins_for_auth(&self, mut plugins: Vec<LoadedPlugin>) -> PluginLoadOutcome {
let auth_mode = self.auth_mode();
fn resolve_loaded_plugins_for_auth(
&self,
mut plugins: Vec<LoadedPlugin>,
auth_mode: Option<AuthMode>,
) -> PluginLoadOutcome {
let target_curated_marketplace = target_curated_marketplace(auth_mode);
plugins.retain(|plugin| {
plugin_is_eligible_for_target_marketplace(
@@ -822,7 +940,10 @@ impl PluginsManager {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
let Some(plugins) = cache.as_ref() else {
if !self.remote_installed_plugins_cache_matches_current_auth(&cache) {
return HashMap::new();
}
let Some(plugins) = cache.plugins.as_ref() else {
return HashMap::new();
};
@@ -834,7 +955,10 @@ impl PluginsManager {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
let Some(plugins) = cache.as_ref() else {
if !self.remote_installed_plugins_cache_matches_current_auth(&cache) {
return RemoteInstalledPluginsSnapshot::default();
}
let Some(plugins) = cache.plugins.as_ref() else {
return RemoteInstalledPluginsSnapshot::default();
};
@@ -849,7 +973,9 @@ impl PluginsManager {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
if let Some(plugins) = cache.as_ref() {
if self.remote_installed_plugins_cache_matches_current_auth(&cache)
&& let Some(plugins) = cache.plugins.as_ref()
{
return plugins.iter().find_map(|plugin| {
(plugin.name == plugin_id.plugin_name
&& plugin.marketplace_name == plugin_id.marketplace_name)
@@ -862,6 +988,25 @@ impl PluginsManager {
persisted_remote_plugin_id_for_installation(&installation)
}
fn remote_installed_plugins_cache_matches_current_auth(
&self,
cache: &RemoteInstalledPluginsCache,
) -> bool {
let auth = self.auth_manager.auth_cached();
cache.auth_identity.as_ref()
== Some(&RemoteInstalledPluginsAuthIdentity::from_auth(
auth.as_ref(),
))
}
fn remote_installed_plugins_auth_is_current(
&self,
auth_identity: &RemoteInstalledPluginsAuthIdentity,
) -> bool {
let auth = self.auth_manager.auth_cached();
RemoteInstalledPluginsAuthIdentity::from_auth(auth.as_ref()) == *auth_identity
}
pub async fn telemetry_metadata_for_installed_plugin(
&self,
plugin_id: &PluginId,
@@ -939,12 +1084,29 @@ impl PluginsManager {
pub fn build_remote_installed_plugin_marketplaces_from_cache(
&self,
visible_marketplaces: &[&str],
) -> Option<Vec<crate::remote::RemoteMarketplace>> {
let auth = self.auth_manager.auth_cached();
self.build_remote_installed_plugin_marketplaces_from_cache_for_auth(
auth.as_ref(),
visible_marketplaces,
)
}
fn build_remote_installed_plugin_marketplaces_from_cache_for_auth(
&self,
auth: Option<&CodexAuth>,
visible_marketplaces: &[&str],
) -> Option<Vec<crate::remote::RemoteMarketplace>> {
let cache = match self.remote_installed_plugins_cache.read() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
let plugins = cache.as_ref()?;
if cache.auth_identity.as_ref()
!= Some(&RemoteInstalledPluginsAuthIdentity::from_auth(auth))
{
return None;
}
let plugins = cache.plugins.as_ref()?;
Some(
crate::remote::group_remote_installed_plugins_by_marketplaces(
plugins,
@@ -985,6 +1147,9 @@ impl PluginsManager {
visible_marketplaces: &[&str],
on_effective_plugins_changed: Option<EffectivePluginsChangedCallback>,
) -> Result<Vec<crate::remote::RemoteMarketplace>, RemotePluginCatalogError> {
let generation = self
.prepare_remote_installed_plugins_cache_generation(auth)
.ok_or(RemotePluginCatalogError::AuthRequired)?;
let plugins = crate::remote::fetch_remote_installed_plugins(
&remote_plugin_service_config(config),
auth,
@@ -994,25 +1159,173 @@ impl PluginsManager {
&plugins,
visible_marketplaces,
);
let changed = self.write_remote_installed_plugins_cache(plugins);
let Some(changed) = self.write_remote_installed_plugins_cache_snapshot(
generation,
plugins,
auth,
RemoteInstalledPluginsCachePublication::Refresh,
) else {
let auth_identity = RemoteInstalledPluginsAuthIdentity::from_auth(auth);
if !self.remote_installed_plugins_auth_is_current(&auth_identity) {
return Err(RemotePluginCatalogError::AuthRequired);
}
return Ok(self
.build_remote_installed_plugin_marketplaces_from_cache_for_auth(
auth,
visible_marketplaces,
)
.unwrap_or(marketplaces));
};
if changed && let Some(on_effective_plugins_changed) = on_effective_plugins_changed {
on_effective_plugins_changed(EffectivePluginsChange::default());
}
Ok(marketplaces)
Ok(self
.build_remote_installed_plugin_marketplaces_from_cache_for_auth(
auth,
visible_marketplaces,
)
.unwrap_or(marketplaces))
}
fn write_remote_installed_plugins_cache(&self, plugins: Vec<RemoteInstalledPlugin>) -> bool {
fn prepare_remote_installed_plugins_cache_generation(
&self,
auth: Option<&CodexAuth>,
) -> Option<u64> {
let auth_identity = RemoteInstalledPluginsAuthIdentity::from_auth(auth);
if !self.remote_installed_plugins_auth_is_current(&auth_identity) {
return None;
}
let mut cache = match self.remote_installed_plugins_cache.write() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
if cache.as_ref().is_some_and(|cache| cache.eq(&plugins)) {
return false;
let cache_changed = if cache.auth_identity.as_ref() != Some(&auth_identity) {
cache.generation = cache.generation.wrapping_add(1);
cache.auth_identity = Some(auth_identity);
cache.reconciliation_generation = None;
cache.needs_effective_plugins_refresh = false;
cache.plugins.take().is_some()
} else {
false
};
let generation = cache.generation;
drop(cache);
if cache_changed {
self.clear_loaded_plugins_cache();
}
*cache = Some(plugins);
Some(generation)
}
fn begin_remote_installed_plugins_reconcile(&self, auth: Option<&CodexAuth>) -> Option<u64> {
let generation = self.prepare_remote_installed_plugins_cache_generation(auth)?;
let auth_identity = RemoteInstalledPluginsAuthIdentity::from_auth(auth);
let mut cache = match self.remote_installed_plugins_cache.write() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
if cache.generation != generation || cache.auth_identity.as_ref() != Some(&auth_identity) {
return None;
}
cache.generation = cache.generation.wrapping_add(1);
cache.reconciliation_generation = Some(cache.generation);
Some(cache.generation)
}
fn abandon_remote_installed_plugins_reconcile(&self, generation: u64) {
let mut cache = match self.remote_installed_plugins_cache.write() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
if cache.reconciliation_generation != Some(generation) {
return;
}
cache.reconciliation_generation = None;
cache.generation = cache.generation.wrapping_add(1);
cache.needs_effective_plugins_refresh = true;
drop(cache);
self.clear_loaded_plugins_cache();
true
}
fn remote_installed_plugins_cache_generation_if_current(
&self,
auth: Option<&CodexAuth>,
) -> Option<u64> {
let auth_identity = RemoteInstalledPluginsAuthIdentity::from_auth(auth);
if !self.remote_installed_plugins_auth_is_current(&auth_identity) {
return None;
}
let cache = match self.remote_installed_plugins_cache.read() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
(cache.auth_identity.as_ref() == Some(&auth_identity)).then_some(cache.generation)
}
fn write_remote_installed_plugins_cache_snapshot(
&self,
generation: u64,
plugins: Vec<RemoteInstalledPlugin>,
auth: Option<&CodexAuth>,
publication: RemoteInstalledPluginsCachePublication,
) -> Option<bool> {
let auth_identity = RemoteInstalledPluginsAuthIdentity::from_auth(auth);
if !self.remote_installed_plugins_auth_is_current(&auth_identity) {
return None;
}
let mut cache = match self.remote_installed_plugins_cache.write() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
if cache.generation != generation
|| cache
.auth_identity
.as_ref()
.is_some_and(|current| current != &auth_identity)
{
return None;
}
let is_reconcile = matches!(
publication,
RemoteInstalledPluginsCachePublication::Reconcile
);
if !is_reconcile && cache.reconciliation_generation == Some(generation) {
return None;
}
if cache.auth_identity.is_none() {
cache.auth_identity = Some(auth_identity);
}
if is_reconcile {
cache.reconciliation_generation = None;
cache.generation = cache.generation.wrapping_add(1);
}
let needs_effective_plugins_refresh =
is_reconcile && std::mem::take(&mut cache.needs_effective_plugins_refresh);
if cache.plugins.as_ref() == Some(&plugins) {
drop(cache);
if needs_effective_plugins_refresh {
self.clear_loaded_plugins_cache();
}
return Some(needs_effective_plugins_refresh);
}
cache.plugins = Some(plugins);
drop(cache);
self.clear_loaded_plugins_cache();
Some(true)
}
#[cfg(test)]
fn write_remote_installed_plugins_cache(&self, plugins: Vec<RemoteInstalledPlugin>) -> bool {
let auth = self.auth_manager.auth_cached();
let generation = self
.prepare_remote_installed_plugins_cache_generation(auth.as_ref())
.expect("test cache identity should remain current");
self.write_remote_installed_plugins_cache_snapshot(
generation,
plugins,
auth.as_ref(),
RemoteInstalledPluginsCachePublication::Refresh,
)
.expect("test cache generation should remain current")
}
pub fn clear_remote_installed_plugins_cache(&self) -> bool {
@@ -1020,15 +1333,38 @@ impl PluginsManager {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
if cache.is_none() {
cache.generation = cache.generation.wrapping_add(1);
cache.auth_identity = None;
cache.reconciliation_generation = None;
cache.needs_effective_plugins_refresh = false;
if cache.plugins.take().is_none() {
return false;
}
*cache = None;
drop(cache);
self.clear_loaded_plugins_cache();
true
}
fn clear_remote_installed_plugins_cache_if_current(&self, generation: u64) -> Option<bool> {
let mut cache = match self.remote_installed_plugins_cache.write() {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
if cache.generation != generation {
return None;
}
cache.generation = cache.generation.wrapping_add(1);
cache.auth_identity = None;
cache.reconciliation_generation = None;
cache.needs_effective_plugins_refresh = false;
if cache.plugins.take().is_none() {
return Some(false);
}
drop(cache);
self.clear_loaded_plugins_cache();
Some(true)
}
pub fn maybe_start_remote_plugin_caches_refresh(
self: &Arc<Self>,
config: &PluginsConfigInput,
@@ -1079,8 +1415,14 @@ impl PluginsManager {
return;
}
let Some(generation) =
self.prepare_remote_installed_plugins_cache_generation(auth.as_ref())
else {
return;
};
self.schedule_remote_installed_plugins_cache_refresh(
RemoteInstalledPluginsCacheRefreshRequest {
generation,
service_config: remote_plugin_service_config(config),
auth,
notify,
@@ -1109,34 +1451,27 @@ impl PluginsManager {
return;
};
let manager = Arc::clone(self);
let config_for_refresh = config.clone();
let service_config = remote_plugin_service_config(config);
let config = config.clone();
tokio::spawn(async move {
let _permit = permit;
let result = crate::remote::sync_remote_installed_plugin_bundles_once(
manager.codex_home.clone(),
&service_config,
Some(&auth),
)
.await;
let result = manager
.reconcile_remote_installed_plugins_after_acquiring_gate(&config, Some(&auth))
.await;
match result {
Ok(outcome) => {
Ok((outcome, effective_plugins_changed)) => {
tracing::info!(
materialized_remote_plugins = ?outcome.materialized_remote_plugins,
removed_cache_plugin_ids = ?outcome.removed_cache_plugin_ids,
failed_remote_plugin_ids = ?outcome.failed_remote_plugin_ids,
failed_materialization_remote_plugin_ids = ?outcome.failed_materialization_remote_plugin_ids,
"completed remote installed plugin bundle sync"
);
if outcome.changed_local_cache() {
manager.maybe_start_remote_installed_plugins_cache_refresh_with_notify(
&config_for_refresh,
Some(auth),
RemoteInstalledPluginsCacheRefreshNotify::AfterSuccessfulRefresh,
on_effective_plugins_changed,
EffectivePluginsChange {
materialized_remote_plugins: outcome.materialized_remote_plugins,
},
);
if (effective_plugins_changed || outcome.changed_local_cache())
&& let Some(on_effective_plugins_changed) = on_effective_plugins_changed
{
on_effective_plugins_changed(EffectivePluginsChange {
materialized_remote_plugins: outcome.materialized_remote_plugins,
});
}
}
Err(err) => {
@@ -1144,11 +1479,86 @@ impl PluginsManager {
error = %err,
"remote installed plugin bundle sync failed"
);
manager.maybe_start_remote_installed_plugins_cache_refresh_with_notify(
&config,
Some(auth),
RemoteInstalledPluginsCacheRefreshNotify::IfCacheChanged,
on_effective_plugins_changed,
EffectivePluginsChange::default(),
);
}
}
});
}
/// Acquires the cache-root gate shared by full installed-bundle sync, reconciliation, and
/// direct remote plugin mutations.
pub async fn acquire_remote_installed_plugin_sync_guard(
&self,
) -> Result<RemoteInstalledPluginSyncGuard, RemoteInstalledPluginBundleSyncError> {
let permit = tokio::time::timeout(
REMOTE_INSTALLED_PLUGIN_SYNC_WAIT_TIMEOUT,
Arc::clone(&self.remote_installed_plugin_bundle_sync_gate).acquire_owned(),
)
.await
.map_err(|_| RemoteInstalledPluginBundleSyncError::LockTimeout)?
.map_err(|_| RemoteInstalledPluginBundleSyncError::Superseded)?;
Ok(RemoteInstalledPluginSyncGuard { _permit: permit })
}
/// Synchronizes remote bundles and publishes the same installed-plugin snapshot.
///
/// The shared gate orders this blocking path with background bundle synchronization. A
/// reconciled publication advances the cache generation so metadata fetches that started
/// before or during the bundle pass cannot overwrite its snapshot afterward.
pub async fn reconcile_remote_installed_plugins(
&self,
config: &PluginsConfigInput,
auth: Option<&CodexAuth>,
) -> Result<RemoteInstalledPluginBundleSyncOutcome, RemoteInstalledPluginBundleSyncError> {
let _guard = self.acquire_remote_installed_plugin_sync_guard().await?;
let (outcome, _) = self
.reconcile_remote_installed_plugins_after_acquiring_gate(config, auth)
.await?;
Ok(outcome)
}
async fn reconcile_remote_installed_plugins_after_acquiring_gate(
&self,
config: &PluginsConfigInput,
auth: Option<&CodexAuth>,
) -> Result<(RemoteInstalledPluginBundleSyncOutcome, bool), RemoteInstalledPluginBundleSyncError>
{
let generation = self
.begin_remote_installed_plugins_reconcile(auth)
.ok_or(RemoteInstalledPluginBundleSyncError::Superseded)?;
let mut reconciliation = RemoteInstalledPluginsReconciliationGuard {
manager: self,
generation,
committed: false,
};
let result = crate::remote::sync_remote_installed_plugin_bundles_once_with_snapshot(
self.codex_home.clone(),
&remote_plugin_service_config(config),
auth,
)
.await?;
let outcome = result.outcome;
let Some(effective_plugins_changed) = self.write_remote_installed_plugins_cache_snapshot(
generation,
result.installed_plugins,
auth,
RemoteInstalledPluginsCachePublication::Reconcile,
) else {
return Err(RemoteInstalledPluginBundleSyncError::Superseded);
};
reconciliation.committed = true;
if !effective_plugins_changed && outcome.changed_local_cache() {
self.clear_loaded_plugins_cache();
}
Ok((outcome, effective_plugins_changed))
}
fn maybe_start_remote_catalog_cache_refresh(
self: &Arc<Self>,
config: &PluginsConfigInput,
@@ -1372,9 +1782,12 @@ impl PluginsManager {
Ok(cache) => cache,
Err(err) => err.into_inner(),
};
cache
.as_deref()
.unwrap_or_default()
let plugins = self
.remote_installed_plugins_cache_matches_current_auth(&cache)
.then_some(cache.plugins.as_deref())
.flatten()
.unwrap_or_default();
plugins
.iter()
.filter(|plugin| plugin.marketplace_name == REMOTE_GLOBAL_MARKETPLACE_NAME)
.map(|plugin| plugin.id.clone())
@@ -2379,12 +2792,24 @@ impl PluginsManager {
self: &Arc<Self>,
mut request: RemoteInstalledPluginsCacheRefreshRequest,
) {
if self.remote_installed_plugins_cache_generation_if_current(request.auth.as_ref())
!= Some(request.generation)
{
return;
}
let should_spawn = {
let mut state = match self.remote_installed_plugins_cache_refresh_state.write() {
Ok(state) => state,
Err(err) => err.into_inner(),
};
if let Some(existing_request) = state.requested.as_ref() {
if self.remote_installed_plugins_cache_generation_if_current(request.auth.as_ref())
!= Some(request.generation)
{
return;
}
if let Some(existing_request) = state.requested.as_ref()
&& existing_request.generation == request.generation
{
if matches!(
existing_request.notify,
RemoteInstalledPluginsCacheRefreshNotify::AfterSuccessfulRefresh
@@ -2725,16 +3150,21 @@ impl PluginsManager {
}
};
let installed_plugins = crate::remote::fetch_remote_installed_plugins(
let plugins = crate::remote::fetch_remote_installed_plugins(
&request.service_config,
request.auth.as_ref(),
)
.await;
match installed_plugins {
Ok(installed_plugins) => {
// 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);
match plugins {
Ok(plugins) => {
let Some(changed) = self.write_remote_installed_plugins_cache_snapshot(
request.generation,
plugins,
request.auth.as_ref(),
RemoteInstalledPluginsCachePublication::Refresh,
) else {
continue;
};
let should_notify = changed
|| !request.change.materialized_remote_plugins.is_empty()
|| matches!(
@@ -2752,7 +3182,11 @@ impl PluginsManager {
RemotePluginCatalogError::AuthRequired
| RemotePluginCatalogError::UnsupportedAuthMode,
) => {
let changed = self.clear_remote_installed_plugins_cache();
let Some(changed) =
self.clear_remote_installed_plugins_cache_if_current(request.generation)
else {
continue;
};
if changed
&& let Some(on_effective_plugins_changed) =
request.on_effective_plugins_changed
codex-rs/core-plugins/src/manager_tests.rs
+254
View File
@@ -20,6 +20,7 @@ use crate::startup_sync::curated_plugins_repo_path;
use crate::test_support::TEST_CURATED_PLUGIN_CACHE_VERSION;
use crate::test_support::TEST_CURATED_PLUGIN_SHA;
use crate::test_support::load_plugins_config as load_plugins_config_input;
use crate::test_support::set_test_auth;
use crate::test_support::set_test_auth_mode;
use crate::test_support::test_auth_manager;
use crate::test_support::test_http_client_factory;
@@ -3062,6 +3063,7 @@ fn loaded_plugins_cache_invalidation_rejects_stale_load_completion() {
configured_plugins: HashMap::new(),
skill_config_rules: SkillConfigRules::default(),
remote_global_catalog_active: false,
auth_identity: None,
};
let stale_generation = manager.loaded_plugins_cache_generation();
@@ -3076,6 +3078,60 @@ fn loaded_plugins_cache_invalidation_rejects_stale_load_completion() {
assert_eq!(manager.cached_loaded_plugins(&cache_key), None);
}
#[tokio::test]
async fn plugins_for_config_discards_in_flight_load_after_account_change() {
let codex_home = TempDir::new().unwrap();
write_cached_plugin(
codex_home.path(),
REMOTE_GLOBAL_MARKETPLACE_NAME,
"remote-only",
);
let config = PluginsConfigInput::new(
unrestricted_config_layer_stack(),
String::new(),
/*plugins_enabled*/ true,
/*remote_plugin_enabled*/ true,
String::new(),
test_http_client_factory(),
);
let auth_manager = test_auth_manager(Some(AuthMode::ChatgptAuthTokens));
let manager = Arc::new(test_plugins_manager_with_auth_manager(
codex_home.path().to_path_buf(),
Some(Product::Codex),
Arc::clone(&auth_manager),
));
manager.write_remote_installed_plugins_cache(vec![remote_installed_plugin("remote-only")]);
let account_a = auth_manager
.auth_cached()
.expect("capture first ChatGPT auth");
let load_permit = manager
.loaded_plugins_load_semaphore
.try_acquire()
.expect("hold plugin load semaphore");
let load = manager.plugins_for_config(&config);
tokio::pin!(load);
assert!(matches!(
futures::poll!(load.as_mut()),
std::task::Poll::Pending
));
let account_b = CodexAuth::from_external_chatgpt_tokens(
"header.e30.other",
"other-account",
/*chatgpt_plan_type*/ None,
)
.expect("build second ChatGPT auth");
set_test_auth(&auth_manager, Some(account_b)).await;
drop(load_permit);
assert_eq!(load.await, PluginLoadOutcome::default());
set_test_auth(&auth_manager, Some(account_a)).await;
assert_eq!(
loaded_plugin_names(&manager, &config).await,
vec!["remote-only@openai-curated-remote".to_string()]
);
}
#[tokio::test]
async fn load_plugins_rejects_invalid_plugin_keys() {
let codex_home = TempDir::new().unwrap();
@@ -6830,8 +6886,12 @@ fn remote_installed_plugins_cache_refresh_coalesces_materializations() {
});
callback
};
let generation = manager
.prepare_remote_installed_plugins_cache_generation(/*auth*/ None)
.expect("prepare remote installed cache generation");
let request =
|change, on_effective_plugins_changed| RemoteInstalledPluginsCacheRefreshRequest {
generation,
service_config: RemotePluginServiceConfig::new(
"https://example.com".to_string(),
test_http_client_factory(),
@@ -6931,6 +6991,200 @@ remote_plugin = true
server.verify().await;
}
#[tokio::test]
async fn reconcile_remote_installed_plugins_rejects_incomplete_snapshot_without_cleanup() {
let codex_home = TempDir::new().unwrap();
write_file(
&codex_home.path().join(CONFIG_TOML_FILE),
r#"[features]
plugins = true
remote_plugin = true
"#,
);
write_cached_plugin(codex_home.path(), REMOTE_GLOBAL_MARKETPLACE_NAME, "linear");
write_cached_plugin(
codex_home.path(),
REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME,
"malformed",
);
write_cached_plugin(
codex_home.path(),
REMOTE_WORKSPACE_MARKETPLACE_NAME,
"renamed-malformed",
);
let malformed_remote_plugin_id = "plugins~Plugin_malformed";
let malformed_plugin_id = PluginId::new(
"malformed".to_string(),
REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME.to_string(),
)
.expect("valid malformed plugin id");
PluginStore::new(codex_home.path().to_path_buf())
.write_remote_plugin_id(&malformed_plugin_id, malformed_remote_plugin_id)
.expect("persist malformed plugin identity");
let server = MockServer::start().await;
Mock::given(method("GET"))
.and(path("/backend-api/ps/plugins/installed"))
.and(query_param("includeDownloadUrls", "true"))
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({
"plugins": [
{
"id": "plugins~Plugin_linear",
"name": "linear",
"scope": "GLOBAL",
"installation_policy": "AVAILABLE",
"authentication_policy": "ON_USE",
"status": "ENABLED",
"release": {
"version": "local",
"display_name": "",
"description": "Track work",
"interface": {},
},
"enabled": true,
},
{
"id": malformed_remote_plugin_id,
"name": "renamed-malformed",
"scope": "WORKSPACE",
"installation_policy": "AVAILABLE",
"authentication_policy": "ON_USE",
"status": "ENABLED",
"release": {
"version": "local",
"display_name": "Malformed",
"description": "Missing discoverability",
"interface": {},
},
"enabled": true,
},
],
"pagination": {"next_page_token": null},
})))
.expect(1)
.mount(&server)
.await;
let mut config = load_config(codex_home.path(), codex_home.path()).await;
config.chatgpt_base_url = format!("{}/backend-api", server.uri());
let auth_manager = test_auth_manager(Some(AuthMode::Chatgpt));
let auth = auth_manager.auth_cached().expect("test ChatGPT auth");
let manager = std::sync::Arc::new(test_plugins_manager_with_auth_manager(
codex_home.path().to_path_buf(),
Some(Product::Codex),
auth_manager,
));
let mut previous_malformed = remote_installed_plugin_in_marketplace(
"malformed",
REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME,
);
previous_malformed.id = malformed_remote_plugin_id.to_string();
previous_malformed.version = Some("local".to_string());
previous_malformed.enabled = false;
manager.write_remote_installed_plugins_cache(vec![previous_malformed.clone()]);
let result = manager
.reconcile_remote_installed_plugins(&config, Some(&auth))
.await;
assert!(matches!(
result,
Err(RemoteInstalledPluginBundleSyncError::Catalog(
RemotePluginCatalogError::UnexpectedResponse(_)
))
));
server.verify().await;
assert_eq!(
manager
.remote_installed_plugins_cache
.read()
.expect("installed plugin cache lock")
.plugins
.clone(),
Some(vec![previous_malformed])
);
let plugin_cache_root = codex_home.path().join("plugins/cache");
let renamed_malformed_cache = plugin_cache_root
.join(REMOTE_WORKSPACE_MARKETPLACE_NAME)
.join("renamed-malformed");
let previous_malformed_cache = plugin_cache_root
.join(REMOTE_WORKSPACE_SHARED_WITH_ME_MARKETPLACE_NAME)
.join("malformed");
let linear_metadata = plugin_cache_root
.join(REMOTE_GLOBAL_MARKETPLACE_NAME)
.join("linear")
.join(".codex-remote-plugin-install.json");
assert!(renamed_malformed_cache.exists());
assert!(previous_malformed_cache.exists());
assert!(!linear_metadata.exists());
}
#[test]
fn reconciliation_fences_refresh_publication_and_recovers_after_cancellation() {
let codex_home = TempDir::new().unwrap();
let manager = test_plugins_manager(codex_home.path().to_path_buf());
let reconcile_generation = manager
.begin_remote_installed_plugins_reconcile(/*auth*/ None)
.expect("begin reconciliation");
let stale_refresh_generation = manager
.remote_installed_plugins_cache_generation_if_current(/*auth*/ None)
.expect("capture concurrent refresh generation");
assert_eq!(
manager.write_remote_installed_plugins_cache_snapshot(
stale_refresh_generation,
Vec::new(),
/*auth*/ None,
RemoteInstalledPluginsCachePublication::Refresh,
),
None
);
assert_eq!(
manager.write_remote_installed_plugins_cache_snapshot(
reconcile_generation,
vec![remote_installed_linear_plugin()],
/*auth*/ None,
RemoteInstalledPluginsCachePublication::Reconcile,
),
Some(true)
);
let cancelled_generation = manager
.begin_remote_installed_plugins_reconcile(/*auth*/ None)
.expect("begin cancelled reconciliation");
let reconciliation = RemoteInstalledPluginsReconciliationGuard {
manager: &manager,
generation: cancelled_generation,
committed: false,
};
drop(reconciliation);
let refresh_generation = manager
.remote_installed_plugins_cache_generation_if_current(/*auth*/ None)
.expect("capture refresh generation after cancellation");
assert_eq!(
manager.write_remote_installed_plugins_cache_snapshot(
refresh_generation,
vec![remote_installed_plugin("beta")],
/*auth*/ None,
RemoteInstalledPluginsCachePublication::Refresh,
),
Some(true)
);
let retry_generation = manager
.begin_remote_installed_plugins_reconcile(/*auth*/ None)
.expect("begin retry after cancellation");
assert_eq!(
manager.write_remote_installed_plugins_cache_snapshot(
retry_generation,
vec![remote_installed_plugin("beta")],
/*auth*/ None,
RemoteInstalledPluginsCachePublication::Reconcile,
),
Some(true)
);
}
#[test]
fn plugin_install_error_preserves_store_io_sub_error_type() {
let error = PluginInstallError::Store(PluginStoreError::Io {
codex-rs/core-plugins/src/remote.rs
+9 -1
View File
@@ -62,6 +62,7 @@ pub use remote_installed_plugin_sync::RemotePluginMaterialization;
pub use remote_installed_plugin_sync::mark_remote_plugin_cache_mutation_in_flight;
pub(crate) use remote_installed_plugin_sync::remote_installed_plugin_bundle_sync_gate;
pub use remote_installed_plugin_sync::sync_remote_installed_plugin_bundles_once;
pub(crate) use remote_installed_plugin_sync::sync_remote_installed_plugin_bundles_once_with_snapshot;
pub use search::RemotePluginSearchPage;
pub use search::RemotePluginSearchRequest;
pub use search::search_remote_plugins;
@@ -1815,11 +1816,18 @@ fn remote_installed_plugin_to_cache_entry(
installed_plugin: &RemotePluginInstalledItem,
) -> Result<RemoteInstalledPlugin, RemotePluginCatalogError> {
let plugin = &installed_plugin.plugin;
let marketplace_name = remote_plugin_canonical_marketplace_name(plugin)?.to_string();
PluginId::new(plugin.name.clone(), marketplace_name.clone()).map_err(|err| {
RemotePluginCatalogError::UnexpectedResponse(format!(
"invalid remote plugin config id for `{}` in `{marketplace_name}`: {err}",
plugin.name
))
})?;
// Remote per-skill disabled state (`disabled_skill_names`) is intentionally
// not projected into skills/list yet; local skills.config remains the
// supported source for skill enablement.
Ok(RemoteInstalledPlugin {
marketplace_name: remote_plugin_canonical_marketplace_name(plugin)?.to_string(),
marketplace_name,
id: plugin.id.clone(),
version: plugin.release.version.clone(),
name: plugin.name.clone(),
codex-rs/core-plugins/src/remote/remote_installed_plugin_sync.rs
+150 -57
View File
@@ -4,6 +4,7 @@ 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::RemoteInstalledPlugin;
use super::RemoteInstalledPluginScope;
use super::RemotePluginCatalogError;
use super::RemotePluginScope;
@@ -11,7 +12,6 @@ use super::RemotePluginServiceConfig;
use super::RemotePluginShareDiscoverability;
use super::ensure_chatgpt_auth;
use super::fetch_installed_plugins;
use super::remote_plugin_canonical_marketplace_name;
use crate::store::PLUGINS_CACHE_DIR;
use crate::store::PluginStore;
use crate::store::PluginStoreError;
@@ -51,6 +51,13 @@ pub struct RemoteInstalledPluginBundleSyncOutcome {
pub materialized_remote_plugins: Vec<RemotePluginMaterialization>,
pub removed_cache_plugin_ids: Vec<String>,
pub failed_remote_plugin_ids: Vec<String>,
/// Failures that leave an otherwise valid installed plugin unavailable locally.
pub failed_materialization_remote_plugin_ids: Vec<String>,
}
pub(crate) struct RemoteInstalledPluginBundleSyncResult {
pub(crate) outcome: RemoteInstalledPluginBundleSyncOutcome,
pub(crate) installed_plugins: Vec<RemoteInstalledPlugin>,
}
impl RemoteInstalledPluginBundleSyncOutcome {
@@ -66,6 +73,12 @@ pub enum RemoteInstalledPluginBundleSyncError {
#[error("{0}")]
Store(#[from] PluginStoreError),
#[error("timed out waiting for another remote plugin cache mutation; retry")]
LockTimeout,
#[error("remote plugin state changed during reconciliation; retry reconciliation")]
Superseded,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -100,15 +113,44 @@ pub async fn sync_remote_installed_plugin_bundles_once(
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
) -> Result<RemoteInstalledPluginBundleSyncOutcome, RemoteInstalledPluginBundleSyncError> {
let result =
sync_remote_installed_plugin_bundles_once_with_snapshot(codex_home, config, auth).await?;
Ok(result.outcome)
}
pub(crate) async fn sync_remote_installed_plugin_bundles_once_with_snapshot(
codex_home: PathBuf,
config: &RemotePluginServiceConfig,
auth: Option<&CodexAuth>,
) -> Result<RemoteInstalledPluginBundleSyncResult, RemoteInstalledPluginBundleSyncError> {
let auth = ensure_chatgpt_auth(auth)?;
let authenticated_account_id = auth.get_account_id();
let installed_plugins = fetch_installed_plugins(
let fetched_installed_plugins = fetch_installed_plugins(
config,
auth,
RemoteInstalledPluginScope::All,
/*include_download_urls*/ true,
)
.await?;
// `/installed` is an authoritative full snapshot. If any row cannot be
// canonicalized to a local cache key, omitting it makes an installed plugin
// indistinguishable from an uninstall, so reject the pass before downloads,
// publication, or stale cleanup.
let mut validated_installed_plugins = Vec::with_capacity(fetched_installed_plugins.len());
for installed_plugin in fetched_installed_plugins {
let cached_plugin = super::remote_installed_plugin_to_cache_entry(&installed_plugin)?;
let plugin_id = PluginId::new(
installed_plugin.plugin.name.clone(),
cached_plugin.marketplace_name.clone(),
)
.map_err(|err| {
RemotePluginCatalogError::UnexpectedResponse(format!(
"remote installed plugin `{}` has an invalid local cache id: {err}",
installed_plugin.plugin.id
))
})?;
validated_installed_plugins.push((installed_plugin, cached_plugin, plugin_id));
}
let store = PluginStore::try_new(codex_home.clone())?;
let mut installed_plugin_names_by_marketplace =
BTreeMap::<String, BTreeSet<String>>::from_iter([
@@ -136,37 +178,27 @@ pub async fn sync_remote_installed_plugin_bundles_once(
]);
let mut materialized_remote_plugins = BTreeMap::new();
let mut failed_remote_plugin_ids = BTreeSet::new();
let mut failed_materialization_remote_plugin_ids = BTreeSet::new();
let mut installed_plugins = Vec::new();
for installed_plugin in installed_plugins {
for (installed_plugin, cached_plugin, plugin_id) in validated_installed_plugins {
let marketplace_name = cached_plugin.marketplace_name.clone();
let plugin = installed_plugin.plugin;
let scope = plugin.scope;
let discoverability = plugin.discoverability;
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 {
if let Some(release_version) = release_version
&& store.active_plugin_version(&plugin_id).as_deref() == Some(release_version)
{
if let Err(err) = store.write_remote_plugin_id(&plugin_id, &plugin.id) {
warn!(
remote_plugin_id = %plugin.id,
@@ -177,6 +209,7 @@ pub async fn sync_remote_installed_plugin_bundles_once(
);
failed_remote_plugin_ids.insert(plugin.id);
}
installed_plugins.push(cached_plugin);
continue;
}
@@ -197,7 +230,9 @@ pub async fn sync_remote_installed_plugin_bundles_once(
error = %err,
"skipping remote installed plugin bundle download"
);
failed_remote_plugin_ids.insert(plugin.id);
failed_remote_plugin_ids.insert(plugin.id.clone());
failed_materialization_remote_plugin_ids.insert(plugin.id);
installed_plugins.push(cached_plugin);
continue;
}
};
@@ -229,34 +264,51 @@ pub async fn sync_remote_installed_plugin_bundles_once(
error = %err,
"failed to download remote installed plugin bundle"
);
failed_remote_plugin_ids.insert(plugin.id);
failed_remote_plugin_ids.insert(plugin.id.clone());
failed_materialization_remote_plugin_ids.insert(plugin.id);
}
}
installed_plugins.push(cached_plugin);
}
installed_plugins.sort_by(|left, right| {
left.marketplace_name
.cmp(&right.marketplace_name)
.then_with(|| left.id.cmp(&right.id))
});
let stale_cache_cleanup = tokio::task::spawn_blocking(move || {
remove_stale_remote_plugin_caches(
let mut removed_cache_plugin_ids = Vec::new();
let cleanup_error = remove_stale_remote_plugin_caches(
codex_home.as_path(),
&installed_plugin_names_by_marketplace,
&mut removed_cache_plugin_ids,
)
.err();
removed_cache_plugin_ids.sort();
(removed_cache_plugin_ids, cleanup_error)
})
.await;
let removed_cache_plugin_ids = match stale_cache_cleanup {
Ok(Ok(removed_cache_plugin_ids)) => removed_cache_plugin_ids,
Ok(Err(err)) => {
warn!(error = %err, "failed to remove stale remote plugin cache entries");
Vec::new()
}
let (removed_cache_plugin_ids, cleanup_error) = match stale_cache_cleanup {
Ok(result) => result,
Err(err) => {
warn!(error = %err, "failed to join stale remote plugin cache cleanup task");
Vec::new()
(Vec::new(), Some(err.to_string()))
}
};
if let Some(err) = cleanup_error {
warn!(error = %err, "failed to remove stale remote plugin cache entries");
}
Ok(RemoteInstalledPluginBundleSyncOutcome {
materialized_remote_plugins: materialized_remote_plugins.into_values().collect(),
removed_cache_plugin_ids,
failed_remote_plugin_ids: failed_remote_plugin_ids.into_iter().collect(),
Ok(RemoteInstalledPluginBundleSyncResult {
outcome: RemoteInstalledPluginBundleSyncOutcome {
materialized_remote_plugins: materialized_remote_plugins.into_values().collect(),
removed_cache_plugin_ids,
failed_remote_plugin_ids: failed_remote_plugin_ids.into_iter().collect(),
failed_materialization_remote_plugin_ids: failed_materialization_remote_plugin_ids
.into_iter()
.collect(),
},
installed_plugins,
})
}
@@ -301,8 +353,8 @@ impl Drop for RemotePluginCacheMutationGuard {
fn remove_stale_remote_plugin_caches(
codex_home: &Path,
installed_plugin_names_by_marketplace: &BTreeMap<String, BTreeSet<String>>,
) -> Result<Vec<String>, String> {
let mut removed_cache_plugin_ids = Vec::new();
removed_cache_plugin_ids: &mut Vec<String>,
) -> Result<(), String> {
for marketplace_name in [
REMOTE_GLOBAL_MARKETPLACE_NAME,
REMOTE_CREATED_BY_ME_MARKETPLACE_NAME,
@@ -369,8 +421,7 @@ fn remove_stale_remote_plugin_caches(
}
}
removed_cache_plugin_ids.sort();
Ok(removed_cache_plugin_ids)
Ok(())
}
fn remote_plugin_cache_root(codex_home: &Path) -> PathBuf {
@@ -410,7 +461,7 @@ mod tests {
use wiremock::matchers::query_param_is_missing;
#[tokio::test]
async fn sync_same_version_backfills_metadata_without_materialization() {
async fn sync_same_version_backfills_metadata_and_missing_version_requires_materialization() {
let server = MockServer::start().await;
let codex_home = tempfile::tempdir().expect("create codex home");
let cached_manifest = codex_home
@@ -432,21 +483,37 @@ mod tests {
.and(query_param("limit", "200"))
.and(query_param("includeDownloadUrls", "true"))
.respond_with(ResponseTemplate::new(200).set_body_json(json!({
"plugins": [{
"id": remote_plugin_id,
"name": "linear",
"scope": "GLOBAL",
"installation_policy": "AVAILABLE",
"authentication_policy": "ON_USE",
"status": "ENABLED",
"release": {
"version": "1.2.3",
"display_name": "Linear",
"description": "Track work",
"interface": {},
"plugins": [
{
"id": remote_plugin_id,
"name": "linear",
"scope": "GLOBAL",
"installation_policy": "AVAILABLE",
"authentication_policy": "ON_USE",
"status": "ENABLED",
"release": {
"version": "1.2.3",
"display_name": "Linear",
"description": "Track work",
"interface": {},
},
"enabled": true,
},
"enabled": true,
}],
{
"id": "plugins~Plugin_missing_version",
"name": "missing-version",
"scope": "GLOBAL",
"installation_policy": "AVAILABLE",
"authentication_policy": "ON_USE",
"status": "ENABLED",
"release": {
"display_name": "Missing version",
"description": "Needs materialization",
"interface": {},
},
"enabled": true,
},
],
"pagination": {"next_page_token": null},
})))
.expect(1)
@@ -466,7 +533,16 @@ mod tests {
.await
.expect("sync current remote plugin bundle");
assert_eq!(outcome, RemoteInstalledPluginBundleSyncOutcome::default());
assert_eq!(
outcome,
RemoteInstalledPluginBundleSyncOutcome {
failed_remote_plugin_ids: vec!["plugins~Plugin_missing_version".to_string()],
failed_materialization_remote_plugin_ids: vec![
"plugins~Plugin_missing_version".to_string()
],
..RemoteInstalledPluginBundleSyncOutcome::default()
}
);
let plugin_id = PluginId::new(
"linear".to_string(),
REMOTE_GLOBAL_MARKETPLACE_NAME.to_string(),
@@ -486,6 +562,14 @@ mod tests {
"remote_plugin_id": remote_plugin_id,
})
);
assert!(
!codex_home
.path()
.join(PLUGINS_CACHE_DIR)
.join(REMOTE_GLOBAL_MARKETPLACE_NAME)
.join("missing-version")
.exists()
);
}
#[tokio::test]
@@ -611,6 +695,7 @@ mod tests {
materialized_remote_plugins: Vec::new(),
removed_cache_plugin_ids,
failed_remote_plugin_ids: Vec::new(),
failed_materialization_remote_plugin_ids: Vec::new(),
}
);
assert_eq!(
@@ -701,27 +786,33 @@ mod tests {
REMOTE_GLOBAL_MARKETPLACE_NAME,
"linear",
);
let removed = remove_stale_remote_plugin_caches(
let mut removed = Vec::new();
remove_stale_remote_plugin_caches(
codex_home.path(),
&installed_plugin_names_by_marketplace,
&mut removed,
)
.expect("cleanup while install is guarded");
assert_eq!(removed, Vec::<String>::new());
assert!(cached_manifest.is_file());
drop(guard);
let removed = remove_stale_remote_plugin_caches(
let mut removed = Vec::new();
remove_stale_remote_plugin_caches(
codex_home.path(),
&installed_plugin_names_by_marketplace,
&mut removed,
)
.expect("cleanup while second install guard is still active");
assert_eq!(removed, Vec::<String>::new());
assert!(cached_manifest.is_file());
drop(second_guard);
let removed = remove_stale_remote_plugin_caches(
let mut removed = Vec::new();
remove_stale_remote_plugin_caches(
codex_home.path(),
&installed_plugin_names_by_marketplace,
&mut removed,
)
.expect("cleanup after install guard is dropped");
assert_eq!(removed, vec!["linear@openai-curated-remote".to_string()]);
@@ -799,9 +890,11 @@ mod tests {
),
]);
let removed = remove_stale_remote_plugin_caches(
let mut removed = Vec::new();
remove_stale_remote_plugin_caches(
codex_home.path(),
&installed_plugin_names_by_marketplace,
&mut removed,
)
.expect("cleanup private shared-with-me cache");
codex-rs/core-plugins/src/test_support.rs
+5 -1
View File
@@ -89,7 +89,11 @@ pub(crate) fn test_auth_manager(auth_mode: Option<AuthMode>) -> Arc<AuthManager>
}
pub(crate) async fn set_test_auth_mode(auth_manager: &AuthManager, auth_mode: Option<AuthMode>) {
let Some(auth) = test_codex_auth(auth_mode) else {
set_test_auth(auth_manager, test_codex_auth(auth_mode)).await;
}
pub(crate) async fn set_test_auth(auth_manager: &AuthManager, auth: Option<CodexAuth>) {
let Some(auth) = auth else {
auth_manager.clear_external_auth();
return;
};