From c34e07d94a074e527cfdafe1ced58232bf9c2843 Mon Sep 17 00:00:00 2001 From: Matthew Zeng Date: Wed, 17 Jun 2026 10:27:22 -0700 Subject: [PATCH] Add revision-gated runtime tool catalog --- codex-rs/codex-mcp/src/connection_manager.rs | 124 ++++++++++++++++- .../codex-mcp/src/connection_manager_tests.rs | 59 ++++++++ codex-rs/codex-mcp/src/lib.rs | 2 + codex-rs/core-plugins/src/discoverable.rs | 47 ++++++- codex-rs/core-plugins/src/lib.rs | 3 + codex-rs/core-plugins/src/manager.rs | 124 ++++++++++++++++- codex-rs/core-plugins/src/manager_tests.rs | 71 ++++++++++ .../src/plugin_catalog_revision.rs | 34 +++++ codex-rs/core-plugins/src/startup_sync.rs | 35 +++++ .../core-plugins/src/startup_sync_tests.rs | 7 + codex-rs/core/src/connectors.rs | 3 + codex-rs/core/src/connectors_tests.rs | 2 + codex-rs/core/src/lib.rs | 1 + codex-rs/core/src/plugins/discoverable.rs | 4 +- .../core/src/plugins/discoverable_tests.rs | 1 + codex-rs/core/src/runtime_tool_catalog.rs | 104 ++++++++++++++ .../core/src/runtime_tool_catalog_tests.rs | 130 ++++++++++++++++++ codex-rs/core/src/session/session.rs | 1 + codex-rs/core/src/session/tests.rs | 2 + codex-rs/core/src/session/turn.rs | 114 ++++++++++++++- codex-rs/core/src/state/service.rs | 2 + codex-rs/core/tests/suite/search_tool.rs | 60 ++++++++ 22 files changed, 907 insertions(+), 23 deletions(-) create mode 100644 codex-rs/core-plugins/src/plugin_catalog_revision.rs create mode 100644 codex-rs/core/src/runtime_tool_catalog.rs create mode 100644 codex-rs/core/src/runtime_tool_catalog_tests.rs diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index 3596486568..c46ec1dbbe 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -9,6 +9,7 @@ use std::collections::HashMap; use std::path::PathBuf; use std::sync::Arc; +use std::sync::atomic::AtomicU64; use std::sync::atomic::Ordering; use std::time::Duration; use std::time::Instant; @@ -79,6 +80,35 @@ const MCP_UI_META_KEY: &str = "ui"; const MCP_UI_VISIBILITY_META_KEY: &str = "visibility"; const MCP_UI_MODEL_VISIBILITY: &str = "model"; +static NEXT_HOSTED_CONNECTOR_RUNTIME_REVISION: AtomicU64 = AtomicU64::new(1); + +/// Monotonic revision of the host-owned Codex Apps runtime tool snapshot. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct HostedConnectorRuntimeRevision(u64); + +impl HostedConnectorRuntimeRevision { + pub fn generation(self) -> u64 { + self.0 + } +} + +/// Complete host-owned Codex Apps tools paired with the revision that covers them. +#[derive(Clone, Debug)] +pub struct HostedConnectorToolsSnapshot { + pub revision: HostedConnectorRuntimeRevision, + pub tools: Vec, +} + +fn next_hosted_connector_runtime_revision() -> u64 { + NEXT_HOSTED_CONNECTOR_RUNTIME_REVISION.fetch_add(1, Ordering::Relaxed) +} + +#[derive(Clone, Copy)] +enum McpToolSelection { + All, + HostedConnectors, +} + /// Returns whether a tool may be included in model-facing tool declarations. /// /// Tools without visibility metadata remain visible. @@ -110,6 +140,7 @@ pub struct McpConnectionManager { required_servers: Vec, tool_plugin_provenance: Arc, host_owned_codex_apps_enabled: bool, + hosted_connector_runtime_revision: Arc, prefix_mcp_tool_names: bool, elicitation_requests: ElicitationRequestManager, startup_cancellation_token: CancellationToken, @@ -151,6 +182,8 @@ impl McpConnectionManager { initial_permission_profile, elicitation_reviewer, ); + let hosted_connector_runtime_revision = + Arc::new(AtomicU64::new(next_hosted_connector_runtime_revision())); let tool_plugin_provenance = Arc::new(tool_plugin_provenance); let startup_submit_id = submit_id.clone(); let codex_apps_auth_provider = auth @@ -214,8 +247,15 @@ impl McpConnectionManager { let tx_event = tx_event.clone(); let submit_id = startup_submit_id.clone(); let auth_entry = auth_entries.get(&server_name).cloned(); + let advances_hosted_connector_revision = + host_owned_codex_apps_enabled && server_name == CODEX_APPS_MCP_SERVER_NAME; + let hosted_connector_runtime_revision = Arc::clone(&hosted_connector_runtime_revision); join_set.spawn(async move { let mut outcome = async_managed_client.client().await; + if advances_hosted_connector_revision && outcome.is_ok() { + hosted_connector_runtime_revision + .store(next_hosted_connector_runtime_revision(), Ordering::Release); + } if cancel_token.is_cancelled() { outcome = Err(StartupOutcomeError::Cancelled); } @@ -251,6 +291,7 @@ impl McpConnectionManager { required_servers, tool_plugin_provenance, host_owned_codex_apps_enabled, + hosted_connector_runtime_revision, prefix_mcp_tool_names, elicitation_requests: elicitation_requests.clone(), startup_cancellation_token: startup_cancellation_token.clone(), @@ -337,6 +378,9 @@ impl McpConnectionManager { required_servers: Vec::new(), tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()), host_owned_codex_apps_enabled: false, + hosted_connector_runtime_revision: Arc::new(AtomicU64::new( + next_hosted_connector_runtime_revision(), + )), prefix_mcp_tool_names, elicitation_requests: ElicitationRequestManager::new( approval_policy.value(), @@ -401,6 +445,24 @@ impl McpConnectionManager { self.host_owned_codex_apps_enabled && server_name == CODEX_APPS_MCP_SERVER_NAME } + pub fn hosted_connector_runtime_revision(&self) -> Option { + (self.host_owned_codex_apps_enabled + && self.clients.contains_key(CODEX_APPS_MCP_SERVER_NAME)) + .then(|| { + HostedConnectorRuntimeRevision( + self.hosted_connector_runtime_revision + .load(Ordering::Acquire), + ) + }) + } + + fn record_successful_hosted_tools_refresh(&self) { + if self.hosted_connector_runtime_revision().is_some() { + self.hosted_connector_runtime_revision + .store(next_hosted_connector_runtime_revision(), Ordering::Release); + } + } + pub fn set_approval_policy(&self, approval_policy: &Constrained) { if let Ok(mut policy) = self.elicitation_requests.approval_policy.lock() { *policy = approval_policy.value(); @@ -446,8 +508,61 @@ impl McpConnectionManager { /// Returns all tools with model-visible names normalized. #[instrument(level = "trace", skip_all, fields(mcp_server_count = self.clients.len()))] pub async fn list_all_tools(&self) -> Vec { + let tools = self + .list_tools(McpToolSelection::All, /*hosted_connector_tools*/ None) + .await; + normalize_tools_for_model_with_prefix(tools, self.prefix_mcp_tool_names) + } + + /// Reads a complete host-owned tool snapshot without racing a successful refresh. + pub async fn hosted_connector_tools_snapshot(&self) -> Option { + loop { + let revision = self.hosted_connector_runtime_revision()?; + let tools = self + .list_tools( + McpToolSelection::HostedConnectors, + /*hosted_connector_tools*/ None, + ) + .await; + if self.hosted_connector_runtime_revision() == Some(revision) { + return Some(HostedConnectorToolsSnapshot { revision, tools }); + } + } + } + + /// Combines a reusable hosted fragment with freshly listed non-hosted MCP tools. + pub async fn list_all_tools_with_hosted_connector_tools( + &self, + hosted_connector_tools: &[ToolInfo], + ) -> Vec { + let tools = self + .list_tools(McpToolSelection::All, Some(hosted_connector_tools)) + .await; + normalize_tools_for_model_with_prefix(tools, self.prefix_mcp_tool_names) + } + + async fn list_tools( + &self, + selection: McpToolSelection, + hosted_connector_tools: Option<&[ToolInfo]>, + ) -> Vec { let mut tools = Vec::new(); for (server_name, managed_client) in &self.clients { + let is_hosted_connector_source = + self.host_owned_codex_apps_enabled && server_name == CODEX_APPS_MCP_SERVER_NAME; + if is_hosted_connector_source + && let Some(hosted_connector_tools) = hosted_connector_tools + { + tools.extend_from_slice(hosted_connector_tools); + continue; + } + let selected = match selection { + McpToolSelection::All => true, + McpToolSelection::HostedConnectors => is_hosted_connector_source, + }; + if !selected { + continue; + } let has_cached_tool_info_snapshot = managed_client.cached_tool_info_snapshot.is_some(); let startup_complete = managed_client .startup_complete @@ -481,7 +596,7 @@ impl McpConnectionManager { .map(|tool| self.with_server_metadata(tool)), ); } - normalize_tools_for_model_with_prefix(tools, self.prefix_mcp_tool_names) + tools } /// Force-refresh codex apps tools by bypassing the in-process cache. @@ -533,10 +648,9 @@ impl McpConnectionManager { tool.tool = tool_with_model_visible_input_schema(&tool.tool); self.with_server_metadata(tool) }); - Ok(normalize_tools_for_model_with_prefix( - tools, - self.prefix_mcp_tool_names, - )) + let tools = normalize_tools_for_model_with_prefix(tools, self.prefix_mcp_tool_names); + self.record_successful_hosted_tools_refresh(); + Ok(tools) } /// Returns a single map that contains all resources. Each key is the diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index bfbc09c473..0308425560 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/codex-rs/codex-mcp/src/connection_manager_tests.rs @@ -115,6 +115,65 @@ fn model_tool_name_len(name: &ToolName) -> usize { + name.name.len() } +fn manager_with_pending_hosted_tools(tools: Vec) -> McpConnectionManager { + let pending_client = futures::future::pending::>() + .boxed() + .shared(); + let approval_policy = Constrained::allow_any(AskForApproval::OnFailure); + let permission_profile = Constrained::allow_any(PermissionProfile::default()); + let mut manager = McpConnectionManager::new_uninitialized( + &approval_policy, + &permission_profile, + /*prefix_mcp_tool_names*/ true, + ); + manager.host_owned_codex_apps_enabled = true; + manager.clients.insert( + CODEX_APPS_MCP_SERVER_NAME.to_string(), + AsyncManagedClient { + client: pending_client, + cached_tool_info_snapshot: Some(tools), + cached_server_info: None, + startup_complete: Arc::new(std::sync::atomic::AtomicBool::new(false)), + tool_plugin_provenance: Arc::new(ToolPluginProvenance::default()), + cancel_token: CancellationToken::new(), + }, + ); + manager +} + +#[tokio::test] +async fn hosted_revision_covers_refresh_replacement_and_snapshot_tools() { + let tool = create_test_tool_with_connector( + CODEX_APPS_MCP_SERVER_NAME, + "calendar_create_event", + "calendar", + Some("Calendar"), + ); + let manager = manager_with_pending_hosted_tools(vec![tool.clone()]); + let replacement = manager_with_pending_hosted_tools(Vec::new()); + assert_ne!( + manager.hosted_connector_runtime_revision(), + replacement.hosted_connector_runtime_revision() + ); + let before_refresh = manager.hosted_connector_runtime_revision(); + manager.record_successful_hosted_tools_refresh(); + assert_ne!(before_refresh, manager.hosted_connector_runtime_revision()); + + let snapshot = manager + .hosted_connector_tools_snapshot() + .await + .expect("hosted snapshot"); + + assert_eq!( + snapshot.revision, + manager.hosted_connector_runtime_revision().unwrap() + ); + assert_eq!( + serde_json::to_value(snapshot.tools).expect("serialize snapshot tools"), + serde_json::to_value(vec![tool]).expect("serialize expected tools") + ); +} + fn is_code_mode_compatible_tool_name(name: &ToolName) -> bool { name.namespace .as_deref() diff --git a/codex-rs/codex-mcp/src/lib.rs b/codex-rs/codex-mcp/src/lib.rs index b56a4f9072..b2ad1433d0 100644 --- a/codex-rs/codex-mcp/src/lib.rs +++ b/codex-rs/codex-mcp/src/lib.rs @@ -1,3 +1,5 @@ +pub use connection_manager::HostedConnectorRuntimeRevision; +pub use connection_manager::HostedConnectorToolsSnapshot; pub use connection_manager::McpConnectionManager; pub use connection_manager::tool_is_model_visible; pub use elicitation::ElicitationReviewRequest; diff --git a/codex-rs/core-plugins/src/discoverable.rs b/codex-rs/core-plugins/src/discoverable.rs index 0379819c47..bcd995ae94 100644 --- a/codex-rs/core-plugins/src/discoverable.rs +++ b/codex-rs/core-plugins/src/discoverable.rs @@ -9,8 +9,10 @@ use tracing::warn; use crate::OPENAI_API_CURATED_MARKETPLACE_NAME; use crate::OPENAI_CURATED_MARKETPLACE_NAME; +use crate::PluginCatalogRevision; use crate::PluginsConfigInput; use crate::PluginsManager; +use crate::marketplace::MarketplaceListOutcome; use crate::marketplace::MarketplacePluginInstallPolicy; use crate::remote::REMOTE_GLOBAL_MARKETPLACE_NAME; @@ -66,11 +68,34 @@ pub struct ToolSuggestDiscoverablePlugin { pub app_connector_ids: Vec, } +/// Selects the plugin catalog input used by the legacy tool-suggestion projection. +pub enum ToolSuggestPluginCatalog<'a> { + RebuildAll, + Revisioned { + revision: &'a PluginCatalogRevision, + marketplaces: &'a MarketplaceListOutcome, + }, +} + impl PluginsManager { pub async fn list_tool_suggest_discoverable_plugins( &self, input: &ToolSuggestPluginDiscoveryInput, auth: Option<&CodexAuth>, + ) -> anyhow::Result> { + self.list_tool_suggest_discoverable_plugins_with_catalog( + input, + auth, + ToolSuggestPluginCatalog::RebuildAll, + ) + .await + } + + pub async fn list_tool_suggest_discoverable_plugins_with_catalog( + &self, + input: &ToolSuggestPluginDiscoveryInput, + auth: Option<&CodexAuth>, + catalog: ToolSuggestPluginCatalog<'_>, ) -> anyhow::Result> { if !input.plugins.plugins_enabled { return Ok(Vec::new()); @@ -78,14 +103,22 @@ impl PluginsManager { let use_remote_global_catalog = input.plugins.remote_plugin_enabled && auth.is_some_and(CodexAuth::uses_codex_backend); - let marketplaces = self - .list_marketplaces_for_config( + let include_openai_curated = !use_remote_global_catalog; + let marketplaces = match catalog { + ToolSuggestPluginCatalog::RebuildAll => { + self.list_marketplaces_for_config(&input.plugins, &[], include_openai_curated) + } + ToolSuggestPluginCatalog::Revisioned { + revision, + marketplaces, + } => self.list_marketplaces_for_config_with_revisioned_plugin_catalog( &input.plugins, - &[], - /*include_openai_curated*/ !use_remote_global_catalog, - ) - .context("failed to list plugin marketplaces for tool suggestions")? - .marketplaces; + include_openai_curated, + (revision, marketplaces), + ), + } + .context("failed to list plugin marketplaces for tool suggestions")? + .marketplaces; let remote_installed_marketplaces = if use_remote_global_catalog { self.build_remote_installed_plugin_marketplaces_from_cache(&[ REMOTE_GLOBAL_MARKETPLACE_NAME, diff --git a/codex-rs/core-plugins/src/lib.rs b/codex-rs/core-plugins/src/lib.rs index 356c0f8bdc..c2a11ea55b 100644 --- a/codex-rs/core-plugins/src/lib.rs +++ b/codex-rs/core-plugins/src/lib.rs @@ -9,6 +9,7 @@ pub mod marketplace_add; pub mod marketplace_remove; pub mod marketplace_upgrade; mod plugin_bundle_archive; +mod plugin_catalog_revision; mod provider; pub mod remote; pub mod remote_bundle; @@ -33,6 +34,7 @@ pub type PluginLoadOutcome = codex_plugin::PluginLoadOutcome Option { + if !config.plugins_enabled + || !include_openai_curated + || matches!( + self.auth_mode(), + Some(AuthMode::ApiKey | AuthMode::BedrockApiKey) + ) + { + return None; + } + let curated_repo_root = curated_plugins_repo_path(self.codex_home.as_path()); + read_curated_plugins_catalog_revision(&curated_repo_root) + } + + pub fn build_revisioned_plugin_catalog( + &self, + revision: &PluginCatalogRevision, + ) -> Result { + let incomplete = || { + MarketplaceError::InvalidPlugin( + "revisioned plugin catalog did not load completely".to_string(), + ) + }; + if !revision.is_current() { + return Err(incomplete()); + } + let root = AbsolutePathBuf::try_from(revision.source_root().to_path_buf()) + .map_err(|_| incomplete())?; + let outcome = + list_marketplaces_with_home(std::slice::from_ref(&root), /*home_dir*/ None)?; + if !revision.is_current() || outcome.marketplaces.is_empty() || !outcome.errors.is_empty() { + return Err(incomplete()); + } + Ok(outcome) + } + + pub(crate) fn list_marketplaces_for_config_with_revisioned_plugin_catalog( + &self, + config: &PluginsConfigInput, + include_openai_curated: bool, + revisioned_catalog: (&PluginCatalogRevision, &MarketplaceListOutcome), + ) -> Result { + if !config.plugins_enabled { + return Ok(ConfiguredMarketplaceListOutcome::default()); + } + + let Some(current_revision) = + self.revisioned_plugin_catalog_revision_for_config(config, include_openai_curated) + else { + return self.list_marketplaces_for_config(config, &[], include_openai_curated); + }; + let (revision, revisioned_catalog) = revisioned_catalog; + if current_revision != *revision || !revision.is_current() { + return self.list_marketplaces_for_config(config, &[], include_openai_curated); + } + let roots = self.marketplace_roots(config, &[], include_openai_curated); + let Some(revisioned_index) = roots + .iter() + .position(|root| root.as_path() == current_revision.source_root()) + else { + return Err(MarketplaceError::InvalidPlugin( + "revisioned plugin catalog source is missing".to_string(), + )); + }; + let mut marketplace_outcome = + list_marketplaces_with_home(&roots[..revisioned_index], home_dir().as_deref())?; + extend_marketplace_outcome(&mut marketplace_outcome, revisioned_catalog.clone()); + let remaining = + list_marketplaces_with_home(&roots[revisioned_index + 1..], /*home_dir*/ None)?; + extend_marketplace_outcome(&mut marketplace_outcome, remaining); + + Ok(self.configured_marketplaces_from_outcome(config, marketplace_outcome)) + } + + fn configured_marketplaces_from_outcome( + &self, + config: &PluginsConfigInput, + marketplace_outcome: MarketplaceListOutcome, + ) -> ConfiguredMarketplaceListOutcome { + let (installed_plugins, enabled_plugins) = self.configured_plugin_states(config); let mut seen_plugin_keys = HashSet::new(); let marketplaces = marketplace_outcome .marketplaces @@ -1310,10 +1400,10 @@ impl PluginsManager { }) .collect(); - Ok(ConfiguredMarketplaceListOutcome { + ConfiguredMarketplaceListOutcome { marketplaces, errors: marketplace_outcome.errors, - }) + } } pub fn discover_marketplaces_for_config( @@ -2136,6 +2226,34 @@ fn remote_plugin_install_required_description(source: &MarketplacePluginSource) ) } +fn extend_marketplace_outcome( + aggregate: &mut MarketplaceListOutcome, + outcome: MarketplaceListOutcome, +) { + let mut marketplace_paths = aggregate + .marketplaces + .iter() + .map(|marketplace| marketplace.path.clone()) + .collect::>(); + aggregate.marketplaces.extend( + outcome + .marketplaces + .into_iter() + .filter(|marketplace| marketplace_paths.insert(marketplace.path.clone())), + ); + let mut error_paths = aggregate + .errors + .iter() + .map(|error| error.path.clone()) + .collect::>(); + aggregate.errors.extend( + outcome + .errors + .into_iter() + .filter(|error| error_paths.insert(error.path.clone())), + ); +} + #[derive(Debug, thiserror::Error)] pub enum PluginInstallError { #[error("{0}")] diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index c2917211d2..49614d98db 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/codex-rs/core-plugins/src/manager_tests.rs @@ -3163,6 +3163,77 @@ plugins = true ); } +#[tokio::test] +async fn curated_catalog_reuses_only_a_current_proven_revision() { + let tmp = tempfile::tempdir().unwrap(); + let curated_root = curated_plugins_repo_path(tmp.path()); + write_file( + &tmp.path().join(CONFIG_TOML_FILE), + r#"[features] +plugins = true +"#, + ); + write_openai_curated_marketplace(&curated_root, &["linear"]); + let config = load_config(tmp.path(), tmp.path()).await; + let manager = PluginsManager::new(tmp.path().to_path_buf()); + let plugin_ids = |outcome: ConfiguredMarketplaceListOutcome| { + outcome + .marketplaces + .into_iter() + .flat_map(|marketplace| marketplace.plugins) + .map(|plugin| plugin.id) + .collect::>() + }; + + let first = manager + .list_marketplaces_for_config(&config, &[], /*include_openai_curated*/ true) + .expect("first unrevisioned read"); + write_openai_curated_marketplace(&curated_root, &["github"]); + let second = manager + .list_marketplaces_for_config(&config, &[], /*include_openai_curated*/ true) + .expect("second unrevisioned read"); + assert_eq!(plugin_ids(first), vec!["linear@openai-curated"]); + assert_eq!(plugin_ids(second), vec!["github@openai-curated"]); + + write_file( + &curated_root.join(".codex-plugin-catalog-revision"), + "revision-one\n", + ); + let revision = manager + .revisioned_plugin_catalog_revision_for_config( + &config, /*include_openai_curated*/ true, + ) + .expect("revisioned curated source"); + + let outcome = manager + .build_revisioned_plugin_catalog(&revision) + .expect("complete revisioned catalog"); + assert_eq!(outcome.marketplaces.len(), 1); + write_openai_curated_marketplace(&curated_root, &["linear"]); + let reused = manager + .list_marketplaces_for_config_with_revisioned_plugin_catalog( + &config, + /*include_openai_curated*/ true, + (&revision, &outcome), + ) + .expect("reuse revisioned catalog"); + assert_eq!(plugin_ids(reused), vec!["github@openai-curated"]); + + write_file( + &curated_root.join(".codex-plugin-catalog-revision"), + "revision-two\n", + ); + assert!(manager.build_revisioned_plugin_catalog(&revision).is_err()); + let rebuilt = manager + .list_marketplaces_for_config_with_revisioned_plugin_catalog( + &config, + /*include_openai_curated*/ true, + (&revision, &outcome), + ) + .expect("rebuild stale catalog"); + assert_eq!(plugin_ids(rebuilt), vec!["linear@openai-curated"]); +} + #[tokio::test] async fn list_marketplaces_can_skip_openai_curated_before_loading() { let tmp = tempfile::tempdir().unwrap(); diff --git a/codex-rs/core-plugins/src/plugin_catalog_revision.rs b/codex-rs/core-plugins/src/plugin_catalog_revision.rs new file mode 100644 index 0000000000..d2762eb3f0 --- /dev/null +++ b/codex-rs/core-plugins/src/plugin_catalog_revision.rs @@ -0,0 +1,34 @@ +use std::fs; +use std::path::Path; +use std::path::PathBuf; + +/// A revision marker materialized inside the plugin catalog source it identifies. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PluginCatalogRevision { + source_root: PathBuf, + marker_path: PathBuf, + marker_contents: String, +} + +impl PluginCatalogRevision { + pub(crate) fn read(source_root: &Path, marker_path: PathBuf) -> Option { + let marker_contents = fs::read_to_string(&marker_path).ok()?; + Some(Self { + source_root: source_root.to_path_buf(), + marker_path, + marker_contents, + }) + } + + pub fn source_root(&self) -> &Path { + &self.source_root + } + + pub fn marker_contents(&self) -> &str { + &self.marker_contents + } + + pub(crate) fn is_current(&self) -> bool { + fs::read_to_string(&self.marker_path).is_ok_and(|contents| contents == self.marker_contents) + } +} diff --git a/codex-rs/core-plugins/src/startup_sync.rs b/codex-rs/core-plugins/src/startup_sync.rs index 44fa957940..e527e9d241 100644 --- a/codex-rs/core-plugins/src/startup_sync.rs +++ b/codex-rs/core-plugins/src/startup_sync.rs @@ -14,6 +14,7 @@ use tempfile::TempDir; use tracing::warn; use zip::ZipArchive; +use crate::plugin_catalog_revision::PluginCatalogRevision; use codex_login::default_client::build_reqwest_client; const GITHUB_API_BASE_URL: &str = "https://api.github.com"; @@ -27,6 +28,7 @@ const OPENAI_PLUGINS_GIT_URL: &str = "https://github.com/openai/plugins.git"; const CURATED_PLUGINS_FETCH_REF: &str = "refs/codex/curated-sync"; const CURATED_PLUGINS_RELATIVE_DIR: &str = ".tmp/plugins"; const CURATED_PLUGINS_SHA_FILE: &str = ".tmp/plugins.sha"; +const CURATED_PLUGINS_CATALOG_REVISION_FILE: &str = ".codex-plugin-catalog-revision"; const CURATED_PLUGINS_SYNC_LOCK_FILE: &str = ".tmp/plugins.sync.lock"; const CURATED_PLUGINS_BACKUP_ARCHIVE_FALLBACK_VERSION: &str = "export-backup"; const CURATED_PLUGINS_GIT_TIMEOUT: Duration = Duration::from_secs(30); @@ -67,10 +69,20 @@ pub fn read_curated_plugins_sha(codex_home: &Path) -> Option { read_sha_file(curated_plugins_sha_path(codex_home).as_path()) } +pub(crate) fn read_curated_plugins_catalog_revision( + repo_path: &Path, +) -> Option { + PluginCatalogRevision::read(repo_path, curated_plugins_catalog_revision_path(repo_path)) +} + fn curated_plugins_sha_path(codex_home: &Path) -> PathBuf { codex_home.join(CURATED_PLUGINS_SHA_FILE) } +fn curated_plugins_catalog_revision_path(repo_path: &Path) -> PathBuf { + repo_path.join(CURATED_PLUGINS_CATALOG_REVISION_FILE) +} + pub fn sync_openai_plugins_repo(codex_home: &Path) -> Result { sync_openai_plugins_repo_with_transport_overrides( codex_home, @@ -168,6 +180,7 @@ fn sync_openai_plugins_repo_via_git(codex_home: &Path, git_binary: &str) -> Resu let local_sha = read_local_git_or_sha_file(&repo_path, &sha_path, git_binary); if local_sha.as_deref() == Some(remote_sha.as_str()) && repo_path.join(".git").is_dir() { + ensure_curated_plugins_catalog_revision(&repo_path, &remote_sha)?; return Ok(remote_sha); } @@ -200,6 +213,7 @@ fn sync_openai_plugins_repo_via_git(codex_home: &Path, git_binary: &str) -> Resu } ensure_marketplace_manifest_exists(staged_repo_dir.path())?; + write_curated_plugins_catalog_revision(staged_repo_dir.path(), &remote_sha)?; activate_curated_repo(&repo_path, staged_repo_dir)?; write_curated_plugins_sha(&sha_path, &remote_sha)?; Ok(remote_sha) @@ -303,6 +317,7 @@ fn sync_openai_plugins_repo_via_http( let local_sha = read_sha_file(&sha_path); if local_sha.as_deref() == Some(remote_sha.as_str()) && repo_path.is_dir() { + ensure_curated_plugins_catalog_revision(&repo_path, &remote_sha)?; return Ok(remote_sha); } @@ -310,6 +325,7 @@ fn sync_openai_plugins_repo_via_http( let zipball_bytes = runtime.block_on(fetch_curated_repo_zipball(api_base_url, &remote_sha))?; extract_zipball_to_dir(&zipball_bytes, staged_repo_dir.path())?; ensure_marketplace_manifest_exists(staged_repo_dir.path())?; + write_curated_plugins_catalog_revision(staged_repo_dir.path(), &remote_sha)?; activate_curated_repo(&repo_path, staged_repo_dir)?; write_curated_plugins_sha(&sha_path, &remote_sha)?; Ok(remote_sha) @@ -333,6 +349,7 @@ fn sync_openai_plugins_repo_via_backup_archive( ensure_marketplace_manifest_exists(staged_repo_dir.path())?; let export_version = read_extracted_backup_archive_git_sha(staged_repo_dir.path())? .unwrap_or_else(|| CURATED_PLUGINS_BACKUP_ARCHIVE_FALLBACK_VERSION.to_string()); + write_curated_plugins_catalog_revision(staged_repo_dir.path(), &export_version)?; activate_curated_repo(&repo_path, staged_repo_dir)?; write_curated_plugins_sha(&sha_path, &export_version)?; Ok(export_version) @@ -568,6 +585,24 @@ fn write_curated_plugins_sha(sha_path: &Path, remote_sha: &str) -> Result<(), St }) } +fn ensure_curated_plugins_catalog_revision(repo_path: &Path, revision: &str) -> Result<(), String> { + let revision_path = curated_plugins_catalog_revision_path(repo_path); + if read_sha_file(&revision_path).as_deref() == Some(revision) { + return Ok(()); + } + write_curated_plugins_catalog_revision(repo_path, revision) +} + +fn write_curated_plugins_catalog_revision(repo_path: &Path, revision: &str) -> Result<(), String> { + let revision_path = curated_plugins_catalog_revision_path(repo_path); + std::fs::write(&revision_path, format!("{revision}\n")).map_err(|err| { + format!( + "failed to write curated plugins catalog revision file {}: {err}", + revision_path.display() + ) + }) +} + fn read_local_git_or_sha_file( repo_path: &Path, sha_path: &Path, diff --git a/codex-rs/core-plugins/src/startup_sync_tests.rs b/codex-rs/core-plugins/src/startup_sync_tests.rs index f73e4ad9d1..2d636aede3 100644 --- a/codex-rs/core-plugins/src/startup_sync_tests.rs +++ b/codex-rs/core-plugins/src/startup_sync_tests.rs @@ -600,6 +600,12 @@ async fn sync_openai_plugins_repo_falls_back_to_http_when_git_is_unavailable() { assert_eq!(synced_sha, sha); assert_curated_gmail_repo(&repo_path); assert_eq!(read_curated_plugins_sha(tmp.path()).as_deref(), Some(sha)); + let revision = read_curated_plugins_catalog_revision(&repo_path) + .expect("synced repository should carry a catalog revision"); + assert_eq!(revision.marker_contents(), format!("{sha}\n")); + assert!(revision.is_current()); + write_curated_plugins_catalog_revision(&repo_path, "replacement").expect("replace revision"); + assert!(!revision.is_current()); } #[cfg(unix)] @@ -796,6 +802,7 @@ async fn sync_openai_plugins_repo_skips_archive_download_when_sha_matches() { assert_eq!(read_curated_plugins_sha(tmp.path()).as_deref(), Some(sha)); assert!(repo_path.join(".agents/plugins/marketplace.json").is_file()); + assert!(read_curated_plugins_catalog_revision(&repo_path).is_some()); } #[tokio::test] diff --git a/codex-rs/core/src/connectors.rs b/codex-rs/core/src/connectors.rs index f1c8c5607b..38fc73f660 100644 --- a/codex-rs/core/src/connectors.rs +++ b/codex-rs/core/src/connectors.rs @@ -28,6 +28,7 @@ use crate::session::INITIAL_SUBMIT_ID; use codex_config::types::ApprovalsReviewer; use codex_config::types::ToolSuggestDiscoverableType; use codex_core_plugins::PluginsManager; +use codex_core_plugins::ToolSuggestPluginCatalog; use codex_features::Feature; use codex_login::AuthManager; use codex_login::CodexAuth; @@ -100,6 +101,7 @@ pub(crate) async fn list_tool_suggest_discoverable_tools_with_auth( auth: Option<&CodexAuth>, accessible_connectors: &[AppInfo], loaded_plugin_app_connector_ids: &[String], + plugin_catalog: ToolSuggestPluginCatalog<'_>, ) -> anyhow::Result> { let connector_ids = tool_suggest_connector_ids(config, loaded_plugin_app_connector_ids); let directory_connectors = codex_connectors::merge::merge_plugin_connectors( @@ -120,6 +122,7 @@ pub(crate) async fn list_tool_suggest_discoverable_tools_with_auth( plugins_manager, auth, loaded_plugin_app_connector_ids, + plugin_catalog, ) .await? .into_iter() diff --git a/codex-rs/core/src/connectors_tests.rs b/codex-rs/core/src/connectors_tests.rs index 4e6cb55aaa..ccadff4dd2 100644 --- a/codex-rs/core/src/connectors_tests.rs +++ b/codex-rs/core/src/connectors_tests.rs @@ -495,6 +495,7 @@ discoverables = [ Some(&auth), &[], &[], + ToolSuggestPluginCatalog::RebuildAll, ) .await .expect("discoverable tools should load"); @@ -533,6 +534,7 @@ apps = true Some(&auth), &[], &loaded_plugin_app_connector_ids, + ToolSuggestPluginCatalog::RebuildAll, ) .await .expect("discoverable tools should load"); diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 073c06cede..10db58dffe 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -55,6 +55,7 @@ mod mcp_tool_approval_templates; mod mcp_tool_exposure; mod network_policy_decision; pub(crate) mod network_proxy_loader; +mod runtime_tool_catalog; pub use mcp::McpManager; pub use network_proxy_loader::MtimeConfigReloader; pub use network_proxy_loader::build_network_proxy_state; diff --git a/codex-rs/core/src/plugins/discoverable.rs b/codex-rs/core/src/plugins/discoverable.rs index 960e244e97..7ad4b415fb 100644 --- a/codex-rs/core/src/plugins/discoverable.rs +++ b/codex-rs/core/src/plugins/discoverable.rs @@ -1,6 +1,7 @@ use crate::config::Config; use codex_config::types::ToolSuggestDiscoverableType; use codex_core_plugins::PluginsManager; +use codex_core_plugins::ToolSuggestPluginCatalog; use codex_core_plugins::ToolSuggestPluginDiscoveryInput; use codex_login::CodexAuth; use codex_tools::DiscoverablePluginInfo; @@ -13,6 +14,7 @@ pub(crate) async fn list_tool_suggest_discoverable_plugins( plugins_manager: &PluginsManager, auth: Option<&CodexAuth>, loaded_plugin_app_connector_ids: &[String], + catalog: ToolSuggestPluginCatalog<'_>, ) -> anyhow::Result> { let input = ToolSuggestPluginDiscoveryInput { plugins: config.plugins_config_input(), @@ -36,7 +38,7 @@ pub(crate) async fn list_tool_suggest_discoverable_plugins( .collect::>(), }; plugins_manager - .list_tool_suggest_discoverable_plugins(&input, auth) + .list_tool_suggest_discoverable_plugins_with_catalog(&input, auth, catalog) .await .map(|plugins| { plugins diff --git a/codex-rs/core/src/plugins/discoverable_tests.rs b/codex-rs/core/src/plugins/discoverable_tests.rs index 2d92e70964..fa7f37835b 100644 --- a/codex-rs/core/src/plugins/discoverable_tests.rs +++ b/codex-rs/core/src/plugins/discoverable_tests.rs @@ -49,6 +49,7 @@ async fn list_discoverable_plugins_with_manager_and_auth( plugins_manager, auth, loaded_plugin_app_connector_ids, + codex_core_plugins::ToolSuggestPluginCatalog::RebuildAll, ) .await } diff --git a/codex-rs/core/src/runtime_tool_catalog.rs b/codex-rs/core/src/runtime_tool_catalog.rs new file mode 100644 index 0000000000..da2d04791c --- /dev/null +++ b/codex-rs/core/src/runtime_tool_catalog.rs @@ -0,0 +1,104 @@ +use std::path::PathBuf; +use std::sync::Arc; +use std::sync::PoisonError; +use std::sync::RwLock; + +use codex_app_server_protocol::AppInfo; +use codex_core_plugins::PluginCatalogRevision; +use codex_core_plugins::marketplace::MarketplaceListOutcome; +use codex_mcp::HostedConnectorRuntimeRevision; +use codex_mcp::ToolInfo; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct HostedConnectorsRevision(pub(crate) u64); + +impl From for HostedConnectorsRevision { + fn from(revision: HostedConnectorRuntimeRevision) -> Self { + Self(revision.generation()) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct RevisionedPluginsRevision(PathBuf, String); + +impl From<&PluginCatalogRevision> for RevisionedPluginsRevision { + fn from(revision: &PluginCatalogRevision) -> Self { + Self( + revision.source_root().to_path_buf(), + revision.marker_contents().to_string(), + ) + } +} + +pub(crate) struct HostedConnectorRuntimeFragment { + pub(crate) revision: HostedConnectorsRevision, + pub(crate) tools: Vec, + pub(crate) connectors: Vec, +} + +pub(crate) struct RevisionedPluginCatalogFragment { + pub(crate) revision: RevisionedPluginsRevision, + pub(crate) marketplaces: MarketplaceListOutcome, +} + +#[derive(Default)] +pub(crate) struct RuntimeToolCatalogSnapshot { + pub(crate) hosted_connectors: Option>, + pub(crate) revisioned_plugins: Option>, +} + +impl RuntimeToolCatalogSnapshot { + pub(crate) fn hosted_connectors_for( + &self, + revision: HostedConnectorsRevision, + ) -> Option> { + self.hosted_connectors + .as_ref() + .filter(|fragment| fragment.revision == revision) + .map(Arc::clone) + } + + pub(crate) fn revisioned_plugins_for( + &self, + revision: &RevisionedPluginsRevision, + ) -> Option> { + self.revisioned_plugins + .as_ref() + .filter(|fragment| fragment.revision == *revision) + .map(Arc::clone) + } +} + +#[derive(Default)] +pub(crate) struct RuntimeToolCatalogManager { + snapshot: RwLock>, +} + +impl RuntimeToolCatalogManager { + pub(crate) fn snapshot(&self) -> Arc { + self.snapshot + .read() + .unwrap_or_else(PoisonError::into_inner) + .clone() + } + + pub(crate) fn publish_if_current( + &self, + base: &Arc, + next: Result, + ) -> Result, E> { + let next = Arc::new(next?); + let mut current = self + .snapshot + .write() + .unwrap_or_else(PoisonError::into_inner); + if Arc::ptr_eq(¤t, base) { + *current = next; + } + Ok(Arc::clone(¤t)) + } +} + +#[cfg(test)] +#[path = "runtime_tool_catalog_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/runtime_tool_catalog_tests.rs b/codex-rs/core/src/runtime_tool_catalog_tests.rs new file mode 100644 index 0000000000..7c4add9b67 --- /dev/null +++ b/codex-rs/core/src/runtime_tool_catalog_tests.rs @@ -0,0 +1,130 @@ +use super::*; +use codex_core_plugins::marketplace::MarketplaceListOutcome; +use std::path::PathBuf; +use std::sync::Arc; + +fn hosted(revision: u64) -> Arc { + Arc::new(HostedConnectorRuntimeFragment { + revision: HostedConnectorsRevision(revision), + tools: Vec::new(), + connectors: Vec::new(), + }) +} + +fn plugin_revision(revision: &str) -> RevisionedPluginsRevision { + RevisionedPluginsRevision(PathBuf::from("/plugin-catalog"), revision.to_string()) +} + +fn plugins(revision: &str) -> Arc { + Arc::new(RevisionedPluginCatalogFragment { + revision: plugin_revision(revision), + marketplaces: MarketplaceListOutcome::default(), + }) +} + +fn catalog( + hosted_connectors: Option>, + revisioned_plugins: Option>, +) -> RuntimeToolCatalogSnapshot { + RuntimeToolCatalogSnapshot { + hosted_connectors, + revisioned_plugins, + } +} + +#[test] +fn hosted_revision_controls_fragment_reuse() { + let fragment = hosted(1); + let snapshot = catalog(Some(Arc::clone(&fragment)), None); + + assert!(Arc::ptr_eq( + &fragment, + &snapshot + .hosted_connectors_for(HostedConnectorsRevision(1)) + .unwrap() + )); + assert!( + snapshot + .hosted_connectors_for(HostedConnectorsRevision(2)) + .is_none() + ); +} + +#[test] +fn only_proven_unchanged_plugin_revisions_are_reused() { + let fragment = plugins("one"); + let snapshot = catalog(None, Some(Arc::clone(&fragment))); + + assert!(Arc::ptr_eq( + &fragment, + &snapshot + .revisioned_plugins_for(&plugin_revision("one")) + .unwrap() + )); + assert!( + snapshot + .revisioned_plugins_for(&plugin_revision("two")) + .is_none() + ); +} + +#[test] +fn source_changes_are_independent_and_change_the_aggregate() { + let manager = RuntimeToolCatalogManager::default(); + let empty = manager.snapshot(); + let plugins = plugins("one"); + let populated = manager + .publish_if_current( + &empty, + Ok::<_, ()>(catalog(Some(hosted(1)), Some(Arc::clone(&plugins)))), + ) + .unwrap(); + let hosted_changed = manager + .publish_if_current( + &populated, + Ok::<_, ()>(catalog( + Some(hosted(2)), + populated.revisioned_plugins_for(&plugin_revision("one")), + )), + ) + .unwrap(); + let hosted_only = manager + .publish_if_current( + &hosted_changed, + Ok::<_, ()>(catalog(hosted_changed.hosted_connectors.clone(), None)), + ) + .unwrap(); + + assert!(Arc::ptr_eq( + &plugins, + hosted_changed.revisioned_plugins.as_ref().unwrap() + )); + assert_eq!( + ( + empty.hosted_connectors.is_some(), + populated.revisioned_plugins.is_some(), + hosted_only.revisioned_plugins.is_some(), + ), + (false, true, false) + ); +} + +#[test] +fn failed_rebuild_does_not_publish_partial_state() { + let manager = RuntimeToolCatalogManager::default(); + let empty = manager.snapshot(); + let initial = manager + .publish_if_current( + &empty, + Ok::<_, &str>(catalog(Some(hosted(1)), Some(plugins("one")))), + ) + .unwrap(); + + let result = manager.publish_if_current( + &initial, + Err::("plugin rebuild failed"), + ); + + assert!(matches!(result, Err("plugin rebuild failed"))); + assert!(Arc::ptr_eq(&initial, &manager.snapshot())); +} diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 578152814b..a4dfeeb90d 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -980,6 +980,7 @@ impl Session { // changing this to use Option or OnceCell, though the current // setup is straightforward enough and performs well. mcp_connection_manager, + runtime_tool_catalog_manager: Default::default(), mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()), unified_exec_manager: UnifiedExecProcessManager::new( config.background_terminal_max_timeout, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 473adaa835..b03fc89e0d 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -4973,6 +4973,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { config.prefix_mcp_tool_names(), ), )), + runtime_tool_catalog_manager: Default::default(), mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()), unified_exec_manager: UnifiedExecProcessManager::new( config.background_terminal_max_timeout, @@ -7018,6 +7019,7 @@ where config.prefix_mcp_tool_names(), ), )), + runtime_tool_catalog_manager: Default::default(), mcp_startup_cancellation_token: Mutex::new(CancellationToken::new()), unified_exec_manager: UnifiedExecProcessManager::new( config.background_terminal_max_timeout, diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 7da157cf7b..12b64b70b6 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -39,6 +39,11 @@ use crate::responses_metadata::CodexResponsesMetadata; use crate::responses_metadata::CodexResponsesRequestKind; use crate::responses_retry::ResponsesStreamRequest; use crate::responses_retry::handle_retryable_response_stream_error; +use crate::runtime_tool_catalog::HostedConnectorRuntimeFragment; +use crate::runtime_tool_catalog::HostedConnectorsRevision; +use crate::runtime_tool_catalog::RevisionedPluginCatalogFragment; +use crate::runtime_tool_catalog::RevisionedPluginsRevision; +use crate::runtime_tool_catalog::RuntimeToolCatalogSnapshot; use crate::session::PreviousTurnSettings; use crate::session::TurnInput; use crate::session::session::Session; @@ -74,6 +79,7 @@ use codex_analytics::TurnResolvedConfigFact; use codex_analytics::build_track_events_context; use codex_async_utils::OrCancelExt; use codex_core_plugins::RecommendedPluginCandidatesInput; +use codex_core_plugins::ToolSuggestPluginCatalog; use codex_core_skills::injection::InjectedHostSkillPrompts; use codex_extension_api::TurnInputContext; use codex_extension_api::TurnInputEnvironment; @@ -1156,11 +1162,44 @@ pub(crate) async fn built_tools( cancellation_token: &CancellationToken, ) -> CodexResult> { let mcp_connection_manager = sess.services.mcp_connection_manager.load_full(); + let runtime_catalog_manager = &sess.services.runtime_tool_catalog_manager; + let base_catalog = runtime_catalog_manager.snapshot(); + let hosted_connectors = if let Some(revision) = mcp_connection_manager + .hosted_connector_runtime_revision() + .map(HostedConnectorsRevision::from) + { + if let Some(fragment) = base_catalog.hosted_connectors_for(revision) { + Some(fragment) + } else { + mcp_connection_manager + .hosted_connector_tools_snapshot() + .or_cancel(cancellation_token) + .await? + .map(|snapshot| { + let connectors = + connectors::accessible_connectors_from_mcp_tools(&snapshot.tools); + Arc::new(HostedConnectorRuntimeFragment { + revision: snapshot.revision.into(), + tools: snapshot.tools, + connectors, + }) + }) + } + } else { + None + }; let has_mcp_servers = mcp_connection_manager.has_servers(); - let all_mcp_tools = mcp_connection_manager - .list_all_tools() - .or_cancel(cancellation_token) - .await?; + let all_mcp_tools = if let Some(hosted_connectors) = hosted_connectors.as_ref() { + mcp_connection_manager + .list_all_tools_with_hosted_connector_tools(&hosted_connectors.tools) + .or_cancel(cancellation_token) + .await? + } else { + mcp_connection_manager + .list_all_tools() + .or_cancel(cancellation_token) + .await? + }; let loaded_plugins = sess .services .plugins_manager @@ -1169,8 +1208,12 @@ pub(crate) async fn built_tools( .await; let apps_enabled = turn_context.apps_enabled(); - let accessible_connectors = - apps_enabled.then(|| connectors::accessible_connectors_from_mcp_tools(&all_mcp_tools)); + let accessible_connectors = apps_enabled.then(|| { + hosted_connectors.as_ref().map_or_else( + || connectors::accessible_connectors_from_mcp_tools(&all_mcp_tools), + |fragment| fragment.connectors.clone(), + ) + }); let accessible_connectors_with_enabled_state = accessible_connectors.as_ref().map(|connectors| { connectors::with_app_enabled_state(connectors.clone(), &turn_context.config) @@ -1196,8 +1239,44 @@ pub(crate) async fn built_tools( } else { None }; + let plugins_config = turn_context.config.plugins_config_input(); + let include_openai_curated = !(plugins_config.remote_plugin_enabled + && auth + .as_ref() + .is_some_and(codex_login::CodexAuth::uses_codex_backend)); + let plugin_catalog_revision = if tool_suggest_is_enabled { + sess.services + .plugins_manager + .revisioned_plugin_catalog_revision_for_config(&plugins_config, include_openai_curated) + } else { + None + }; + let mut reusable_fragment_build_failed = false; + let revisioned_plugins = if let Some(revision) = plugin_catalog_revision.as_ref() { + let revision_key = RevisionedPluginsRevision::from(revision); + if let Some(fragment) = base_catalog.revisioned_plugins_for(&revision_key) { + Some(fragment) + } else { + match sess + .services + .plugins_manager + .build_revisioned_plugin_catalog(revision) + { + Ok(marketplaces) => Some(Arc::new(RevisionedPluginCatalogFragment { + revision: revision.into(), + marketplaces, + })), + Err(err) => { + reusable_fragment_build_failed = true; + warn!("failed to build revisioned plugin catalog fragment: {err:#}"); + None + } + } + } + } else { + None + }; let endpoint_recommended_plugin_candidates = if tool_suggest_is_enabled { - let plugins_config = turn_context.config.plugins_config_input(); sess.services .plugins_manager .recommended_plugin_candidates_for_config(RecommendedPluginCandidatesInput { @@ -1211,6 +1290,16 @@ pub(crate) async fn built_tools( } else { None }; + let plugin_catalog = match ( + plugin_catalog_revision.as_ref(), + revisioned_plugins.as_deref(), + ) { + (Some(revision), Some(fragment)) => ToolSuggestPluginCatalog::Revisioned { + revision, + marketplaces: &fragment.marketplaces, + }, + _ => ToolSuggestPluginCatalog::RebuildAll, + }; let tool_suggest_candidates = if let Some(recommended_plugin_candidates) = endpoint_recommended_plugin_candidates { Some(ToolSuggestCandidates { @@ -1234,6 +1323,7 @@ pub(crate) async fn built_tools( auth.as_ref(), accessible_connectors.as_slice(), &loaded_plugin_app_connector_ids, + plugin_catalog, ) .await .map(|discoverable_tools| { @@ -1270,6 +1360,16 @@ pub(crate) async fn built_tools( ); let mcp_tools = has_mcp_servers.then_some(mcp_tool_exposure.direct_tools); let deferred_mcp_tools = mcp_tool_exposure.deferred_tools; + let next_catalog = RuntimeToolCatalogSnapshot { + hosted_connectors, + revisioned_plugins, + }; + let next_catalog = if reusable_fragment_build_failed { + Err(()) + } else { + Ok(next_catalog) + }; + let _ = runtime_catalog_manager.publish_if_current(&base_catalog, next_catalog); Ok(Arc::new(ToolRouter::from_turn_context( turn_context, ToolRouterParams { diff --git a/codex-rs/core/src/state/service.rs b/codex-rs/core/src/state/service.rs index 00380c6d64..0bb4f8060c 100644 --- a/codex-rs/core/src/state/service.rs +++ b/codex-rs/core/src/state/service.rs @@ -12,6 +12,7 @@ use crate::exec_policy::ExecPolicyManager; use crate::guardian::GuardianRejection; use crate::guardian::GuardianRejectionCircuitBreaker; use crate::mcp::McpManager; +use crate::runtime_tool_catalog::RuntimeToolCatalogManager; use crate::tools::code_mode::CodeModeService; use crate::tools::handlers::ToolSearchHandlerCache; use crate::tools::network_approval::NetworkApprovalService; @@ -42,6 +43,7 @@ use tokio_util::sync::CancellationToken; pub(crate) struct SessionServices { /// The latest manager; callers retain an owned handle while performing MCP I/O. pub(crate) mcp_connection_manager: Arc>, + pub(crate) runtime_tool_catalog_manager: RuntimeToolCatalogManager, pub(crate) mcp_startup_cancellation_token: Mutex, pub(crate) unified_exec_manager: UnifiedExecProcessManager, #[cfg_attr(not(unix), allow(dead_code))] diff --git a/codex-rs/core/tests/suite/search_tool.rs b/codex-rs/core/tests/suite/search_tool.rs index 284ee49b8f..257acf3bad 100644 --- a/codex-rs/core/tests/suite/search_tool.rs +++ b/codex-rs/core/tests/suite/search_tool.rs @@ -224,6 +224,66 @@ async fn always_defer_feature_hides_small_app_tool_sets() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn repeated_built_tools_preserves_visible_and_searchable_tool_identities() -> Result<()> { + skip_if_no_network!(Ok(())); + + let server = start_mock_server().await; + let apps_server = AppsTestServer::mount_searchable(&server).await?; + let search_event = |response_id: &str, call_id: &str| { + sse(vec![ + ev_response_created(response_id), + ev_tool_search_call( + call_id, + &json!({"query": "create calendar event", "limit": 2}), + ), + ev_completed(response_id), + ]) + }; + let completion_event = |response_id: &str, message_id: &str| { + sse(vec![ + ev_response_created(response_id), + ev_assistant_message(message_id, "done"), + ev_completed(response_id), + ]) + }; + let mock = mount_sse_sequence( + &server, + vec![ + search_event("resp-1", "search-1"), + completion_event("resp-2", "msg-1"), + search_event("resp-3", "search-2"), + completion_event("resp-4", "msg-2"), + ], + ) + .await; + let mut builder = configured_builder(apps_server.chatgpt_base_url.clone()); + let test = builder.build(&server).await?; + + for prompt in ["find the calendar tool", "find it again"] { + test.submit_turn_with_approval_and_permission_profile( + prompt, + AskForApproval::Never, + PermissionProfile::Disabled, + ) + .await?; + } + + let requests = mock.requests(); + assert_eq!( + requests[0].body_json()["tools"], + requests[2].body_json()["tools"] + ); + let first_search_results = tool_search_output_tools(&requests[1], "search-1"); + assert!(!first_search_results.is_empty()); + assert_eq!( + first_search_results, + tool_search_output_tools(&requests[3], "search-2") + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn app_only_tools_are_not_visible_or_runnable_by_direct_model_calls() -> Result<()> { skip_if_no_network!(Ok(()));