diff --git a/codex-rs/app-server/tests/suite/v2/marketplace_upgrade.rs b/codex-rs/app-server/tests/suite/v2/marketplace_upgrade.rs index d38d2c9a8e..066c89de61 100644 --- a/codex-rs/app-server/tests/suite/v2/marketplace_upgrade.rs +++ b/codex-rs/app-server/tests/suite/v2/marketplace_upgrade.rs @@ -17,6 +17,7 @@ use codex_app_server_protocol::RequestId; use codex_config::MarketplaceConfigUpdate; use codex_config::record_user_marketplace; use codex_utils_absolute_path::AbsolutePathBuf; +use core_test_support::fs_wait::wait_for_path_exists; use pretty_assertions::assert_eq; use tempfile::TempDir; use test_case::test_case; @@ -51,10 +52,15 @@ fn write_marketplace_files(root: &Path, marketplace_name: &str, marker: &str) -> Ok(()) } -fn init_marketplace_repo(root: &Path, marketplace_name: &str, marker: &str) -> Result { +fn init_git_repo(root: &Path) -> Result<()> { run_git(root, &["init"])?; run_git(root, &["config", "user.email", "codex@example.com"])?; run_git(root, &["config", "user.name", "Codex Tests"])?; + Ok(()) +} + +fn init_marketplace_repo(root: &Path, marketplace_name: &str, marker: &str) -> Result { + init_git_repo(root)?; write_marketplace_files(root, marketplace_name, marker)?; run_git(root, &["add", "."])?; run_git(root, &["commit", "-m", "initial marketplace"])?; @@ -205,6 +211,108 @@ async fn marketplace_upgrade_all_configured_git_marketplaces() -> Result<()> { Ok(()) } +fn write_plugin_release(root: &Path, name: &str, version: &str, instructions: &str) -> Result<()> { + std::fs::create_dir_all(root.join(".codex-plugin"))?; + std::fs::create_dir_all(root.join("skills/deploy"))?; + let manifest = serde_json::json!({ "name": name, "version": version }); + std::fs::write(root.join(".codex-plugin/plugin.json"), manifest.to_string())?; + std::fs::write(root.join("skills/deploy/SKILL.md"), instructions)?; + Ok(()) +} + +fn publish_stable(repo: &Path) -> Result<()> { + // Git runs shell aliases with its own shell, including on Windows. + let publish = "alias.publish-stable=!set -e +git add . +git commit -qm release +git tag -f -a stable -m release"; + run_git(repo, &["-c", publish, "publish-stable"])?; + Ok(()) +} + +async fn list_local_plugins(server: &mut TestAppServer, force_refetch: bool) -> Result<()> { + let _: PluginListResponse = server + .request(|request_id| ClientRequest::PluginList { + request_id, + params: PluginListParams { + cwds: None, + marketplace_kinds: Some(vec![PluginListMarketplaceKind::Local]), + force_refetch, + }, + }) + .await?; + Ok(()) +} + +#[tokio::test] +async fn first_full_plugin_list_installs_only_new_plugin_versions() -> Result<()> { + let home = TempDir::new()?; + let catalog = TempDir::new()?; + let repo = TempDir::new()?; + init_git_repo(repo.path())?; + let repo_url = url::Url::from_directory_path(repo.path()).expect("absolute plugin repo path"); + let cache = home.path().join("plugins/cache/team"); + let plugins = ["docs", "release-guide"]; + for name in plugins { + write_plugin_release(&repo.path().join(name), name, "1.0.0", "production")?; + write_plugin_release(&cache.join(name).join("1.0.0"), name, "1.0.0", "production")?; + } + publish_stable(repo.path())?; + let entries = plugins.map(|name| serde_json::json!({ + "name": name, + "source": { "source": "git-subdir", "url": repo_url.as_str(), "path": name, "ref": "stable" } + })); + std::fs::create_dir_all(catalog.path().join(".agents/plugins"))?; + std::fs::write( + catalog.path().join(".agents/plugins/marketplace.json"), + serde_json::json!({ "name": "team", "plugins": entries }).to_string(), + )?; + std::fs::write( + home.path().join("config.toml"), + r#"[features] +plugins = true +[plugins."docs@team"] +enabled = true +[plugins."release-guide@team"] +enabled = true +"#, + )?; + record_user_marketplace( + home.path(), + "team", + &configured_local_marketplace_update( + catalog.path().to_str().context("UTF-8 catalog path")?, + ), + )?; + let mut server = TestAppServer::builder() + .with_codex_home(home.path()) + .build_initialized() + .await?; + + write_plugin_release(&repo.path().join("docs"), "docs", "1.0.0", "staging")?; + write_plugin_release( + &repo.path().join("release-guide"), + "release-guide", + "1.1.0", + "staging", + )?; + publish_stable(repo.path())?; + list_local_plugins(&mut server, /*force_refetch*/ false).await?; + // The worker processes docs before release-guide. Seeing the new release means it checked both. + let skill = "skills/deploy/SKILL.md"; + let updated = wait_for_path_exists( + cache.join("release-guide/1.1.0").join(skill), + DEFAULT_TIMEOUT, + ) + .await?; + assert_eq!(std::fs::read_to_string(updated)?, "staging"); + assert_eq!( + std::fs::read_to_string(cache.join("docs/1.0.0").join(skill))?, + "production" + ); + Ok(()) +} + #[test_case(false; "user config")] #[test_case(true; "invalid user config fallback")] #[tokio::test] @@ -349,16 +457,7 @@ async fn automatic_upgrade_isolates_git_while_explicit_install_preserves_configu )?; run_git(plugin_source.path(), &["add", "."])?; run_git(plugin_source.path(), &["commit", "-qm", "upgrade plugin"])?; - let _: PluginListResponse = server - .request(|request_id| ClientRequest::PluginList { - request_id, - params: PluginListParams { - cwds: None, - marketplace_kinds: Some(vec![PluginListMarketplaceKind::Local]), - force_refetch: true, - }, - }) - .await?; + list_local_plugins(&mut server, /*force_refetch*/ true).await?; assert!( codex_home .path() diff --git a/codex-rs/core-plugins/src/loader.rs b/codex-rs/core-plugins/src/loader.rs index 896e20bea8..1e76918a80 100644 --- a/codex-rs/core-plugins/src/loader.rs +++ b/codex-rs/core-plugins/src/loader.rs @@ -106,7 +106,7 @@ enum PluginLoadScope<'a> { } #[derive(Clone, Copy, PartialEq, Eq)] -enum NonCuratedCacheRefreshMode { +pub(crate) enum NonCuratedCacheRefreshMode { IfVersionChanged, ForceReinstall, } @@ -534,67 +534,24 @@ pub(crate) fn refresh_non_curated_plugin_cache( additional_roots: &[AbsolutePathBuf], configured_plugin_keys: &[String], ) -> Result { - collapse_non_curated_cache_refresh(refresh_non_curated_plugin_cache_detailed( + collapse_non_curated_cache_refresh(refresh_non_curated_plugin_cache_with_mode( codex_home, additional_roots, - configured_plugin_keys, - PluginGitMode::Automatic, - )) -} - -pub(crate) fn refresh_non_curated_plugin_cache_detailed( - codex_home: &Path, - additional_roots: &[AbsolutePathBuf], - configured_plugin_keys: &[String], - git_mode: PluginGitMode, -) -> Result { - refresh_non_curated_plugin_cache_with_mode( - codex_home, - additional_roots, - configured_plugin_keys, + configured_plugin_keys.iter().map(String::as_str), NonCuratedCacheRefreshMode::IfVersionChanged, - git_mode, - ) -} - -#[cfg(test)] -pub(crate) fn refresh_non_curated_plugin_cache_force_reinstall( - codex_home: &Path, - additional_roots: &[AbsolutePathBuf], - configured_plugin_keys: &[String], -) -> Result { - collapse_non_curated_cache_refresh(refresh_non_curated_plugin_cache_force_reinstall_detailed( - codex_home, - additional_roots, - configured_plugin_keys, PluginGitMode::Automatic, )) } -pub(crate) fn refresh_non_curated_plugin_cache_force_reinstall_detailed( +pub(crate) fn refresh_non_curated_plugin_cache_with_mode<'a>( codex_home: &Path, additional_roots: &[AbsolutePathBuf], - configured_plugin_keys: &[String], - git_mode: PluginGitMode, -) -> Result { - refresh_non_curated_plugin_cache_with_mode( - codex_home, - additional_roots, - configured_plugin_keys, - NonCuratedCacheRefreshMode::ForceReinstall, - git_mode, - ) -} - -fn refresh_non_curated_plugin_cache_with_mode( - codex_home: &Path, - additional_roots: &[AbsolutePathBuf], - configured_plugin_keys: &[String], + configured_plugin_keys: impl IntoIterator, mode: NonCuratedCacheRefreshMode, git_mode: PluginGitMode, ) -> Result { let mut configured_non_curated_plugin_ids = configured_plugin_keys - .iter() + .into_iter() .filter_map(|plugin_key| match PluginId::parse(plugin_key) { Ok(plugin_id) if !is_openai_curated_marketplace_name(&plugin_id.marketplace_name) => { Some(plugin_id) diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 02b7f35608..1841741d5c 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -28,6 +28,7 @@ use crate::installed_marketplaces::installed_marketplace_roots_from_layer_stack; use crate::is_openai_curated_marketplace_name; use crate::loaded_cache_metrics; use crate::loaded_cache_metrics::RequestOutcome; +use crate::loader::NonCuratedCacheRefreshMode; use crate::loader::PluginHookLoadOutcome; use crate::loader::TargetCuratedMarketplace; use crate::loader::configured_curated_plugin_ids_from_codex_home; @@ -43,8 +44,7 @@ use crate::loader::materialize_marketplace_plugin_source; use crate::loader::plugin_capability_summary_from_root; use crate::loader::plugin_is_eligible_for_target_marketplace; use crate::loader::refresh_curated_plugin_cache; -use crate::loader::refresh_non_curated_plugin_cache_detailed; -use crate::loader::refresh_non_curated_plugin_cache_force_reinstall_detailed; +use crate::loader::refresh_non_curated_plugin_cache_with_mode; use crate::loader::remote_installed_plugins_to_config; use crate::manifest::PluginManifestFormat; use crate::manifest::PluginManifestInterface; @@ -166,6 +166,20 @@ const LOADED_PLUGINS_CACHE_CAPACITY: usize = 8; type EffectivePluginsChangedCallback = Arc; +/// Plugin refresh runs on a background OS thread. Capture the caller's Tokio runtime +/// so the worker can notify open chats there when their plugins change. +fn plugin_change_callback_for_worker( + notify: EffectivePluginsChangedCallback, +) -> EffectivePluginsChangedCallback { + let Ok(runtime) = tokio::runtime::Handle::try_current() else { + return notify; + }; + Arc::new(move |change| { + let notify = Arc::clone(¬ify); + runtime.spawn(async move { notify(change) }); + }) +} + #[derive(Debug, Clone)] pub struct PluginsConfigInput { pub config_layer_stack: ConfigLayerStack, @@ -429,12 +443,19 @@ pub struct PluginListBackgroundTaskOptions { #[derive(Clone, PartialEq, Eq)] struct NonCuratedCacheRefreshRequest { roots: Vec, - configured_plugin_keys: Vec, configured_plugin_sources: Vec, mode: NonCuratedCacheRefreshMode, git_mode: PluginGitMode, } +impl NonCuratedCacheRefreshRequest { + fn covers_sources_for(&self, other: &Self) -> bool { + self.roots == other.roots + && self.configured_plugin_sources == other.configured_plugin_sources + && (self.git_mode == other.git_mode || self.git_mode == PluginGitMode::Manual) + } +} + #[derive(Clone, PartialEq, Eq)] struct NonCuratedPluginSource { marketplace_path: AbsolutePathBuf, @@ -443,19 +464,21 @@ struct NonCuratedPluginSource { local_version: Option, } -#[derive(Clone, Copy, PartialEq, Eq)] -enum NonCuratedCacheRefreshMode { - IfVersionChanged, - ForceReinstall, -} - #[derive(Default)] struct NonCuratedCacheRefreshState { - requested: Option, + // Refreshes configured plugins for the list, including ones that still need to be installed. + // Keep this request while it runs so a matching list can reuse it. If a different list + // replaces it, the worker finishes its current pass before handling the newer one. + pending_configured_plugins_refresh: Option, last_refreshed: Option, in_flight: bool, } +struct NonCuratedPluginBatchOutcome { + completed_without_errors: bool, + cache_changed: bool, +} + #[derive(Clone, Copy, Default)] struct NonCuratedCacheRefreshCompletion { sequence: u64, @@ -3135,17 +3158,15 @@ impl PluginsManager { )); } if !outcome.upgraded_roots.is_empty() { - let mut configured_plugin_keys = configured_plugins_from_stack( + let configured_plugins = configured_plugins_from_stack( &config.config_layer_stack, self.codex_home.as_path(), - ) - .into_keys() - .collect::>(); - configured_plugin_keys.sort_unstable(); - match refresh_non_curated_plugin_cache_force_reinstall_detailed( + ); + match refresh_non_curated_plugin_cache_with_mode( self.codex_home.as_path(), &outcome.upgraded_roots, - &configured_plugin_keys, + configured_plugins.keys().map(String::as_str), + NonCuratedCacheRefreshMode::ForceReinstall, mode, ) { Ok(refresh_outcome) => { @@ -3376,36 +3397,33 @@ impl PluginsManager { request.mode = NonCuratedCacheRefreshMode::ForceReinstall; } if request.mode == NonCuratedCacheRefreshMode::IfVersionChanged - && state.requested.as_ref().is_some_and(|requested| { - requested.mode == NonCuratedCacheRefreshMode::ForceReinstall - && requested.roots == request.roots - }) + && state + .pending_configured_plugins_refresh + .as_ref() + .is_some_and(|requested| { + requested.mode == NonCuratedCacheRefreshMode::ForceReinstall + && requested.roots == request.roots + }) { request.mode = NonCuratedCacheRefreshMode::ForceReinstall; } // Reconcile each canonical plugin generation once before publishing its resource. - if state.requested.as_ref().is_some_and(|requested| { - requested == &request - || (requested.git_mode == PluginGitMode::Manual - && request.git_mode == PluginGitMode::Automatic - && requested.roots == request.roots - && requested.configured_plugin_keys == request.configured_plugin_keys - && requested.configured_plugin_sources == request.configured_plugin_sources - && requested.mode == request.mode) - }) || (request.mode == NonCuratedCacheRefreshMode::IfVersionChanged - && !state.in_flight - && state.last_refreshed.as_ref().is_some_and(|last_refreshed| { - last_refreshed.roots == request.roots - && last_refreshed.configured_plugin_keys == request.configured_plugin_keys - && last_refreshed.configured_plugin_sources - == request.configured_plugin_sources - && (last_refreshed.git_mode == request.git_mode - || last_refreshed.git_mode == PluginGitMode::Manual) - })) + if state + .pending_configured_plugins_refresh + .as_ref() + .is_some_and(|pending| { + pending.mode == request.mode && pending.covers_sources_for(&request) + }) + || (request.mode == NonCuratedCacheRefreshMode::IfVersionChanged + && !state.in_flight + && state + .last_refreshed + .as_ref() + .is_some_and(|last| last.covers_sources_for(&request))) { return; } - state.requested = Some(request); + state.pending_configured_plugins_refresh = Some(request); if state.in_flight { false } else { @@ -3427,7 +3445,7 @@ impl PluginsManager { Err(err) => err.into_inner(), }; state.in_flight = false; - state.requested = None; + state.pending_configured_plugins_refresh = None; self.non_curated_cache_refresh_completion .send_modify(|completion| { completion.sequence = completion.sequence.wrapping_add(1); @@ -3445,18 +3463,7 @@ impl PluginsManager { return; } let on_effective_plugins_changed = - on_effective_plugins_changed.map(|on_effective_plugins_changed| { - let Ok(runtime) = tokio::runtime::Handle::try_current() else { - return on_effective_plugins_changed; - }; - let callback: EffectivePluginsChangedCallback = Arc::new(move |change| { - let on_effective_plugins_changed = Arc::clone(&on_effective_plugins_changed); - runtime.spawn(async move { - on_effective_plugins_changed(change); - }); - }); - callback - }); + on_effective_plugins_changed.map(plugin_change_callback_for_worker); let manager = Arc::clone(self); let codex_home = self.codex_home.clone(); if let Err(err) = std::thread::Builder::new() @@ -3625,6 +3632,47 @@ impl PluginsManager { } } + fn refresh_non_curated_plugin_batch<'a>( + &self, + roots: &[AbsolutePathBuf], + plugin_keys: impl IntoIterator, + mode: NonCuratedCacheRefreshMode, + git_mode: PluginGitMode, + ) -> NonCuratedPluginBatchOutcome { + match refresh_non_curated_plugin_cache_with_mode( + self.codex_home.as_path(), + roots, + plugin_keys, + mode, + git_mode, + ) { + Ok(refresh_outcome) => { + if refresh_outcome.cache_refreshed { + self.clear_cache(); + } + for error in &refresh_outcome.errors { + warn!( + marketplace = error.marketplace_name, + error = %error.message, + "failed to refresh configured plugin cache" + ); + } + NonCuratedPluginBatchOutcome { + completed_without_errors: refresh_outcome.errors.is_empty(), + cache_changed: refresh_outcome.cache_refreshed, + } + } + Err(err) => { + self.clear_cache(); + warn!("failed to refresh non-curated plugin cache: {err}"); + NonCuratedPluginBatchOutcome { + completed_without_errors: false, + cache_changed: false, + } + } + } + } + fn run_non_curated_plugin_cache_refresh_loop(self: Arc) { loop { let request = { @@ -3632,7 +3680,7 @@ impl PluginsManager { Ok(state) => state, Err(err) => err.into_inner(), }; - state.requested.clone() + state.pending_configured_plugins_refresh.clone() }; let Some(request) = request else { @@ -3648,64 +3696,32 @@ impl PluginsManager { return; }; - let refresh_result = match request.mode { - NonCuratedCacheRefreshMode::IfVersionChanged => { - refresh_non_curated_plugin_cache_detailed( - self.codex_home.as_path(), - &request.roots, - &request.configured_plugin_keys, - request.git_mode, - ) - } - NonCuratedCacheRefreshMode::ForceReinstall => { - refresh_non_curated_plugin_cache_force_reinstall_detailed( - self.codex_home.as_path(), - &request.roots, - &request.configured_plugin_keys, - request.git_mode, - ) - } - }; - let (refreshed, cache_changed) = match refresh_result { - Ok(refresh_outcome) => { - if refresh_outcome.cache_refreshed { - self.clear_cache(); - } - for error in &refresh_outcome.errors { - warn!( - marketplace = error.marketplace_name, - error = %error.message, - "failed to refresh configured plugin cache" - ); - } - ( - refresh_outcome.errors.is_empty(), - refresh_outcome.cache_refreshed, - ) - } - Err(err) => { - self.clear_cache(); - warn!("failed to refresh non-curated plugin cache: {err}"); - (false, false) - } - }; + let outcome = self.refresh_non_curated_plugin_batch( + &request.roots, + request + .configured_plugin_sources + .iter() + .map(|plugin| plugin.plugin_key.as_str()), + request.mode, + request.git_mode, + ); let mut state = match self.non_curated_cache_refresh_state.write() { Ok(state) => state, Err(err) => err.into_inner(), }; - if refreshed { + if outcome.completed_without_errors { state.last_refreshed = Some(request.clone()); } - let complete = state.requested.as_ref() == Some(&request); + let complete = state.pending_configured_plugins_refresh.as_ref() == Some(&request); if complete { - state.requested = None; + state.pending_configured_plugins_refresh = None; state.in_flight = false; } self.non_curated_cache_refresh_completion .send_modify(|completion| { completion.sequence = completion.sequence.wrapping_add(1); - if cache_changed { + if outcome.cache_changed { completion.changed_sequence = completion.changed_sequence.wrapping_add(1); } }); diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index 8c1320e72c..c12ce00bea 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/codex-rs/core-plugins/src/manager_tests.rs @@ -10,7 +10,6 @@ use crate::loader::load_plugin_skill_inventory; use crate::loader::load_plugins_from_layer_stack; use crate::loader::plugin_capability_summary_from_root; use crate::loader::refresh_non_curated_plugin_cache; -use crate::loader::refresh_non_curated_plugin_cache_force_reinstall; use crate::marketplace::MarketplacePluginInstallPolicy; use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME; use crate::remote::REMOTE_WORKSPACE_MARKETPLACE_NAME; @@ -6882,10 +6881,13 @@ enabled = true .write() .expect("refresh state lock"); assert!(matches!( - state.requested.as_ref().map(|request| request.git_mode), + state + .pending_configured_plugins_refresh + .as_ref() + .map(|request| request.git_mode), Some(PluginGitMode::Manual) )); - state.last_refreshed = state.requested.take(); + state.last_refreshed = state.pending_configured_plugins_refresh.take(); state.in_flight = false; } @@ -6897,8 +6899,8 @@ enabled = true .write() .expect("refresh state lock"); assert!(!state.in_flight); - assert!(state.requested.is_none()); - state.requested = state.last_refreshed.clone(); + assert!(state.pending_configured_plugins_refresh.is_none()); + state.pending_configured_plugins_refresh = state.last_refreshed.clone(); state.in_flight = true; } @@ -6918,7 +6920,10 @@ enabled = true .non_curated_cache_refresh_state .read() .expect("refresh state lock"); - let request = state.requested.as_ref().expect("pending refresh"); + let request = state + .pending_configured_plugins_refresh + .as_ref() + .expect("pending refresh"); assert!(matches!(request.git_mode, PluginGitMode::Automatic)); assert_eq!( request.configured_plugin_sources[0] @@ -7129,12 +7134,15 @@ enabled = true ); assert!( - refresh_non_curated_plugin_cache_force_reinstall( + refresh_non_curated_plugin_cache_with_mode( tmp.path(), &[AbsolutePathBuf::try_from(repo_root).unwrap()], - &["sample-plugin@debug".to_string()], + ["sample-plugin@debug"], + NonCuratedCacheRefreshMode::ForceReinstall, + PluginGitMode::Automatic, ) .expect("cache refresh should reinstall unchanged local version") + .cache_refreshed ); assert_eq!( diff --git a/codex-rs/core-plugins/src/marketplace_context.rs b/codex-rs/core-plugins/src/marketplace_context.rs index bab5781777..af8c1d6008 100644 --- a/codex-rs/core-plugins/src/marketplace_context.rs +++ b/codex-rs/core-plugins/src/marketplace_context.rs @@ -112,7 +112,6 @@ impl PluginMarketplaceContext { .collect::>(); let mut roots = Vec::new(); - let mut configured_plugin_keys = Vec::new(); let mut configured_plugin_sources = Vec::new(); for marketplace in marketplaces { @@ -133,7 +132,6 @@ impl PluginMarketplaceContext { } else { plugin.local_version.clone() }; - configured_plugin_keys.push(plugin.id.clone()); configured_plugin_sources.push(NonCuratedPluginSource { marketplace_path: marketplace.path.clone(), plugin_key: plugin.id.clone(), @@ -144,12 +142,11 @@ impl PluginMarketplaceContext { roots.push(marketplace.path.clone()); } - if roots.is_empty() || configured_plugin_keys.is_empty() { + if configured_plugin_sources.is_empty() { return None; } // Refresh rediscovers sources in root order; retain listing's precedence. - configured_plugin_keys.sort_unstable(); configured_plugin_sources.sort_by(|left, right| { left.marketplace_path .cmp(&right.marketplace_path) @@ -158,7 +155,6 @@ impl PluginMarketplaceContext { Some(NonCuratedCacheRefreshRequest { roots, - configured_plugin_keys, configured_plugin_sources, mode, git_mode,