Load plugin skill roots through the host skills service (#37440)

## What changed

- Implement `SkillRootLoader<PluginSkillRoot>` for `HostSkillsService`.
- Load roots concurrently while preserving request order, product restrictions, cached snapshots, skill errors, and per-skill filesystem metadata.
- Deduplicate overlapping roots and prefer native plugin skills over migrated commands with the same plugin ID and skill name.

## Testing

- Cover native and migrated skill precedence, nested and overlapping roots, product filtering, snapshot reuse, error propagation, and symlinked migration roots.

GitOrigin-RevId: 2270d10616b3a0463852c563ee071c524cd11e6b
This commit is contained in:
felixxia-oai authored and copyberry committed 2026-08-07 13:23:23 +00:00
1 parent a4b129eb3e
commit e75a1888d7
10 files changed
+634 -24

No files matched your search

+32
View File
@@ -7,6 +7,7 @@ use std::sync::Weak;
use codex_config::ConfigLayerStack;
use codex_exec_server::ExecutorFileSystem;
use codex_exec_server::LOCAL_FS;
use codex_protocol::protocol::Product;
use codex_protocol::protocol::SkillScope;
use codex_utils_absolute_path::AbsolutePathBuf;
@@ -31,11 +32,16 @@ use codex_core_skills::loader::MAX_CONCURRENT_ROOT_SCANS;
use codex_core_skills::loader::SkillRoot;
use codex_core_skills::loader::SkillRootSnapshot;
use codex_core_skills::loader::load_skill_root_snapshot;
use codex_skills::LoadedSkills;
use codex_skills::SkillLoadFuture;
use codex_skills::SkillRootLoadRequest;
use codex_skills::SkillRootLoader;
use codex_skills::install_system_skills;
use crate::HostSkillsSnapshot;
use crate::host_roots::resolve_skill_roots;
use crate::loader::HostSkillRoot;
use crate::loader::load_and_merge_host_skill_roots;
use crate::loader::load_host_skill_root;
#[derive(Debug, Clone)]
@@ -405,6 +411,32 @@ impl HostSkillsService {
}
}
impl SkillRootLoader<PluginSkillRoot> for HostSkillsService {
fn load_roots(
&self,
request: SkillRootLoadRequest<PluginSkillRoot>,
) -> SkillLoadFuture<'_, LoadedSkills> {
Box::pin(async move {
let roots = request
.roots
.into_iter()
.map(|root| HostSkillRoot::plugin(root, Arc::clone(&LOCAL_FS)))
.collect();
let outcome = load_and_merge_host_skill_roots(
roots,
&self.root_scan_slots,
request.restriction_product,
request.snapshots.as_ref(),
)
.await;
LoadedSkills {
skills: outcome.skills,
errors: outcome.errors,
}
})
}
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
struct ConfigSkillsCacheKey {
roots: Vec<ConfigSkillRootCacheKey>,
+17 -11
View File
@@ -4,12 +4,12 @@ use std::sync::Arc;
use codex_exec_server::ExecutorFileSystem;
use codex_protocol::protocol::SkillScope;
use codex_skills::ParsedSkillFrontmatter;
use codex_skills::SkillError;
use codex_skills::SkillMetadata;
use codex_skills::parse_skill_frontmatter_metadata;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_path_uri::PathUri;
use codex_utils_plugins::PluginIdentity;
#[cfg(test)]
use codex_utils_plugins::PluginSkillRoot;
use codex_utils_plugins::SkillDiscoveryMode;
use futures::StreamExt;
@@ -61,7 +61,6 @@ impl HostSkillRoot {
}
}
#[cfg(test)]
pub(crate) fn plugin(root: PluginSkillRoot, file_system: Arc<dyn ExecutorFileSystem>) -> Self {
Self {
path: root.path,
@@ -88,6 +87,16 @@ impl HostSkillRoot {
self.plugin.as_ref().map(|plugin| &plugin.root)
}
pub(crate) fn plugin_skill_root(&self) -> Option<PluginSkillRoot> {
self.plugin.as_ref().map(|plugin| PluginSkillRoot {
path: self.path.clone(),
plugin_identity: plugin.identity.clone(),
plugin_namespace: plugin.namespace.clone(),
plugin_root: plugin.root.clone(),
discovery_mode: plugin.discovery_mode,
})
}
pub(crate) fn discovery_mode(&self) -> SkillDiscoveryMode {
self.plugin
.as_ref()
@@ -97,20 +106,15 @@ impl HostSkillRoot {
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct HostSkillError {
pub path: AbsolutePathBuf,
pub message: String,
}
/// Skills and errors loaded from one canonical host root.
#[derive(Clone)]
pub struct HostSkillRootSnapshot {
pub root: AbsolutePathBuf,
pub skills: Vec<SkillMetadata>,
pub skill_discovery_path_by_path: Arc<HashMap<AbsolutePathBuf, AbsolutePathBuf>>,
pub errors: Vec<HostSkillError>,
pub errors: Vec<SkillError>,
pub file_system: Arc<dyn ExecutorFileSystem>,
pub is_agent_plugin: bool,
}
struct ResolvedDiscoveredSkill {
@@ -120,6 +124,7 @@ struct ResolvedDiscoveredSkill {
}
pub async fn load_host_skill_root(root: HostSkillRoot) -> HostSkillRootSnapshot {
let is_agent_plugin = root.discovery_mode() == SkillDiscoveryMode::DirectChildren;
let canonical_root =
canonicalize_for_skill_identity(root.file_system.as_ref(), &root.path).await;
let (skills, skill_discovery_path_by_path, errors) =
@@ -130,6 +135,7 @@ pub async fn load_host_skill_root(root: HostSkillRoot) -> HostSkillRootSnapshot
skill_discovery_path_by_path,
errors,
file_system: root.file_system,
is_agent_plugin,
}
}
@@ -139,7 +145,7 @@ async fn load_skills_under_root(
) -> (
Vec<SkillMetadata>,
Arc<HashMap<AbsolutePathBuf, AbsolutePathBuf>>,
Vec<HostSkillError>,
Vec<SkillError>,
) {
let file_system = skill_root.file_system.as_ref();
let plugin_identity = skill_root.plugin_identity();
@@ -309,7 +315,7 @@ async fn load_skills_under_root(
loaded_skills.push(skill);
}
Err(message) if skill_root.scope != SkillScope::System => {
errors.push(HostSkillError { path, message });
errors.push(SkillError { path, message });
}
Err(_) => {}
}
@@ -0,0 +1,232 @@
use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use codex_core_skills::SkillLoadOutcome;
use codex_core_skills::loader::MAX_CONCURRENT_ROOT_SCANS;
use codex_exec_server::ExecutorFileSystem;
use codex_protocol::protocol::Product;
use codex_protocol::protocol::SkillScope;
use codex_skills::LoadedSkillRoot;
use codex_skills::SkillRootSnapshots;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_path_uri::PathUri;
use codex_utils_plugins::PluginSkillRoot;
use codex_utils_plugins::migrated_command_skills_root;
use futures::StreamExt;
use tokio::sync::OnceCell;
use tokio::sync::Semaphore;
use super::HostSkillRoot;
use super::host::HostSkillRootSnapshot;
use super::load_host_skill_root;
pub(crate) async fn load_and_merge_host_skill_roots(
roots: Vec<HostSkillRoot>,
root_scan_slots: &Semaphore,
restriction_product: Option<Product>,
plugin_skill_snapshots: Option<&SkillRootSnapshots<PluginSkillRoot>>,
) -> SkillLoadOutcome {
let migrated_command_roots_by_plugin = roots
.iter()
.filter_map(HostSkillRoot::plugin_root)
.map(|plugin_root| (plugin_root.clone(), Arc::new(OnceCell::new())))
.collect::<HashMap<_, _>>();
let migrated_command_roots = &migrated_command_roots_by_plugin;
let mut indexed_snapshots = futures::stream::iter(roots.into_iter().enumerate())
.map(|(root_index, root)| async move {
let plugin_root = root.plugin_root().cloned();
let migrated_command_root = plugin_root
.as_ref()
.and_then(|plugin_root| migrated_command_roots.get(plugin_root))
.cloned();
let plugin_skill_root = root.plugin_skill_root();
let file_system = Arc::clone(&root.file_system);
let _root_scan_slot = root_scan_slots
.acquire()
.await
.unwrap_or_else(|_| unreachable!());
let cached_snapshot = plugin_skill_root
.as_ref()
.and_then(|plugin_skill_root| plugin_skill_snapshots?.get(plugin_skill_root));
let snapshot = match cached_snapshot {
Some(snapshot) => HostSkillRootSnapshot {
root: snapshot.root,
skills: snapshot.skills,
skill_discovery_path_by_path: snapshot.skill_discovery_path_by_path,
errors: snapshot.errors,
file_system: Arc::clone(&file_system),
is_agent_plugin: snapshot.is_agent_plugin,
},
None => {
let snapshot = load_host_skill_root(root).await;
if let Some(plugin_skill_snapshots) = plugin_skill_snapshots
&& let Some(plugin_skill_root) = plugin_skill_root
{
plugin_skill_snapshots.insert(
plugin_skill_root,
LoadedSkillRoot {
root: snapshot.root.clone(),
skills: snapshot.skills.clone(),
skill_discovery_path_by_path: Arc::clone(
&snapshot.skill_discovery_path_by_path,
),
errors: snapshot.errors.clone(),
is_agent_plugin: snapshot.is_agent_plugin,
},
);
}
snapshot
}
};
let migrated_command_root = match (plugin_root.as_ref(), migrated_command_root) {
(Some(plugin_root), Some(migrated_command_root)) => Some(
migrated_command_root
.get_or_init(|| async {
let migrated_root = migrated_command_skills_root(plugin_root);
file_system
.canonicalize(
&PathUri::from_abs_path(&migrated_root),
/*sandbox*/ None,
)
.await
.and_then(|path| path.to_abs_path())
.unwrap_or(migrated_root)
})
.await
.clone(),
),
_ => None,
};
(root_index, migrated_command_root, snapshot)
})
.buffer_unordered(MAX_CONCURRENT_ROOT_SCANS)
.collect::<Vec<_>>()
.await;
indexed_snapshots.sort_unstable_by_key(|(root_index, _, _)| *root_index);
for (_, _, snapshot) in &mut indexed_snapshots {
snapshot
.skills
.retain(|skill| skill.matches_product_restriction_for_product(restriction_product));
}
let mut native_plugin_skill_names = HashMap::<String, HashSet<String>>::new();
for skill in indexed_snapshots
.iter()
.flat_map(|(_, migrated_command_root, snapshot)| {
snapshot.skills.iter().filter(|skill| {
!migrated_command_root.as_ref().is_some_and(|migrated_root| {
skill
.path_to_skills_md
.as_path()
.starts_with(migrated_root.as_path())
})
})
})
{
if let Some(plugin_id) = &skill.plugin_id {
native_plugin_skill_names
.entry(plugin_id.clone())
.or_default()
.insert(skill.name.clone());
}
}
for (_, migrated_command_root, snapshot) in &mut indexed_snapshots {
let Some(migrated_command_root) = migrated_command_root else {
continue;
};
snapshot.skills.retain(|skill| {
!skill
.path_to_skills_md
.as_path()
.starts_with(migrated_command_root.as_path())
|| skill.plugin_id.as_ref().is_none_or(|plugin_id| {
native_plugin_skill_names
.get(plugin_id)
.is_none_or(|names| !names.contains(&skill.name))
})
});
}
merge_host_skill_root_snapshots(
indexed_snapshots
.into_iter()
.map(|(_, _, snapshot)| snapshot)
.collect(),
)
}
fn merge_host_skill_root_snapshots(snapshots: Vec<HostSkillRootSnapshot>) -> SkillLoadOutcome {
let mut skills = Vec::new();
let mut errors = Vec::new();
let mut skill_roots = Vec::new();
let mut skill_root_by_path = HashMap::new();
let mut skill_discovery_path_by_path = HashMap::new();
let mut agent_plugin_skill_paths = HashSet::new();
let mut file_systems_by_skill_path =
HashMap::<AbsolutePathBuf, Arc<dyn ExecutorFileSystem>>::new();
for snapshot in snapshots {
if !snapshot.skills.is_empty() && !skill_roots.contains(&snapshot.root) {
skill_roots.push(snapshot.root.clone());
}
for skill in &snapshot.skills {
let path = skill.path_to_skills_md.clone();
if !skill_root_by_path.contains_key(&path) {
skill_root_by_path.insert(path.clone(), snapshot.root.clone());
if let Some(discovery_path) = snapshot.skill_discovery_path_by_path.get(&path) {
skill_discovery_path_by_path.insert(path.clone(), discovery_path.clone());
}
file_systems_by_skill_path.insert(path.clone(), Arc::clone(&snapshot.file_system));
if snapshot.is_agent_plugin {
agent_plugin_skill_paths.insert(path);
}
}
}
skills.extend(snapshot.skills);
errors.extend(snapshot.errors);
}
let mut seen_paths = HashSet::new();
skills.retain(|skill| seen_paths.insert(skill.path_to_skills_md.clone()));
let retained_paths = skills
.iter()
.map(|skill| skill.path_to_skills_md.clone())
.collect::<HashSet<_>>();
skill_root_by_path.retain(|path, _| retained_paths.contains(path));
skill_discovery_path_by_path.retain(|path, _| retained_paths.contains(path));
let retained_roots = skill_root_by_path.values().collect::<HashSet<_>>();
skill_roots.retain(|root| retained_roots.contains(root));
file_systems_by_skill_path.retain(|path, _| retained_paths.contains(path));
agent_plugin_skill_paths.retain(|path| retained_paths.contains(path));
skills.sort_by(|left, right| {
scope_rank(left.scope)
.cmp(&scope_rank(right.scope))
.then_with(|| left.name.cmp(&right.name))
.then_with(|| left.path_to_skills_md.cmp(&right.path_to_skills_md))
});
SkillLoadOutcome::from_parts(
skills,
errors,
skill_roots,
skill_root_by_path,
skill_discovery_path_by_path,
agent_plugin_skill_paths,
file_systems_by_skill_path,
)
}
fn scope_rank(scope: SkillScope) -> u8 {
match scope {
SkillScope::Repo => 0,
SkillScope::User => 1,
SkillScope::System => 2,
SkillScope::Admin => 3,
}
}
#[cfg(test)]
#[path = "host_merge_tests.rs"]
mod tests;
@@ -0,0 +1,313 @@
use std::collections::HashMap;
use std::fs;
use std::sync::Arc;
use std::sync::Mutex;
use codex_exec_server::LOCAL_FS;
use codex_protocol::protocol::Product;
use codex_skills::LoadedSkillRoot;
use codex_skills::SkillRootSnapshotCache;
use codex_skills::SkillRootSnapshots;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_plugins::PluginIdentity;
use codex_utils_plugins::PluginSkillRoot;
use codex_utils_plugins::SkillDiscoveryMode;
use codex_utils_plugins::migrated_command_skills_root;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use tokio::sync::Semaphore;
use super::HostSkillRoot;
use super::load_and_merge_host_skill_roots;
#[derive(Default)]
struct TestPluginSkillSnapshots {
snapshots: Mutex<HashMap<PluginSkillRoot, LoadedSkillRoot>>,
}
impl SkillRootSnapshotCache<PluginSkillRoot> for TestPluginSkillSnapshots {
fn get(&self, root: &PluginSkillRoot) -> Option<LoadedSkillRoot> {
self.snapshots.lock().unwrap().get(root).cloned()
}
fn insert(&self, root: PluginSkillRoot, snapshot: LoadedSkillRoot) {
self.snapshots.lock().unwrap().insert(root, snapshot);
}
}
fn write_skill(root: &AbsolutePathBuf, directory: &str, name: &str, description: &str) {
let skill_dir = root.join(directory);
fs::create_dir_all(&skill_dir).expect("create skill directory");
fs::write(
skill_dir.join("SKILL.md"),
format!("---\nname: {name}\ndescription: {description}\n---\n"),
)
.expect("write skill");
}
struct PluginSkillsFixture {
_temp_dir: TempDir,
plugin_path: AbsolutePathBuf,
native_root: AbsolutePathBuf,
migrated_root: AbsolutePathBuf,
identity: PluginIdentity,
}
impl PluginSkillsFixture {
fn new() -> Self {
let temp_dir = TempDir::new().expect("temp dir");
let plugin_path = AbsolutePathBuf::from_absolute_path_checked(temp_dir.path())
.expect("absolute plugin path");
let native_root = plugin_path.join("skills");
let migrated_root = migrated_command_skills_root(&plugin_path);
write_skill(&native_root, "review", "source-command-review", "native");
write_skill(
&migrated_root,
"review",
"source-command-review",
"migrated duplicate",
);
write_skill(
&migrated_root,
"only",
"source-command-only",
"migrated only",
);
Self {
_temp_dir: temp_dir,
plugin_path,
native_root,
migrated_root,
identity: PluginIdentity {
plugin_id: "sample@test".to_string(),
remote_plugin_id: Some("remote-sample".to_string()),
},
}
}
fn root(&self, path: AbsolutePathBuf) -> HostSkillRoot {
HostSkillRoot::plugin(
PluginSkillRoot {
path,
plugin_identity: self.identity.clone(),
plugin_namespace: "sample".to_string(),
plugin_root: self.plugin_path.clone(),
discovery_mode: SkillDiscoveryMode::Recursive,
},
Arc::clone(&LOCAL_FS),
)
}
fn roots(&self) -> Vec<HostSkillRoot> {
[self.migrated_root.clone(), self.native_root.clone()]
.into_iter()
.map(|path| self.root(path))
.collect()
}
}
#[tokio::test]
async fn native_plugin_skill_replaces_migrated_command_with_the_same_name() {
let fixture = PluginSkillsFixture::new();
let outcome = load_and_merge_host_skill_roots(
fixture.roots(),
&Semaphore::new(2),
/*restriction_product*/ None,
/*plugin_skill_snapshots*/ None,
)
.await;
assert_eq!(
outcome
.skills
.iter()
.map(|skill| (skill.name.as_str(), skill.description.as_str()))
.collect::<Vec<_>>(),
vec![
("sample:source-command-only", "migrated only"),
("sample:source-command-review", "native"),
]
);
}
#[tokio::test]
async fn nested_migrated_command_root_preserves_native_skill_precedence() {
let fixture = PluginSkillsFixture::new();
let outcome = load_and_merge_host_skill_roots(
vec![
fixture.root(fixture.migrated_root.join("review")),
fixture.root(fixture.native_root.clone()),
fixture.root(fixture.migrated_root.clone()),
],
&Semaphore::new(3),
/*restriction_product*/ None,
/*plugin_skill_snapshots*/ None,
)
.await;
assert_eq!(
outcome
.skills
.iter()
.map(|skill| (skill.name.as_str(), skill.description.as_str()))
.collect::<Vec<_>>(),
vec![
("sample:source-command-only", "migrated only"),
("sample:source-command-review", "native"),
]
);
}
#[tokio::test]
async fn product_filtered_native_skill_does_not_hide_migrated_command() {
let fixture = PluginSkillsFixture::new();
let metadata_dir = fixture.native_root.join("review/agents");
fs::create_dir_all(&metadata_dir).expect("create metadata directory");
fs::write(
metadata_dir.join("openai.yaml"),
"policy:\n products: [CHATGPT]\n",
)
.expect("write product policy");
let outcome = load_and_merge_host_skill_roots(
fixture.roots(),
&Semaphore::new(2),
Some(Product::Codex),
/*plugin_skill_snapshots*/ None,
)
.await;
assert_eq!(
outcome
.skills
.iter()
.map(|skill| (skill.name.as_str(), skill.description.as_str()))
.collect::<Vec<_>>(),
vec![
("sample:source-command-only", "migrated only"),
("sample:source-command-review", "migrated duplicate"),
]
);
}
#[tokio::test]
async fn overlapping_plugin_roots_keep_one_canonical_skill() {
let fixture = PluginSkillsFixture::new();
let outcome = load_and_merge_host_skill_roots(
vec![
fixture.root(fixture.native_root.clone()),
fixture.root(fixture.native_root.join("review")),
],
&Semaphore::new(2),
/*restriction_product*/ None,
/*plugin_skill_snapshots*/ None,
)
.await;
assert_eq!(
outcome
.skills
.iter()
.map(|skill| skill.name.as_str())
.collect::<Vec<_>>(),
vec!["sample:source-command-review"]
);
}
#[tokio::test]
async fn reuses_owner_managed_plugin_root_snapshots() {
let fixture = PluginSkillsFixture::new();
let snapshots = SkillRootSnapshots::new(Arc::new(TestPluginSkillSnapshots::default()));
let first_outcome = load_and_merge_host_skill_roots(
vec![fixture.root(fixture.native_root.clone())],
&Semaphore::new(2),
/*restriction_product*/ None,
Some(&snapshots),
)
.await;
write_skill(
&fixture.native_root,
"review",
"source-command-review",
"updated",
);
let cached_outcome = load_and_merge_host_skill_roots(
vec![fixture.root(fixture.native_root.clone())],
&Semaphore::new(2),
/*restriction_product*/ None,
Some(&snapshots),
)
.await;
let refreshed_outcome = load_and_merge_host_skill_roots(
vec![fixture.root(fixture.native_root.clone())],
&Semaphore::new(2),
/*restriction_product*/ None,
/*plugin_skill_snapshots*/ None,
)
.await;
assert_eq!(first_outcome.skills, cached_outcome.skills);
assert_eq!(cached_outcome.skills[0].description, "native");
assert_eq!(refreshed_outcome.skills[0].description, "updated");
assert_eq!(
cached_outcome.skills[0].remote_plugin_id.as_deref(),
Some("remote-sample")
);
}
#[tokio::test]
async fn preserves_plugin_skill_errors() {
let fixture = PluginSkillsFixture::new();
fs::write(
fixture.native_root.join("review/SKILL.md"),
"missing frontmatter",
)
.expect("write invalid plugin skill");
let outcome = load_and_merge_host_skill_roots(
vec![fixture.root(fixture.native_root.clone())],
&Semaphore::new(2),
/*restriction_product*/ None,
/*plugin_skill_snapshots*/ None,
)
.await;
assert_eq!(outcome.skills, Vec::new());
assert_eq!(outcome.errors.len(), 1);
}
#[cfg(unix)]
#[tokio::test]
async fn recognizes_symlinked_migrated_command_roots() {
let fixture = PluginSkillsFixture::new();
let alias = fixture.plugin_path.join("migrated-alias");
std::os::unix::fs::symlink(fixture.migrated_root.as_path(), alias.as_path())
.expect("create migrated root alias");
let outcome = load_and_merge_host_skill_roots(
vec![
fixture.root(alias),
fixture.root(fixture.native_root.clone()),
],
&Semaphore::new(2),
/*restriction_product*/ None,
/*plugin_skill_snapshots*/ None,
)
.await;
assert_eq!(
outcome
.skills
.iter()
.map(|skill| (skill.name.as_str(), skill.description.as_str()))
.collect::<Vec<_>>(),
vec![
("sample:source-command-only", "migrated only"),
("sample:source-command-review", "native"),
]
);
}
+2
View File
@@ -1,6 +1,7 @@
mod discovery;
mod environment;
mod host;
mod host_merge;
mod metadata;
mod namespace;
@@ -8,6 +9,7 @@ pub(crate) use environment::load_environment_skills_from_discovery;
pub(crate) use environment::load_environment_skills_from_root;
pub(crate) use host::HostSkillRoot;
pub(crate) use host::load_host_skill_root;
pub(crate) use host_merge::load_and_merge_host_skill_roots;
pub(super) const SKILLS_FILENAME: &str = "SKILL.md";
pub(super) const SKILLS_METADATA_DIR: &str = "agents";