mirror of
https://github.com/openai/codex.git
synced 2026-11-11 21:10:39 +00:00
defer plugin and MCP initialization
This commit is contained in:
@@ -9,6 +9,7 @@
|
||||
use std::collections::HashMap;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::RwLock;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
@@ -104,7 +105,7 @@ pub fn tool_is_model_visible(tool: &ToolInfo) -> bool {
|
||||
pub struct McpConnectionManager {
|
||||
clients: HashMap<String, AsyncManagedClient>,
|
||||
server_metadata: HashMap<String, McpServerMetadata>,
|
||||
tool_plugin_provenance: Arc<ToolPluginProvenance>,
|
||||
tool_plugin_provenance: Arc<RwLock<ToolPluginProvenance>>,
|
||||
host_owned_codex_apps_enabled: bool,
|
||||
prefix_mcp_tool_names: bool,
|
||||
elicitation_requests: ElicitationRequestManager,
|
||||
@@ -132,7 +133,7 @@ impl McpConnectionManager {
|
||||
Self {
|
||||
clients: HashMap::new(),
|
||||
server_metadata: HashMap::new(),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
host_owned_codex_apps_enabled: false,
|
||||
prefix_mcp_tool_names,
|
||||
elicitation_requests: ElicitationRequestManager::new(
|
||||
@@ -179,9 +180,12 @@ impl McpConnectionManager {
|
||||
.is_none_or(|metadata| metadata.pollutes_memory)
|
||||
}
|
||||
|
||||
pub fn plugin_id_for_mcp_server_name(&self, server_name: &str) -> Option<&str> {
|
||||
pub fn plugin_id_for_mcp_server_name(&self, server_name: &str) -> Option<String> {
|
||||
self.tool_plugin_provenance
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.plugin_id_for_mcp_server_name(server_name)
|
||||
.map(str::to_string)
|
||||
}
|
||||
|
||||
pub fn is_host_owned_codex_apps_server(&self, server_name: &str) -> bool {
|
||||
@@ -227,27 +231,85 @@ impl McpConnectionManager {
|
||||
elicitation_reviewer: Option<ElicitationReviewerHandle>,
|
||||
) -> (Self, CancellationToken) {
|
||||
let cancel_token = CancellationToken::new();
|
||||
let mut clients = HashMap::new();
|
||||
let mut server_metadata = HashMap::new();
|
||||
let mut join_set = JoinSet::new();
|
||||
let elicitation_requests = ElicitationRequestManager::new(
|
||||
approval_policy.value(),
|
||||
initial_permission_profile,
|
||||
elicitation_reviewer,
|
||||
);
|
||||
let tool_plugin_provenance = Arc::new(tool_plugin_provenance);
|
||||
let mut manager = Self {
|
||||
clients: HashMap::new(),
|
||||
server_metadata: HashMap::new(),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
host_owned_codex_apps_enabled,
|
||||
prefix_mcp_tool_names,
|
||||
elicitation_requests,
|
||||
startup_cancellation_token: cancel_token.clone(),
|
||||
};
|
||||
let startup_was_empty = !mcp_servers.values().any(EffectiveMcpServer::enabled);
|
||||
let empty_startup_event = startup_was_empty.then(|| Event {
|
||||
id: submit_id.clone(),
|
||||
msg: EventMsg::McpStartupComplete(McpStartupCompleteEvent::default()),
|
||||
});
|
||||
manager
|
||||
.add_servers(
|
||||
mcp_servers,
|
||||
store_mode,
|
||||
submit_id,
|
||||
tx_event.clone(),
|
||||
runtime_context,
|
||||
codex_home,
|
||||
codex_apps_tools_cache_key,
|
||||
client_elicitation_capability,
|
||||
tool_plugin_provenance,
|
||||
auth,
|
||||
)
|
||||
.await;
|
||||
if let Some(event) = empty_startup_event {
|
||||
let _ = tx_event.send(event).await;
|
||||
}
|
||||
(manager, cancel_token)
|
||||
}
|
||||
|
||||
/// Start newly discovered MCP servers without reconnecting servers already owned by this
|
||||
/// manager. Plugin discovery uses this after thread readiness has been published.
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
pub async fn add_servers(
|
||||
&mut self,
|
||||
mcp_servers: &HashMap<String, EffectiveMcpServer>,
|
||||
store_mode: OAuthCredentialsStoreMode,
|
||||
submit_id: String,
|
||||
tx_event: Sender<Event>,
|
||||
runtime_context: McpRuntimeContext,
|
||||
codex_home: PathBuf,
|
||||
codex_apps_tools_cache_key: CodexAppsToolsCacheKey,
|
||||
client_elicitation_capability: ElicitationCapability,
|
||||
tool_plugin_provenance: ToolPluginProvenance,
|
||||
auth: Option<&CodexAuth>,
|
||||
) {
|
||||
*self
|
||||
.tool_plugin_provenance
|
||||
.write()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner) = tool_plugin_provenance;
|
||||
let mut join_set = JoinSet::new();
|
||||
let startup_submit_id = submit_id.clone();
|
||||
let codex_apps_auth_provider = auth
|
||||
.filter(|auth| auth.uses_codex_backend())
|
||||
.map(codex_model_provider::auth_provider_from_auth);
|
||||
let mcp_servers = mcp_servers.clone();
|
||||
for (server_name, server) in mcp_servers
|
||||
.into_iter()
|
||||
.filter(|(_, server)| server.enabled())
|
||||
{
|
||||
let new_servers = mcp_servers
|
||||
.iter()
|
||||
.filter(|(server_name, server)| {
|
||||
server.enabled() && !self.clients.contains_key(*server_name)
|
||||
})
|
||||
.map(|(server_name, server)| (server_name.clone(), server.clone()))
|
||||
.collect::<Vec<_>>();
|
||||
if new_servers.is_empty() {
|
||||
return;
|
||||
}
|
||||
for (server_name, server) in new_servers {
|
||||
let configured_server = server.configured_config().cloned();
|
||||
server_metadata.insert(server_name.clone(), McpServerMetadata::from(&server));
|
||||
let cancel_token = cancel_token.child_token();
|
||||
self.server_metadata
|
||||
.insert(server_name.clone(), McpServerMetadata::from(&server));
|
||||
let cancel_token = self.startup_cancellation_token.child_token();
|
||||
let _ = emit_update(
|
||||
startup_submit_id.as_str(),
|
||||
&tx_event,
|
||||
@@ -287,14 +349,15 @@ impl McpConnectionManager {
|
||||
store_mode,
|
||||
cancel_token.clone(),
|
||||
tx_event.clone(),
|
||||
elicitation_requests.clone(),
|
||||
self.elicitation_requests.clone(),
|
||||
codex_apps_tools_cache_context,
|
||||
Arc::clone(&tool_plugin_provenance),
|
||||
Arc::clone(&self.tool_plugin_provenance),
|
||||
runtime_context.clone(),
|
||||
runtime_auth_provider,
|
||||
client_elicitation_capability.clone(),
|
||||
);
|
||||
clients.insert(server_name.clone(), async_managed_client.clone());
|
||||
self.clients
|
||||
.insert(server_name.clone(), async_managed_client.clone());
|
||||
let tx_event = tx_event.clone();
|
||||
let submit_id = startup_submit_id.clone();
|
||||
join_set.spawn(async move {
|
||||
@@ -328,15 +391,6 @@ impl McpConnectionManager {
|
||||
(server_name, outcome)
|
||||
});
|
||||
}
|
||||
let manager = Self {
|
||||
clients,
|
||||
server_metadata,
|
||||
tool_plugin_provenance,
|
||||
host_owned_codex_apps_enabled,
|
||||
prefix_mcp_tool_names,
|
||||
elicitation_requests: elicitation_requests.clone(),
|
||||
startup_cancellation_token: cancel_token.clone(),
|
||||
};
|
||||
tokio::spawn(async move {
|
||||
let outcomes = join_set.join_all().await;
|
||||
let mut summary = McpStartupCompleteEvent::default();
|
||||
@@ -359,7 +413,6 @@ impl McpConnectionManager {
|
||||
})
|
||||
.await;
|
||||
});
|
||||
(manager, cancel_token)
|
||||
}
|
||||
|
||||
pub async fn resolve_elicitation(
|
||||
@@ -829,9 +882,13 @@ fn mcp_init_error_display(
|
||||
"The {server_name} MCP server is not logged in. Run `codex mcp login {server_name}`."
|
||||
)
|
||||
} else if is_mcp_client_startup_timeout_error(err) {
|
||||
let startup_timeout_secs = config
|
||||
.and_then(|config| config.startup_timeout_sec)
|
||||
.unwrap_or(DEFAULT_STARTUP_TIMEOUT)
|
||||
let startup_timeout_secs = match config {
|
||||
Some(config) => match config.startup_timeout_sec {
|
||||
Some(timeout) => timeout,
|
||||
None => DEFAULT_STARTUP_TIMEOUT,
|
||||
},
|
||||
None => DEFAULT_STARTUP_TIMEOUT,
|
||||
}
|
||||
.as_secs();
|
||||
format!(
|
||||
"MCP client for `{server_name}` timed out after {startup_timeout_secs} seconds. Add or adjust `startup_timeout_sec` in your config.toml:\n[mcp_servers.{server_name}]\nstartup_timeout_sec = XX"
|
||||
|
||||
@@ -36,6 +36,7 @@ use rmcp::model::NumberOrString;
|
||||
use rmcp::model::Tool;
|
||||
use std::collections::HashSet;
|
||||
use std::sync::Arc;
|
||||
use std::sync::RwLock;
|
||||
use tempfile::tempdir;
|
||||
|
||||
fn create_test_tool(server_name: &str, tool_name: &str) -> ToolInfo {
|
||||
@@ -810,7 +811,7 @@ async fn list_all_tools_uses_cached_tool_info_snapshot_while_client_is_pending()
|
||||
cached_tool_info_snapshot: Some(startup_tools),
|
||||
cached_server_info: None,
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
);
|
||||
@@ -847,7 +848,7 @@ async fn list_available_server_infos_uses_cache_while_client_is_pending() {
|
||||
cached_tool_info_snapshot: Some(Vec::new()),
|
||||
cached_server_info: Some(server_info.clone()),
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
);
|
||||
@@ -884,7 +885,7 @@ async fn list_all_tools_accepts_canonical_namespaced_tool_names() {
|
||||
cached_tool_info_snapshot: Some(startup_tools),
|
||||
cached_server_info: None,
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
);
|
||||
@@ -927,7 +928,7 @@ async fn list_all_tools_applies_legacy_mcp_prefix_by_default() {
|
||||
cached_tool_info_snapshot: Some(startup_tools),
|
||||
cached_server_info: None,
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
);
|
||||
@@ -969,7 +970,7 @@ async fn list_all_tools_blocks_while_client_is_pending_without_cached_tool_info_
|
||||
cached_tool_info_snapshot: None,
|
||||
cached_server_info: None,
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
);
|
||||
@@ -998,7 +999,7 @@ async fn list_all_tools_does_not_block_when_cached_tool_info_snapshot_is_empty()
|
||||
cached_tool_info_snapshot: Some(Vec::new()),
|
||||
cached_server_info: None,
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
);
|
||||
@@ -1038,7 +1039,7 @@ async fn list_all_tools_uses_cached_tool_info_snapshot_when_client_startup_fails
|
||||
cached_tool_info_snapshot: Some(startup_tools),
|
||||
cached_server_info: Some(server_info.clone()),
|
||||
startup_complete,
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
);
|
||||
@@ -1093,7 +1094,7 @@ async fn list_all_tools_adds_server_metadata_to_cached_tools() {
|
||||
cached_tool_info_snapshot: Some(startup_tools),
|
||||
cached_server_info: None,
|
||||
startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)),
|
||||
tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()),
|
||||
tool_plugin_provenance: Arc::new(RwLock::new(ToolPluginProvenance::default())),
|
||||
cancel_token: CancellationToken::new(),
|
||||
},
|
||||
);
|
||||
|
||||
@@ -11,6 +11,7 @@ use std::collections::HashMap;
|
||||
use std::env;
|
||||
use std::ffi::OsString;
|
||||
use std::sync::Arc;
|
||||
use std::sync::RwLock;
|
||||
use std::sync::atomic::AtomicBool;
|
||||
use std::sync::atomic::Ordering;
|
||||
use std::time::Duration;
|
||||
@@ -129,7 +130,7 @@ pub(crate) struct AsyncManagedClient {
|
||||
pub(crate) cached_tool_info_snapshot: Option<Vec<ToolInfo>>,
|
||||
pub(crate) cached_server_info: Option<McpServerInfo>,
|
||||
pub(crate) startup_complete: Arc<AtomicBool>,
|
||||
pub(crate) tool_plugin_provenance: Arc<ToolPluginProvenance>,
|
||||
pub(crate) tool_plugin_provenance: Arc<RwLock<ToolPluginProvenance>>,
|
||||
pub(crate) cancel_token: CancellationToken,
|
||||
}
|
||||
|
||||
@@ -145,7 +146,7 @@ impl AsyncManagedClient {
|
||||
tx_event: Sender<Event>,
|
||||
elicitation_requests: ElicitationRequestManager,
|
||||
codex_apps_tools_cache_context: Option<CodexAppsToolsCacheContext>,
|
||||
tool_plugin_provenance: Arc<ToolPluginProvenance>,
|
||||
tool_plugin_provenance: Arc<RwLock<ToolPluginProvenance>>,
|
||||
runtime_context: McpRuntimeContext,
|
||||
runtime_auth_provider: Option<SharedAuthProvider>,
|
||||
client_elicitation_capability: ElicitationCapability,
|
||||
@@ -257,6 +258,10 @@ impl AsyncManagedClient {
|
||||
|
||||
pub(crate) async fn listed_tools(&self) -> Option<Vec<ToolInfo>> {
|
||||
let annotate_tools = |tools: Vec<ToolInfo>| {
|
||||
let tool_plugin_provenance = self
|
||||
.tool_plugin_provenance
|
||||
.read()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut tools = tools;
|
||||
for tool in &mut tools {
|
||||
if tool.server_name == CODEX_APPS_MCP_SERVER_NAME {
|
||||
@@ -264,11 +269,10 @@ impl AsyncManagedClient {
|
||||
}
|
||||
|
||||
let plugin_names = match tool.connector_id.as_deref() {
|
||||
Some(connector_id) => self
|
||||
.tool_plugin_provenance
|
||||
.plugin_display_names_for_connector_id(connector_id),
|
||||
None => self
|
||||
.tool_plugin_provenance
|
||||
Some(connector_id) => {
|
||||
tool_plugin_provenance.plugin_display_names_for_connector_id(connector_id)
|
||||
}
|
||||
None => tool_plugin_provenance
|
||||
.plugin_display_names_for_mcp_server_name(tool.server_name.as_str()),
|
||||
};
|
||||
tool.plugin_display_names = plugin_names.to_vec();
|
||||
|
||||
@@ -1383,6 +1383,17 @@ impl Config {
|
||||
) -> McpConfig {
|
||||
let plugins_input = self.plugins_config_input();
|
||||
let loaded_plugins = plugins_manager.plugins_for_config(&plugins_input).await;
|
||||
self.to_mcp_config_with_plugins(&loaded_plugins)
|
||||
}
|
||||
|
||||
pub(crate) fn to_mcp_config_without_plugins(&self) -> McpConfig {
|
||||
self.to_mcp_config_with_plugins(&codex_core_plugins::PluginLoadOutcome::default())
|
||||
}
|
||||
|
||||
pub(crate) fn to_mcp_config_with_plugins(
|
||||
&self,
|
||||
loaded_plugins: &codex_core_plugins::PluginLoadOutcome,
|
||||
) -> McpConfig {
|
||||
let mut configured_mcp_servers = self.mcp_servers.get().clone();
|
||||
let mut plugin_ids_by_mcp_server_name = HashMap::new();
|
||||
for plugin in loaded_plugins
|
||||
|
||||
@@ -35,6 +35,14 @@ impl McpManager {
|
||||
effective_mcp_servers(&mcp_config, auth)
|
||||
}
|
||||
|
||||
pub fn effective_servers_without_plugins(
|
||||
&self,
|
||||
config: &Config,
|
||||
auth: Option<&CodexAuth>,
|
||||
) -> HashMap<String, EffectiveMcpServer> {
|
||||
effective_mcp_servers(&config.to_mcp_config_without_plugins(), auth)
|
||||
}
|
||||
|
||||
pub async fn tool_plugin_provenance(&self, config: &Config) -> ToolPluginProvenance {
|
||||
let mcp_config = config.to_mcp_config(self.plugins_manager.as_ref()).await;
|
||||
collect_tool_plugin_provenance(&mcp_config)
|
||||
|
||||
@@ -1426,9 +1426,7 @@ pub(crate) async fn lookup_mcp_tool_metadata(
|
||||
tool_name: &str,
|
||||
) -> Option<McpToolApprovalMetadata> {
|
||||
let manager = sess.services.mcp_connection_manager.read().await;
|
||||
let plugin_id = manager
|
||||
.plugin_id_for_mcp_server_name(server)
|
||||
.map(str::to_string);
|
||||
let plugin_id = manager.plugin_id_for_mcp_server_name(server);
|
||||
let tools = manager.list_all_tools().await;
|
||||
let tool_info = tools
|
||||
.into_iter()
|
||||
|
||||
@@ -1,4 +1,5 @@
|
||||
use super::*;
|
||||
use crate::environment_selection::ResolvedTurnEnvironments;
|
||||
use crate::session::session::BaseInstructionsOrigin;
|
||||
use std::sync::atomic::Ordering;
|
||||
|
||||
@@ -93,6 +94,7 @@ impl Session {
|
||||
self.deferred_initialization_ready
|
||||
.get_or_init(|| async {
|
||||
self.ensure_model_catalog_ready().await;
|
||||
self.initialize_plugins_and_skills().await;
|
||||
let initial_history = self.initial_history.lock().await.take();
|
||||
if let Some(initial_history) = initial_history {
|
||||
self.record_initial_history(initial_history).await;
|
||||
@@ -101,6 +103,97 @@ impl Session {
|
||||
.await;
|
||||
}
|
||||
|
||||
#[expect(
|
||||
clippy::await_holding_invalid_type,
|
||||
reason = "the session-owned MCP manager must remain exclusively locked while adding servers"
|
||||
)]
|
||||
async fn initialize_plugins_and_skills(&self) {
|
||||
let config = self.get_config().await;
|
||||
let plugin_outcome = self
|
||||
.services
|
||||
.plugins_manager
|
||||
.plugins_for_config(&config.plugins_config_input())
|
||||
.await;
|
||||
let effective_skill_roots = plugin_outcome.effective_plugin_skill_roots();
|
||||
let skills_input = skills_load_input_from_config(config.as_ref(), effective_skill_roots);
|
||||
let environments = {
|
||||
let state = self.state.lock().await;
|
||||
state
|
||||
.session_configuration
|
||||
.environment_selections()
|
||||
.to_vec()
|
||||
};
|
||||
let resolved_environments = crate::environment_selection::resolve_environment_selections(
|
||||
self.services.environment_manager.as_ref(),
|
||||
&environments,
|
||||
)
|
||||
.ok();
|
||||
let fs = resolved_environments
|
||||
.as_ref()
|
||||
.and_then(ResolvedTurnEnvironments::primary_filesystem);
|
||||
let skills = self
|
||||
.services
|
||||
.skills_manager
|
||||
.skills_for_config(&skills_input, fs)
|
||||
.await;
|
||||
for err in skills.errors {
|
||||
error!(
|
||||
"failed to load skill {}: {}",
|
||||
err.path.display(),
|
||||
err.message
|
||||
);
|
||||
}
|
||||
|
||||
let hooks = super::build_hooks_with_plugins(
|
||||
config.as_ref(),
|
||||
self.services.user_shell.as_ref(),
|
||||
&plugin_outcome,
|
||||
);
|
||||
for warning in hooks.startup_warnings() {
|
||||
self.send_event_raw(Event {
|
||||
id: INITIAL_SUBMIT_ID.to_owned(),
|
||||
msg: EventMsg::Warning(WarningEvent {
|
||||
message: warning.clone(),
|
||||
}),
|
||||
})
|
||||
.await;
|
||||
}
|
||||
self.services.hooks.store(Arc::new(hooks));
|
||||
|
||||
let mcp_config = config.to_mcp_config_with_plugins(&plugin_outcome);
|
||||
let auth = self.services.auth_manager.auth().await;
|
||||
let mcp_servers = codex_mcp::effective_mcp_servers(&mcp_config, auth.as_ref());
|
||||
let tool_plugin_provenance = codex_mcp::tool_plugin_provenance(&mcp_config);
|
||||
let runtime_context =
|
||||
match resolved_environments.and_then(|resolved| resolved.primary().cloned()) {
|
||||
Some(turn_environment) => McpRuntimeContext::new(
|
||||
Arc::clone(&self.services.environment_manager),
|
||||
turn_environment.cwd.to_path_buf(),
|
||||
),
|
||||
None => McpRuntimeContext::new(
|
||||
Arc::clone(&self.services.environment_manager),
|
||||
config.cwd.to_path_buf(),
|
||||
),
|
||||
};
|
||||
self.services
|
||||
.mcp_connection_manager
|
||||
.write()
|
||||
.await
|
||||
.add_servers(
|
||||
&mcp_servers,
|
||||
mcp_config.mcp_oauth_credentials_store_mode,
|
||||
INITIAL_SUBMIT_ID.to_owned(),
|
||||
self.get_tx_event(),
|
||||
runtime_context,
|
||||
config.codex_home.to_path_buf(),
|
||||
codex_apps_tools_cache_key(auth.as_ref()),
|
||||
mcp_config.client_elicitation_capability,
|
||||
tool_plugin_provenance,
|
||||
auth.as_ref(),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[expect(
|
||||
clippy::await_holding_invalid_type,
|
||||
reason = "the prewarm gate must stay locked until the handle is installed"
|
||||
|
||||
@@ -3417,21 +3417,35 @@ async fn build_hooks_for_config(
|
||||
config: &Config,
|
||||
plugins_manager: &PluginsManager,
|
||||
user_shell: &crate::shell::Shell,
|
||||
) -> Hooks {
|
||||
let plugins_input = config.plugins_config_input();
|
||||
let plugin_outcome = plugins_manager.plugins_for_config(&plugins_input).await;
|
||||
build_hooks_with_plugins(config, user_shell, &plugin_outcome)
|
||||
}
|
||||
|
||||
fn build_hooks_without_plugins(config: &Config, user_shell: &crate::shell::Shell) -> Hooks {
|
||||
build_hooks_with_plugins(
|
||||
config,
|
||||
user_shell,
|
||||
&codex_core_plugins::PluginLoadOutcome::default(),
|
||||
)
|
||||
}
|
||||
|
||||
fn build_hooks_with_plugins(
|
||||
config: &Config,
|
||||
user_shell: &crate::shell::Shell,
|
||||
plugin_outcome: &codex_core_plugins::PluginLoadOutcome,
|
||||
) -> Hooks {
|
||||
let mut hook_shell_argv = user_shell.derive_exec_args("", /*use_login_shell*/ false);
|
||||
let hook_shell_program = hook_shell_argv.remove(0);
|
||||
let _ = hook_shell_argv.pop();
|
||||
let plugins_input = config.plugins_config_input();
|
||||
let plugin_outcome = plugins_manager.plugins_for_config(&plugins_input).await;
|
||||
let plugin_hook_sources = plugin_outcome.effective_plugin_hook_sources();
|
||||
let plugin_hook_load_warnings = plugin_outcome.effective_plugin_hook_warnings();
|
||||
Hooks::new(HooksConfig {
|
||||
legacy_notify_argv: config.notify.clone(),
|
||||
feature_enabled: config.features.enabled(Feature::CodexHooks),
|
||||
bypass_hook_trust: config.bypass_hook_trust,
|
||||
config_layer_stack: Some(config.config_layer_stack.clone()),
|
||||
plugin_hook_sources,
|
||||
plugin_hook_load_warnings,
|
||||
plugin_hook_sources: plugin_outcome.effective_plugin_hook_sources(),
|
||||
plugin_hook_load_warnings: plugin_outcome.effective_plugin_hook_warnings(),
|
||||
shell_program: Some(hook_shell_program),
|
||||
shell_args: hook_shell_argv,
|
||||
})
|
||||
|
||||
@@ -2,7 +2,6 @@ use super::input_queue::InputQueue;
|
||||
use super::*;
|
||||
use crate::agents_md::LoadedAgentsMd;
|
||||
use crate::config::ConstraintError;
|
||||
use crate::skills::SkillError;
|
||||
use crate::state::ActiveTurn;
|
||||
use crate::tools::registry::LateToolRegistry;
|
||||
use codex_protocol::SessionId;
|
||||
@@ -453,29 +452,6 @@ pub(crate) struct AppServerClientMetadata {
|
||||
pub(crate) client_version: Option<String>,
|
||||
}
|
||||
|
||||
async fn warm_plugins_and_skills_for_session_init(
|
||||
config: Arc<Config>,
|
||||
environment_manager: Arc<EnvironmentManager>,
|
||||
plugins_manager: Arc<PluginsManager>,
|
||||
skills_manager: Arc<SkillsManager>,
|
||||
environments: Vec<TurnEnvironmentSelection>,
|
||||
) -> Vec<SkillError> {
|
||||
let fs = crate::environment_selection::resolve_environment_selections(
|
||||
environment_manager.as_ref(),
|
||||
&environments,
|
||||
)
|
||||
.ok()
|
||||
.and_then(|resolved| resolved.primary_filesystem());
|
||||
let plugins_input = config.plugins_config_input();
|
||||
let plugin_outcome = plugins_manager.plugins_for_config(&plugins_input).await;
|
||||
let effective_skill_roots = plugin_outcome.effective_plugin_skill_roots();
|
||||
let skills_input = skills_load_input_from_config(config.as_ref(), effective_skill_roots);
|
||||
skills_manager
|
||||
.skills_for_config(&skills_input, fs)
|
||||
.await
|
||||
.errors
|
||||
}
|
||||
|
||||
impl Session {
|
||||
/// Returns the concrete identity for this thread.
|
||||
pub(crate) fn thread_id(&self) -> ThreadId {
|
||||
@@ -650,8 +626,7 @@ impl Session {
|
||||
let auth_and_mcp_fut = async move {
|
||||
let auth = auth_manager_clone.auth().await;
|
||||
let mcp_servers = mcp_manager_for_mcp
|
||||
.effective_servers(&config_for_mcp, auth.as_ref())
|
||||
.await;
|
||||
.effective_servers_without_plugins(&config_for_mcp, auth.as_ref());
|
||||
(auth, mcp_servers)
|
||||
}
|
||||
.instrument(info_span!(
|
||||
@@ -659,33 +634,9 @@ impl Session {
|
||||
otel.name = "session_init.auth_mcp",
|
||||
));
|
||||
|
||||
let plugin_and_skill_warmup_fut = warm_plugins_and_skills_for_session_init(
|
||||
Arc::clone(&config),
|
||||
Arc::clone(&environment_manager),
|
||||
Arc::clone(&plugins_manager),
|
||||
Arc::clone(&skills_manager),
|
||||
session_configuration.environment_selections().to_vec(),
|
||||
)
|
||||
.instrument(info_span!(
|
||||
"session_init.plugin_skill_warmup",
|
||||
otel.name = "session_init.plugin_skill_warmup",
|
||||
));
|
||||
|
||||
// Join all independent futures.
|
||||
let (thread_persistence_result, state_db_ctx, (auth, mcp_servers), plugin_skill_errors) = tokio::join!(
|
||||
thread_persistence_fut,
|
||||
state_db_fut,
|
||||
auth_and_mcp_fut,
|
||||
plugin_and_skill_warmup_fut
|
||||
);
|
||||
|
||||
for err in &plugin_skill_errors {
|
||||
error!(
|
||||
"failed to load skill {}: {}",
|
||||
err.path.display(),
|
||||
err.message
|
||||
);
|
||||
}
|
||||
let (thread_persistence_result, state_db_ctx, (auth, mcp_servers)) =
|
||||
tokio::join!(thread_persistence_fut, state_db_fut, auth_and_mcp_fut);
|
||||
|
||||
let mut live_thread_init =
|
||||
LiveThreadInitGuard::new(thread_persistence_result.map_err(|e| {
|
||||
@@ -949,8 +900,7 @@ impl Session {
|
||||
(None, None)
|
||||
};
|
||||
|
||||
let hooks =
|
||||
build_hooks_for_config(&config, plugins_manager.as_ref(), &default_shell).await;
|
||||
let hooks = build_hooks_without_plugins(&config, &default_shell);
|
||||
for warning in hooks.startup_warnings() {
|
||||
post_session_configured_events.push(Event {
|
||||
id: INITIAL_SUBMIT_ID.to_owned(),
|
||||
@@ -1145,7 +1095,6 @@ impl Session {
|
||||
for event in events {
|
||||
sess.send_event_raw(event).await;
|
||||
}
|
||||
|
||||
let mut required_mcp_servers: Vec<String> = mcp_servers
|
||||
.iter()
|
||||
.filter(|(_, server)| server.enabled() && server.required())
|
||||
@@ -1155,7 +1104,7 @@ impl Session {
|
||||
let enabled_mcp_server_count =
|
||||
mcp_servers.values().filter(|server| server.enabled()).count();
|
||||
let required_mcp_server_count = required_mcp_servers.len();
|
||||
let tool_plugin_provenance = mcp_manager.tool_plugin_provenance(config.as_ref()).await;
|
||||
let tool_plugin_provenance = codex_mcp::ToolPluginProvenance::default();
|
||||
let host_owned_codex_apps_enabled = config
|
||||
.features
|
||||
.apps_enabled_for_auth(auth.as_ref().is_some_and(|auth| auth.uses_codex_backend()));
|
||||
|
||||
Reference in New Issue
Block a user