openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Cache executor skill catalogs for stable roots

Browse Source
This commit is contained in:
jif-oai
2026-06-25 17:41:49 +01:00
parent 3a27647da3
commit 5998bdd78e
12 changed files with 328 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
-2
View File
@@ -2813,9 +2813,7 @@ name = "codex-core-skills"
version = "0.0.0"
dependencies = [
"anyhow",
"codex-analytics",
"codex-config",
"codex-context-fragments",
"codex-exec-server",
"codex-login",
"codex-model-provider",
codex-rs/core-skills/Cargo.toml
-2
View File
@@ -14,9 +14,7 @@ workspace = true
[dependencies]
anyhow = { workspace = true }
codex-analytics = { workspace = true }
codex-config = { workspace = true }
codex-context-fragments = { workspace = true }
codex-exec-server = { workspace = true }
codex-login = { workspace = true }
codex-model-provider = { workspace = true }
codex-rs/core-skills/src/executor_runtime.rs
+228
View File
@@ -0,0 +1,228 @@
use std::sync::Arc;
use std::sync::Mutex;
use codex_exec_server::ResolvedSelectedCapabilityRoot;
use codex_protocol::capabilities::CapabilityRootLocation;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use codex_protocol::protocol::Product;
use crate::loader::EnvironmentSkillMetadata;
use crate::loader::load_environment_skills_from_root;
use crate::runtime::SkillAuthority;
use crate::runtime::SkillCatalog;
use crate::runtime::SkillCatalogEntry;
use crate::runtime::SkillPackageId;
use crate::runtime::SkillReadRequest;
use crate::runtime::SkillReadResult;
use crate::runtime::SkillResourceId;
use crate::runtime::SkillSource;
use crate::runtime::SkillSourceError;
use crate::runtime::SkillSourceFuture;
use crate::runtime::SkillSourceIdentity;
use crate::runtime::SkillSourceKind;
/// Session-lifetime cache for catalogs discovered from stable executor capability roots.
///
/// Cache entries are keyed by the complete selected root and product restriction, not by the
/// process-local [`ResolvedSelectedCapabilityRoot`]. Environment availability therefore controls
/// whether a cached catalog is projected into a model step, but temporarily losing an environment
/// does not invalidate its catalog. A different root identity, environment ID, path, or product
/// restriction produces a cache miss and a new discovery.
///
/// There is intentionally no filesystem-based invalidation. Selected environment contents are
/// treated as stable for the lifetime of a session. Dropping this cache at session shutdown is the
/// only operation that invalidates successful entries. Catalogs containing warnings are not
/// cached, so transient discovery failures can recover on a later model step.
#[derive(Default)]
pub struct ExecutorSkillCatalogCache {
entries: Mutex<Vec<CachedExecutorSkillCatalog>>,
}
struct CachedExecutorSkillCatalog {
key: ExecutorSkillCatalogCacheKey,
catalog: Arc<SkillCatalog>,
}
#[derive(Clone, Debug, PartialEq, Eq)]
struct ExecutorSkillCatalogCacheKey {
selected_root: SelectedCapabilityRoot,
restriction_product: Option<Product>,
}
impl ExecutorSkillCatalogCache {
pub(crate) async fn catalog_for_stable_root(
&self,
source: &ExecutorSkillSource,
) -> Arc<SkillCatalog> {
let key = source.catalog_cache_key();
if let Some(catalog) = self.catalog(&key) {
return catalog;
}
// Do not hold the cache lock across executor I/O. Concurrent first loads may duplicate
// discovery, but the second check below ensures only one result becomes authoritative.
let discovered = Arc::new(source.load_catalog().await);
if !discovered.warnings.is_empty() {
return discovered;
}
let mut entries = self
.entries
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
if let Some(entry) = entries.iter().find(|entry| entry.key == key) {
return Arc::clone(&entry.catalog);
}
entries.push(CachedExecutorSkillCatalog {
key,
catalog: Arc::clone(&discovered),
});
discovered
}
fn catalog(&self, key: &ExecutorSkillCatalogCacheKey) -> Option<Arc<SkillCatalog>> {
self.entries
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.iter()
.find(|entry| &entry.key == key)
.map(|entry| Arc::clone(&entry.catalog))
}
}
pub(crate) struct ExecutorSkillSource {
root: ResolvedSelectedCapabilityRoot,
authority: SkillAuthority,
restriction_product: Option<Product>,
identity: SkillSourceIdentity,
}
impl ExecutorSkillSource {
pub(crate) fn new(
root: ResolvedSelectedCapabilityRoot,
restriction_product: Option<Product>,
) -> Self {
Self {
authority: SkillAuthority::new(
SkillSourceKind::Executor,
root.selected_root().id.clone(),
),
identity: SkillSourceIdentity::from_owner(Arc::clone(root.environment())),
root,
restriction_product,
}
}
fn catalog_cache_key(&self) -> ExecutorSkillCatalogCacheKey {
ExecutorSkillCatalogCacheKey {
selected_root: self.root.selected_root().clone(),
restriction_product: self.restriction_product,
}
}
async fn load_catalog(&self) -> SkillCatalog {
let CapabilityRootLocation::Environment {
environment_id,
path,
} = &self.root.selected_root().location;
let outcome = load_environment_skills_from_root(
self.root.file_system().as_ref(),
path,
self.restriction_product,
)
.await;
let mut catalog = SkillCatalog {
warnings: outcome.warnings,
..Default::default()
};
for skill in outcome.skills {
catalog.push_entry(executor_catalog_entry(
skill,
self.authority.clone(),
&self.root.selected_root().id,
environment_id,
));
}
catalog
}
}
impl SkillSource for ExecutorSkillSource {
fn identity(&self) -> SkillSourceIdentity {
self.identity.clone()
}
fn list(&self) -> SkillSourceFuture<'_, SkillCatalog> {
Box::pin(async move { Ok(self.load_catalog().await) })
}
fn read(&self, request: SkillReadRequest) -> SkillSourceFuture<'_, SkillReadResult> {
Box::pin(async move {
if request.authority != self.authority || request.package.0 != request.resource.as_str()
{
return Err(SkillSourceError::new(
"executor skill resource does not match its captured source",
));
}
let CapabilityRootLocation::Environment { environment_id, .. } =
&self.root.selected_root().location;
let Some((resource_environment, path)) = request.resource.environment_path() else {
return Err(SkillSourceError::new(
"executor skill resource has no environment path",
));
};
if resource_environment != environment_id {
return Err(SkillSourceError::new(
"executor skill resource belongs to a different environment",
));
}
let contents = self
.root
.file_system()
.read_file_text(path, /*sandbox*/ None)
.await
.map_err(|err| {
SkillSourceError::new(format!(
"failed to read executor skill resource {}: {err}",
request.resource.as_str()
))
})?;
Ok(SkillReadResult {
resource: request.resource,
contents,
})
})
}
}
fn executor_catalog_entry(
skill: EnvironmentSkillMetadata,
authority: SkillAuthority,
root_id: &str,
environment_id: &str,
) -> SkillCatalogEntry {
let prompt_visible = skill.allows_implicit_invocation();
let path = skill.path_to_skills_md.inferred_native_path_string();
let display_path = format!(
"skill://{root_id}/{}",
path.replace('\\', "/").trim_start_matches('/')
);
let entry = SkillCatalogEntry::new(
SkillPackageId(display_path.clone()),
authority,
skill.name,
skill.description,
SkillResourceId::environment(
display_path.clone(),
environment_id,
skill.path_to_skills_md,
),
)
.with_short_description(skill.short_description)
.with_display_path(display_path)
.with_dependencies(skill.dependencies);
if prompt_visible {
entry
} else {
entry.hidden_from_prompt()
}
}
codex-rs/core-skills/src/injection_tests.rs
+1 -1
View File
@@ -37,7 +37,7 @@ fn extract_tool_mentions_requires_link_syntax() {
#[test]
fn extract_tool_mentions_trims_linked_paths_and_allows_spacing() {
assert_mentions("use [$beta] ( /tmp/beta )", &["beta"], &["/tmp/beta"]);
assert_mentions("use [$beta] ( /tmp/beta )", &[], &["/tmp/beta"]);
}
#[test]
codex-rs/core-skills/src/lib.rs
+2
View File
@@ -1,4 +1,5 @@
pub mod config_rules;
mod executor_runtime;
pub mod injection;
pub(crate) mod invocation_utils;
pub mod loader;
@@ -13,6 +14,7 @@ mod runtime_snapshot;
pub mod service;
pub mod system;
pub use executor_runtime::ExecutorSkillCatalogCache;
pub(crate) use invocation_utils::build_implicit_skill_path_indexes;
pub use invocation_utils::detect_implicit_skill_invocation_for_command;
pub use mention_counts::build_skill_name_counts;
codex-rs/core-skills/src/runtime_snapshot.rs
+53 -136
View File
@@ -4,17 +4,16 @@ use std::fmt;
use std::sync::Arc;
use codex_exec_server::ResolvedSelectedCapabilityRoot;
use codex_protocol::capabilities::CapabilityRootLocation;
use codex_protocol::protocol::Product;
use codex_protocol::user_input::UserInput;
use crate::AvailableSkills;
use crate::ExecutorSkillCatalogCache;
use crate::HostSkillsSnapshot;
use crate::SkillMetadata;
use crate::collect_runtime_skill_mentions;
use crate::default_skill_metadata_budget;
use crate::loader::EnvironmentSkillMetadata;
use crate::loader::load_environment_skills_from_root;
use crate::executor_runtime::ExecutorSkillSource;
use crate::render::SkillRenderSideEffects;
use crate::render::build_available_skills_from_catalog;
use crate::runtime::SkillAuthority;
@@ -85,24 +84,50 @@ impl SkillsSnapshot {
pub async fn load(
host: HostSkillsSnapshot,
executor_catalog_cache: &ExecutorSkillCatalogCache,
executor_roots: &[ResolvedSelectedCapabilityRoot],
extra_sources: Option<&SkillSources>,
restriction_product: Option<Product>,
context_window: Option<i64>,
) -> Self {
let mut sources =
SkillSources::new().with_source("host", Arc::new(HostSkillSource::new(host.clone())));
let host_source: Arc<dyn SkillSource> = Arc::new(HostSkillSource::new(host.clone()));
let mut catalog = SkillCatalog::default();
let mut source_by_entry = HashMap::new();
merge_bound_catalog(
&mut catalog,
&mut source_by_entry,
&host_catalog(&host),
&host_source,
);
for root in executor_roots {
sources = sources.with_source(
format!("executor root `{}`", root.selected_root().id),
Arc::new(ExecutorSkillSource::new(root.clone(), restriction_product)),
let executor_source =
Arc::new(ExecutorSkillSource::new(root.clone(), restriction_product));
let source_catalog = executor_catalog_cache
.catalog_for_stable_root(executor_source.as_ref())
.await;
let source: Arc<dyn SkillSource> = executor_source;
merge_bound_catalog(
&mut catalog,
&mut source_by_entry,
source_catalog.as_ref(),
&source,
);
}
if let Some(extra_sources) = extra_sources {
sources.extend(extra_sources.clone());
let (extra_catalog, extra_source_by_entry) = extra_sources.list_with_sources().await;
catalog.warnings.extend(extra_catalog.warnings);
for entry in extra_catalog.entries {
let key = (entry.authority.clone(), entry.id.clone());
if let std::collections::hash_map::Entry::Vacant(route) =
source_by_entry.entry(key.clone())
&& let Some(source) = extra_source_by_entry.get(&key)
{
route.insert(Arc::clone(source));
catalog.entries.push(entry);
}
}
}
let (catalog, source_by_entry) = sources.list_with_sources().await;
Self::new(host, catalog, source_by_entry, context_window)
}
@@ -225,6 +250,24 @@ impl SkillsSnapshot {
}
}
fn merge_bound_catalog(
catalog: &mut SkillCatalog,
source_by_entry: &mut HashMap<EntryKey, Arc<dyn SkillSource>>,
source_catalog: &SkillCatalog,
source: &Arc<dyn SkillSource>,
) {
catalog
.warnings
.extend(source_catalog.warnings.iter().cloned());
for entry in &source_catalog.entries {
let key = (entry.authority.clone(), entry.id.clone());
if let std::collections::hash_map::Entry::Vacant(route) = source_by_entry.entry(key) {
route.insert(Arc::clone(source));
catalog.entries.push(entry.clone());
}
}
}
impl fmt::Debug for SkillsSnapshot {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
@@ -323,129 +366,3 @@ fn host_catalog(snapshot: &HostSkillsSnapshot) -> SkillCatalog {
}
catalog
}
struct ExecutorSkillSource {
root: ResolvedSelectedCapabilityRoot,
authority: SkillAuthority,
restriction_product: Option<Product>,
identity: SkillSourceIdentity,
}
impl ExecutorSkillSource {
fn new(root: ResolvedSelectedCapabilityRoot, restriction_product: Option<Product>) -> Self {
Self {
authority: SkillAuthority::new(
SkillSourceKind::Executor,
root.selected_root().id.clone(),
),
identity: SkillSourceIdentity::from_owner(Arc::clone(root.environment())),
root,
restriction_product,
}
}
}
impl SkillSource for ExecutorSkillSource {
fn identity(&self) -> SkillSourceIdentity {
self.identity.clone()
}
fn list(&self) -> SkillSourceFuture<'_, SkillCatalog> {
Box::pin(async move {
let CapabilityRootLocation::Environment {
environment_id,
path,
} = &self.root.selected_root().location;
let outcome = load_environment_skills_from_root(
self.root.file_system().as_ref(),
path,
self.restriction_product,
)
.await;
let mut catalog = SkillCatalog {
warnings: outcome.warnings,
..Default::default()
};
for skill in outcome.skills {
catalog.push_entry(executor_catalog_entry(
skill,
self.authority.clone(),
&self.root.selected_root().id,
environment_id,
));
}
Ok(catalog)
})
}
fn read(&self, request: SkillReadRequest) -> SkillSourceFuture<'_, SkillReadResult> {
Box::pin(async move {
if request.authority != self.authority || request.package.0 != request.resource.as_str()
{
return Err(SkillSourceError::new(
"executor skill resource does not match its captured source",
));
}
let CapabilityRootLocation::Environment { environment_id, .. } =
&self.root.selected_root().location;
let Some((resource_environment, path)) = request.resource.environment_path() else {
return Err(SkillSourceError::new(
"executor skill resource has no environment path",
));
};
if resource_environment != environment_id {
return Err(SkillSourceError::new(
"executor skill resource belongs to a different environment",
));
}
let contents = self
.root
.file_system()
.read_file_text(path, /*sandbox*/ None)
.await
.map_err(|err| {
SkillSourceError::new(format!(
"failed to read executor skill resource {}: {err}",
request.resource.as_str()
))
})?;
Ok(SkillReadResult {
resource: request.resource,
contents,
})
})
}
}
fn executor_catalog_entry(
skill: EnvironmentSkillMetadata,
authority: SkillAuthority,
root_id: &str,
environment_id: &str,
) -> SkillCatalogEntry {
let prompt_visible = skill.allows_implicit_invocation();
let path = skill.path_to_skills_md.inferred_native_path_string();
let display_path = format!(
"skill://{root_id}/{}",
path.replace('\\', "/").trim_start_matches('/')
);
let entry = SkillCatalogEntry::new(
SkillPackageId(display_path.clone()),
authority,
skill.name,
skill.description,
SkillResourceId::environment(
display_path.clone(),
environment_id,
skill.path_to_skills_md,
),
)
.with_short_description(skill.short_description)
.with_display_path(display_path)
.with_dependencies(skill.dependencies);
if prompt_visible {
entry
} else {
entry.hidden_from_prompt()
}
}
codex-rs/core/src/session/mod.rs
+1
View File
@@ -2840,6 +2840,7 @@ impl Session {
let skills = Arc::new(
codex_core_skills::SkillsSnapshot::load(
turn_context.turn_skills.snapshot.clone(),
&self.services.executor_skill_catalog_cache,
&selected_capability_roots,
extra_skill_sources.as_deref(),
turn_context.session_source.restriction_product(),
codex-rs/core/src/session/session.rs
+1
View File
@@ -1040,6 +1040,7 @@ impl Session {
guardian_rejection_circuit_breaker: Mutex::new(Default::default()),
runtime_handle: tokio::runtime::Handle::current(),
skills_service,
executor_skill_catalog_cache: Default::default(),
agents_md_manager,
plugins_manager: Arc::clone(&plugins_manager),
mcp_manager: Arc::clone(&mcp_manager),
codex-rs/core/src/session/tests.rs
+2
View File
@@ -5404,6 +5404,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) {
guardian_rejection_circuit_breaker: Mutex::new(Default::default()),
runtime_handle: tokio::runtime::Handle::current(),
skills_service,
executor_skill_catalog_cache: Default::default(),
agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)),
plugins_manager,
mcp_manager,
@@ -7481,6 +7482,7 @@ where
guardian_rejection_circuit_breaker: Mutex::new(Default::default()),
runtime_handle: tokio::runtime::Handle::current(),
skills_service,
executor_skill_catalog_cache: Default::default(),
agents_md_manager: Arc::new(AgentsMdManager::new(/*user_instructions*/ None)),
plugins_manager,
mcp_manager,
codex-rs/core/src/state/service.rs
+4
View File
@@ -25,6 +25,7 @@ use arc_swap::ArcSwap;
use arc_swap::ArcSwapOption;
use codex_analytics::AnalyticsEventsClient;
use codex_core_plugins::PluginsManager;
use codex_core_skills::ExecutorSkillCatalogCache;
use codex_extension_api::ExtensionData;
use codex_extension_api::ExtensionDataInit;
use codex_extension_api::ExtensionRegistry;
@@ -66,6 +67,9 @@ pub(crate) struct SessionServices {
pub(crate) guardian_rejection_circuit_breaker: Mutex<GuardianRejectionCircuitBreaker>,
pub(crate) runtime_handle: Handle,
pub(crate) skills_service: Arc<SkillsService>,
/// Catalogs discovered from executor capability roots. The cache is session-scoped because
/// selected environment contents are stable for the session lifetime.
pub(crate) executor_skill_catalog_cache: ExecutorSkillCatalogCache,
pub(crate) agents_md_manager: Arc<AgentsMdManager>,
pub(crate) plugins_manager: Arc<PluginsManager>,
pub(crate) mcp_manager: Arc<McpManager>,
codex-rs/core/tests/common/test_codex.rs
+9
View File
@@ -37,6 +37,7 @@ use codex_login::CodexAuth;
use codex_model_provider_info::ModelProviderInfo;
use codex_model_provider_info::built_in_model_providers;
use codex_models_manager::bundled_models_response;
use codex_protocol::capabilities::SelectedCapabilityRoot;
use codex_protocol::models::PermissionProfile;
use codex_protocol::openai_models::ModelInfo;
use codex_protocol::openai_models::ModelsResponse;
@@ -385,6 +386,14 @@ impl TestCodexBuilder {
self
}
pub fn with_selected_capability_roots(
mut self,
selected_capability_roots: Vec<SelectedCapabilityRoot>,
) -> Self {
self.thread_extension_init.insert(selected_capability_roots);
self
}
pub fn with_user_instructions_provider(
mut self,
provider: Arc<dyn UserInstructionsProvider>,
codex-rs/core/tests/suite/remote_env.rs
+27 -2
View File
@@ -488,6 +488,10 @@ async fn serve_environment_with_skill(
_ = &mut shutdown => return reads,
};
let response = match request["method"].as_str() {
Some("fs/canonicalize") => json!({
"id": request["id"],
"result": { "path": request["params"]["path"] }
}),
Some("fs/walk") => json!({
"id": request["id"],
"result": {
@@ -781,7 +785,7 @@ async fn deferred_executor_loads_agents_md_when_environment_becomes_ready() -> R
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn deferred_executor_replaces_a_colliding_host_skill_for_the_next_step() -> Result<()> {
async fn deferred_executor_caches_replacement_skill_catalog_for_later_steps() -> Result<()> {
const HOST_BODY: &str = "HOST_SKILL_BODY_MARKER";
const EXECUTOR_BODY: &str = "EXECUTOR_SKILL_BODY_MARKER";
const SKILL_REPLACEMENT_NOTICE: &str =
@@ -818,6 +822,11 @@ async fn deferred_executor_replaces_a_colliding_host_skill_for_the_next_step() -
ev_assistant_message("msg-2", "done"),
ev_completed("resp-2"),
]),
sse(vec![
ev_response_created("resp-3"),
ev_assistant_message("msg-3", "still done"),
ev_completed("resp-3"),
]),
],
)
.await;
@@ -870,11 +879,27 @@ async fn deferred_executor_replaces_a_colliding_host_skill_for_the_next_step() -
matches!(event, EventMsg::TurnComplete(_))
})
.await;
test.codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: "Continue without loading another skill".into(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
})
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
shutdown_tx.send(()).expect("stop exec server");
assert_eq!(exec_server.await?, 2);
let requests = response_mock.requests();
assert_eq!(requests.len(), 2);
assert_eq!(requests.len(), 3);
let first_user_context = requests[0].message_input_texts("user");
let host_skill = first_user_context
.iter()