diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index bc5953803e..eb151ebe70 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -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", diff --git a/codex-rs/core-skills/Cargo.toml b/codex-rs/core-skills/Cargo.toml index f86ecdf75b..d10e556415 100644 --- a/codex-rs/core-skills/Cargo.toml +++ b/codex-rs/core-skills/Cargo.toml @@ -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 } diff --git a/codex-rs/core-skills/src/executor_runtime.rs b/codex-rs/core-skills/src/executor_runtime.rs new file mode 100644 index 0000000000..e742ce8b6a --- /dev/null +++ b/codex-rs/core-skills/src/executor_runtime.rs @@ -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>, +} + +struct CachedExecutorSkillCatalog { + key: ExecutorSkillCatalogCacheKey, + catalog: Arc, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct ExecutorSkillCatalogCacheKey { + selected_root: SelectedCapabilityRoot, + restriction_product: Option, +} + +impl ExecutorSkillCatalogCache { + pub(crate) async fn catalog_for_stable_root( + &self, + source: &ExecutorSkillSource, + ) -> Arc { + 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> { + 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, + identity: SkillSourceIdentity, +} + +impl ExecutorSkillSource { + pub(crate) fn new( + root: ResolvedSelectedCapabilityRoot, + restriction_product: Option, + ) -> 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() + } +} diff --git a/codex-rs/core-skills/src/injection_tests.rs b/codex-rs/core-skills/src/injection_tests.rs index d81104b3fa..cb99aba4af 100644 --- a/codex-rs/core-skills/src/injection_tests.rs +++ b/codex-rs/core-skills/src/injection_tests.rs @@ -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] diff --git a/codex-rs/core-skills/src/lib.rs b/codex-rs/core-skills/src/lib.rs index fa41a48518..1734e2bef0 100644 --- a/codex-rs/core-skills/src/lib.rs +++ b/codex-rs/core-skills/src/lib.rs @@ -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; diff --git a/codex-rs/core-skills/src/runtime_snapshot.rs b/codex-rs/core-skills/src/runtime_snapshot.rs index 834a623bc4..a5cecda731 100644 --- a/codex-rs/core-skills/src/runtime_snapshot.rs +++ b/codex-rs/core-skills/src/runtime_snapshot.rs @@ -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, context_window: Option, ) -> Self { - let mut sources = - SkillSources::new().with_source("host", Arc::new(HostSkillSource::new(host.clone()))); + let host_source: Arc = 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 = 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>, + source_catalog: &SkillCatalog, + source: &Arc, +) { + 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, - identity: SkillSourceIdentity, -} - -impl ExecutorSkillSource { - fn new(root: ResolvedSelectedCapabilityRoot, restriction_product: Option) -> 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() - } -} diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 61a5dc7f87..b3a0f1434c 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -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(), diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 96dffed888..85daa021b5 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -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), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index ad61c1e6bd..d055277a28 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -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, diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 3721cb1848..a685a239c2 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -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, pub(crate) runtime_handle: Handle, pub(crate) skills_service: Arc, + /// 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, pub(crate) plugins_manager: Arc, pub(crate) mcp_manager: Arc, diff --git a/codex-rs/core/tests/common/test_codex.rs b/codex-rs/core/tests/common/test_codex.rs index f7004bfe14..8a4f5cca7d 100644 --- a/codex-rs/core/tests/common/test_codex.rs +++ b/codex-rs/core/tests/common/test_codex.rs @@ -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, + ) -> Self { + self.thread_extension_init.insert(selected_capability_roots); + self + } + pub fn with_user_instructions_provider( mut self, provider: Arc, diff --git a/codex-rs/core/tests/suite/remote_env.rs b/codex-rs/core/tests/suite/remote_env.rs index d28ed9608b..2a2675f016 100644 --- a/codex-rs/core/tests/suite/remote_env.rs +++ b/codex-rs/core/tests/suite/remote_env.rs @@ -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()