diff --git a/codex-rs/core-skills/src/executor_runtime.rs b/codex-rs/core-skills/src/executor_runtime.rs index e742ce8b6a..3dcd9908f3 100644 --- a/codex-rs/core-skills/src/executor_runtime.rs +++ b/codex-rs/core-skills/src/executor_runtime.rs @@ -28,6 +28,8 @@ use crate::runtime::SkillSourceKind; /// 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. +/// The entry also owns the source identity used for injection deduplication, so reconnecting the +/// same stable environment does not make an unchanged skill look like a different instruction. /// /// 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 @@ -38,9 +40,11 @@ pub struct ExecutorSkillCatalogCache { entries: Mutex>, } -struct CachedExecutorSkillCatalog { +#[derive(Clone)] +pub(crate) struct CachedExecutorSkillCatalog { key: ExecutorSkillCatalogCacheKey, catalog: Arc, + identity: SkillSourceIdentity, } #[derive(Clone, Debug, PartialEq, Eq)] @@ -52,17 +56,28 @@ struct ExecutorSkillCatalogCacheKey { 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; + root: &ResolvedSelectedCapabilityRoot, + restriction_product: Option, + ) -> CachedExecutorSkillCatalog { + let key = ExecutorSkillCatalogCacheKey { + selected_root: root.selected_root().clone(), + restriction_product, + }; + if let Some(cached) = self.catalog(&key) { + return cached; } // 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 identity = SkillSourceIdentity::from_owner(Arc::new(key.clone())); + let source = ExecutorSkillSource::new(root.clone(), restriction_product, identity.clone()); let discovered = Arc::new(source.load_catalog().await); - if !discovered.warnings.is_empty() { + let discovered = CachedExecutorSkillCatalog { + key: key.clone(), + catalog: discovered, + identity, + }; + if !discovered.warnings().is_empty() { return discovered; } let mut entries = self @@ -70,22 +85,33 @@ impl ExecutorSkillCatalogCache { .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); + return entry.clone(); } - entries.push(CachedExecutorSkillCatalog { - key, - catalog: Arc::clone(&discovered), - }); + entries.push(discovered.clone()); discovered } - fn catalog(&self, key: &ExecutorSkillCatalogCacheKey) -> Option> { + 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)) + .cloned() + } +} + +impl CachedExecutorSkillCatalog { + pub(crate) fn catalog(&self) -> &SkillCatalog { + self.catalog.as_ref() + } + + pub(crate) fn identity(&self) -> SkillSourceIdentity { + self.identity.clone() + } + + fn warnings(&self) -> &[String] { + &self.catalog.warnings } } @@ -100,25 +126,19 @@ impl ExecutorSkillSource { pub(crate) fn new( root: ResolvedSelectedCapabilityRoot, restriction_product: Option, + identity: SkillSourceIdentity, ) -> Self { Self { authority: SkillAuthority::new( SkillSourceKind::Executor, root.selected_root().id.clone(), ), - identity: SkillSourceIdentity::from_owner(Arc::clone(root.environment())), + identity, 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, diff --git a/codex-rs/core-skills/src/runtime_snapshot.rs b/codex-rs/core-skills/src/runtime_snapshot.rs index a5cecda731..ebeebc4e1f 100644 --- a/codex-rs/core-skills/src/runtime_snapshot.rs +++ b/codex-rs/core-skills/src/runtime_snapshot.rs @@ -100,16 +100,19 @@ impl SkillsSnapshot { &host_source, ); for root in executor_roots { - 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()) + let cached = executor_catalog_cache + .catalog_for_stable_root(root, restriction_product) .await; + let executor_source = Arc::new(ExecutorSkillSource::new( + root.clone(), + restriction_product, + cached.identity(), + )); let source: Arc = executor_source; merge_bound_catalog( &mut catalog, &mut source_by_entry, - source_catalog.as_ref(), + cached.catalog(), &source, ); } diff --git a/codex-rs/core/tests/suite/remote_env.rs b/codex-rs/core/tests/suite/remote_env.rs index 2a2675f016..d34b8ffa6b 100644 --- a/codex-rs/core/tests/suite/remote_env.rs +++ b/codex-rs/core/tests/suite/remote_env.rs @@ -785,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_caches_replacement_skill_catalog_for_later_steps() -> Result<()> { +async fn deferred_executor_reuses_skill_catalog_after_environment_ready() -> Result<()> { const HOST_BODY: &str = "HOST_SKILL_BODY_MARKER"; const EXECUTOR_BODY: &str = "EXECUTOR_SKILL_BODY_MARKER"; const SKILL_REPLACEMENT_NOTICE: &str =