From 1c3cb14fb5e45657c4d8cecf4294f7f4820a8e75 Mon Sep 17 00:00:00 2001 From: xli-oai Date: Thu, 7 May 2026 12:43:22 -0700 Subject: [PATCH] Prototype lower-level skills list coordination --- .../src/protocol/common.rs | 7 +- codex-rs/app-server/src/config_manager.rs | 13 ++ .../app-server/src/config_manager_service.rs | 2 + .../request_processors/catalog_processor.rs | 143 ++++++++++++------ .../request_processors/config_processor.rs | 17 ++- .../marketplace_processor.rs | 3 + .../src/request_processors/plugins.rs | 2 + 7 files changed, 128 insertions(+), 59 deletions(-) diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index 8c1f02570a..91ea8b7d52 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -589,7 +589,7 @@ client_request_definitions! { }, SkillsList => "skills/list" { params: v2::SkillsListParams, - serialization: global_shared_read("config"), + serialization: None, response: v2::SkillsListResponse, }, HooksList => "hooks/list" { @@ -1667,10 +1667,7 @@ mod tests { per_cwd_extra_user_roots: None, }, }; - assert_eq!( - skills_list.serialization_scope(), - Some(ClientRequestSerializationScope::GlobalSharedRead("config")) - ); + assert_eq!(skills_list.serialization_scope(), None); let plugin_list = ClientRequest::PluginList { request_id: request_id(), diff --git a/codex-rs/app-server/src/config_manager.rs b/codex-rs/app-server/src/config_manager.rs index 030829fa4b..55835b5523 100644 --- a/codex-rs/app-server/src/config_manager.rs +++ b/codex-rs/app-server/src/config_manager.rs @@ -20,6 +20,9 @@ use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::sync::RwLock; +use tokio::sync::RwLock as AsyncRwLock; +use tokio::sync::RwLockReadGuard; +use tokio::sync::RwLockWriteGuard; use toml::Value as TomlValue; use tracing::warn; @@ -33,6 +36,7 @@ pub(crate) struct ConfigManager { cloud_requirements: Arc>, arg0_paths: Arg0DispatchPaths, thread_config_loader: Arc>>, + shared_state: Arc>, } impl ConfigManager { @@ -52,6 +56,7 @@ impl ConfigManager { cloud_requirements: Arc::new(RwLock::new(cloud_requirements)), arg0_paths, thread_config_loader: Arc::new(RwLock::new(thread_config_loader)), + shared_state: Arc::new(AsyncRwLock::new(())), } } @@ -73,6 +78,14 @@ impl ConfigManager { .unwrap_or_default() } + pub(crate) async fn read_shared_state(&self) -> RwLockReadGuard<'_, ()> { + self.shared_state.read().await + } + + pub(crate) async fn write_shared_state(&self) -> RwLockWriteGuard<'_, ()> { + self.shared_state.write().await + } + pub(crate) fn extend_runtime_feature_enablement(&self, enablement: I) -> Result<(), ()> where I: IntoIterator, diff --git a/codex-rs/app-server/src/config_manager_service.rs b/codex-rs/app-server/src/config_manager_service.rs index aef4393a09..6658fc044f 100644 --- a/codex-rs/app-server/src/config_manager_service.rs +++ b/codex-rs/app-server/src/config_manager_service.rs @@ -171,6 +171,7 @@ impl ConfigManager { &self, params: ConfigValueWriteParams, ) -> Result { + let _guard = self.write_shared_state().await; let edits = vec![(params.key_path, params.value, params.merge_strategy)]; self.apply_edits(params.file_path, params.expected_version, edits) .await @@ -180,6 +181,7 @@ impl ConfigManager { &self, params: ConfigBatchWriteParams, ) -> Result { + let _guard = self.write_shared_state().await; let edits = params .edits .into_iter() diff --git a/codex-rs/app-server/src/request_processors/catalog_processor.rs b/codex-rs/app-server/src/request_processors/catalog_processor.rs index b26748d551..3e33274e1f 100644 --- a/codex-rs/app-server/src/request_processors/catalog_processor.rs +++ b/codex-rs/app-server/src/request_processors/catalog_processor.rs @@ -1,5 +1,6 @@ use super::*; use futures::StreamExt; +use std::time::Instant; #[derive(Clone)] pub(crate) struct CatalogRequestProcessor { @@ -12,6 +13,19 @@ pub(crate) struct CatalogRequestProcessor { const SKILLS_LIST_CWD_CONCURRENCY: usize = 5; +enum PreparedSkillsListEntry { + Ready { + index: usize, + cwd: PathBuf, + skills_input: codex_core::skills::SkillsLoadInput, + extra_roots: Vec, + }, + Error { + index: usize, + entry: codex_app_server_protocol::SkillsListEntry, + }, +} + fn skills_to_info( skills: &[codex_core::skills::SkillMetadata], disabled_paths: &HashSet, @@ -421,68 +435,102 @@ impl CatalogRequestProcessor { .extend(valid_extra_roots); } - let config = self.load_latest_config(/*fallback_cwd*/ None).await?; - let auth = self.auth_manager.auth().await; - let workspace_codex_plugins_enabled = self - .workspace_codex_plugins_enabled(&config, auth.as_ref()) - .await; + let snapshot_started_at = Instant::now(); + let prepared_entries = { + let _guard = self.config_manager.read_shared_state().await; + let config = self.load_latest_config(/*fallback_cwd*/ None).await?; + let auth = self.auth_manager.auth().await; + let workspace_codex_plugins_enabled = self + .workspace_codex_plugins_enabled(&config, auth.as_ref()) + .await; + let plugins_manager = self.thread_manager.plugins_manager(); + futures::stream::iter(cwds.into_iter().enumerate()) + .map(|(index, cwd)| { + let config = &config; + let extra_roots_by_cwd = &extra_roots_by_cwd; + let plugins_manager = &plugins_manager; + async move { + let (cwd_abs, config_layer_stack) = match self + .resolve_cwd_config(&cwd) + .await + { + Ok(resolved) => resolved, + Err(message) => { + let error_path = cwd.clone(); + return PreparedSkillsListEntry::Error { + index, + entry: codex_app_server_protocol::SkillsListEntry { + cwd, + skills: Vec::new(), + errors: vec![codex_app_server_protocol::SkillErrorInfo { + path: error_path, + message, + }], + }, + }; + } + }; + let extra_roots = extra_roots_by_cwd.get(&cwd).cloned().unwrap_or_default(); + let effective_skill_roots = if workspace_codex_plugins_enabled { + let plugins_input = config.plugins_config_input(); + plugins_manager + .effective_skill_roots_for_layer_stack( + &config_layer_stack, + &plugins_input, + ) + .await + } else { + Vec::new() + }; + PreparedSkillsListEntry::Ready { + index, + cwd, + skills_input: codex_core::skills::SkillsLoadInput::new( + cwd_abs.clone(), + effective_skill_roots, + config_layer_stack, + config.bundled_skills_enabled(), + ), + extra_roots, + } + } + }) + .buffer_unordered(SKILLS_LIST_CWD_CONCURRENCY) + .collect::>() + .await + }; + warn!( + elapsed_ms = snapshot_started_at.elapsed().as_millis(), + entry_count = prepared_entries.len(), + "skills/list prepared lower-level snapshot" + ); let skills_manager = self.thread_manager.skills_manager(); - let plugins_manager = self.thread_manager.plugins_manager(); let fs = self .thread_manager .environment_manager() .default_environment() .map(|environment| environment.get_filesystem()); - let mut data = futures::stream::iter(cwds.into_iter().enumerate()) - .map(|(index, cwd)| { - let config = &config; - let extra_roots_by_cwd = &extra_roots_by_cwd; + let mut data = futures::stream::iter(prepared_entries.into_iter()) + .map(|prepared_entry| { let fs = fs.clone(); - let plugins_manager = &plugins_manager; let skills_manager = &skills_manager; async move { - let (cwd_abs, config_layer_stack) = match self.resolve_cwd_config(&cwd).await { - Ok(resolved) => resolved, - Err(message) => { - let error_path = cwd.clone(); - return ( - index, - codex_app_server_protocol::SkillsListEntry { - cwd, - skills: Vec::new(), - errors: vec![codex_app_server_protocol::SkillErrorInfo { - path: error_path, - message, - }], - }, - ); + let (index, cwd, skills_input, extra_roots) = match prepared_entry { + PreparedSkillsListEntry::Ready { + index, + cwd, + skills_input, + extra_roots, + } => (index, cwd, skills_input, extra_roots), + PreparedSkillsListEntry::Error { index, entry } => { + return (index, entry); } }; - let extra_roots = extra_roots_by_cwd - .get(&cwd) - .map_or(&[][..], std::vec::Vec::as_slice); - let effective_skill_roots = if workspace_codex_plugins_enabled { - let plugins_input = config.plugins_config_input(); - plugins_manager - .effective_skill_roots_for_layer_stack( - &config_layer_stack, - &plugins_input, - ) - .await - } else { - Vec::new() - }; - let skills_input = codex_core::skills::SkillsLoadInput::new( - cwd_abs.clone(), - effective_skill_roots, - config_layer_stack, - config.bundled_skills_enabled(), - ); let outcome = skills_manager .skills_for_cwd_with_extra_user_roots( &skills_input, force_reload, - extra_roots, + &extra_roots, fs, ) .await; @@ -585,6 +633,7 @@ impl CatalogRequestProcessor { &self, params: SkillsConfigWriteParams, ) -> Result { + let _guard = self.config_manager.write_shared_state().await; let SkillsConfigWriteParams { path, name, diff --git a/codex-rs/app-server/src/request_processors/config_processor.rs b/codex-rs/app-server/src/request_processors/config_processor.rs index 840542419a..addae1cdbc 100644 --- a/codex-rs/app-server/src/request_processors/config_processor.rs +++ b/codex-rs/app-server/src/request_processors/config_processor.rs @@ -363,13 +363,16 @@ impl ConfigRequestProcessor { return Ok(ExperimentalFeatureEnablementSetResponse { enablement }); } - self.config_manager - .extend_runtime_feature_enablement( - enablement - .iter() - .map(|(name, enabled)| (name.clone(), *enabled)), - ) - .map_err(|_| internal_error("failed to update feature enablement"))?; + { + let _guard = self.config_manager.write_shared_state().await; + self.config_manager + .extend_runtime_feature_enablement( + enablement + .iter() + .map(|(name, enabled)| (name.clone(), *enabled)), + ) + .map_err(|_| internal_error("failed to update feature enablement"))?; + } self.load_latest_config(/*fallback_cwd*/ None).await?; self.reload_user_config().await; diff --git a/codex-rs/app-server/src/request_processors/marketplace_processor.rs b/codex-rs/app-server/src/request_processors/marketplace_processor.rs index 1a09507418..451418ecc4 100644 --- a/codex-rs/app-server/src/request_processors/marketplace_processor.rs +++ b/codex-rs/app-server/src/request_processors/marketplace_processor.rs @@ -51,6 +51,7 @@ impl MarketplaceRequestProcessor { &self, params: MarketplaceRemoveParams, ) -> Result { + let _guard = self.config_manager.write_shared_state().await; remove_marketplace( self.config.codex_home.to_path_buf(), CoreMarketplaceRemoveRequest { @@ -72,6 +73,7 @@ impl MarketplaceRequestProcessor { &self, params: MarketplaceUpgradeParams, ) -> Result { + let _guard = self.config_manager.write_shared_state().await; let config = self.load_latest_config(/*fallback_cwd*/ None).await?; let plugins_manager = self.thread_manager.plugins_manager(); let MarketplaceUpgradeParams { marketplace_name } = params; @@ -105,6 +107,7 @@ impl MarketplaceRequestProcessor { &self, params: MarketplaceAddParams, ) -> Result { + let _guard = self.config_manager.write_shared_state().await; add_marketplace_to_codex_home( self.config.codex_home.to_path_buf(), MarketplaceAddRequest { diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index a36077d86b..93ebfacab6 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -261,6 +261,7 @@ impl PluginRequestProcessor { &self, params: PluginInstallParams, ) -> Result, JSONRPCErrorError> { + let _guard = self.config_manager.write_shared_state().await; self.plugin_install_response(params) .await .map(|response| Some(response.into())) @@ -270,6 +271,7 @@ impl PluginRequestProcessor { &self, params: PluginUninstallParams, ) -> Result, JSONRPCErrorError> { + let _guard = self.config_manager.write_shared_state().await; self.plugin_uninstall_response(params) .await .map(|response| Some(response.into()))