mirror of
https://github.com/openai/codex.git
synced 2026-11-09 20:50:33 +00:00
Reuse plugin discovery during startup
This commit is contained in:
@@ -647,9 +647,11 @@ impl CatalogRequestProcessor {
|
||||
config.features.enabled(Feature::Plugins) && workspace_codex_plugins_enabled;
|
||||
let plugin_hooks = if plugins_enabled {
|
||||
let plugins_input = config.plugins_config_input();
|
||||
plugins_manager
|
||||
.plugin_hooks_for_layer_stack(&config.config_layer_stack, &plugins_input)
|
||||
.await
|
||||
let plugin_outcome = plugins_manager.plugins_for_config(&plugins_input).await;
|
||||
codex_core_plugins::PluginHookLoadOutcome {
|
||||
hook_sources: plugin_outcome.effective_plugin_hook_sources(),
|
||||
hook_load_warnings: plugin_outcome.effective_plugin_hook_warnings(),
|
||||
}
|
||||
} else {
|
||||
codex_core_plugins::PluginHookLoadOutcome::default()
|
||||
};
|
||||
|
||||
@@ -264,6 +264,84 @@ async fn hooks_list_shows_discovered_plugin_hook() -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn hooks_list_warms_plugin_capabilities_for_thread_start() -> Result<()> {
|
||||
let codex_home = TempDir::new()?;
|
||||
let cwd = TempDir::new()?;
|
||||
write_plugin_hook_config(
|
||||
codex_home.path(),
|
||||
r#"{
|
||||
"hooks": {
|
||||
"PreToolUse": [
|
||||
{
|
||||
"hooks": [
|
||||
{
|
||||
"type": "command",
|
||||
"command": "echo plugin hook"
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}"#,
|
||||
)?;
|
||||
let plugin_mcp_path = codex_home
|
||||
.path()
|
||||
.join("plugins/cache/test/demo/local/.mcp.json");
|
||||
std::fs::write(
|
||||
&plugin_mcp_path,
|
||||
r#"{
|
||||
"mcpServers": {
|
||||
"plugin-server": {
|
||||
"url": "http://127.0.0.1:1/mcp"
|
||||
}
|
||||
}
|
||||
}"#,
|
||||
)?;
|
||||
|
||||
let mut mcp = TestAppServer::new(codex_home.path()).await?;
|
||||
timeout(DEFAULT_TIMEOUT, mcp.initialize()).await??;
|
||||
|
||||
let hooks_list_id = mcp
|
||||
.send_hooks_list_request(HooksListParams {
|
||||
cwds: vec![cwd.path().to_path_buf()],
|
||||
})
|
||||
.await?;
|
||||
timeout(
|
||||
DEFAULT_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(hooks_list_id)),
|
||||
)
|
||||
.await??;
|
||||
|
||||
std::fs::remove_file(plugin_mcp_path)?;
|
||||
|
||||
let thread_start_id = mcp
|
||||
.send_thread_start_request(ThreadStartParams::default())
|
||||
.await?;
|
||||
let _: ThreadStartResponse = to_response(
|
||||
timeout(
|
||||
DEFAULT_TIMEOUT,
|
||||
mcp.read_stream_until_response_message(RequestId::Integer(thread_start_id)),
|
||||
)
|
||||
.await??,
|
||||
)?;
|
||||
timeout(
|
||||
DEFAULT_TIMEOUT,
|
||||
mcp.read_stream_until_matching_notification("plugin MCP server starting", |notification| {
|
||||
notification.method == "mcpServer/startupStatus/updated"
|
||||
&& notification
|
||||
.params
|
||||
.as_ref()
|
||||
.and_then(|params| params.get("name"))
|
||||
.and_then(serde_json::Value::as_str)
|
||||
== Some("plugin-server")
|
||||
}),
|
||||
)
|
||||
.await??;
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn hooks_list_shows_plugin_hook_load_warnings() -> Result<()> {
|
||||
let codex_home = TempDir::new()?;
|
||||
|
||||
@@ -57,8 +57,9 @@ use codex_config::apply_user_plugin_config_edits;
|
||||
use codex_config::clear_user_plugin;
|
||||
use codex_config::set_user_plugin_enabled;
|
||||
use codex_config::types::PluginConfig;
|
||||
use codex_config::version_for_toml;
|
||||
use codex_core_skills::SkillMetadata;
|
||||
use codex_core_skills::config_rules::SkillConfigRules;
|
||||
use codex_core_skills::config_rules::skill_config_rules_from_stack;
|
||||
use codex_hooks::plugin_hook_declarations;
|
||||
use codex_login::AuthManager;
|
||||
use codex_login::CodexAuth;
|
||||
@@ -77,6 +78,7 @@ use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::RwLock;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::AtomicU64;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Instant;
|
||||
use tokio::sync::Semaphore;
|
||||
@@ -404,6 +406,8 @@ pub struct PluginsManager {
|
||||
configured_marketplace_upgrade_state: RwLock<ConfiguredMarketplaceUpgradeState>,
|
||||
non_curated_cache_refresh_state: RwLock<NonCuratedCacheRefreshState>,
|
||||
cached_enabled_outcome: RwLock<Option<CachedPluginLoadOutcome>>,
|
||||
enabled_outcome_load_semaphore: Semaphore,
|
||||
enabled_outcome_cache_generation: AtomicU64,
|
||||
remote_installed_plugins_cache: RwLock<Option<Vec<RemoteInstalledPlugin>>>,
|
||||
remote_installed_plugins_cache_refresh_state: RwLock<RemoteInstalledPluginsCacheRefreshState>,
|
||||
remote_sync_lock: Semaphore,
|
||||
@@ -413,10 +417,17 @@ pub struct PluginsManager {
|
||||
|
||||
#[derive(Clone)]
|
||||
struct CachedPluginLoadOutcome {
|
||||
config_version: String,
|
||||
key: PluginLoadCacheKey,
|
||||
outcome: PluginLoadOutcome,
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Eq)]
|
||||
struct PluginLoadCacheKey {
|
||||
configured_plugins: HashMap<String, PluginConfig>,
|
||||
skill_config_rules: SkillConfigRules,
|
||||
remote_plugin_enabled: bool,
|
||||
}
|
||||
|
||||
impl PluginsManager {
|
||||
pub fn new(codex_home: PathBuf) -> Self {
|
||||
Self::new_with_restriction_product(codex_home, Some(Product::Codex))
|
||||
@@ -442,6 +453,8 @@ impl PluginsManager {
|
||||
),
|
||||
non_curated_cache_refresh_state: RwLock::new(NonCuratedCacheRefreshState::default()),
|
||||
cached_enabled_outcome: RwLock::new(None),
|
||||
enabled_outcome_load_semaphore: Semaphore::new(/*permits*/ 1),
|
||||
enabled_outcome_cache_generation: AtomicU64::new(0),
|
||||
remote_installed_plugins_cache: RwLock::new(None),
|
||||
remote_installed_plugins_cache_refresh_state: RwLock::new(
|
||||
RemoteInstalledPluginsCacheRefreshState::default(),
|
||||
@@ -484,11 +497,25 @@ impl PluginsManager {
|
||||
return PluginLoadOutcome::default();
|
||||
}
|
||||
|
||||
let config_version = version_for_toml(&config.config_layer_stack.effective_config());
|
||||
if !force_reload && let Some(outcome) = self.cached_enabled_outcome(&config_version) {
|
||||
let cache_key = PluginLoadCacheKey {
|
||||
configured_plugins: configured_plugins_from_stack(&config.config_layer_stack),
|
||||
skill_config_rules: skill_config_rules_from_stack(&config.config_layer_stack),
|
||||
remote_plugin_enabled: config.remote_plugin_enabled,
|
||||
};
|
||||
if !force_reload && let Some(outcome) = self.cached_enabled_outcome(&cache_key) {
|
||||
return outcome;
|
||||
}
|
||||
|
||||
let Ok(_load_permit) = self.enabled_outcome_load_semaphore.acquire().await else {
|
||||
warn!("plugin load semaphore closed");
|
||||
return PluginLoadOutcome::default();
|
||||
};
|
||||
if !force_reload && let Some(outcome) = self.cached_enabled_outcome(&cache_key) {
|
||||
return outcome;
|
||||
}
|
||||
let cache_generation = self
|
||||
.enabled_outcome_cache_generation
|
||||
.load(Ordering::Acquire);
|
||||
let outcome = load_plugins_from_layer_stack(
|
||||
&config.config_layer_stack,
|
||||
self.remote_installed_plugin_configs(),
|
||||
@@ -498,14 +525,20 @@ impl PluginsManager {
|
||||
)
|
||||
.await;
|
||||
log_plugin_load_errors(&outcome);
|
||||
let mut cache = match self.cached_enabled_outcome.write() {
|
||||
Ok(cache) => cache,
|
||||
Err(err) => err.into_inner(),
|
||||
};
|
||||
*cache = Some(CachedPluginLoadOutcome {
|
||||
config_version,
|
||||
outcome: outcome.clone(),
|
||||
});
|
||||
if self
|
||||
.enabled_outcome_cache_generation
|
||||
.load(Ordering::Acquire)
|
||||
== cache_generation
|
||||
{
|
||||
let mut cache = match self.cached_enabled_outcome.write() {
|
||||
Ok(cache) => cache,
|
||||
Err(err) => err.into_inner(),
|
||||
};
|
||||
*cache = Some(CachedPluginLoadOutcome {
|
||||
key: cache_key,
|
||||
outcome: outcome.clone(),
|
||||
});
|
||||
}
|
||||
outcome
|
||||
}
|
||||
|
||||
@@ -519,6 +552,8 @@ impl PluginsManager {
|
||||
}
|
||||
|
||||
fn clear_enabled_outcome_cache(&self) {
|
||||
self.enabled_outcome_cache_generation
|
||||
.fetch_add(1, Ordering::AcqRel);
|
||||
let mut cached_enabled_outcome = match self.cached_enabled_outcome.write() {
|
||||
Ok(cache) => cache,
|
||||
Err(err) => err.into_inner(),
|
||||
@@ -574,16 +609,16 @@ impl PluginsManager {
|
||||
.effective_plugin_skill_roots()
|
||||
}
|
||||
|
||||
fn cached_enabled_outcome(&self, config_version: &str) -> Option<PluginLoadOutcome> {
|
||||
fn cached_enabled_outcome(&self, key: &PluginLoadCacheKey) -> Option<PluginLoadOutcome> {
|
||||
match self.cached_enabled_outcome.read() {
|
||||
Ok(cache) => cache
|
||||
.as_ref()
|
||||
.filter(|cached| cached.config_version == config_version)
|
||||
.filter(|cached| cached.key == *key)
|
||||
.map(|cached| cached.outcome.clone()),
|
||||
Err(err) => err
|
||||
.into_inner()
|
||||
.as_ref()
|
||||
.filter(|cached| cached.config_version == config_version)
|
||||
.filter(|cached| cached.key == *key)
|
||||
.map(|cached| cached.outcome.clone()),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1300,6 +1300,76 @@ async fn load_plugins_returns_empty_when_feature_disabled() {
|
||||
assert_eq!(outcome, PluginLoadOutcome::default());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn plugin_cache_ignores_unrelated_session_overrides() {
|
||||
let codex_home = TempDir::new().unwrap();
|
||||
let plugin_root = codex_home
|
||||
.path()
|
||||
.join("plugins/cache")
|
||||
.join("test/sample/local");
|
||||
write_plugin(
|
||||
codex_home.path().join("plugins/cache/test").as_path(),
|
||||
"sample/local",
|
||||
"sample",
|
||||
);
|
||||
write_file(
|
||||
&plugin_root.join(".mcp.json"),
|
||||
r#"{
|
||||
"mcpServers": {
|
||||
"sample": {
|
||||
"url": "https://sample.example/mcp"
|
||||
}
|
||||
}
|
||||
}"#,
|
||||
);
|
||||
|
||||
let user_file = codex_home.path().join(CONFIG_TOML_FILE).abs();
|
||||
let user_config: toml::Value = toml::from_str(&plugin_config_toml(
|
||||
/*enabled*/ true, /*plugins_feature_enabled*/ true,
|
||||
))
|
||||
.expect("user config should parse");
|
||||
let stack = |session_config: &str| {
|
||||
ConfigLayerStack::new(
|
||||
vec![
|
||||
ConfigLayerEntry::new(
|
||||
ConfigLayerSource::User {
|
||||
file: user_file.clone(),
|
||||
profile: None,
|
||||
},
|
||||
user_config.clone(),
|
||||
),
|
||||
ConfigLayerEntry::new(
|
||||
ConfigLayerSource::SessionFlags,
|
||||
toml::from_str(session_config).expect("session config should parse"),
|
||||
),
|
||||
],
|
||||
ConfigRequirements::default(),
|
||||
ConfigRequirementsToml::default(),
|
||||
)
|
||||
.expect("config layer stack should build")
|
||||
};
|
||||
let config = |session_config| {
|
||||
PluginsConfigInput::new(
|
||||
stack(session_config),
|
||||
/*plugins_enabled*/ true,
|
||||
/*remote_plugin_enabled*/ false,
|
||||
"https://chatgpt.com".to_string(),
|
||||
)
|
||||
};
|
||||
let manager = PluginsManager::new(codex_home.path().to_path_buf());
|
||||
|
||||
let first = manager
|
||||
.plugins_for_config(&config(r#"model = "first""#))
|
||||
.await;
|
||||
std::fs::remove_file(plugin_root.join(".mcp.json")).unwrap();
|
||||
let second = manager
|
||||
.plugins_for_config(&config(r#"model = "second""#))
|
||||
.await;
|
||||
|
||||
assert_eq!(second, first);
|
||||
assert_eq!(second.plugins()[0].mcp_servers.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn load_plugins_rejects_invalid_plugin_keys() {
|
||||
let codex_home = TempDir::new().unwrap();
|
||||
|
||||
Reference in New Issue
Block a user