diff --git a/codex-rs/app-server/src/effective_plugin_change.rs b/codex-rs/app-server/src/effective_plugin_change.rs index 8c07c55dd7..1c0f394f8a 100644 --- a/codex-rs/app-server/src/effective_plugin_change.rs +++ b/codex-rs/app-server/src/effective_plugin_change.rs @@ -1,3 +1,4 @@ +use codex_core::McpRefreshReason; use std::collections::BTreeSet; use std::sync::Arc; @@ -32,8 +33,14 @@ pub(crate) fn effective_plugins_changed_callback( thread_manager.skills_service().clear_cache(); let refresh_thread_manager = Arc::clone(&thread_manager); + let reason = change.metadata_change_id.map_or( + McpRefreshReason::EffectivePluginsChanged, + McpRefreshReason::InstalledPluginMetadata, + ); tokio::spawn(async move { - refresh_thread_manager.invalidate_mcp_runtimes().await; + refresh_thread_manager + .invalidate_mcp_runtimes_for(reason) + .await; refresh_thread_manager.refresh_hook_runtimes().await; }); diff --git a/codex-rs/app-server/src/request_processors/account_processor.rs b/codex-rs/app-server/src/request_processors/account_processor.rs index 38bdda4260..92fb152787 100644 --- a/codex-rs/app-server/src/request_processors/account_processor.rs +++ b/codex-rs/app-server/src/request_processors/account_processor.rs @@ -10,6 +10,7 @@ use crate::outgoing_message::AccountNotification; use chrono::DateTime; use codex_app_server_protocol::DesktopOnboardingEntrypoint; use codex_app_server_protocol::GetAccountRateLimitsParams; +use codex_core::McpRefreshReason; use codex_login::LoginOnboardingEntrypoint; use codex_login::login_with_bedrock_access_keys; use codex_mcp::ema_auth_scope; @@ -297,16 +298,21 @@ impl AccountRequestProcessor { Self::spawn_effective_plugins_changed_task( Arc::clone(thread_manager), config_manager.clone(), + McpRefreshReason::AccountPluginCacheChanged, ); let plugins_config = config.plugins_config_input(); let refresh_thread_manager = Arc::clone(thread_manager); let refresh_config_manager = config_manager.clone(); let on_effective_plugins_changed: Arc< dyn Fn(codex_core_plugins::EffectivePluginsChange) + Send + Sync, - > = Arc::new(move |_change| { + > = Arc::new(move |change| { Self::spawn_effective_plugins_changed_task( Arc::clone(&refresh_thread_manager), refresh_config_manager.clone(), + change.metadata_change_id.map_or( + McpRefreshReason::AccountPluginCacheChanged, + McpRefreshReason::InstalledPluginMetadata, + ), ); }); thread_manager @@ -338,13 +344,14 @@ impl AccountRequestProcessor { fn spawn_effective_plugins_changed_task( thread_manager: Arc, config_manager: ConfigManager, + reason: McpRefreshReason, ) { tokio::spawn(async move { thread_manager.plugins_manager().clear_cache(); thread_manager.skills_service().clear_cache(); crate::mcp_refresh::reload_mcp_config_best_effort(&thread_manager, &config_manager) .await; - thread_manager.invalidate_mcp_runtimes().await; + thread_manager.invalidate_mcp_runtimes_for(reason).await; }); } @@ -954,7 +961,9 @@ impl AccountRequestProcessor { } async fn send_login_success_notifications(&self, login_id: Option) { - self.thread_manager.invalidate_mcp_runtimes().await; + self.thread_manager + .invalidate_mcp_runtimes_for(McpRefreshReason::AccountLoginCompleted) + .await; self.send_account_login_notifications(AccountLoginCompletedNotification { login_id: login_id.map(|id| id.to_string()), success: true, @@ -1016,7 +1025,9 @@ impl AccountRequestProcessor { self.auth_manager.reload().await; let auth_changes = self.auth_manager.auth_change_state_receiver(); let owner_generation = auth_changes.borrow().owner_generation; - self.thread_manager.invalidate_mcp_runtimes().await; + self.thread_manager + .invalidate_mcp_runtimes_for(McpRefreshReason::AccountAuthReloaded) + .await; self.config_manager .update_managed_config_loader(self.auth_manager.clone(), &self.config); self.config_manager @@ -1041,7 +1052,9 @@ impl AccountRequestProcessor { let config = logout_configured_account(&self.config_manager, &self.auth_manager, &self.config) .await?; - self.thread_manager.invalidate_mcp_runtimes().await; + self.thread_manager + .invalidate_mcp_runtimes_for(McpRefreshReason::AccountLoggedOut) + .await; clear_account_configuration(&self.config_manager, &config).await?; self.model_catalog.invalidate().await; diff --git a/codex-rs/app-server/src/request_processors/account_processor/workspace_switching.rs b/codex-rs/app-server/src/request_processors/account_processor/workspace_switching.rs index c518cf371b..accb051298 100644 --- a/codex-rs/app-server/src/request_processors/account_processor/workspace_switching.rs +++ b/codex-rs/app-server/src/request_processors/account_processor/workspace_switching.rs @@ -8,6 +8,7 @@ use codex_app_server_protocol::AccountSessionsListParams; use codex_app_server_protocol::AccountSessionsResponse; use codex_app_server_protocol::AccountSessionsSwitchParams; use codex_app_server_protocol::AccountSessionsSwitchResponse; +use codex_core::McpRefreshReason; impl AccountRequestProcessor { pub(crate) async fn list_account_sessions( @@ -204,7 +205,9 @@ impl AccountRequestProcessor { "Workspace validation failed: {error}" ))); } - self.thread_manager.invalidate_mcp_runtimes().await; + self.thread_manager + .invalidate_mcp_runtimes_for(McpRefreshReason::AccountWorkspaceSwitched) + .await; self.model_catalog.invalidate().await; Self::maybe_refresh_plugin_caches_for_current_config( &self.config_manager, diff --git a/codex-rs/app-server/src/request_processors/marketplace_processor.rs b/codex-rs/app-server/src/request_processors/marketplace_processor.rs index 5d961528cf..5932678776 100644 --- a/codex-rs/app-server/src/request_processors/marketplace_processor.rs +++ b/codex-rs/app-server/src/request_processors/marketplace_processor.rs @@ -1,5 +1,6 @@ use super::*; use crate::plugin_config_reload; +use codex_core::McpRefreshReason; #[derive(Clone)] pub(crate) struct MarketplaceRequestProcessor { @@ -89,7 +90,9 @@ impl MarketplaceRequestProcessor { if !outcome.upgraded_roots.is_empty() { self.thread_manager.plugins_manager().clear_cache(); self.thread_manager.skills_service().clear_cache(); - self.thread_manager.invalidate_mcp_runtimes().await; + self.thread_manager + .invalidate_mcp_runtimes_for(McpRefreshReason::MarketplaceUpgraded) + .await; self.thread_manager.refresh_hook_runtimes().await; } diff --git a/codex-rs/app-server/src/request_processors/mcp_processor.rs b/codex-rs/app-server/src/request_processors/mcp_processor.rs index 2d6b065491..a7ccf1c331 100644 --- a/codex-rs/app-server/src/request_processors/mcp_processor.rs +++ b/codex-rs/app-server/src/request_processors/mcp_processor.rs @@ -3,6 +3,7 @@ use super::account_processor::EnterpriseLoginTarget; use super::thread_input::ensure_direct_input_allowed; use super::*; use codex_config::types::McpServerAuth; +use codex_core::McpRefreshReason; use codex_mcp::McpServerSource; use codex_mcp::ReadResourceRequestParams; use codex_mcp::ema_auth_scope; @@ -351,7 +352,9 @@ impl McpRequestProcessor { Err(err) => (false, Some(err.to_string())), }; if success && !enterprise { - thread_manager.invalidate_mcp_runtimes().await; + thread_manager + .invalidate_mcp_runtimes_for(McpRefreshReason::McpOAuthLoginCompleted) + .await; } let notification = ServerNotification::McpServerOauthLoginCompleted( diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 141b8cc5fb..ed74f924a7 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -9,6 +9,7 @@ use codex_app_server_protocol::PluginSharePrincipalRole; use codex_app_server_protocol::PluginShareTargetRole; use codex_config::types::McpServerConfig; use codex_config::types::PluginsConfigToml; +use codex_core::McpRefreshReason; use codex_core_plugins::OPENAI_CURATED_MARKETPLACE_NAME; use codex_core_plugins::PluginListBackgroundTaskOptions; use codex_core_plugins::PluginMarketplaceContext; @@ -533,9 +534,11 @@ impl PluginRequestProcessor { Arc::clone(&self.on_effective_plugins_changed) } - async fn on_effective_plugins_changed(&self) { + async fn on_effective_plugins_changed(&self, reason: McpRefreshReason) { self.clear_plugin_related_caches(); - self.thread_manager.invalidate_mcp_runtimes().await; + self.thread_manager + .invalidate_mcp_runtimes_for(reason) + .await; self.thread_manager.refresh_hook_runtimes().await; } @@ -596,7 +599,8 @@ impl PluginRequestProcessor { .refresh_non_curated_plugin_cache_for_context(&context) .await { - self.on_effective_plugins_changed().await; + self.on_effective_plugins_changed(McpRefreshReason::NonCuratedPluginCacheRefreshed) + .await; } let include_vertical = context.global_config.plugins_enabled && marketplace_kinds.contains(&PluginListMarketplaceKind::Vertical); @@ -875,6 +879,7 @@ impl PluginRequestProcessor { &context.global_config, auth.clone(), Some(self.effective_plugins_changed_callback()), + codex_core_plugins::InstalledPluginSyncSource::PluginInstalled, ); let (mut data, marketplace_load_errors) = self @@ -1584,7 +1589,9 @@ impl PluginRequestProcessor { self.clear_plugin_related_caches(); reload_user_config(&self.config_manager, &self.thread_manager).await; - self.thread_manager.invalidate_mcp_runtimes().await; + self.thread_manager + .invalidate_mcp_runtimes_for(McpRefreshReason::PluginInstalled) + .await; self.thread_manager.refresh_hook_runtimes().await; let plugin_mcp_servers = load_configured_plugin_mcp_servers( @@ -2037,7 +2044,9 @@ impl PluginRequestProcessor { Err(err) => (false, Some(err.to_string())), }; if success { - thread_manager.invalidate_mcp_runtimes().await; + thread_manager + .invalidate_mcp_runtimes_for(McpRefreshReason::PluginOAuthLoginCompleted) + .await; } let notification = ServerNotification::McpServerOauthLoginCompleted( @@ -2074,7 +2083,10 @@ impl PluginRequestProcessor { .await .map_err(Self::plugin_uninstall_error)?; match self.load_latest_config(/*fallback_cwd*/ None).await { - Ok(_) => self.on_effective_plugins_changed().await, + Ok(_) => { + self.on_effective_plugins_changed(McpRefreshReason::PluginUninstalled) + .await + } Err(err) => { warn!( "failed to reload config after plugin uninstall, clearing plugin-related caches only: {err:?}" @@ -2169,7 +2181,8 @@ impl PluginRequestProcessor { self.analytics_events_client .track_plugin_uninstalled(outcome.telemetry); if outcome.effective_plugins_changed { - self.on_effective_plugins_changed().await; + self.on_effective_plugins_changed(McpRefreshReason::RemotePluginUninstalled) + .await; } if let Some(err) = outcome.cache_removal_error { return Err(remote_plugin_catalog_error_to_jsonrpc( diff --git a/codex-rs/core-plugins/src/lib.rs b/codex-rs/core-plugins/src/lib.rs index 8b8f57f836..8384333809 100644 --- a/codex-rs/core-plugins/src/lib.rs +++ b/codex-rs/core-plugins/src/lib.rs @@ -86,6 +86,7 @@ pub use manager::ConfiguredMarketplace; pub use manager::ConfiguredMarketplaceListOutcome; pub use manager::ConfiguredMarketplacePlugin; pub use manager::EffectivePluginsChange; +pub use manager::InstalledPluginSyncSource; pub use manager::PluginDetail; pub use manager::PluginDetailsUnavailableReason; pub use manager::PluginInstallError; diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index a0845699fc..d76fe985c3 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -254,9 +254,19 @@ impl PluginsConfigInput { } } +/// Operation that requested a background sync of installed plugin bundles. +#[derive(Clone, Copy, Debug)] +pub enum InstalledPluginSyncSource { + PluginInstalled, + PluginList, + Startup, +} + /// Effective-plugin changes that downstream composition layers may act on. #[derive(Debug, Clone, Default, PartialEq, Eq)] pub struct EffectivePluginsChange { + /// Joins this notification to its installed-plugin cache publication in diagnostics. + pub metadata_change_id: Option, /// Remote bundles installed or updated by background installed-plugin sync. pub materialized_remote_plugins: Vec, } @@ -320,6 +330,10 @@ struct RemoteInstalledPluginsCache { display_assets: InstalledPluginDisplayAssets, } +struct InstalledPluginCacheUpdate { + metadata_change_id: Option, +} + enum RemoteInstalledPluginsCachePublication { Refresh(InstalledPluginDisplayAssets), Reconcile, @@ -1435,7 +1449,7 @@ impl PluginsManager { &plugins, visible_marketplaces, ); - let Some(changed) = self.write_remote_installed_plugins_cache_snapshot( + let Some(update) = self.write_remote_installed_plugins_cache_snapshot( generation, plugins, auth, @@ -1453,8 +1467,13 @@ impl PluginsManager { ) .unwrap_or(marketplaces)); }; - if changed && let Some(on_effective_plugins_changed) = on_effective_plugins_changed { - on_effective_plugins_changed(EffectivePluginsChange::default()); + if update.metadata_change_id.is_some() + && let Some(on_effective_plugins_changed) = on_effective_plugins_changed + { + on_effective_plugins_changed(EffectivePluginsChange { + metadata_change_id: update.metadata_change_id, + ..Default::default() + }); } Ok(self .build_remote_installed_plugin_marketplaces_from_cache_for_auth( @@ -1545,7 +1564,7 @@ impl PluginsManager { plugins: Vec, auth: Option<&CodexAuth>, publication: RemoteInstalledPluginsCachePublication, - ) -> Option { + ) -> Option { let auth_identity = RemoteInstalledPluginsAuthIdentity::from_auth(auth); if !self.remote_installed_plugins_auth_is_current(&auth_identity) { return None; @@ -1584,23 +1603,32 @@ impl PluginsManager { InstalledPluginDisplayAssets::Include } }; - if cache.plugins.as_ref() == Some(&plugins) { - drop(cache); - if needs_effective_plugins_refresh { - self.clear_loaded_plugins_cache(); - } - return Some(needs_effective_plugins_refresh); - } - let metadata_changed = cache.plugins.as_ref().is_none_or(|previous| { - !crate::remote_metadata::installed_plugin_metadata_eq(previous, &plugins) - }); - cache.plugins = Some(plugins); - drop(cache); + let previous = cache.plugins.as_deref(); + let payload_changed = previous != Some(plugins.as_slice()); + let metadata_changed = payload_changed + && previous.is_none_or(|old| { + !crate::remote_metadata::installed_plugin_metadata_eq(old, &plugins) + }); let changed = needs_effective_plugins_refresh || metadata_changed; - if changed { + let metadata_change_id = changed.then(uuid::Uuid::new_v4); + let initial_snapshot = previous.is_none(); + let changes = if metadata_changed && tracing::enabled!(tracing::Level::INFO) { + crate::remote_metadata::installed_plugin_metadata_changes( + previous.unwrap_or_default(), + &plugins, + ) + } else { + Default::default() + }; + if payload_changed { + cache.plugins = Some(plugins); + } + drop(cache); + if let Some(change_id) = metadata_change_id { + tracing::info!(%change_id, ?changes, initial_snapshot, reconciliation_retry = needs_effective_plugins_refresh, is_reconcile, "installed plugin runtime metadata invalidated"); self.clear_loaded_plugins_cache(); } - Some(changed) + Some(InstalledPluginCacheUpdate { metadata_change_id }) } #[cfg(test)] @@ -1616,6 +1644,8 @@ impl PluginsManager { RemoteInstalledPluginsCachePublication::Refresh(InstalledPluginDisplayAssets::Include), ) .expect("test cache generation should remain current") + .metadata_change_id + .is_some() } pub fn clear_remote_installed_plugins_cache(&self) -> bool { @@ -1727,6 +1757,7 @@ impl PluginsManager { config: &PluginsConfigInput, auth: Option, on_effective_plugins_changed: Option, + source: InstalledPluginSyncSource, ) { if !config.plugins_enabled { return; @@ -1748,18 +1779,21 @@ impl PluginsManager { .reconcile_remote_installed_plugins_after_acquiring_gate(&config, Some(&auth)) .await; match result { - Ok((outcome, effective_plugins_changed)) => { + Ok((outcome, update)) => { tracing::info!( materialized_remote_plugins = ?outcome.materialized_remote_plugins, changed_plugins = ?outcome.changed_plugins, failed_remote_plugin_ids = ?outcome.failed_remote_plugin_ids, failed_materialization_remote_plugin_ids = ?outcome.failed_materialization_remote_plugin_ids, + ?source, + metadata_change_id = ?update.metadata_change_id, "completed remote installed plugin bundle sync" ); - if (effective_plugins_changed || !outcome.changed_plugins.is_empty()) + if (update.metadata_change_id.is_some() || !outcome.changed_plugins.is_empty()) && let Some(on_effective_plugins_changed) = on_effective_plugins_changed { on_effective_plugins_changed(EffectivePluginsChange { + metadata_change_id: update.metadata_change_id, materialized_remote_plugins: outcome.materialized_remote_plugins, }); } @@ -1817,8 +1851,13 @@ impl PluginsManager { &self, config: &PluginsConfigInput, auth: Option<&CodexAuth>, - ) -> Result<(RemoteInstalledPluginBundleSyncOutcome, bool), RemoteInstalledPluginBundleSyncError> - { + ) -> Result< + ( + RemoteInstalledPluginBundleSyncOutcome, + InstalledPluginCacheUpdate, + ), + RemoteInstalledPluginBundleSyncError, + > { let generation = self .begin_remote_installed_plugins_reconcile(auth) .ok_or(RemoteInstalledPluginBundleSyncError::Superseded)?; @@ -1893,7 +1932,7 @@ impl PluginsManager { outcome .changed_plugins .sort_unstable_by(|left, right| left.plugin_id.cmp(&right.plugin_id)); - let Some(effective_plugins_changed) = self.write_remote_installed_plugins_cache_snapshot( + let Some(update) = self.write_remote_installed_plugins_cache_snapshot( generation, result.installed_plugins, auth, @@ -1902,10 +1941,10 @@ impl PluginsManager { return Err(RemoteInstalledPluginBundleSyncError::Superseded); }; reconciliation.committed = true; - if !effective_plugins_changed && !outcome.changed_plugins.is_empty() { + if update.metadata_change_id.is_none() && !outcome.changed_plugins.is_empty() { self.clear_loaded_plugins_cache(); } - Ok((outcome, effective_plugins_changed)) + Ok((outcome, update)) } fn maybe_start_remote_catalog_cache_refresh( @@ -1947,6 +1986,7 @@ impl PluginsManager { &context.global_config, auth, on_effective_plugins_changed, + InstalledPluginSyncSource::PluginList, ); } @@ -3041,6 +3081,7 @@ impl PluginsManager { &config_for_remote_sync, auth, on_effective_plugins_changed, + InstalledPluginSyncSource::Startup, ); }); } @@ -3467,7 +3508,7 @@ impl PluginsManager { .await; match plugins { Ok(plugins) => { - let Some(changed) = self.write_remote_installed_plugins_cache_snapshot( + let Some(update) = self.write_remote_installed_plugins_cache_snapshot( request.generation, plugins, request.auth.as_ref(), @@ -3477,7 +3518,7 @@ impl PluginsManager { ) else { continue; }; - let should_notify = changed + let should_notify = update.metadata_change_id.is_some() || !request.change.materialized_remote_plugins.is_empty() || matches!( request.notify, @@ -3487,7 +3528,9 @@ impl PluginsManager { && let Some(on_effective_plugins_changed) = request.on_effective_plugins_changed { - on_effective_plugins_changed(request.change); + let mut change = request.change; + change.metadata_change_id = update.metadata_change_id; + on_effective_plugins_changed(change); } } Err( diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index c4189fdebc..c0c7ce1bd5 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/codex-rs/core-plugins/src/manager_tests.rs @@ -1860,12 +1860,14 @@ async fn unresolved_display_assets_does_not_satisfy_display_cache_reads() { .prepare_remote_installed_plugins_cache_generation(auth.as_ref()) .unwrap(); assert_eq!( - manager.write_remote_installed_plugins_cache_snapshot( - generation, - plugins.clone(), - auth.as_ref(), - RemoteInstalledPluginsCachePublication::Refresh(InstalledPluginDisplayAssets::Skip), - ), + manager + .write_remote_installed_plugins_cache_snapshot( + generation, + plugins.clone(), + auth.as_ref(), + RemoteInstalledPluginsCachePublication::Refresh(InstalledPluginDisplayAssets::Skip), + ) + .map(|update| update.metadata_change_id.is_some()), Some(true) ); assert!( @@ -1883,12 +1885,16 @@ async fn unresolved_display_assets_does_not_satisfy_display_cache_reads() { .expect("runtime metadata is ready"); assert_eq!(runtime[0].plugins.len(), 1); assert_eq!( - manager.write_remote_installed_plugins_cache_snapshot( - generation, - plugins.clone(), - auth.as_ref(), - RemoteInstalledPluginsCachePublication::Refresh(InstalledPluginDisplayAssets::Include), - ), + manager + .write_remote_installed_plugins_cache_snapshot( + generation, + plugins.clone(), + auth.as_ref(), + RemoteInstalledPluginsCachePublication::Refresh( + InstalledPluginDisplayAssets::Include + ), + ) + .map(|update| update.metadata_change_id.is_some()), Some(false) ); assert_eq!( @@ -1898,12 +1904,14 @@ async fn unresolved_display_assets_does_not_satisfy_display_cache_reads() { Some(runtime) ); assert_eq!( - manager.write_remote_installed_plugins_cache_snapshot( - generation, - plugins, - auth.as_ref(), - RemoteInstalledPluginsCachePublication::Refresh(InstalledPluginDisplayAssets::Skip), - ), + manager + .write_remote_installed_plugins_cache_snapshot( + generation, + plugins, + auth.as_ref(), + RemoteInstalledPluginsCachePublication::Refresh(InstalledPluginDisplayAssets::Skip), + ) + .map(|update| update.metadata_change_id.is_some()), Some(false) ); assert!( @@ -7454,6 +7462,7 @@ fn remote_installed_plugins_cache_refresh_coalesces_materializations() { authenticated_account_id: Some("account-123".to_string()), }; let change = |name: &str| EffectivePluginsChange { + metadata_change_id: None, materialized_remote_plugins: vec![materialization(name)], }; let callback = |count: std::sync::Arc| { @@ -7496,6 +7505,7 @@ fn remote_installed_plugins_cache_refresh_coalesces_materializations() { assert_eq!( request.change, EffectivePluginsChange { + metadata_change_id: None, materialized_remote_plugins: vec![materialization("alpha"), materialization("beta"),], } ); @@ -7557,11 +7567,13 @@ remote_plugin = true &config, Some(auth.clone()), /*on_effective_plugins_changed*/ None, + InstalledPluginSyncSource::PluginInstalled, ); second_manager.maybe_start_remote_installed_plugin_bundle_sync( &config, Some(auth), /*on_effective_plugins_changed*/ None, + InstalledPluginSyncSource::PluginInstalled, ); let _guard = first_manager @@ -7853,6 +7865,7 @@ async fn failed_installed_plugin_pagination_preserves_inventory_and_releases_mut &config, Some(auth), /*on_effective_plugins_changed*/ None, + InstalledPluginSyncSource::PluginInstalled, ); // Starting the sync acquires the gate synchronously, so this waits for its release. let _guard = tokio::time::timeout( @@ -7905,21 +7918,27 @@ fn reconciliation_fences_refresh_publication_and_recovers_after_cancellation() { .expect("capture concurrent refresh generation"); assert_eq!( - manager.write_remote_installed_plugins_cache_snapshot( - stale_refresh_generation, - Vec::new(), - /*auth*/ None, - RemoteInstalledPluginsCachePublication::Refresh(InstalledPluginDisplayAssets::Include), - ), + manager + .write_remote_installed_plugins_cache_snapshot( + stale_refresh_generation, + Vec::new(), + /*auth*/ None, + RemoteInstalledPluginsCachePublication::Refresh( + InstalledPluginDisplayAssets::Include + ), + ) + .map(|update| update.metadata_change_id.is_some()), None ); assert_eq!( - manager.write_remote_installed_plugins_cache_snapshot( - reconcile_generation, - vec![remote_installed_linear_plugin()], - /*auth*/ None, - RemoteInstalledPluginsCachePublication::Reconcile, - ), + manager + .write_remote_installed_plugins_cache_snapshot( + reconcile_generation, + vec![remote_installed_linear_plugin()], + /*auth*/ None, + RemoteInstalledPluginsCachePublication::Reconcile, + ) + .map(|update| update.metadata_change_id.is_some()), Some(true) ); let cancelled_generation = manager @@ -7937,24 +7956,30 @@ fn reconciliation_fences_refresh_publication_and_recovers_after_cancellation() { .remote_installed_plugins_cache_generation_if_current(/*auth*/ None) .expect("capture refresh generation after cancellation"); assert_eq!( - manager.write_remote_installed_plugins_cache_snapshot( - refresh_generation, - vec![remote_installed_plugin("beta")], - /*auth*/ None, - RemoteInstalledPluginsCachePublication::Refresh(InstalledPluginDisplayAssets::Include), - ), + manager + .write_remote_installed_plugins_cache_snapshot( + refresh_generation, + vec![remote_installed_plugin("beta")], + /*auth*/ None, + RemoteInstalledPluginsCachePublication::Refresh( + InstalledPluginDisplayAssets::Include + ), + ) + .map(|update| update.metadata_change_id.is_some()), Some(true) ); let retry_generation = manager .begin_remote_installed_plugins_reconcile(/*auth*/ None) .expect("begin retry after cancellation"); assert_eq!( - manager.write_remote_installed_plugins_cache_snapshot( - retry_generation, - vec![remote_installed_plugin("beta")], - /*auth*/ None, - RemoteInstalledPluginsCachePublication::Reconcile, - ), + manager + .write_remote_installed_plugins_cache_snapshot( + retry_generation, + vec![remote_installed_plugin("beta")], + /*auth*/ None, + RemoteInstalledPluginsCachePublication::Reconcile, + ) + .map(|update| update.metadata_change_id.is_some()), Some(true) ); } diff --git a/codex-rs/core-plugins/src/remote_metadata.rs b/codex-rs/core-plugins/src/remote_metadata.rs index a4db4055c1..0ab68a14c4 100644 --- a/codex-rs/core-plugins/src/remote_metadata.rs +++ b/codex-rs/core-plugins/src/remote_metadata.rs @@ -5,29 +5,83 @@ use crate::remote::RemoteInstalledPlugin; use crate::remote::RemoteMarketplace; +use std::collections::BTreeMap; +use std::collections::HashMap; + +/// Use the same comparisons for invalidation and its diagnostics. Expose field names only. +fn changed_fields<'a>( + previous: &'a RemoteInstalledPlugin, + current: &'a RemoteInstalledPlugin, +) -> impl Iterator + 'a { + macro_rules! compare { + ($($field:ident),+ $(,)?) => { + [$( (stringify!($field), previous.$field != current.$field) ),+] + }; + } + compare!( + marketplace_name, + id, + name, + canonical_app_id, + version, + installed_at, + enabled, + install_policy, + install_policy_source, + must_show_installation_interstitial, + auth_policy, + availability, + disabled_reason, + eligible_plan_types, + ) + .into_iter() + .filter_map(|(field, changed)| changed.then_some(field)) +} pub(crate) fn installed_plugin_metadata_eq( previous: &[RemoteInstalledPlugin], current: &[RemoteInstalledPlugin], ) -> bool { previous.len() == current.len() - && previous.iter().zip(current).all(|(previous, current)| { - previous.marketplace_name == current.marketplace_name - && previous.id == current.id - && previous.name == current.name - && previous.canonical_app_id == current.canonical_app_id - && previous.version == current.version - && previous.installed_at == current.installed_at - && previous.enabled == current.enabled - && previous.install_policy == current.install_policy - && previous.install_policy_source == current.install_policy_source - && previous.must_show_installation_interstitial - == current.must_show_installation_interstitial - && previous.auth_policy == current.auth_policy - && previous.availability == current.availability - && previous.disabled_reason == current.disabled_reason - && previous.eligible_plan_types == current.eligible_plan_types - }) + && previous + .iter() + .zip(current) + .all(|(previous, current)| changed_fields(previous, current).next().is_none()) +} + +/// Called only for a changed snapshot. Match by remote ID so an insertion does +/// not misattribute every subsequent plugin's fields to its neighbor. +pub(crate) fn installed_plugin_metadata_changes( + previous: &[RemoteInstalledPlugin], + current: &[RemoteInstalledPlugin], +) -> BTreeMap> { + let mut old: HashMap<_, _> = previous + .iter() + .enumerate() + .map(|(index, p)| (p.id.as_str(), (index, p))) + .collect(); + // A removed plugin shifts subsequent indices without reordering the surviving list. + let same_members = + previous.len() == current.len() && current.iter().all(|p| old.contains_key(p.id.as_str())); + let mut changes = BTreeMap::new(); + for (index, plugin) in current.iter().enumerate() { + let fields = if let Some((old_index, prior)) = old.remove(plugin.id.as_str()) { + let mut fields: Vec<_> = changed_fields(prior, plugin).collect(); + if old_index != index && same_members { + fields.push("position"); + } + fields + } else { + vec!["added"] + }; + if !fields.is_empty() { + changes.insert(plugin.id.clone(), fields); + } + } + for id in old.into_keys() { + changes.insert(id.to_owned(), vec!["removed"]); + } + changes } /// Compare installed catalog identity, versions, enablement, policy and availability. diff --git a/codex-rs/core-plugins/src/remote_metadata_tests.rs b/codex-rs/core-plugins/src/remote_metadata_tests.rs index 3e61f3a2ea..6397addc1d 100644 --- a/codex-rs/core-plugins/src/remote_metadata_tests.rs +++ b/codex-rs/core-plugins/src/remote_metadata_tests.rs @@ -121,6 +121,13 @@ fn installed_ordering_remains_significant() { let mut current = previous.clone(); current.reverse(); assert!(!installed_plugin_metadata_eq(&previous, ¤t)); + assert_eq!( + installed_plugin_metadata_changes(&previous, ¤t), + BTreeMap::from([ + (previous[0].id.clone(), vec!["position"]), + (previous[1].id.clone(), vec!["position"]), + ]) + ); } #[test] @@ -179,3 +186,33 @@ fn catalog_display_reordering_does_not_invalidate() { &catalog(¤t) )); } + +#[test] +fn diagnostics_attribute_fields_by_plugin_id_across_membership_changes() { + let previous = [ + plugin("old"), + RemoteInstalledPlugin { + id: "removed".into(), + ..plugin("old") + }, + ]; + let mut updated = previous[0].clone(); + updated.enabled = !updated.enabled; + updated.version = Some("2".into()); + let id = updated.id.clone(); + let current = [ + RemoteInstalledPlugin { + id: "added".into(), + ..plugin("new") + }, + updated, + ]; + assert_eq!( + installed_plugin_metadata_changes(&previous, ¤t), + std::collections::BTreeMap::from([ + ("added".into(), vec!["added"]), + ("removed".into(), vec!["removed"]), + (id, vec!["version", "enabled"]), + ]) + ); +} diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index ec127194f3..d684b39426 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -188,6 +188,7 @@ pub use thread_manager::AgentTreeShutdownFailureReason; pub use thread_manager::AgentTreeShutdownReport; pub use thread_manager::ForkSnapshot; pub use thread_manager::InternalSessionParent; +pub use thread_manager::McpRefreshReason; pub use thread_manager::NewThread; pub use thread_manager::PromptCacheKey; pub use thread_manager::StartThreadOptions; diff --git a/codex-rs/core/src/session/config_refresh.rs b/codex-rs/core/src/session/config_refresh.rs index 016eae3978..7464df91e6 100644 --- a/codex-rs/core/src/session/config_refresh.rs +++ b/codex-rs/core/src/session/config_refresh.rs @@ -1,5 +1,6 @@ //! Publishes resolved MCP configuration and host timing rollouts for future turns. +use crate::McpRefreshReason; use std::sync::Arc; use crate::ConfigRefreshOutcome; @@ -25,7 +26,7 @@ impl Session { config.disable_mcp_enterprise_auth(); state.session_configuration.original_config_do_not_use = Arc::new(config); self.services.mcp_runtime.invalidate_resource_caches(); - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::EnterpriseAuthDisabled); } self.services.mcp_runtime.quiesce_ema().await; self.schedule_mcp_prewarm(); @@ -192,7 +193,11 @@ impl Session { if matches!(scope, RuntimeConfigRefresh::Mcp) { self.services.mcp_runtime.invalidate_resource_caches(); } - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(match scope { + RuntimeConfigRefresh::User => McpRefreshReason::UserConfigReloaded, + RuntimeConfigRefresh::Mcp => McpRefreshReason::McpConfigReloaded, + RuntimeConfigRefresh::UserFiles => McpRefreshReason::UserConfigFilesReloaded, + }); let new_config = notify_contributors .then(|| self.build_effective_session_config(&state.session_configuration)); (previous_config, new_config) diff --git a/codex-rs/core/src/session/environment.rs b/codex-rs/core/src/session/environment.rs index 3820b8b7d8..e4469afbc9 100644 --- a/codex-rs/core/src/session/environment.rs +++ b/codex-rs/core/src/session/environment.rs @@ -1,3 +1,4 @@ +use crate::McpRefreshReason; use std::collections::HashSet; use std::sync::Arc; @@ -369,7 +370,7 @@ impl Session { state.session_configuration.environments = future; if update_current { // Invalidate MCP before installed configuration can wake a waiting turn. - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::CurrentEnvironmentsChanged); self.services.turn_environments.update_selections(¤t); } Ok(()) diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index ec9ac801bf..1a4424c89e 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -1,4 +1,5 @@ use super::Submission; +use crate::McpRefreshReason; use crate::WithTurnExtensionData; use crate::realtime_conversation::handle_audio as handle_realtime_conversation_audio; use crate::realtime_conversation::handle_close as handle_realtime_conversation_close; @@ -260,7 +261,7 @@ pub async fn dynamic_tool_response(sess: &Arc, id: String, response: Dy pub fn refresh_mcp_servers(sess: &Session) { sess.services.mcp_runtime.reconnect_on_next_refresh(); - sess.request_mcp_runtime_refresh(); + sess.request_mcp_runtime_refresh(McpRefreshReason::ExplicitServerReconnect); } pub async fn reload_user_config(sess: &Arc) { diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index 82aa3b8802..4865b61d38 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -1,5 +1,5 @@ -use super::mcp_refresh::McpRefreshInvalidationGuard; use super::*; +use crate::McpRefreshReason; use crate::environment_selection::TurnEnvironmentSnapshot; use crate::environment_selection::combine_selected_capability_roots; use crate::mcp_tool_call::McpToolApprovalContext; @@ -119,8 +119,11 @@ impl Session { /*skill_turn*/ None, ) .await; - if inventory_changed || discovery_recovered { - self.mark_mcp_runtime_dirty(); + if inventory_changed { + self.mark_mcp_runtime_dirty(McpRefreshReason::ExecutorCapabilityInventoryChanged); + } + if discovery_recovered { + self.mark_mcp_runtime_dirty(McpRefreshReason::ExecutorCapabilityDiscoveryRecovered); } let mcp_projection = self.services @@ -199,7 +202,7 @@ impl Session { &environment_selections, &environments.ready_environment_handles(), ) { - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::EnvironmentBindingsChanged); } let auth = self.services.auth_manager.auth_cached(); if !self @@ -207,15 +210,11 @@ impl Session { .mcp_runtime .current_auth_matches(auth.as_ref()) { - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::RuntimeAuthChanged); } - if !self.mcp_refresh.claim() { + let Some(mut refresh_invalidation) = self.mcp_refresh.claim(self.thread_id) else { return; - } - let mut refresh_invalidation = McpRefreshInvalidationGuard { - refresh: &self.mcp_refresh, - published: false, }; let auth = self.services.auth_manager.auth().await; // Capture after claiming the refresh so a new invalidation cannot be consumed @@ -282,7 +281,7 @@ impl Session { self: &Arc, ) -> anyhow::Result { // Reconcile unchanged config so failed or closed clients can be replaced. - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::CodexAppsToolRefresh); self.refresh_mcp_if_dirty().await; let _refresh = self .mcp_refresh @@ -302,11 +301,9 @@ impl Session { .await .map_err(|_| anyhow::anyhow!("MCP runtime refresh semaphore closed"))?; // This is forced work even when there was no pending invalidation. - self.mcp_refresh.claim(); - let mut refresh_invalidation = McpRefreshInvalidationGuard { - refresh: &self.mcp_refresh, - published: false, - }; + let mut refresh_invalidation = self + .mcp_refresh + .claim_forced(self.thread_id, McpRefreshReason::HardRefreshCodexAppsTools); let auth = self.services.auth_manager.auth().await; let environments = self.services.turn_environments.snapshot().await; let environment_selections = environments.all_selections(); @@ -380,8 +377,8 @@ impl Session { refreshed } - pub(super) fn mark_mcp_runtime_dirty(&self) { - self.mcp_refresh.invalidate(); + pub(super) fn mark_mcp_runtime_dirty(&self, reason: McpRefreshReason) { + self.mcp_refresh.invalidate(reason); } #[tracing::instrument( @@ -404,7 +401,7 @@ impl Session { .current_ready_selected_capability_roots() != ready_selected_capability_roots { - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::SelectedCapabilityRootsChanged); } let recovered_oauth_servers = @@ -423,7 +420,7 @@ impl Session { .has_authentication_failed_servers(&recovered_oauth_servers) .await { - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::OAuthCredentialsRecovered); } self.refresh_mcp_if_dirty().await; // Prewarm and other context-only captures must not start cloud discovery. @@ -450,7 +447,7 @@ impl Session { .await { // Preserve MCP updates if a later refresh is cancelled. - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::ExtensionCatalogChanged); } } self.refresh_mcp_if_dirty().await; @@ -861,11 +858,9 @@ impl Session { state.session_configuration.original_config_do_not_use = Arc::new(config); } // Claim after config mutation, before capturing the state this forced refresh handles. - self.mcp_refresh.claim(); - let mut refresh_invalidation = McpRefreshInvalidationGuard { - refresh: &self.mcp_refresh, - published: false, - }; + let mut refresh_invalidation = self + .mcp_refresh + .claim_forced(self.thread_id, McpRefreshReason::SkillMcpDependenciesAdded); let auth = self.services.auth_manager.auth().await; let ready_selected_capability_roots = self .services diff --git a/codex-rs/core/src/session/mcp_prewarm.rs b/codex-rs/core/src/session/mcp_prewarm.rs index 435699122a..49e828c863 100644 --- a/codex-rs/core/src/session/mcp_prewarm.rs +++ b/codex-rs/core/src/session/mcp_prewarm.rs @@ -4,17 +4,18 @@ //! newest thread state; exact model steps remain the correctness path. use super::*; +use crate::McpRefreshReason; impl Session { - pub(crate) fn request_mcp_runtime_refresh(&self) { + pub(crate) fn request_mcp_runtime_refresh(&self, reason: McpRefreshReason) { // Plugin changes can reuse connections but still change their skill resources. self.services.mcp_runtime.invalidate_resource_caches(); - self.request_mcp_runtime_reprojection(); + self.request_mcp_runtime_reprojection(reason); } /// Reproject contributor state without invalidating cached MCP resources. - pub(crate) fn request_mcp_runtime_reprojection(&self) { - self.mark_mcp_runtime_dirty(); + pub(crate) fn request_mcp_runtime_reprojection(&self, reason: McpRefreshReason) { + self.mark_mcp_runtime_dirty(reason); self.schedule_mcp_prewarm(); } @@ -47,7 +48,7 @@ impl Session { break; }; if auth_changed { - session.mark_mcp_runtime_dirty(); + session.mark_mcp_runtime_dirty(McpRefreshReason::AuthChangeNotification); } tokio::select! { biased; diff --git a/codex-rs/core/src/session/mcp_refresh.rs b/codex-rs/core/src/session/mcp_refresh.rs index f70a11d157..6e33c577be 100644 --- a/codex-rs/core/src/session/mcp_refresh.rs +++ b/codex-rs/core/src/session/mcp_refresh.rs @@ -1,34 +1,78 @@ -use std::sync::atomic::AtomicBool; -use std::sync::atomic::Ordering; +use crate::McpRefreshReason; +use codex_protocol::ThreadId; +use std::collections::BTreeSet; +use std::sync::Mutex; use tokio::sync::AcquireError; use tokio::sync::Semaphore; use tokio::sync::SemaphorePermit; /// Owns MCP invalidation and the single gate used to publish runtime updates. pub(super) struct McpRefresh { - pending: AtomicBool, + pending: Mutex>, gate: Semaphore, } impl McpRefresh { pub(super) fn new() -> Self { Self { - pending: AtomicBool::new(false), + pending: Mutex::default(), gate: Semaphore::new(/*permits*/ 1), } } - pub(super) fn invalidate(&self) { - self.pending.store(true, Ordering::Release); + pub(super) fn invalidate(&self, reason: McpRefreshReason) { + self.pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(reason); } #[cfg(test)] pub(super) fn is_pending(&self) -> bool { - self.pending.load(Ordering::Acquire) + !self + .pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_empty() } - pub(super) fn claim(&self) -> bool { - self.pending.swap(false, Ordering::AcqRel) + /// Take only the causes known at the start of this refresh. Later invalidations + /// belong to the next publication, even if they have the same cause. + pub(super) fn claim(&self, thread_id: ThreadId) -> Option> { + let reasons = self.take_pending(); + (!reasons.is_empty()).then(|| self.start(thread_id, reasons)) + } + + pub(super) fn claim_forced( + &self, + thread_id: ThreadId, + reason: McpRefreshReason, + ) -> McpRefreshInvalidationGuard<'_> { + let mut reasons = self.take_pending(); + reasons.insert(reason); + self.start(thread_id, reasons) + } + + pub(super) fn take_pending(&self) -> BTreeSet { + std::mem::take( + &mut *self + .pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ) + } + + fn start( + &self, + thread_id: ThreadId, + reasons: BTreeSet, + ) -> McpRefreshInvalidationGuard<'_> { + tracing::info!(%thread_id, ?reasons, "refreshing MCP runtime"); + McpRefreshInvalidationGuard { + refresh: self, + reasons, + published: false, + } } #[tracing::instrument(name = "mcp.runtime.refresh_wait", skip_all)] @@ -43,14 +87,21 @@ impl McpRefresh { /// Restores a claimed refresh when its task is cancelled before publication. pub(super) struct McpRefreshInvalidationGuard<'a> { - pub(super) refresh: &'a McpRefresh, + refresh: &'a McpRefresh, + reasons: BTreeSet, pub(super) published: bool, } impl Drop for McpRefreshInvalidationGuard<'_> { fn drop(&mut self) { if !self.published { - self.refresh.invalidate(); + let mut pending = self + .refresh + .pending + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + pending.append(&mut self.reasons); + pending.insert(McpRefreshReason::PreviousRefreshUnpublished); } } } diff --git a/codex-rs/core/src/session/mcp_refresh_tests.rs b/codex-rs/core/src/session/mcp_refresh_tests.rs index f6a9d5dd23..cb2d402b96 100644 --- a/codex-rs/core/src/session/mcp_refresh_tests.rs +++ b/codex-rs/core/src/session/mcp_refresh_tests.rs @@ -1,27 +1,36 @@ -//! Checks that successful refreshes preserve newly pending work. +//! Verifies that refreshes consume their own causes and preserve concurrent work. use super::*; use pretty_assertions::assert_eq; #[test] -fn successful_manual_claim_consumes_only_preexisting_work() { - for initially_pending in [false, true] { - for invalidated_during_refresh in [false, true] { - let refresh = McpRefresh::new(); - if initially_pending { - refresh.invalidate(); - } - assert_eq!(refresh.claim(), initially_pending); - let mut guard = McpRefreshInvalidationGuard { - refresh: &refresh, - published: false, - }; - if invalidated_during_refresh { - refresh.invalidate(); - } - guard.published = true; - drop(guard); - assert_eq!(refresh.is_pending(), invalidated_during_refresh); - } - } +fn refresh_keeps_concurrent_invalidations_and_restores_unpublished_causes() { + let thread_id = ThreadId::new(); + let refresh = McpRefresh::new(); + refresh.invalidate(McpRefreshReason::McpConfigReloaded); + let claimed = refresh.claim(thread_id).expect("pending"); + refresh.invalidate(McpRefreshReason::RuntimeAuthChanged); + drop(claimed); + + let mut retry = refresh.claim(thread_id).expect("retry"); + assert_eq!( + retry.reasons, + BTreeSet::from([ + McpRefreshReason::RuntimeAuthChanged, + McpRefreshReason::McpConfigReloaded, + McpRefreshReason::PreviousRefreshUnpublished + ]) + ); + refresh.invalidate(McpRefreshReason::RuntimeAuthChanged); + retry.published = true; + drop(retry); + + let mut next = refresh.claim(thread_id).expect("new invalidation"); + assert_eq!( + next.reasons, + BTreeSet::from([McpRefreshReason::RuntimeAuthChanged]) + ); + next.published = true; + drop(next); + assert!(refresh.claim(thread_id).is_none()); } diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 9d68f4a359..3f32564982 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1,5 +1,6 @@ pub(crate) mod startup; +use crate::McpRefreshReason; use crate::PromptCacheKey; use crate::WithTurnExtensionData; use crate::runtime_auth::RuntimeAuthSlot; @@ -2207,7 +2208,7 @@ impl Session { != updated.step_settings.service_tier; let mcp_inputs_changed = self.mcp_inputs_differ(&state.session_configuration, &updated); if mcp_inputs_changed { - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::SessionMcpInputsChanged); } // Save new environment defaults for future turns. The running turn keeps its own. windows_sandbox_changed = @@ -4378,8 +4379,11 @@ impl Session { ) .await }; - if inventory_changed || discovery_recovered { - self.mark_mcp_runtime_dirty(); + if inventory_changed { + self.mark_mcp_runtime_dirty(McpRefreshReason::ExecutorCapabilityInventoryChanged); + } + if discovery_recovered { + self.mark_mcp_runtime_dirty(McpRefreshReason::ExecutorCapabilityDiscoveryRecovered); } let extension_data = codex_extension_api::ExtensionData::new(turn_context.sub_id.clone()); diff --git a/codex-rs/core/src/session/plugin_selection.rs b/codex-rs/core/src/session/plugin_selection.rs index 0a9580db0c..99e5e3cb64 100644 --- a/codex-rs/core/src/session/plugin_selection.rs +++ b/codex-rs/core/src/session/plugin_selection.rs @@ -1,5 +1,6 @@ //! Activates an admitted task's plugin selection without changing pending settings. +use crate::McpRefreshReason; use std::sync::Arc; use super::Session; @@ -11,7 +12,7 @@ impl Session { let mut state = self.state.lock().await; if state.active_disabled_plugin_ids != turn_context.disabled_plugin_ids { state.active_disabled_plugin_ids = turn_context.disabled_plugin_ids.clone(); - self.mark_mcp_runtime_dirty(); + self.mark_mcp_runtime_dirty(McpRefreshReason::DisabledPluginSelectionChanged); } Arc::clone(&state.session_configuration.original_config_do_not_use) }; diff --git a/codex-rs/core/src/session/step_activation_tests.rs b/codex-rs/core/src/session/step_activation_tests.rs index 4ad17f2300..1923cd29b5 100644 --- a/codex-rs/core/src/session/step_activation_tests.rs +++ b/codex-rs/core/src/session/step_activation_tests.rs @@ -1,4 +1,5 @@ use super::*; +use crate::McpRefreshReason; use crate::agents_md_manager::AgentsMdManager; use crate::agents_md_manager::SessionInstructions; use crate::session::Submission; @@ -612,7 +613,7 @@ async fn submitted_sparse_updates_preserve_captured_steps_and_ordering() { .acquire() .await .expect("MCP refresh lock"); - session.mark_mcp_runtime_dirty(); + session.mark_mcp_runtime_dirty(McpRefreshReason::ExplicitInvalidation); let capture_cancel = CancellationToken::new(); let mut during = Box::pin(tokio::task::unconstrained( session.capture_step_context(Arc::clone(&turn), &capture_cancel), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 7f96937e30..c10635e708 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -1,3 +1,4 @@ +use crate::McpRefreshReason; use crate::agent::LocalAgentControl; use codex_protocol::ContextWindowId; #[path = "compaction_replay_tests.rs"] @@ -2329,7 +2330,7 @@ async fn refresh_config_rejects_stale_owner_without_effects(scope: RuntimeConfig crate::ConfigRefreshOutcome::Published ); let current_owner = session.get_config().await; - assert!(session.mcp_refresh.claim()); + assert!(!session.mcp_refresh.take_pending().is_empty()); assert_eq!( session @@ -7073,7 +7074,7 @@ pub(crate) async fn make_session_and_context() -> (Session, TurnContext) { "turn_id".to_string(), skills_snapshot, ); - session.mark_mcp_runtime_dirty(); + session.mark_mcp_runtime_dirty(McpRefreshReason::ExplicitInvalidation); crate::guardian::test_host::install(&session, &turn_context.config); (session, turn_context) } @@ -9417,7 +9418,7 @@ where "turn_id".to_string(), skills_snapshot, )); - session.mark_mcp_runtime_dirty(); + session.mark_mcp_runtime_dirty(McpRefreshReason::ExplicitInvalidation); crate::guardian::test_host::install(&session, &turn_context.config); (session, turn_context, rx_event) } @@ -9604,7 +9605,7 @@ async fn refresh_mcp_servers_uses_latest_state_for_existing_turns() { .expect("enable secret auth storage"); state.session_configuration.original_config_do_not_use = Arc::new(config); } - session.mark_mcp_runtime_dirty(); + session.mark_mcp_runtime_dirty(McpRefreshReason::ExplicitInvalidation); let next_turn = session.new_default_turn().await; let new_step = session @@ -9774,7 +9775,7 @@ async fn mcp_elicitation_reviewer_uses_active_reviewer_and_latest_runtime_policy .spawn_task(Arc::clone(&old_turn), Vec::new(), task) .await; - session.mark_mcp_runtime_dirty(); + session.mark_mcp_runtime_dirty(McpRefreshReason::ExplicitInvalidation); session.refresh_mcp_if_dirty().await; if live_update { assert_eq!( @@ -9990,7 +9991,7 @@ async fn mcp_elicitation_reviewer_is_reused_across_runtime_refreshes() { let session = Arc::new(session); let previous = session.mcp_elicitation_reviewer(); - session.mark_mcp_runtime_dirty(); + session.mark_mcp_runtime_dirty(McpRefreshReason::ExplicitInvalidation); session.refresh_mcp_if_dirty().await; assert!(Arc::ptr_eq(&previous, &session.mcp_elicitation_reviewer())); @@ -10336,7 +10337,7 @@ async fn step_context_keeps_its_mcp_runtime_for_tools() -> anyhow::Result<()> { ); assert_ne!(step_context.mcp.config().plugins, current_config.plugins); - session.mark_mcp_runtime_dirty(); + session.mark_mcp_runtime_dirty(McpRefreshReason::ExplicitInvalidation); session.refresh_mcp_if_dirty().await; let current = session .services diff --git a/codex-rs/core/src/session/turn_input_tests.rs b/codex-rs/core/src/session/turn_input_tests.rs index 78cd5d7f01..219a4229a4 100644 --- a/codex-rs/core/src/session/turn_input_tests.rs +++ b/codex-rs/core/src/session/turn_input_tests.rs @@ -616,7 +616,7 @@ async fn automatic_admission_rechecks_plan_mode_without_committing_sparse_settin .acquire() .await .expect("acquire MCP refresh gate"); - session.mcp_refresh.claim(); + session.mcp_refresh.take_pending(); assert!(!session.mcp_refresh.is_pending()); for (submission_id, prepared) in [ diff --git a/codex-rs/core/src/session/turn_start_mcp_tests.rs b/codex-rs/core/src/session/turn_start_mcp_tests.rs index 46d6ec4144..65eae0ba8c 100644 --- a/codex-rs/core/src/session/turn_start_mcp_tests.rs +++ b/codex-rs/core/src/session/turn_start_mcp_tests.rs @@ -1,6 +1,7 @@ //! Turn-start MCP reconciliation, discovery cancellation, and prewarm refresh boundaries. use super::*; +use crate::McpRefreshReason; use codex_mcp::McpResourceClient; use codex_mcp::McpResourceClientCacheKey; use pretty_assertions::assert_eq; @@ -73,10 +74,10 @@ async fn turn_start_refreshes_dirty_mcp_reuses_clean_runtime_and_cancels_discove .requires_mcp .store(turn > 0, std::sync::atomic::Ordering::Relaxed); if turn == 0 { - session.mark_mcp_runtime_dirty(); + session.mark_mcp_runtime_dirty(McpRefreshReason::ExplicitInvalidation); } else if turn == 3 { session.services.mcp_runtime.reconnect_on_next_refresh(); - session.request_mcp_runtime_refresh(); + session.request_mcp_runtime_refresh(McpRefreshReason::ExplicitInvalidation); } session .spawn_task( diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index af0f10df1d..5d15403492 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -5,6 +5,7 @@ mod regular; mod review; mod user_shell; +use crate::McpRefreshReason; use std::sync::Arc; use std::time::Duration; use std::time::Instant; @@ -304,7 +305,7 @@ impl Session { .enabled(Feature::ExecutorCapabilityDiscoveryV2) { // MCP contributors must project the new inventory even if their other inputs match. - self.request_mcp_runtime_reprojection(); + self.request_mcp_runtime_reprojection(McpRefreshReason::TaskStartedExecutorDiscovery); } self.activate_plugin_selection(&turn_context).await; // Inherited or recovered roots are applied before task start. Otherwise this diff --git a/codex-rs/core/src/tasks/regular.rs b/codex-rs/core/src/tasks/regular.rs index fd91a1d557..a693630786 100644 --- a/codex-rs/core/src/tasks/regular.rs +++ b/codex-rs/core/src/tasks/regular.rs @@ -1,3 +1,4 @@ +use crate::McpRefreshReason; use std::sync::Arc; use codex_async_utils::OrCancelExt; @@ -92,7 +93,9 @@ impl SessionTask for RegularTask { .await; // Even cancelled discovery may have cleared the previous account's catalog. if prepares_mcp { - sess.request_mcp_runtime_reprojection(); + sess.request_mcp_runtime_reprojection( + McpRefreshReason::TurnStartContributorPreparedMcp, + ); } if preparation.is_err() { return SessionStartupPrewarmResolution::Cancelled; diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index af6c240112..18bfc6af7b 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -263,6 +263,51 @@ enum ShutdownOutcome { TimedOut, } +type StartupMcpInvalidations = std::sync::Mutex>; + +/// Identifies the cause of an MCP runtime publication in CLI diagnostics. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +pub enum McpRefreshReason { + AccountAuthReloaded, + AccountLoggedOut, + AccountLoginCompleted, + AccountPluginCacheChanged, + AccountWorkspaceSwitched, + AuthChangeNotification, + CodexAppsToolRefresh, + CurrentEnvironmentsChanged, + DisabledPluginSelectionChanged, + EffectivePluginsChanged, + EnterpriseAuthDisabled, + EnvironmentBindingsChanged, + ExecutorCapabilityDiscoveryRecovered, + ExecutorCapabilityInventoryChanged, + ExplicitInvalidation, + ExplicitServerReconnect, + ExtensionCatalogChanged, + HardRefreshCodexAppsTools, + MarketplaceUpgraded, + McpConfigReloaded, + McpOAuthLoginCompleted, + NonCuratedPluginCacheRefreshed, + OAuthCredentialsRecovered, + PluginInstalled, + PluginOAuthLoginCompleted, + PluginUninstalled, + PreviousRefreshUnpublished, + RemotePluginUninstalled, + RuntimeAuthChanged, + SelectedCapabilityRootsChanged, + SessionMcpInputsChanged, + SkillMcpDependenciesAdded, + TaskStartedExecutorDiscovery, + TurnStartContributorPreparedMcp, + UserConfigFilesReloaded, + UserConfigReloaded, + /// Links to the field-level diff logged when the installed plugin cache changed. + InstalledPluginMetadata(uuid::Uuid), +} + /// [`ThreadManager`] is responsible for creating threads and maintaining /// them in memory. pub struct ThreadManager { @@ -545,7 +590,7 @@ pub(crate) struct ThreadManagerState { model_catalog: model_catalog::ThreadModelCatalog, git_root_discovery: Arc, environment_manager: Arc, - starting_mcp_runtimes: std::sync::Mutex>>, + starting_mcp_runtimes: std::sync::Mutex>>, skills_service: Arc, plugins_manager: Arc, mcp_manager: Arc, @@ -990,7 +1035,12 @@ impl ThreadManager { /// Refreshes every loaded thread and marks threads that are still being created. pub async fn invalidate_mcp_runtimes(&self) { - self.invalidate_starting_mcp_runtimes(); + self.invalidate_mcp_runtimes_for(McpRefreshReason::ExplicitInvalidation) + .await; + } + + pub async fn invalidate_mcp_runtimes_for(&self, reason: McpRefreshReason) { + self.invalidate_starting_mcp_runtimes(reason); let threads = self .state .threads @@ -1000,7 +1050,7 @@ impl ThreadManager { .cloned() .collect::>(); for thread in threads { - thread.session.request_mcp_runtime_refresh(); + thread.session.request_mcp_runtime_refresh(reason); } } @@ -1020,7 +1070,7 @@ impl ThreadManager { } } - fn invalidate_starting_mcp_runtimes(&self) { + fn invalidate_starting_mcp_runtimes(&self, reason: McpRefreshReason) { let mut starting = self .state .starting_mcp_runtimes @@ -1030,7 +1080,10 @@ impl ThreadManager { let Some(runtime) = runtime.upgrade() else { return false; }; - runtime.store(true, Ordering::Release); + runtime + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .insert(reason); true }); } @@ -2702,14 +2755,14 @@ impl ThreadManagerState { (Arc::clone(&self.extensions), Arc::clone(&self.mcp_manager)) }; let tracked_session_source = session_source.clone(); - let source_changed_during_startup = Arc::new(AtomicBool::new(false)); + let startup_invalidations = Arc::new(StartupMcpInvalidations::default()); { let mut starting = self .starting_mcp_runtimes .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); starting.retain(|runtime| runtime.strong_count() != 0); - starting.push(Arc::downgrade(&source_changed_during_startup)); + starting.push(Arc::downgrade(&startup_invalidations)); } let windows_sandbox_proxy_settings_mode = if matches!( &session_source, @@ -2907,8 +2960,15 @@ impl ThreadManagerState { if matches!(ready_notification, ThreadReadyNotification::Emit) { new_thread.thread.emit_thread_ready_lifecycle().await; } - if source_changed_during_startup.load(Ordering::Acquire) { - new_thread.thread.session.request_mcp_runtime_refresh(); + for reason in std::mem::take( + &mut *startup_invalidations + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner), + ) { + new_thread + .thread + .session + .request_mcp_runtime_refresh(reason); } if is_resumed_thread { new_thread.thread.emit_thread_resume_lifecycle().await;