Trace MCP refresh causes and plugin metadata changes (#60505)

## What changed

- Track and log `McpRefreshReason` values for runtime refreshes, including invalidations received during thread startup and forced refreshes. Restore claimed reasons when a refresh is cancelled before publication, while preserving concurrent invalidations.
- Correlate installed-plugin metadata invalidations with downstream MCP refreshes using a shared change ID. Log changed field names by plugin ID, additions, removals, and reordering without logging field values.
- Record whether background installed-plugin bundle updates were requested by startup, plugin listing, or installation.

## Testing

Add coverage for metadata change attribution across membership changes and reordering, and for retaining refresh causes across cancellation and concurrent invalidation.

GitOrigin-RevId: 3d4485ccfcc1d8788eb9909a060ac200c467af13
This commit is contained in:
Charlie Marsh
2026-11-10 02:45:30 +00:00
committed by copyberry
parent 0e6f153732
commit aa39ded153
28 changed files with 534 additions and 196 deletions
@@ -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;
});
@@ -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<ThreadManager>,
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<Uuid>) {
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;
@@ -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,
@@ -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;
}
@@ -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(
@@ -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(
+1
View File
@@ -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;
+71 -28
View File
@@ -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<uuid::Uuid>,
/// Remote bundles installed or updated by background installed-plugin sync.
pub materialized_remote_plugins: Vec<RemotePluginMaterialization>,
}
@@ -320,6 +330,10 @@ struct RemoteInstalledPluginsCache {
display_assets: InstalledPluginDisplayAssets,
}
struct InstalledPluginCacheUpdate {
metadata_change_id: Option<uuid::Uuid>,
}
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<RemoteInstalledPlugin>,
auth: Option<&CodexAuth>,
publication: RemoteInstalledPluginsCachePublication,
) -> Option<bool> {
) -> Option<InstalledPluginCacheUpdate> {
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<CodexAuth>,
on_effective_plugins_changed: Option<EffectivePluginsChangedCallback>,
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(
+67 -42
View File
@@ -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<std::sync::atomic::AtomicUsize>| {
@@ -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)
);
}
+71 -17
View File
@@ -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<Item = &'static str> + '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<String, Vec<&'static str>> {
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.
@@ -121,6 +121,13 @@ fn installed_ordering_remains_significant() {
let mut current = previous.clone();
current.reverse();
assert!(!installed_plugin_metadata_eq(&previous, &current));
assert_eq!(
installed_plugin_metadata_changes(&previous, &current),
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(&current)
));
}
#[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, &current),
std::collections::BTreeMap::from([
("added".into(), vec!["added"]),
("removed".into(), vec!["removed"]),
(id, vec!["version", "enabled"]),
])
);
}
+1
View File
@@ -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;
+7 -2
View File
@@ -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)
+2 -1
View File
@@ -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(&current);
}
Ok(())
+2 -1
View File
@@ -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<Session>, 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<Session>) {
+21 -26
View File
@@ -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<Self>,
) -> anyhow::Result<codex_mcp::CodexAppsToolSnapshot> {
// 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
+6 -5
View File
@@ -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;
+62 -11
View File
@@ -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<BTreeSet<McpRefreshReason>>,
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<McpRefreshInvalidationGuard<'_>> {
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<McpRefreshReason> {
std::mem::take(
&mut *self
.pending
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner),
)
}
fn start(
&self,
thread_id: ThreadId,
reasons: BTreeSet<McpRefreshReason>,
) -> 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<McpRefreshReason>,
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);
}
}
}
+30 -21
View File
@@ -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());
}
+7 -3
View File
@@ -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());
@@ -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)
};
@@ -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),
+8 -7
View File
@@ -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
@@ -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 [
@@ -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(
+2 -1
View File
@@ -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
+4 -1
View File
@@ -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;
+69 -9
View File
@@ -263,6 +263,51 @@ enum ShutdownOutcome {
TimedOut,
}
type StartupMcpInvalidations = std::sync::Mutex<std::collections::BTreeSet<McpRefreshReason>>;
/// 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<GitRootDiscovery>,
environment_manager: Arc<EnvironmentManager>,
starting_mcp_runtimes: std::sync::Mutex<Vec<std::sync::Weak<AtomicBool>>>,
starting_mcp_runtimes: std::sync::Mutex<Vec<std::sync::Weak<StartupMcpInvalidations>>>,
skills_service: Arc<HostSkillsService>,
plugins_manager: Arc<PluginsManager>,
mcp_manager: Arc<McpManager>,
@@ -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::<Vec<_>>();
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;