use std::collections::HashMap; use std::collections::HashSet; use std::future::Future; use std::sync::Arc; use std::sync::Mutex; use codex_exec_server::FileSystemSandboxContext; use codex_extension_api::ExtensionMetrics; use codex_mcp::McpResourceClient; use codex_mcp::McpResourceClientCacheKey; use codex_protocol::capabilities::SelectedCapabilityRoot; use tokio::sync::OnceCell; use crate::SkillsExtensionConfig; use crate::catalog::SkillAuthority; use crate::catalog::SkillCatalog; use crate::catalog::SkillCatalogEntry; use crate::catalog::SkillPackageId; use crate::catalog::SkillProviderError; use crate::catalog::SkillProviderResult; use crate::catalog::SkillReadResult; use crate::catalog::SkillResourceId; use crate::catalog::SkillSourceKind; use crate::provider::SkillListQuery; use crate::provider::SkillReadRequest; use crate::shadow_selection_experiment::ShadowSelectionTurnState; use crate::sources::SkillProviders; const MAX_CACHED_ORCHESTRATOR_RESOURCES: usize = 100; const MAX_CACHED_ORCHESTRATOR_CONTENT_BYTES: usize = 8 * 1024 * 1024; pub(crate) struct SkillsSessionState { pub(crate) mcp_resources: Option>, pub(crate) extension_metrics: Option>, } pub(crate) struct SkillsThreadState { config: Mutex, orchestrator_skills_available: bool, executor_cache: Mutex>, executor_discovery_cache: Mutex>, orchestrator_cache: Mutex>>, shadow_selection_turn: Mutex>, } impl SkillsThreadState { pub(crate) fn new(config: SkillsExtensionConfig, orchestrator_skills_available: bool) -> Self { Self { config: Mutex::new(config), orchestrator_skills_available, executor_cache: Mutex::new(Vec::new()), executor_discovery_cache: Mutex::new(None), orchestrator_cache: Mutex::new(None), shadow_selection_turn: Mutex::new(None), } } pub(crate) fn config(&self) -> SkillsExtensionConfig { self.config .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .clone() } pub(crate) fn set_config(&self, config: SkillsExtensionConfig) { *self .config .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) = config; } pub(crate) fn orchestrator_skills_enabled(&self) -> bool { self.orchestrator_skills_available && self.config().orchestrator_skills_enabled } pub(crate) fn replace_shadow_selection_turn( &self, turn_id: String, state: Option, ) { *self .shadow_selection_turn .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) = state.map(|state| ShadowSelectionTurn { turn_id, state: Arc::new(state), }); } pub(crate) fn shadow_selection_turn( &self, turn_id: &str, ) -> Option> { self.shadow_selection_turn .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .as_ref() .filter(|turn| turn.turn_id == turn_id) .map(|turn| Arc::clone(&turn.state)) } /// Returns catalogs for stable selected roots. /// /// The first catalog returned for a root remains cached until this thread state is dropped. /// Environment availability only controls whether the root is projected into the current /// step; it never invalidates the cache. There is intentionally no filesystem watcher or /// content-based invalidation because selected environment roots are treated as stable. #[tracing::instrument( name = "skills.executor.catalog_snapshot", level = "info", skip_all, fields(root_count = query.executor_roots.len()) )] pub(crate) async fn executor_catalog_snapshot( &self, providers: &SkillProviders, mut query: SkillListQuery, ) -> SkillCatalog { if query.executor_capability_discovery.is_some() { return self .executor_discovery_catalog_snapshot(providers, query) .await; } let roots = std::mem::take(&mut query.executor_roots); let mut catalog = SkillCatalog::default(); for root in roots { query.executor_roots = vec![root.clone()]; catalog.extend( self.executor_root_catalog(providers, root, query.clone()) .await, ); } catalog } async fn executor_discovery_catalog_snapshot( &self, providers: &SkillProviders, query: SkillListQuery, ) -> SkillCatalog { let sandbox_contexts = query .executor_capability_discovery .as_ref() .map(|discovery| discovery.sandbox_contexts().clone()) .unwrap_or_default(); if let Some(cached) = self .executor_discovery_cache .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .as_ref() .filter(|cached| { cached.roots == query.executor_roots && cached.sandbox_contexts == sandbox_contexts }) { return cached.catalog.clone(); } let roots = query.executor_roots.clone(); let discovered = providers.list_executor_for_turn(query).await; let mut cache = self .executor_discovery_cache .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); if let Some(cached) = cache .as_ref() .filter(|cached| cached.roots == roots && cached.sandbox_contexts == sandbox_contexts) { return cached.catalog.clone(); } *cache = Some(CachedExecutorDiscoveryCatalog { roots, sandbox_contexts, catalog: discovered.clone(), }); discovered } pub(crate) async fn orchestrator_catalog_snapshot( &self, mcp_resources: Option<&McpResourceClient>, initialize: impl Future> + Send, ) -> SkillCatalog { self.orchestrator_cache(mcp_resources) .catalog .get_or_init(|| async { initialize.await.unwrap_or_else(|err| SkillCatalog { warnings: vec![err.message], ..Default::default() }) }) .await .clone() } pub(crate) async fn read_skill( &self, providers: &SkillProviders, request: SkillReadRequest, ) -> SkillProviderResult { if request.authority.kind != SkillSourceKind::Orchestrator { return providers.read(request).await; } let cache = self.orchestrator_cache(request.mcp_resources.as_deref()); let cache_key = SkillReadCacheKey::from(&request); if let Some(result) = cache .resources .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .get(&cache_key) { return Ok(result); } let result = providers.read(request).await?; if result.resource != cache_key.resource { return Ok(result); } Ok(cache .resources .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .insert(cache_key, result)) } fn orchestrator_cache( &self, mcp_resources: Option<&McpResourceClient>, ) -> Arc { let mut cache = self .orchestrator_cache .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let cache_key = mcp_resources.map(McpResourceClient::cache_key); if let Some(cache) = cache .as_ref() .filter(|cache| cache.mcp_cache_key == cache_key) { return Arc::clone(cache); } let next_cache = Arc::new(OrchestratorGenerationCache { mcp_cache_key: cache_key, catalog: OnceCell::new(), resources: Mutex::new(OrchestratorResourceCache::default()), }); *cache = Some(Arc::clone(&next_cache)); next_cache } #[tracing::instrument(name = "skills.executor.catalog_root", level = "info", skip_all)] async fn executor_root_catalog( &self, providers: &SkillProviders, root: SelectedCapabilityRoot, query: SkillListQuery, ) -> SkillCatalog { if let Some(cached) = self .executor_cache .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .iter() .find(|cached| cached.root == root) { return cached.catalog.clone(); } let discovered = providers.list_executor_for_turn(query).await; let mut cache = self .executor_cache .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); if let Some(cached) = cache.iter().find(|cached| cached.root == root) { return cached.catalog.clone(); } cache.push(CachedExecutorCatalog { root, catalog: discovered.clone(), }); discovered } } struct ShadowSelectionTurn { turn_id: String, state: Arc, } struct CachedExecutorCatalog { root: SelectedCapabilityRoot, catalog: SkillCatalog, } struct CachedExecutorDiscoveryCatalog { roots: Vec, sandbox_contexts: HashMap, catalog: SkillCatalog, } struct OrchestratorGenerationCache { mcp_cache_key: Option, catalog: OnceCell, resources: Mutex, } #[derive(Clone, Debug, PartialEq, Eq, Hash)] struct SkillReadCacheKey { authority: SkillAuthority, package: SkillPackageId, resource: SkillResourceId, } impl From<&SkillReadRequest> for SkillReadCacheKey { fn from(request: &SkillReadRequest) -> Self { Self { authority: request.authority.clone(), package: request.package.clone(), resource: request.resource.clone(), } } } #[derive(Default)] struct OrchestratorResourceCache { entries: HashMap, contents_bytes: usize, } impl OrchestratorResourceCache { fn get(&self, key: &SkillReadCacheKey) -> Option { self.entries.get(key).cloned() } fn insert(&mut self, key: SkillReadCacheKey, result: SkillReadResult) -> SkillReadResult { if let Some(cached) = self.entries.get(&key) { return cached.clone(); } let contents_bytes = result.contents.len(); let Some(next_contents_bytes) = self.contents_bytes.checked_add(contents_bytes) else { return result; }; if self.entries.len() >= MAX_CACHED_ORCHESTRATOR_RESOURCES || next_contents_bytes > MAX_CACHED_ORCHESTRATOR_CONTENT_BYTES { return result; } self.contents_bytes = next_contents_bytes; self.entries.insert(key, result.clone()); result } } #[derive(Default)] pub(crate) struct EmittedCatalogBudgetWarnings(Mutex>); impl EmittedCatalogBudgetWarnings { pub(crate) fn insert(&self, warning: &str) -> bool { self.0 .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) .insert(warning.to_string()) } } #[derive(Clone, Debug, Default, PartialEq, Eq)] pub(crate) struct SkillsTurnState { pub(crate) catalog: SkillCatalog, pub(crate) selected_entries: Vec, pub(crate) warnings: Vec, pub(crate) main_prompts_injected: bool, } #[derive(Clone, Debug, Default)] pub(crate) struct ExecutorSkillsStepState(pub(crate) SkillCatalog); #[derive(Clone, Debug, Default)] pub(crate) struct HostSkillsStepState(pub(crate) SkillCatalog);