diff --git a/codex-rs/app-server/tests/suite/v2/skills_list.rs b/codex-rs/app-server/tests/suite/v2/skills_list.rs index c6880fb0e0..245f10dd4d 100644 --- a/codex-rs/app-server/tests/suite/v2/skills_list.rs +++ b/codex-rs/app-server/tests/suite/v2/skills_list.rs @@ -30,8 +30,10 @@ use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartResponse; use codex_config::types::AuthCredentialsStoreMode; use codex_core::config::set_project_trust_level; +use codex_core_plugins::store::PluginStore; use codex_exec_server::CODEX_EXEC_SERVER_URL_ENV_VAR; use codex_exec_server::CreateDirectoryOptions; +use codex_plugin::PluginId; use codex_protocol::config_types::TrustLevel; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_path_uri::PathUri; @@ -926,6 +928,95 @@ enabled = false Ok(()) } +#[tokio::test] +async fn skills_list_refreshes_externally_updated_plugin_versions() -> Result<()> { + skip_if_wine_exec!( + Ok(()), + "skills/list currently requires host-native cwd paths for workspace config" + ); + let codex_home = TempDir::new()?; + let cwd = TempDir::new()?; + let source = TempDir::new()?; + std::fs::create_dir_all(source.path().join(".codex-plugin"))?; + std::fs::create_dir_all(source.path().join("skills"))?; + std::fs::write( + source.path().join(".codex-plugin/plugin.json"), + r#"{"name":"sample"}"#, + )?; + std::fs::write( + codex_home.path().join("config.toml"), + r#"[features] +plugins = true + +[plugins."sample@test"] +enabled = true +"#, + )?; + let plugin_id = PluginId::parse("sample@test")?; + let store = PluginStore::new(codex_home.path().to_path_buf()); + let mut mcp = TestAppServer::builder() + .with_codex_home(codex_home.path()) + .build_initialized_with_timeout(DEFAULT_TIMEOUT) + .await?; + let file_system = mcp.auto_env()?.environment().get_filesystem(); + file_system + .create_directory( + &PathUri::from_abs_path(&AbsolutePathBuf::try_from(cwd.path())?).join(".git")?, + CreateDirectoryOptions { + recursive: true, + follow_symlinks: true, + }, + /*sandbox*/ None, + ) + .await?; + + // Mutate the shared store from outside app-server, without its cache invalidation callback. + for version in ["1.0.0", "2.0.0", "0.9.0"] { + let body = format!("---\nname: search\ndescription: version {version}\n---\n"); + std::fs::write(source.path().join("skills/SKILL.md"), &body)?; + let installed = store.install_with_version( + AbsolutePathBuf::try_from(source.path())?, + plugin_id.clone(), + version.to_string(), + )?; + let expected_path = AbsolutePathBuf::try_from(std::fs::canonicalize( + installed.installed_path.join("skills/SKILL.md"), + )?)?; + // Exercise both invalidation and the subsequent warm read for the same cwd. + for _ in 0..2 { + let request_id = mcp + .send_skills_list_request(SkillsListParams { + cwds: vec![cwd.path().to_path_buf()], + force_reload: false, + }) + .await?; + let SkillsListResponse { data } = + timeout(DEFAULT_TIMEOUT, mcp.read_response(request_id)).await??; + let skills = data[0] + .skills + .iter() + .filter(|skill| skill.name == "sample:search") + .map(|skill| { + Ok(( + skill.path.clone(), + skill.description.clone(), + std::fs::read_to_string(&skill.path)?, + )) + }) + .collect::>>()?; + assert_eq!( + skills, + vec![( + expected_path.clone(), + format!("version {version}"), + body.clone() + )], + ); + } + } + Ok(()) +} + #[tokio::test] async fn skills_list_force_reload_refreshes_cached_plugin_roots() -> Result<()> { skip_if_wine_exec!( diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 112eeb0d87..69269e67ed 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -564,9 +564,26 @@ struct LoadedPluginsCache { } impl LoadedPluginsCache { - fn get(&mut self, key: &PluginLoadCacheKey) -> Option<&LoadedPluginsCacheEntry> { + fn get( + &mut self, + key: &PluginLoadCacheKey, + store: &PluginStore, + ) -> Option<&LoadedPluginsCacheEntry> { let index = self.entries.iter().position(|entry| &entry.key == key)?; let entry = self.entries.remove(index)?; + // Another process can replace installed versions without invalidating this manager. + // Keep the parsed skills paired with the installation whose paths they advertise. + if entry.plugins.iter().any(|plugin| { + let Ok(plugin_id) = PluginId::parse(&plugin.config_name) else { + return false; + }; + let installed_root = store + .active_plugin_root(&plugin_id) + .unwrap_or_else(|| store.plugin_base_root(&plugin_id)); + installed_root != plugin.root + }) { + return None; + } self.entries.push_front(entry); self.entries.front() } @@ -745,7 +762,7 @@ impl PluginsManager { self.loaded_plugins_cache .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(&key) + .get(&key, &self.store) .map(|cached| cached.plugin_skill_snapshots.clone()) } @@ -968,7 +985,7 @@ impl PluginsManager { self.loaded_plugins_cache .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) - .get(key) + .get(key, &self.store) .map(|cached| cached.plugins.clone()) } diff --git a/codex-rs/ext/skills/src/host_service.rs b/codex-rs/ext/skills/src/host_service.rs index c13f9d07da..3feda36e77 100644 --- a/codex-rs/ext/skills/src/host_service.rs +++ b/codex-rs/ext/skills/src/host_service.rs @@ -1,4 +1,5 @@ use std::collections::HashMap; +use std::collections::VecDeque; use std::hash::Hash; use std::hash::Hasher; use std::sync::Arc; @@ -37,6 +38,13 @@ use crate::loader::MAX_CONCURRENT_ROOT_SCANS; use crate::loader::load_and_merge_host_skill_roots; use crate::loader::load_and_merge_host_skill_roots_with_request_snapshots; +const CONFIG_SKILLS_CACHE_CAPACITY: usize = 32; + +struct ConfigSkillsCacheEntry { + key: ConfigSkillsCacheKey, + snapshot: Arc>, +} + #[derive(Debug, Clone)] pub struct HostSkillsLoadInput { cwd: AbsolutePathBuf, @@ -77,7 +85,7 @@ pub struct HostSkillsService { restriction_product: Option, extra_roots: RwLock>, cache_by_cwd: RwLock>, - cache_by_config: RwLock>>>, + cache_by_config: RwLock>, // Shared across cwds so root scheduling cannot multiply per-root I/O fanout. root_scan_slots: Arc, } @@ -124,7 +132,7 @@ impl HostSkillsService { restriction_product, extra_roots: RwLock::new(Vec::new()), cache_by_cwd: RwLock::new(HashMap::new()), - cache_by_config: RwLock::new(HashMap::new()), + cache_by_config: RwLock::new(VecDeque::new()), root_scan_slots: Arc::new(Semaphore::new(MAX_CONCURRENT_ROOT_SCANS)), }; // The cache is shared by every process using this CODEX_HOME. Disabled services filter @@ -178,10 +186,6 @@ impl HostSkillsService { &skill_config_rules, input.plugin_skill_snapshots.as_ref(), ); - if let Some(snapshot) = self.cached_snapshot_for_config(&cache_key) { - return snapshot; - } - self.snapshot_for_skill_roots( input, roots, @@ -310,17 +314,21 @@ impl HostSkillsService { .cache_by_config .write() .unwrap_or_else(std::sync::PoisonError::into_inner); - if force_reload { - let snapshot_cell = Arc::new(OnceCell::new()); - cache.insert(cache_key, Arc::clone(&snapshot_cell)); - snapshot_cell - } else { - Arc::clone( - cache - .entry(cache_key) - .or_insert_with(|| Arc::new(OnceCell::new())), - ) - } + let snapshot_cell = cache + .iter() + .position(|entry| entry.key == cache_key) + .and_then(|index| cache.remove(index)) + .filter(|_| !force_reload) + .map(|entry| entry.snapshot) + .unwrap_or_else(|| Arc::new(OnceCell::new())); + cache.push_front(ConfigSkillsCacheEntry { + key: cache_key, + snapshot: Arc::clone(&snapshot_cell), + }); + // Keys retain parsed plugin generations; eviction releases obsolete catalogs while + // callers and in-flight loads keep their own snapshots alive. + cache.truncate(CONFIG_SKILLS_CACHE_CAPACITY); + snapshot_cell }; snapshot_cell @@ -394,23 +402,6 @@ impl HostSkillsService { } } - fn cached_snapshot_for_config( - &self, - cache_key: &ConfigSkillsCacheKey, - ) -> Option { - match self.cache_by_config.read() { - Ok(cache) => cache - .get(cache_key) - .and_then(|snapshot| snapshot.get()) - .cloned(), - Err(err) => err - .into_inner() - .get(cache_key) - .and_then(|snapshot| snapshot.get()) - .cloned(), - } - } - fn extra_roots(&self) -> Vec { match self.extra_roots.read() { Ok(roots) => roots.clone(), diff --git a/codex-rs/ext/skills/src/host_service_tests.rs b/codex-rs/ext/skills/src/host_service_tests.rs index 046e5cf549..0b76742089 100644 --- a/codex-rs/ext/skills/src/host_service_tests.rs +++ b/codex-rs/ext/skills/src/host_service_tests.rs @@ -207,6 +207,88 @@ async fn skills_for_config_reuses_cache_for_same_effective_config() { assert_eq!(outcome2.skills, outcome1.skills); } +#[tokio::test] +async fn skills_for_config_bounds_plugin_generations_and_preserves_live_snapshots() { + let codex_home = tempfile::tempdir().expect("tempdir"); + let cwd = tempfile::tempdir().expect("tempdir"); + let skill_path = write_plugin_skill( + &codex_home, + "test", + "sample", + "search", + "search", + "initial description", + ); + let plugin_root = plugin_skill_root_for_skill_path(&skill_path, "sample@test", "sample"); + let base_input = HostSkillsLoadInput::new( + cwd.path().abs(), + vec![plugin_root], + config_stack(&codex_home, "[skills.bundled]\nenabled = false\n"), + ); + let skills_service = HostSkillsService::new( + codex_home.path().abs(), + /*bundled_skills_enabled*/ false, + ); + let mut generations = Vec::new(); + let mut recent = None; + let mut held_snapshot = None; + let mut held_skills = Vec::new(); + + // Reloads allocate new handles even when roots repeat, as they do after a rollback. + for generation in 0..CONFIG_SKILLS_CACHE_CAPACITY + 2 { + fs::write( + &skill_path, + format!("---\nname: search\ndescription: generation {generation}\n---\n\n# Body\n"), + ) + .expect("write updated skill"); + let owner = Arc::new(TestPluginSkillSnapshotCache::default()); + generations.push(Arc::downgrade(&owner)); + let input = base_input + .clone() + .with_plugin_skill_snapshots(Some(SkillRootSnapshots::new(owner))); + let snapshot = skills_service + .snapshot_for_config(&input, Some(Arc::clone(&LOCAL_FS))) + .await; + + if generation == 0 { + recent = Some((input, snapshot)); + } else if generation == 1 { + held_skills = snapshot.outcome().skills.clone(); + held_snapshot = Some(snapshot); + } + if generation == CONFIG_SKILLS_CACHE_CAPACITY - 1 { + let (input, snapshot) = recent.as_ref().expect("first generation"); + let reused = skills_service + .snapshot_for_config(input, Some(Arc::clone(&LOCAL_FS))) + .await; + assert!(std::ptr::eq(snapshot.outcome(), reused.outcome())); + } + } + drop(recent); + + let retained_generations = generations + .iter() + .enumerate() + .filter_map(|(generation, owner)| owner.upgrade().map(|_| generation)) + .collect::>(); + assert_eq!( + retained_generations, + std::iter::once(/*value*/ 0) + .chain(3..CONFIG_SKILLS_CACHE_CAPACITY + 2) + .collect::>() + ); + assert_eq!( + held_snapshot + .expect("evicted live snapshot") + .outcome() + .skills, + held_skills + ); + + skills_service.clear_cache(); + assert_eq!(generations.iter().filter_map(Weak::upgrade).count(), 0); +} + #[tokio::test] async fn watchable_skill_root_paths_exclude_plugin_and_system_roots() { let codex_home = tempfile::tempdir().expect("tempdir");