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
This commit is contained in:
chess
2026-08-05 01:24:35 +00:00
committed by copyberry
parent 431c78eb9c
commit 5d89ab65dc
5 changed files with 251 additions and 44 deletions
@@ -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;
@@ -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 {
+16
View File
@@ -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<H: Hasher>(&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")
+111 -20
View File
@@ -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<Product>,
extra_roots: RwLock<Vec<AbsolutePathBuf>>,
cache_by_cwd: RwLock<HashMap<AbsolutePathBuf, HostSkillsSnapshot>>,
cache_by_config: RwLock<HashMap<ConfigSkillsCacheKey, HostSkillsSnapshot>>,
cache_by_config: RwLock<HashMap<ConfigSkillsCacheKey, Arc<OnceCell<HostSkillsSnapshot>>>>,
// Shared across cwds so root scheduling cannot multiply per-root I/O fanout.
root_scan_slots: Arc<Semaphore>,
}
@@ -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<SkillRoot>,
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<HostSkillsSnapshot> {
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<ConfigSkillRootCacheKey>,
skill_config_rules: SkillConfigRules,
plugin_skill_snapshots: Option<PluginSkillSnapshots>,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
@@ -302,6 +370,24 @@ struct ConfigSkillRootCacheKey {
scope_rank: u8,
plugin_identity: Option<PluginIdentity>,
plugin_namespace: Option<String>,
file_system: FileSystemIdentity,
}
#[derive(Debug, Clone)]
struct FileSystemIdentity(Weak<dyn ExecutorFileSystem>);
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<H: Hasher>(&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(),
}
}
+58 -10
View File
@@ -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::<HashSet<_>>();
assert!(loaded_names.contains("user-skill"));
assert!(loaded_names.contains("repo-skill"));
let other_file_system: Arc<dyn codex_exec_server::ExecutorFileSystem> =
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