openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-28 18:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add revision-gated runtime tool catalog

Browse Source
This commit is contained in:
Matthew Zeng
2026-06-17 10:27:22 -07:00
parent 1315198853
commit c34e07d94a
22 changed files with 907 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/codex-mcp/src/connection_manager.rs
+119 -5
View File
@@ -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<ToolInfo>,
}
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<String>,
tool_plugin_provenance: Arc<ToolPluginProvenance>,
host_owned_codex_apps_enabled: bool,
hosted_connector_runtime_revision: Arc<AtomicU64>,
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<HostedConnectorRuntimeRevision> {
(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<AskForApproval>) {
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<ToolInfo> {
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<HostedConnectorToolsSnapshot> {
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<ToolInfo> {
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<ToolInfo> {
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
codex-rs/codex-mcp/src/connection_manager_tests.rs
+59
View File
@@ -115,6 +115,65 @@ fn model_tool_name_len(name: &ToolName) -> usize {
+ name.name.len()
}
fn manager_with_pending_hosted_tools(tools: Vec<ToolInfo>) -> McpConnectionManager {
let pending_client = futures::future::pending::<Result<ManagedClient, StartupOutcomeError>>()
.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()
codex-rs/codex-mcp/src/lib.rs
+2
View File
@@ -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;
codex-rs/core-plugins/src/discoverable.rs
+40 -7
View File
@@ -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<String>,
}
/// 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<Vec<ToolSuggestDiscoverablePlugin>> {
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<Vec<ToolSuggestDiscoverablePlugin>> {
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,
codex-rs/core-plugins/src/lib.rs
+3
View File
@@ -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<codex_config::McpSe
pub use app_mcp_routing::apps_route_available;
pub use discoverable::ToolSuggestDiscoverablePlugin;
pub use discoverable::ToolSuggestPluginCatalog;
pub use discoverable::ToolSuggestPluginDiscoveryInput;
pub use loader::PluginHookLoadOutcome;
pub use manager::ConfiguredMarketplace;
@@ -52,6 +54,7 @@ pub use manager::PluginsManager;
pub use manager::RecommendedPluginCandidatesInput;
pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeError as PluginMarketplaceUpgradeError;
pub use marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome as PluginMarketplaceUpgradeOutcome;
pub use plugin_catalog_revision::PluginCatalogRevision;
pub use provider::ExecutorPluginProvider;
pub use provider::ExecutorPluginProviderError;
pub use provider::ResolvedExecutorPlugin;
codex-rs/core-plugins/src/manager.rs
+121 -3
View File
@@ -32,12 +32,15 @@ use crate::marketplace::MarketplacePluginSource;
use crate::marketplace::ResolvedMarketplacePlugin;
use crate::marketplace::find_installable_marketplace_plugin;
use crate::marketplace::find_marketplace_plugin;
use crate::marketplace::home_dir;
use crate::marketplace::list_marketplaces;
use crate::marketplace::list_marketplaces_with_home;
use crate::marketplace::plugin_interface_with_marketplace_category;
use crate::marketplace_upgrade::ConfiguredMarketplaceUpgradeError;
use crate::marketplace_upgrade::ConfiguredMarketplaceUpgradeOutcome;
use crate::marketplace_upgrade::configured_git_marketplace_names;
use crate::marketplace_upgrade::upgrade_configured_git_marketplaces;
use crate::plugin_catalog_revision::PluginCatalogRevision;
use crate::remote::RecommendedPluginsMode;
use crate::remote::RemoteInstalledPlugin;
use crate::remote::RemotePluginCatalogError;
@@ -46,6 +49,7 @@ use crate::remote_legacy::RemotePluginFetchError;
use crate::remote_legacy::RemotePluginMutationError;
use crate::startup_sync::curated_plugins_api_marketplace_path;
use crate::startup_sync::curated_plugins_repo_path;
use crate::startup_sync::read_curated_plugins_catalog_revision;
use crate::startup_sync::read_curated_plugins_sha;
use crate::startup_sync::sync_openai_plugins_repo;
use crate::store::PluginInstallResult as StorePluginInstallResult;
@@ -1235,10 +1239,96 @@ impl PluginsManager {
return Ok(ConfiguredMarketplaceListOutcome::default());
}
let (installed_plugins, enabled_plugins) = self.configured_plugin_states(config);
let marketplace_roots =
self.marketplace_roots(config, additional_roots, include_openai_curated);
let marketplace_outcome = list_marketplaces(&marketplace_roots)?;
Ok(self.configured_marketplaces_from_outcome(config, marketplace_outcome))
}
pub fn revisioned_plugin_catalog_revision_for_config(
&self,
config: &PluginsConfigInput,
include_openai_curated: bool,
) -> Option<PluginCatalogRevision> {
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<MarketplaceListOutcome, MarketplaceError> {
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<ConfiguredMarketplaceListOutcome, MarketplaceError> {
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::<HashSet<_>>();
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::<HashSet<_>>();
aggregate.errors.extend(
outcome
.errors
.into_iter()
.filter(|error| error_paths.insert(error.path.clone())),
);
}
#[derive(Debug, thiserror::Error)]
pub enum PluginInstallError {
#[error("{0}")]
codex-rs/core-plugins/src/manager_tests.rs
+71
View File
@@ -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::<Vec<_>>()
};
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();
codex-rs/core-plugins/src/plugin_catalog_revision.rs
+34
View File
@@ -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<Self> {
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)
}
}
codex-rs/core-plugins/src/startup_sync.rs
+35
View File
@@ -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<String> {
read_sha_file(curated_plugins_sha_path(codex_home).as_path())
}
pub(crate) fn read_curated_plugins_catalog_revision(
repo_path: &Path,
) -> Option<PluginCatalogRevision> {
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<String, String> {
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,
codex-rs/core-plugins/src/startup_sync_tests.rs
+7
View File
@@ -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]
codex-rs/core/src/connectors.rs
+3
View File
@@ -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<Vec<DiscoverableTool>> {
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()
codex-rs/core/src/connectors_tests.rs
+2
View File
@@ -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");
codex-rs/core/src/lib.rs
+1
View File
@@ -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;
codex-rs/core/src/plugins/discoverable.rs
+3 -1
View File
@@ -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<Vec<DiscoverablePluginInfo>> {
let input = ToolSuggestPluginDiscoveryInput {
plugins: config.plugins_config_input(),
@@ -36,7 +38,7 @@ pub(crate) async fn list_tool_suggest_discoverable_plugins(
.collect::<HashSet<_>>(),
};
plugins_manager
.list_tool_suggest_discoverable_plugins(&input, auth)
.list_tool_suggest_discoverable_plugins_with_catalog(&input, auth, catalog)
.await
.map(|plugins| {
plugins
codex-rs/core/src/plugins/discoverable_tests.rs
+1
View File
@@ -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
}
codex-rs/core/src/runtime_tool_catalog.rs
+104
View File
@@ -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<HostedConnectorRuntimeRevision> 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<ToolInfo>,
pub(crate) connectors: Vec<AppInfo>,
}
pub(crate) struct RevisionedPluginCatalogFragment {
pub(crate) revision: RevisionedPluginsRevision,
pub(crate) marketplaces: MarketplaceListOutcome,
}
#[derive(Default)]
pub(crate) struct RuntimeToolCatalogSnapshot {
pub(crate) hosted_connectors: Option<Arc<HostedConnectorRuntimeFragment>>,
pub(crate) revisioned_plugins: Option<Arc<RevisionedPluginCatalogFragment>>,
}
impl RuntimeToolCatalogSnapshot {
pub(crate) fn hosted_connectors_for(
&self,
revision: HostedConnectorsRevision,
) -> Option<Arc<HostedConnectorRuntimeFragment>> {
self.hosted_connectors
.as_ref()
.filter(|fragment| fragment.revision == revision)
.map(Arc::clone)
}
pub(crate) fn revisioned_plugins_for(
&self,
revision: &RevisionedPluginsRevision,
) -> Option<Arc<RevisionedPluginCatalogFragment>> {
self.revisioned_plugins
.as_ref()
.filter(|fragment| fragment.revision == *revision)
.map(Arc::clone)
}
}
#[derive(Default)]
pub(crate) struct RuntimeToolCatalogManager {
snapshot: RwLock<Arc<RuntimeToolCatalogSnapshot>>,
}
impl RuntimeToolCatalogManager {
pub(crate) fn snapshot(&self) -> Arc<RuntimeToolCatalogSnapshot> {
self.snapshot
.read()
.unwrap_or_else(PoisonError::into_inner)
.clone()
}
pub(crate) fn publish_if_current<E>(
&self,
base: &Arc<RuntimeToolCatalogSnapshot>,
next: Result<RuntimeToolCatalogSnapshot, E>,
) -> Result<Arc<RuntimeToolCatalogSnapshot>, E> {
let next = Arc::new(next?);
let mut current = self
.snapshot
.write()
.unwrap_or_else(PoisonError::into_inner);
if Arc::ptr_eq(&current, base) {
*current = next;
}
Ok(Arc::clone(&current))
}
}
#[cfg(test)]
#[path = "runtime_tool_catalog_tests.rs"]
mod tests;
codex-rs/core/src/runtime_tool_catalog_tests.rs
+130
View File
@@ -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<HostedConnectorRuntimeFragment> {
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<RevisionedPluginCatalogFragment> {
Arc::new(RevisionedPluginCatalogFragment {
revision: plugin_revision(revision),
marketplaces: MarketplaceListOutcome::default(),
})
}
fn catalog(
hosted_connectors: Option<Arc<HostedConnectorRuntimeFragment>>,
revisioned_plugins: Option<Arc<RevisionedPluginCatalogFragment>>,
) -> 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::<RuntimeToolCatalogSnapshot, _>("plugin rebuild failed"),
);
assert!(matches!(result, Err("plugin rebuild failed")));
assert!(Arc::ptr_eq(&initial, &manager.snapshot()));
}
codex-rs/core/src/session/session.rs
+1
View File
@@ -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,
codex-rs/core/src/session/tests.rs
+2
View File
@@ -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,
codex-rs/core/src/session/turn.rs
+107 -7
View File
@@ -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<Arc<ToolRouter>> {
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 {
codex-rs/core/src/state/service.rs
+2
View File
@@ -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<ArcSwap<McpConnectionManager>>,
pub(crate) runtime_tool_catalog_manager: RuntimeToolCatalogManager,
pub(crate) mcp_startup_cancellation_token: Mutex<CancellationToken>,
pub(crate) unified_exec_manager: UnifiedExecProcessManager,
#[cfg_attr(not(unix), allow(dead_code))]
codex-rs/core/tests/suite/search_tool.rs
+60
View File
@@ -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(()));