openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-08 20:40:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Prototype lower-level skills list coordination

Browse Source
This commit is contained in:
xli-oai
2026-05-07 12:43:22 -07:00
parent 1789ab69d1
commit 1c3cb14fb5
7 changed files with 128 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/src/protocol/common.rs
+2 -5
View File
@@ -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(),
codex-rs/app-server/src/config_manager.rs
+13
View File
@@ -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<RwLock<CloudRequirementsLoader>>,
arg0_paths: Arg0DispatchPaths,
thread_config_loader: Arc<RwLock<Arc<dyn ThreadConfigLoader>>>,
shared_state: Arc<AsyncRwLock<()>>,
}
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<I>(&self, enablement: I) -> Result<(), ()>
where
I: IntoIterator<Item = (String, bool)>,
codex-rs/app-server/src/config_manager_service.rs
+2
View File
@@ -171,6 +171,7 @@ impl ConfigManager {
&self,
params: ConfigValueWriteParams,
) -> Result<ConfigWriteResponse, ConfigManagerError> {
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<ConfigWriteResponse, ConfigManagerError> {
let _guard = self.write_shared_state().await;
let edits = params
.edits
.into_iter()
codex-rs/app-server/src/request_processors/catalog_processor.rs
+96 -47
View File
@@ -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<AbsolutePathBuf>,
},
Error {
index: usize,
entry: codex_app_server_protocol::SkillsListEntry,
},
}
fn skills_to_info(
skills: &[codex_core::skills::SkillMetadata],
disabled_paths: &HashSet<AbsolutePathBuf>,
@@ -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::<Vec<_>>()
.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<SkillsConfigWriteResponse, JSONRPCErrorError> {
let _guard = self.config_manager.write_shared_state().await;
let SkillsConfigWriteParams {
path,
name,
codex-rs/app-server/src/request_processors/config_processor.rs
+10 -7
View File
@@ -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;
codex-rs/app-server/src/request_processors/marketplace_processor.rs
+3
View File
@@ -51,6 +51,7 @@ impl MarketplaceRequestProcessor {
&self,
params: MarketplaceRemoveParams,
) -> Result<MarketplaceRemoveResponse, JSONRPCErrorError> {
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<MarketplaceUpgradeResponse, JSONRPCErrorError> {
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<MarketplaceAddResponse, JSONRPCErrorError> {
let _guard = self.config_manager.write_shared_state().await;
add_marketplace_to_codex_home(
self.config.codex_home.to_path_buf(),
MarketplaceAddRequest {
codex-rs/app-server/src/request_processors/plugins.rs
+2
View File
@@ -261,6 +261,7 @@ impl PluginRequestProcessor {
&self,
params: PluginInstallParams,
) -> Result<Option<ClientResponsePayload>, 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<Option<ClientResponsePayload>, JSONRPCErrorError> {
let _guard = self.config_manager.write_shared_state().await;
self.plugin_uninstall_response(params)
.await
.map(|response| Some(response.into()))