Keep executor skill identity stable across reconnects

This commit is contained in:
jif-oai
2026-06-25 17:41:49 +01:00
parent 5998bdd78e
commit addf6b52e5
3 changed files with 51 additions and 28 deletions
+42 -22
View File
@@ -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<Vec<CachedExecutorSkillCatalog>>,
}
struct CachedExecutorSkillCatalog {
#[derive(Clone)]
pub(crate) struct CachedExecutorSkillCatalog {
key: ExecutorSkillCatalogCacheKey,
catalog: Arc<SkillCatalog>,
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<SkillCatalog> {
let key = source.catalog_cache_key();
if let Some(catalog) = self.catalog(&key) {
return catalog;
root: &ResolvedSelectedCapabilityRoot,
restriction_product: Option<Product>,
) -> 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<Arc<SkillCatalog>> {
fn catalog(&self, key: &ExecutorSkillCatalogCacheKey) -> Option<CachedExecutorSkillCatalog> {
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<Product>,
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,
+8 -5
View File
@@ -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<dyn SkillSource> = executor_source;
merge_bound_catalog(
&mut catalog,
&mut source_by_entry,
source_catalog.as_ref(),
cached.catalog(),
&source,
);
}
+1 -1
View File
@@ -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 =