From ee1a3e594a394764a3aababbcaffbccb68a775ad Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 25 Jun 2026 04:52:07 +0000 Subject: [PATCH 1/6] Parallelize environment skill loading --- .../core-skills/src/loader/environment.rs | 21 ++++++++++++------- 1 file changed, 13 insertions(+), 8 deletions(-) diff --git a/codex-rs/core-skills/src/loader/environment.rs b/codex-rs/core-skills/src/loader/environment.rs index 5dc29ebfb263..2def024267ad 100644 --- a/codex-rs/core-skills/src/loader/environment.rs +++ b/codex-rs/core-skills/src/loader/environment.rs @@ -220,7 +220,9 @@ pub async fn load_environment_skills_from_root( .filter_map(|(plugin_root, namespace)| namespace.map(|namespace| (plugin_root, namespace))) .collect::>(); - for path in discovery.skill_files { + // Remote executors can multiplex these independent per-skill reads, so polling them together + // allows the I/O for each skill and its metadata to happen concurrently. + let skill_results = join_all(discovery.skill_files.into_iter().map(|path| { let mut ancestor = path.parent(); let plugin_namespace = loop { let Some(current) = ancestor else { @@ -231,13 +233,16 @@ pub async fn load_environment_skills_from_root( } ancestor = current.parent(); }; - match EnvironmentSkillMetadata::parse( - file_system, - &path, - /*plugin_namespace*/ plugin_namespace, - ) - .await - { + async move { + let result = + EnvironmentSkillMetadata::parse(file_system, &path, plugin_namespace).await; + (path, result) + } + })) + .await; + + for (path, result) in skill_results { + match result { Ok(skill) if skill.matches_product_restriction(restriction_product) => { outcome.skills.push(skill); } From 04a59fce920d674613b6a95077e7abb3b50c210b Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 25 Jun 2026 05:29:02 +0000 Subject: [PATCH 2/6] codex: address PR review feedback (#29990) --- .../core-skills/src/loader/environment.rs | 41 +++++++++++-------- 1 file changed, 23 insertions(+), 18 deletions(-) diff --git a/codex-rs/core-skills/src/loader/environment.rs b/codex-rs/core-skills/src/loader/environment.rs index 2def024267ad..b7b84d0e8085 100644 --- a/codex-rs/core-skills/src/loader/environment.rs +++ b/codex-rs/core-skills/src/loader/environment.rs @@ -10,6 +10,7 @@ use codex_utils_path_uri::PathUri; use codex_utils_plugins::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; use codex_utils_plugins::plugin_namespace_for_root_uri; use codex_utils_plugins::plugin_namespace_for_skill_uri; +use futures::StreamExt; use futures::future::join_all; use crate::model::SkillDependencies; @@ -31,6 +32,7 @@ use super::sanitize_single_line; use super::validate_len; const MAX_SKILLS_ENTRIES_PER_ROOT: usize = 20_000; +const MAX_CONCURRENT_SKILL_LOADS: usize = 16; /// URI-native metadata for one skill owned by an execution environment. #[derive(Clone, Debug, PartialEq, Eq)] @@ -220,26 +222,29 @@ pub async fn load_environment_skills_from_root( .filter_map(|(plugin_root, namespace)| namespace.map(|namespace| (plugin_root, namespace))) .collect::>(); - // Remote executors can multiplex these independent per-skill reads, so polling them together - // allows the I/O for each skill and its metadata to happen concurrently. - let skill_results = join_all(discovery.skill_files.into_iter().map(|path| { - let mut ancestor = path.parent(); - let plugin_namespace = loop { - let Some(current) = ancestor else { - break None; + // Remote executors can multiplex these independent per-skill reads, so polling a bounded + // number together allows the I/O for each skill and its metadata to happen concurrently. + let skill_results = futures::stream::iter(discovery.skill_files) + .map(|path| { + let mut ancestor = path.parent(); + let plugin_namespace = loop { + let Some(current) = ancestor else { + break None; + }; + if let Some(namespace) = plugin_namespaces.get(¤t) { + break Some(namespace.as_str()); + } + ancestor = current.parent(); }; - if let Some(namespace) = plugin_namespaces.get(¤t) { - break Some(namespace.as_str()); + async move { + let result = + EnvironmentSkillMetadata::parse(file_system, &path, plugin_namespace).await; + (path, result) } - ancestor = current.parent(); - }; - async move { - let result = - EnvironmentSkillMetadata::parse(file_system, &path, plugin_namespace).await; - (path, result) - } - })) - .await; + }) + .buffered(MAX_CONCURRENT_SKILL_LOADS) + .collect::>() + .await; for (path, result) in skill_results { match result { From d1121cf83a5de3c504c3dc9e482348b5fd3eab1e Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 25 Jun 2026 05:35:13 +0000 Subject: [PATCH 3/6] codex: address PR review feedback (#29990) --- .../core-skills/src/loader/environment.rs | 41 ++++++++----------- 1 file changed, 18 insertions(+), 23 deletions(-) diff --git a/codex-rs/core-skills/src/loader/environment.rs b/codex-rs/core-skills/src/loader/environment.rs index b7b84d0e8085..2def024267ad 100644 --- a/codex-rs/core-skills/src/loader/environment.rs +++ b/codex-rs/core-skills/src/loader/environment.rs @@ -10,7 +10,6 @@ use codex_utils_path_uri::PathUri; use codex_utils_plugins::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; use codex_utils_plugins::plugin_namespace_for_root_uri; use codex_utils_plugins::plugin_namespace_for_skill_uri; -use futures::StreamExt; use futures::future::join_all; use crate::model::SkillDependencies; @@ -32,7 +31,6 @@ use super::sanitize_single_line; use super::validate_len; const MAX_SKILLS_ENTRIES_PER_ROOT: usize = 20_000; -const MAX_CONCURRENT_SKILL_LOADS: usize = 16; /// URI-native metadata for one skill owned by an execution environment. #[derive(Clone, Debug, PartialEq, Eq)] @@ -222,29 +220,26 @@ pub async fn load_environment_skills_from_root( .filter_map(|(plugin_root, namespace)| namespace.map(|namespace| (plugin_root, namespace))) .collect::>(); - // Remote executors can multiplex these independent per-skill reads, so polling a bounded - // number together allows the I/O for each skill and its metadata to happen concurrently. - let skill_results = futures::stream::iter(discovery.skill_files) - .map(|path| { - let mut ancestor = path.parent(); - let plugin_namespace = loop { - let Some(current) = ancestor else { - break None; - }; - if let Some(namespace) = plugin_namespaces.get(¤t) { - break Some(namespace.as_str()); - } - ancestor = current.parent(); + // Remote executors can multiplex these independent per-skill reads, so polling them together + // allows the I/O for each skill and its metadata to happen concurrently. + let skill_results = join_all(discovery.skill_files.into_iter().map(|path| { + let mut ancestor = path.parent(); + let plugin_namespace = loop { + let Some(current) = ancestor else { + break None; }; - async move { - let result = - EnvironmentSkillMetadata::parse(file_system, &path, plugin_namespace).await; - (path, result) + if let Some(namespace) = plugin_namespaces.get(¤t) { + break Some(namespace.as_str()); } - }) - .buffered(MAX_CONCURRENT_SKILL_LOADS) - .collect::>() - .await; + ancestor = current.parent(); + }; + async move { + let result = + EnvironmentSkillMetadata::parse(file_system, &path, plugin_namespace).await; + (path, result) + } + })) + .await; for (path, result) in skill_results { match result { From 3db9e2841d2370508fd539220ea0131124fb0dd2 Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 25 Jun 2026 05:36:43 +0000 Subject: [PATCH 4/6] codex: address PR review feedback (#29990) --- .../core-skills/src/loader/environment.rs | 41 +++++++++++-------- 1 file changed, 23 insertions(+), 18 deletions(-) diff --git a/codex-rs/core-skills/src/loader/environment.rs b/codex-rs/core-skills/src/loader/environment.rs index 2def024267ad..e380f0a0e43f 100644 --- a/codex-rs/core-skills/src/loader/environment.rs +++ b/codex-rs/core-skills/src/loader/environment.rs @@ -10,6 +10,7 @@ use codex_utils_path_uri::PathUri; use codex_utils_plugins::DISCOVERABLE_PLUGIN_MANIFEST_PATHS; use codex_utils_plugins::plugin_namespace_for_root_uri; use codex_utils_plugins::plugin_namespace_for_skill_uri; +use futures::StreamExt; use futures::future::join_all; use crate::model::SkillDependencies; @@ -31,6 +32,7 @@ use super::sanitize_single_line; use super::validate_len; const MAX_SKILLS_ENTRIES_PER_ROOT: usize = 20_000; +const MAX_CONCURRENT_SKILL_LOADS: usize = 256; /// URI-native metadata for one skill owned by an execution environment. #[derive(Clone, Debug, PartialEq, Eq)] @@ -220,26 +222,29 @@ pub async fn load_environment_skills_from_root( .filter_map(|(plugin_root, namespace)| namespace.map(|namespace| (plugin_root, namespace))) .collect::>(); - // Remote executors can multiplex these independent per-skill reads, so polling them together - // allows the I/O for each skill and its metadata to happen concurrently. - let skill_results = join_all(discovery.skill_files.into_iter().map(|path| { - let mut ancestor = path.parent(); - let plugin_namespace = loop { - let Some(current) = ancestor else { - break None; + // Remote executors can multiplex these independent per-skill reads, so polling a bounded + // number together allows the I/O for each skill and its metadata to happen concurrently. + let skill_results = futures::stream::iter(discovery.skill_files) + .map(|path| { + let mut ancestor = path.parent(); + let plugin_namespace = loop { + let Some(current) = ancestor else { + break None; + }; + if let Some(namespace) = plugin_namespaces.get(¤t) { + break Some(namespace.as_str()); + } + ancestor = current.parent(); }; - if let Some(namespace) = plugin_namespaces.get(¤t) { - break Some(namespace.as_str()); + async move { + let result = + EnvironmentSkillMetadata::parse(file_system, &path, plugin_namespace).await; + (path, result) } - ancestor = current.parent(); - }; - async move { - let result = - EnvironmentSkillMetadata::parse(file_system, &path, plugin_namespace).await; - (path, result) - } - })) - .await; + }) + .buffered(MAX_CONCURRENT_SKILL_LOADS) + .collect::>() + .await; for (path, result) in skill_results { match result { From be61f7175f2dd4c339f1d9e5cefbcd76960b7d9e Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 25 Jun 2026 05:55:39 +0000 Subject: [PATCH 5/6] codex: address PR review feedback (#29990) --- .../core-skills/src/loader/environment.rs | 37 +++++++++++++------ 1 file changed, 26 insertions(+), 11 deletions(-) diff --git a/codex-rs/core-skills/src/loader/environment.rs b/codex-rs/core-skills/src/loader/environment.rs index e380f0a0e43f..dbb3f007ba6b 100644 --- a/codex-rs/core-skills/src/loader/environment.rs +++ b/codex-rs/core-skills/src/loader/environment.rs @@ -1,5 +1,6 @@ use std::collections::HashMap; use std::collections::HashSet; +use std::future::Future; use std::io; use codex_exec_server::ExecutorFileSystem; @@ -32,7 +33,7 @@ use super::sanitize_single_line; use super::validate_len; const MAX_SKILLS_ENTRIES_PER_ROOT: usize = 20_000; -const MAX_CONCURRENT_SKILL_LOADS: usize = 256; +const MAX_CONCURRENT_SKILL_LOADS: usize = 64; /// URI-native metadata for one skill owned by an execution environment. #[derive(Clone, Debug, PartialEq, Eq)] @@ -70,10 +71,11 @@ impl EnvironmentSkillMetadata { path: &PathUri, plugin_namespace: Option<&str>, ) -> Result { - let contents = file_system - .read_file_text(path, /*sandbox*/ None) - .await - .map_err(|err| format!("failed to read file: {err}"))?; + let contents = retry_once_on_broken_pipe(|| { + file_system.read_file_text(path, /*sandbox*/ None) + }) + .await + .map_err(|err| format!("failed to read file: {err}"))?; let ParsedSkillFrontmatter { name: base_name, description, @@ -279,9 +281,10 @@ async fn load_skill_metadata( else { return (None, None); }; - match file_system - .get_metadata(&metadata_path, /*sandbox*/ None) - .await + match retry_once_on_broken_pipe(|| { + file_system.get_metadata(&metadata_path, /*sandbox*/ None) + }) + .await { Ok(metadata) if metadata.is_file => {} Ok(_) => return (None, None), @@ -291,9 +294,10 @@ async fn load_skill_metadata( return (None, None); } } - let contents = match file_system - .read_file_text(&metadata_path, /*sandbox*/ None) - .await + let contents = match retry_once_on_broken_pipe(|| { + file_system.read_file_text(&metadata_path, /*sandbox*/ None) + }) + .await { Ok(contents) => contents, Err(error) => { @@ -315,6 +319,17 @@ async fn load_skill_metadata( ) } +async fn retry_once_on_broken_pipe(mut operation: F) -> io::Result +where + F: FnMut() -> Fut, + Fut: Future>, +{ + match operation().await { + Err(error) if error.kind() == io::ErrorKind::BrokenPipe => operation().await, + result => result, + } +} + fn default_skill_name(path: &PathUri) -> String { path.parent() .and_then(|parent| parent.basename()) From 56e9e3f11ec9c214efdb8b6cd8b283fe61f1c40d Mon Sep 17 00:00:00 2001 From: Adam Perry Date: Thu, 25 Jun 2026 06:06:14 +0000 Subject: [PATCH 6/6] codex: address PR review feedback (#29990) --- .../core-skills/src/loader/environment.rs | 35 ++++++------------- 1 file changed, 10 insertions(+), 25 deletions(-) diff --git a/codex-rs/core-skills/src/loader/environment.rs b/codex-rs/core-skills/src/loader/environment.rs index dbb3f007ba6b..93ab14e9aa58 100644 --- a/codex-rs/core-skills/src/loader/environment.rs +++ b/codex-rs/core-skills/src/loader/environment.rs @@ -1,6 +1,5 @@ use std::collections::HashMap; use std::collections::HashSet; -use std::future::Future; use std::io; use codex_exec_server::ExecutorFileSystem; @@ -71,11 +70,10 @@ impl EnvironmentSkillMetadata { path: &PathUri, plugin_namespace: Option<&str>, ) -> Result { - let contents = retry_once_on_broken_pipe(|| { - file_system.read_file_text(path, /*sandbox*/ None) - }) - .await - .map_err(|err| format!("failed to read file: {err}"))?; + let contents = file_system + .read_file_text(path, /*sandbox*/ None) + .await + .map_err(|err| format!("failed to read file: {err}"))?; let ParsedSkillFrontmatter { name: base_name, description, @@ -281,10 +279,9 @@ async fn load_skill_metadata( else { return (None, None); }; - match retry_once_on_broken_pipe(|| { - file_system.get_metadata(&metadata_path, /*sandbox*/ None) - }) - .await + match file_system + .get_metadata(&metadata_path, /*sandbox*/ None) + .await { Ok(metadata) if metadata.is_file => {} Ok(_) => return (None, None), @@ -294,10 +291,9 @@ async fn load_skill_metadata( return (None, None); } } - let contents = match retry_once_on_broken_pipe(|| { - file_system.read_file_text(&metadata_path, /*sandbox*/ None) - }) - .await + let contents = match file_system + .read_file_text(&metadata_path, /*sandbox*/ None) + .await { Ok(contents) => contents, Err(error) => { @@ -319,17 +315,6 @@ async fn load_skill_metadata( ) } -async fn retry_once_on_broken_pipe(mut operation: F) -> io::Result -where - F: FnMut() -> Fut, - Fut: Future>, -{ - match operation().await { - Err(error) if error.kind() == io::ErrorKind::BrokenPipe => operation().await, - result => result, - } -} - fn default_skill_name(path: &PathUri) -> String { path.parent() .and_then(|parent| parent.basename())