Skip to content

Keep shared skill caches fresh across plugin loads #37000

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
copyberry merged 1 commit into from
Aug 5, 2026
Merged

Keep shared skill caches fresh across plugin loads #37000

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
46 changes: 34 additions & 12 deletions codex-rs/app-server/src/request_processors/catalog_processor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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;
Expand Down
34 changes: 32 additions & 2 deletions codex-rs/app-server/tests/suite/v2/skills_list.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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?;

Expand All @@ -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(
Expand All @@ -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 {
Expand Down
16 changes: 16 additions & 0 deletions codex-rs/core-skills/src/root_loader.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;

Expand Down Expand Up @@ -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")
Expand Down
131 changes: 111 additions & 20 deletions codex-rs/ext/skills/src/host_service.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;
Expand All @@ -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;
Expand Down Expand Up @@ -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>,
}
Expand Down Expand Up @@ -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(
Expand Down Expand Up @@ -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)
{
Expand All @@ -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()
Expand All @@ -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,
Expand Down Expand Up @@ -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(),
}
}

Expand All @@ -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)]
Expand All @@ -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(
Expand Down Expand Up @@ -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
Expand All @@ -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(),
}
}

Expand Down
Loading
Loading