From 5d89ab65dc9d4d0c55796c11df112b54157922b4 Mon Sep 17 00:00:00 2001 From: chess Date: Wed, 5 Aug 2026 01:19:15 +0000 Subject: [PATCH] Keep shared skill caches fresh across plugin loads (#37000) ## What changed - Key cached skill snapshots by filesystem and plugin snapshot identity so compatible config and working-directory loads can share results without reusing stale plugin data. - Coalesce concurrent loads for the same cache key and replace the cached entry on forced reload. - Clear both plugin and skill caches when `skills/list` forces a reload, and bypass working-directory caching when effective plugin roots are present. ## Testing - Cover concurrent cache sharing, filesystem isolation, forced reloads, and refreshed plugin skill metadata in host service and app server tests. GitOrigin-RevId: 03fed3b40d45bb29206d5a3c3e78f06df04dbb1e --- .../request_processors/catalog_processor.rs | 46 ++++-- .../app-server/tests/suite/v2/skills_list.rs | 34 ++++- codex-rs/core-skills/src/root_loader.rs | 16 +++ codex-rs/ext/skills/src/host_service.rs | 131 +++++++++++++++--- codex-rs/ext/skills/src/host_service_tests.rs | 68 +++++++-- 5 files changed, 251 insertions(+), 44 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/catalog_processor.rs b/codex-rs/app-server/src/request_processors/catalog_processor.rs index 4070254f1a8c..4d80271fa20c 100644 --- a/codex-rs/app-server/src/request_processors/catalog_processor.rs +++ b/codex-rs/app-server/src/request_processors/catalog_processor.rs @@ -506,6 +506,13 @@ impl CatalogRequestProcessor { .await; let skills_service = self.thread_manager.skills_service(); let plugins_manager = self.thread_manager.plugins_manager(); + if force_reload + && workspace_codex_plugins_enabled + && config.features.enabled(Feature::Plugins) + { + plugins_manager.clear_cache(); + skills_service.clear_cache(); + } let fs = self .thread_manager .environment_manager() @@ -535,23 +542,38 @@ impl CatalogRequestProcessor { ); } }; - let effective_skill_roots = if workspace_codex_plugins_enabled { - let plugins_input = config.plugins_config_input(); - plugins_manager - .effective_skill_roots_for_layer_stack( - &config_layer_stack, - &plugins_input, - ) - .await - } else { - Vec::new() - }; + let (effective_skill_roots, plugin_skill_snapshots) = + if workspace_codex_plugins_enabled { + let plugins_input = config.plugins_config_input(); + if config_layer_stack == plugins_input.config_layer_stack { + let plugins = + plugins_manager.plugins_for_config(&plugins_input).await; + ( + plugins.effective_plugin_skill_roots(), + plugins_manager + .plugin_skill_snapshots_for_config(&plugins_input), + ) + } else { + ( + plugins_manager + .effective_skill_roots_for_layer_stack( + &config_layer_stack, + &plugins_input, + ) + .await, + None, + ) + } + } else { + (Vec::new(), None) + }; let skills_input = codex_core::skills::HostSkillsLoadInput::new( cwd_abs.clone(), effective_skill_roots, config_layer_stack, config.bundled_skills_enabled(), - ); + ) + .with_plugin_skill_snapshots(plugin_skill_snapshots); let snapshot = skills_service .snapshot_for_cwd(&skills_input, force_reload, fs) .await; 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 b83be6f7f872..ff4d1e23f21d 100644 --- a/codex-rs/app-server/tests/suite/v2/skills_list.rs +++ b/codex-rs/app-server/tests/suite/v2/skills_list.rs @@ -365,7 +365,6 @@ enabled = true let mut mcp = TestAppServer::builder() .with_codex_home(codex_home.path()) - .without_auto_env() .build_initialized_with_timeout(DEFAULT_TIMEOUT) .await?; @@ -380,7 +379,13 @@ enabled = true let initial_skills_list_request_id = mcp .send_skills_list_request(SkillsListParams { cwds: vec![cwd.path().to_path_buf()], - force_reload: true, + force_reload: false, + }) + .await?; + let thread_start_request_id = mcp + .send_thread_start_request_with_auto_env(ThreadStartParams { + cwd: Some(cwd.path().to_string_lossy().into_owned()), + ..Default::default() }) .await?; let SkillsListResponse { data } = timeout( @@ -394,6 +399,31 @@ enabled = true .iter() .any(|skill| skill.name == "google-calendar:meeting-prep") })); + let _: ThreadStartResponse = + timeout(DEFAULT_TIMEOUT, mcp.read_response(thread_start_request_id)).await??; + + std::fs::write( + codex_home.path().join( + "plugins/cache/openai-curated/google-calendar/local/skills/meeting-prep/SKILL.md", + ), + "---\nname: meeting-prep\ndescription: Updated meeting preparation\n---\n\n# Body\n", + )?; + for force_reload in [true, false] { + let request_id = mcp + .send_skills_list_request(SkillsListParams { + cwds: vec![cwd.path().to_path_buf()], + force_reload, + }) + .await?; + let SkillsListResponse { data } = + timeout(DEFAULT_TIMEOUT, mcp.read_response(request_id)).await??; + assert!(data.iter().any(|entry| { + entry.skills.iter().any(|skill| { + skill.name == "google-calendar:meeting-prep" + && skill.description == "Updated meeting preparation" + }) + })); + } let enablement_request_id = mcp .send_experimental_feature_enablement_set_request(ExperimentalFeatureEnablementSetParams { diff --git a/codex-rs/core-skills/src/root_loader.rs b/codex-rs/core-skills/src/root_loader.rs index f18f96e85d1f..e3cc4fb6d686 100644 --- a/codex-rs/core-skills/src/root_loader.rs +++ b/codex-rs/core-skills/src/root_loader.rs @@ -1,6 +1,8 @@ use std::collections::HashMap; use std::collections::HashSet; use std::fmt; +use std::hash::Hash; +use std::hash::Hasher; use std::sync::Arc; use std::sync::Mutex; @@ -33,6 +35,20 @@ impl PluginSkillSnapshots { } } +impl PartialEq for PluginSkillSnapshots { + fn eq(&self, other: &Self) -> bool { + Arc::ptr_eq(&self.snapshots_by_root, &other.snapshots_by_root) + } +} + +impl Eq for PluginSkillSnapshots {} + +impl Hash for PluginSkillSnapshots { + fn hash(&self, state: &mut H) { + Arc::as_ptr(&self.snapshots_by_root).hash(state); + } +} + impl fmt::Debug for PluginSkillSnapshots { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("PluginSkillSnapshots") diff --git a/codex-rs/ext/skills/src/host_service.rs b/codex-rs/ext/skills/src/host_service.rs index 1663d36d3420..edeb208875c4 100644 --- a/codex-rs/ext/skills/src/host_service.rs +++ b/codex-rs/ext/skills/src/host_service.rs @@ -1,6 +1,9 @@ use std::collections::HashMap; +use std::hash::Hash; +use std::hash::Hasher; use std::sync::Arc; use std::sync::RwLock; +use std::sync::Weak; use codex_config::ConfigLayerStack; use codex_exec_server::ExecutorFileSystem; @@ -9,6 +12,7 @@ use codex_protocol::protocol::SkillScope; use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_plugins::PluginIdentity; use codex_utils_plugins::PluginSkillRoot; +use tokio::sync::OnceCell; use tokio::sync::Semaphore; use tracing::info; use tracing::instrument; @@ -71,7 +75,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, } @@ -132,21 +136,23 @@ impl HostSkillsService { ) -> HostSkillsSnapshot { let roots = self.skill_roots_for_config(input, fs).await; let skill_config_rules = skill_config_rules_from_stack(&input.config_layer_stack); - let cache_key = config_skills_cache_key(&roots, &skill_config_rules); + let cache_key = config_skills_cache_key( + &roots, + &skill_config_rules, + input.plugin_skill_snapshots.as_ref(), + ); if let Some(snapshot) = self.cached_snapshot_for_config(&cache_key) { return snapshot; } - let snapshot = HostSkillsSnapshot::new(Arc::new( - self.build_skill_outcome(input, roots, &skill_config_rules) - .await, - )); - let mut cache = self - .cache_by_config - .write() - .unwrap_or_else(std::sync::PoisonError::into_inner); - cache.insert(cache_key, snapshot.clone()); - snapshot + self.snapshot_for_skill_roots( + input, + roots, + &skill_config_rules, + cache_key, + /*force_reload*/ false, + ) + .await } pub async fn skill_roots_for_config( @@ -182,7 +188,8 @@ impl HostSkillsService { self.ensure_system_skills_installed(); } let use_cwd_cache = fs.is_some(); - if use_cwd_cache + let cache_snapshot_by_cwd = use_cwd_cache && input.effective_skill_roots.is_empty(); + if cache_snapshot_by_cwd && !force_reload && let Some(snapshot) = self.cached_snapshot_for_cwd(&input.cwd) { @@ -201,11 +208,27 @@ impl HostSkillsService { roots.retain(|root| root.scope != SkillScope::System); } let skill_config_rules = skill_config_rules_from_stack(&input.config_layer_stack); - let snapshot = HostSkillsSnapshot::new(Arc::new( - self.build_skill_outcome(input, roots, &skill_config_rules) - .await, - )); - if use_cwd_cache { + let snapshot = if use_cwd_cache { + let cache_key = config_skills_cache_key( + &roots, + &skill_config_rules, + input.plugin_skill_snapshots.as_ref(), + ); + self.snapshot_for_skill_roots( + input, + roots, + &skill_config_rules, + cache_key, + force_reload, + ) + .await + } else { + HostSkillsSnapshot::new(Arc::new( + self.build_skill_outcome(input, roots, &skill_config_rules) + .await, + )) + }; + if cache_snapshot_by_cwd { let mut cache = self .cache_by_cwd .write() @@ -215,6 +238,43 @@ impl HostSkillsService { snapshot } + async fn snapshot_for_skill_roots( + &self, + input: &HostSkillsLoadInput, + roots: Vec, + skill_config_rules: &SkillConfigRules, + cache_key: ConfigSkillsCacheKey, + force_reload: bool, + ) -> HostSkillsSnapshot { + let snapshot_cell = { + let mut cache = self + .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())), + ) + } + }; + + snapshot_cell + .get_or_init(|| async { + HostSkillsSnapshot::new(Arc::new( + self.build_skill_outcome(input, roots, skill_config_rules) + .await, + )) + }) + .await + .clone() + } + #[instrument(level = "trace", skip_all)] async fn build_skill_outcome( &self, @@ -271,8 +331,15 @@ impl HostSkillsService { cache_key: &ConfigSkillsCacheKey, ) -> Option { match self.cache_by_config.read() { - Ok(cache) => cache.get(cache_key).cloned(), - Err(err) => err.into_inner().get(cache_key).cloned(), + 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(), } } @@ -294,6 +361,7 @@ impl HostSkillsService { struct ConfigSkillsCacheKey { roots: Vec, skill_config_rules: SkillConfigRules, + plugin_skill_snapshots: Option, } #[derive(Debug, Clone, PartialEq, Eq, Hash)] @@ -302,6 +370,24 @@ struct ConfigSkillRootCacheKey { scope_rank: u8, plugin_identity: Option, plugin_namespace: Option, + file_system: FileSystemIdentity, +} + +#[derive(Debug, Clone)] +struct FileSystemIdentity(Weak); + +impl PartialEq for FileSystemIdentity { + fn eq(&self, other: &Self) -> bool { + Weak::ptr_eq(&self.0, &other.0) + } +} + +impl Eq for FileSystemIdentity {} + +impl Hash for FileSystemIdentity { + fn hash(&self, state: &mut H) { + (self.0.as_ptr() as *const ()).hash(state); + } } pub fn bundled_skills_enabled_from_stack( @@ -329,6 +415,7 @@ pub fn bundled_skills_enabled_from_stack( fn config_skills_cache_key( roots: &[SkillRoot], skill_config_rules: &SkillConfigRules, + plugin_skill_snapshots: Option<&PluginSkillSnapshots>, ) -> ConfigSkillsCacheKey { ConfigSkillsCacheKey { roots: roots @@ -345,10 +432,14 @@ fn config_skills_cache_key( scope_rank, plugin_identity: root.plugin_identity.clone(), plugin_namespace: root.plugin_namespace.clone(), + file_system: FileSystemIdentity(Arc::downgrade(&root.file_system)), } }) .collect(), skill_config_rules: skill_config_rules.clone(), + plugin_skill_snapshots: plugin_skill_snapshots + .filter(|_| roots.iter().any(|root| root.plugin_identity.is_some())) + .cloned(), } } diff --git a/codex-rs/ext/skills/src/host_service_tests.rs b/codex-rs/ext/skills/src/host_service_tests.rs index 5c9b6afdb0b9..047ea00a5692 100644 --- a/codex-rs/ext/skills/src/host_service_tests.rs +++ b/codex-rs/ext/skills/src/host_service_tests.rs @@ -203,13 +203,48 @@ async fn skills_for_config_refreshes_cache_when_remote_plugin_id_changes() { /*bundled_skills_enabled*/ true, ); - skills_for_config_with_stack( - &skills_service, - &cwd, - &config_layer_stack, - &[plugin_skill_root.clone()], + let plugin_input = HostSkillsLoadInput::new( + cwd.path().abs(), + vec![plugin_skill_root.clone()], + config_layer_stack.clone(), + bundled_skills_enabled_from_stack(&config_layer_stack), ) - .await; + .with_plugin_skill_snapshots(Some(PluginSkillSnapshots::for_plugin_load())); + let plugin_snapshot = skills_service + .snapshot_for_config(&plugin_input, Some(Arc::clone(&LOCAL_FS))) + .await; + fs::write( + &skill_path, + "---\nname: sample-search\ndescription: updated sample data\n---\n\n# Body\n", + ) + .expect("update plugin skill"); + let listing_input = plugin_input + .clone() + .with_plugin_skill_snapshots(/*plugin_skill_snapshots*/ None); + let listing_snapshot = skills_service + .snapshot_for_cwd( + &listing_input, + /*force_reload*/ false, + Some(Arc::clone(&LOCAL_FS)), + ) + .await; + assert_eq!( + ( + plugin_snapshot + .outcome() + .skills + .iter() + .find(|skill| skill.name == "sample:sample-search") + .map(|skill| skill.description.as_str()), + listing_snapshot + .outcome() + .skills + .iter() + .find(|skill| skill.name == "sample:sample-search") + .map(|skill| skill.description.as_str()), + ), + (Some("search sample data"), Some("updated sample data")) + ); plugin_skill_root.plugin_identity.remote_plugin_id = Some("plugins~Plugin_sample".to_string()); let refreshed = skills_for_config_with_stack( @@ -466,6 +501,12 @@ async fn skills_for_cwd_loads_repo_and_user_roots_with_local_fs() { .collect::>(); assert!(loaded_names.contains("user-skill")); assert!(loaded_names.contains("repo-skill")); + let other_file_system: Arc = + Arc::new(codex_exec_server::LocalFileSystem::unsandboxed()); + let other_snapshot = skills_service + .snapshot_for_config(&skills_input, Some(other_file_system)) + .await; + assert!(!std::ptr::eq(snapshot.outcome(), other_snapshot.outcome())); } #[tokio::test] @@ -570,20 +611,27 @@ async fn skills_for_cwd_uses_cached_result_until_force_reload() { codex_home.path().abs(), /*bundled_skills_enabled*/ true, ); - let _ = skills_for_config_with_stack(&skills_service, &cwd, &config_layer_stack, &[]).await; let base_input = HostSkillsLoadInput::new( cwd.path().abs(), Vec::new(), config_layer_stack.clone(), bundled_skills_enabled_from_stack(&config_layer_stack), ); - let snapshot_a = skills_service - .snapshot_for_cwd( + let config_input = base_input + .clone() + .with_plugin_skill_snapshots(Some(PluginSkillSnapshots::for_plugin_load())); + let (config_snapshot, snapshot_a) = tokio::join!( + skills_service.snapshot_for_config(&config_input, Some(Arc::clone(&LOCAL_FS))), + skills_service.snapshot_for_cwd( &base_input, /*force_reload*/ false, Some(Arc::clone(&LOCAL_FS)), ) - .await; + ); + assert!(std::ptr::eq( + config_snapshot.outcome(), + snapshot_a.outcome() + )); let outcome_a = snapshot_a.outcome(); assert!( outcome_a