From ea819e1f4d0759777106e5a6fb9984752b20f972 Mon Sep 17 00:00:00 2001 From: Anthony Tafoya Date: Fri, 23 Oct 2026 17:11:35 +0000 Subject: [PATCH] Record skill and plugin catalog sizes on publication (#56530) ## What changed - Add `codex.skills.catalog.entries` and `codex.plugins.catalog.entries` histograms with `cloud` or `executor` stages and `complete`, `warning`, or `partial` status. Partial results take precedence over warnings. - Exclude cached reads and repeated observations of unchanged failed skill discovery from metric samples. - Mark executor skill catalogs partial when discovery fails, and clear unpublished executor plugin catalogs instead of publishing an empty result. ## Testing Extend skill integration tests to check catalog counts and statuses, fresh discovery snapshots, partial discovery, and repeated reads without duplicate samples. Update plugin state tests to distinguish unpublished catalogs from empty discovery results. GitOrigin-RevId: e25978ba9fce54e9660853b1d9673ddab6db5061 --- .../extension-api/src/capabilities/metrics.rs | 27 ++++ .../ext/extension-api/src/capabilities/mod.rs | 1 + codex-rs/ext/extension-api/src/lib.rs | 1 + codex-rs/ext/mcp/src/catalog_observability.rs | 20 +++ codex-rs/ext/mcp/src/cloud_plugin.rs | 7 +- codex-rs/ext/mcp/src/lib.rs | 1 + codex-rs/ext/mcp/src/plugin.rs | 2 +- .../ext/mcp/src/plugin_contributor_state.rs | 11 ++ .../mcp/src/plugin_contributor_state_tests.rs | 4 +- .../ext/skills/src/catalog_observability.rs | 20 +++ codex-rs/ext/skills/src/catalog_state.rs | 9 +- codex-rs/ext/skills/src/cloud_cache_tests.rs | 1 + codex-rs/ext/skills/src/extension.rs | 11 +- codex-rs/ext/skills/src/lib.rs | 1 + codex-rs/ext/skills/src/provider/executor.rs | 2 + codex-rs/ext/skills/src/state.rs | 74 ++++++---- codex-rs/ext/skills/tests/skills_extension.rs | 20 ++- .../skills_extension/discovery_v2_tests.rs | 128 ++++++++++-------- .../skills_extension/merge_catalog_tests.rs | 33 ++++- 19 files changed, 274 insertions(+), 99 deletions(-) create mode 100644 codex-rs/ext/mcp/src/catalog_observability.rs create mode 100644 codex-rs/ext/skills/src/catalog_observability.rs diff --git a/codex-rs/ext/extension-api/src/capabilities/metrics.rs b/codex-rs/ext/extension-api/src/capabilities/metrics.rs index 67f395b480..1cee3b1d00 100644 --- a/codex-rs/ext/extension-api/src/capabilities/metrics.rs +++ b/codex-rs/ext/extension-api/src/capabilities/metrics.rs @@ -19,3 +19,30 @@ pub trait ExtensionMetrics: Send + Sync { tags: &[(&str, &str)], ); } + +/// Records a published catalog size, with partial results taking precedence over warnings. +pub fn record_catalog_entries( + metrics: Option<&dyn ExtensionMetrics>, + name: &str, + stage: &str, + entry_count: usize, + is_partial: bool, + warning_count: usize, +) { + let Some(metrics) = metrics else { + return; + }; + let status = if is_partial { + "partial" + } else if warning_count > 0 { + "warning" + } else { + "complete" + }; + metrics.histogram_with_boundaries( + name, + i64::try_from(entry_count).unwrap_or(i64::MAX), + &[0., 1., 2., 5., 10., 20., 50., 100., 200., 500., 1000.], + &[("stage", stage), ("status", status)], + ); +} diff --git a/codex-rs/ext/extension-api/src/capabilities/mod.rs b/codex-rs/ext/extension-api/src/capabilities/mod.rs index d933738616..358f44e9e5 100644 --- a/codex-rs/ext/extension-api/src/capabilities/mod.rs +++ b/codex-rs/ext/extension-api/src/capabilities/mod.rs @@ -8,6 +8,7 @@ pub use events::ExtensionEventSink; pub use events::ExtensionWarning; pub use events::NoopExtensionEventSink; pub use metrics::ExtensionMetrics; +pub use metrics::record_catalog_entries; pub use response_items::NoopResponseItemInjector; pub use response_items::ResponseItemInjectionFuture; pub use response_items::ResponseItemInjector; diff --git a/codex-rs/ext/extension-api/src/lib.rs b/codex-rs/ext/extension-api/src/lib.rs index eabf26998b..45fdd9f155 100644 --- a/codex-rs/ext/extension-api/src/lib.rs +++ b/codex-rs/ext/extension-api/src/lib.rs @@ -23,6 +23,7 @@ pub use capabilities::NoopExtensionEventSink; pub use capabilities::NoopResponseItemInjector; pub use capabilities::ResponseItemInjectionFuture; pub use capabilities::ResponseItemInjector; +pub use capabilities::record_catalog_entries; pub use catalog_refresh::CatalogRefreshBackoff; pub use codex_context_fragments::ContextualUserFragment; pub use codex_mcp::McpProtocolMode; diff --git a/codex-rs/ext/mcp/src/catalog_observability.rs b/codex-rs/ext/mcp/src/catalog_observability.rs new file mode 100644 index 0000000000..bad6a0bb54 --- /dev/null +++ b/codex-rs/ext/mcp/src/catalog_observability.rs @@ -0,0 +1,20 @@ +//! Measures published cloud and V2 executor plugin catalogs, excluding cached reads. + +use codex_core_plugins::PluginCatalog; +use codex_extension_api::ExtensionMetrics; +use codex_extension_api::record_catalog_entries; + +pub(crate) fn record_catalog( + metrics: Option<&dyn ExtensionMetrics>, + stage: &'static str, + catalog: &PluginCatalog, +) { + record_catalog_entries( + metrics, + "codex.plugins.catalog.entries", + stage, + catalog.entries.len(), + catalog.is_partial, + catalog.warnings.len(), + ); +} diff --git a/codex-rs/ext/mcp/src/cloud_plugin.rs b/codex-rs/ext/mcp/src/cloud_plugin.rs index 4ef00874f9..38df3ecfcc 100644 --- a/codex-rs/ext/mcp/src/cloud_plugin.rs +++ b/codex-rs/ext/mcp/src/cloud_plugin.rs @@ -141,8 +141,11 @@ impl ThreadLifecycleContributor for PluginContributor { ) -> ExtensionFuture<'a, ()> { Box::pin(async move { let state = input.thread_store.get_or_init(PluginsThreadState::default); - state.contributor_state().plugins_enabled = - input.config.features.enabled(Feature::Plugins); + { + let mut state = state.contributor_state(); + state.plugins_enabled = input.config.features.enabled(Feature::Plugins); + state.extension_metrics = input.extension_metrics; + } if let Some(resources) = &input.mcp_resource_client { input.session_store.insert(resources.as_ref().clone()); } diff --git a/codex-rs/ext/mcp/src/lib.rs b/codex-rs/ext/mcp/src/lib.rs index e8a46b6dd5..9418f490e3 100644 --- a/codex-rs/ext/mcp/src/lib.rs +++ b/codex-rs/ext/mcp/src/lib.rs @@ -7,6 +7,7 @@ use codex_extension_api::McpServerContributor; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; use codex_mcp::hosted_plugin_runtime_mcp_server_config; +mod catalog_observability; mod cloud_plugin; mod plugin; mod plugin_contributor; diff --git a/codex-rs/ext/mcp/src/plugin.rs b/codex-rs/ext/mcp/src/plugin.rs index bb6db0f5ac..e2095bdfbb 100644 --- a/codex-rs/ext/mcp/src/plugin.rs +++ b/codex-rs/ext/mcp/src/plugin.rs @@ -324,7 +324,7 @@ impl McpServerContributor for PluginContributor { } return plugins; } - state.contributor_state().update_executor_catalog(&[]); + state.contributor_state().clear_executor_catalog(); if let Some(snapshot) = context.executor_capability_discovery() { for root in snapshot.roots() { diff --git a/codex-rs/ext/mcp/src/plugin_contributor_state.rs b/codex-rs/ext/mcp/src/plugin_contributor_state.rs index db07aefce9..76e108cb5b 100644 --- a/codex-rs/ext/mcp/src/plugin_contributor_state.rs +++ b/codex-rs/ext/mcp/src/plugin_contributor_state.rs @@ -1,17 +1,20 @@ //! Thread-owned cloud catalog and stable selected-root metadata, guarded by a short-lived mutex. +use crate::catalog_observability::record_catalog; use codex_core_plugins::PluginCatalog; use codex_core_plugins::PluginRootOwnership; use codex_core_plugins::catalog_from_snapshots; use codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot; use codex_exec_server::FileSystemSandboxContext; use codex_extension_api::CatalogRefreshBackoff; +use codex_extension_api::ExtensionMetrics; use codex_extension_api::SelectedPluginContribution; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; use codex_mcp::McpResourceClient; use codex_mcp::McpResourceClientAuthKey; use codex_protocol::capabilities::SelectedCapabilityRoot; use std::collections::HashMap; +use std::sync::Arc; use std::sync::Mutex; use std::sync::MutexGuard; use tokio::sync::Mutex as AsyncMutex; @@ -40,6 +43,7 @@ impl PluginsThreadState { /// `None` is unpublished; an empty catalog is a successful discovery result. #[derive(Default)] pub(crate) struct PluginContributorState { + pub(crate) extension_metrics: Option>, pub(crate) plugins_enabled: bool, pub(crate) cloud_generation: Option, pub(crate) cloud_retry: Option, @@ -90,10 +94,16 @@ impl PluginContributorState { .unwrap_or(usize::MAX) }); } + record_catalog(self.extension_metrics.as_deref(), "executor", &catalog); self.executor_catalog = Some(catalog); self.executor_snapshots = snapshots.to_vec(); } + pub(crate) fn clear_executor_catalog(&mut self) { + self.executor_catalog = None; + self.executor_snapshots.clear(); + } + /// Returns the cloud catalog only while it belongs to the live auth generation. pub(crate) fn cloud_catalog(&self) -> Option<&PluginCatalog> { self.cloud_generation @@ -151,6 +161,7 @@ impl PluginContributorState { "cloud plugin discovery was partial; retaining prior auth-scoped entries as partial" ); } + record_catalog(self.extension_metrics.as_deref(), "cloud", &catalog); generation.catalog = Some(catalog); true } diff --git a/codex-rs/ext/mcp/src/plugin_contributor_state_tests.rs b/codex-rs/ext/mcp/src/plugin_contributor_state_tests.rs index 39f2ade974..4d5164d19b 100644 --- a/codex-rs/ext/mcp/src/plugin_contributor_state_tests.rs +++ b/codex-rs/ext/mcp/src/plugin_contributor_state_tests.rs @@ -353,10 +353,10 @@ async fn projection_respects_discovery_mode_and_preserves_enabled_plugins() -> a "{case}, V2={v2_enabled}, V1={}", legacy_snapshot.is_some() ); - let expected_catalog = catalog_from_snapshots(snapshots.unwrap_or_default()); + let expected_catalog = snapshots.map(catalog_from_snapshots); assert_eq!( state.contributor_state().executor_catalog.as_ref(), - Some(&expected_catalog), + expected_catalog.as_ref(), "{case}" ); } diff --git a/codex-rs/ext/skills/src/catalog_observability.rs b/codex-rs/ext/skills/src/catalog_observability.rs new file mode 100644 index 0000000000..c376c9e2b6 --- /dev/null +++ b/codex-rs/ext/skills/src/catalog_observability.rs @@ -0,0 +1,20 @@ +//! Measures published skill catalogs, excluding cached reads. + +use crate::catalog::SkillCatalog; +use codex_extension_api::ExtensionMetrics; +use codex_extension_api::record_catalog_entries; + +pub(crate) fn record_catalog( + metrics: Option<&dyn ExtensionMetrics>, + stage: &'static str, + catalog: &SkillCatalog, +) { + record_catalog_entries( + metrics, + "codex.skills.catalog.entries", + stage, + catalog.entries.len(), + catalog.is_partial, + catalog.warnings.len(), + ); +} diff --git a/codex-rs/ext/skills/src/catalog_state.rs b/codex-rs/ext/skills/src/catalog_state.rs index d0fe4c8d13..efa4b7442f 100644 --- a/codex-rs/ext/skills/src/catalog_state.rs +++ b/codex-rs/ext/skills/src/catalog_state.rs @@ -108,7 +108,7 @@ pub(crate) struct CachedExecutorV2Catalog { } impl CachedExecutorV2Catalog { - pub(crate) fn matches( + pub(crate) fn same_discovery( &self, snapshots: &[codex_exec_server::ExecutorCapabilityDiscoveryV2Snapshot], ) -> bool { @@ -117,9 +117,7 @@ impl CachedExecutorV2Catalog { .snapshots .iter() .zip(snapshots) - .all(|(previous, current)| { - current.result.is_ok() && previous.same_discovery(current) - }) + .all(|(previous, current)| previous.same_discovery(current)) } } @@ -180,7 +178,8 @@ impl CachedExecutorDiscoveryCatalog { previous.selected_root == current.selected_root && match (&previous.result, ¤t.result) { (Ok(previous), Ok(current)) => Arc::ptr_eq(previous, current), - _ => false, + (Err(previous), Err(current)) => previous == current, + (Ok(_), Err(_)) | (Err(_), Ok(_)) => false, } }, ) diff --git a/codex-rs/ext/skills/src/cloud_cache_tests.rs b/codex-rs/ext/skills/src/cloud_cache_tests.rs index 4dd3b7fd34..0c64158d8e 100644 --- a/codex-rs/ext/skills/src/cloud_cache_tests.rs +++ b/codex-rs/ext/skills/src/cloud_cache_tests.rs @@ -100,6 +100,7 @@ fn state() -> SkillsThreadState { shadow_selection_enabled: false, }, /*cloud_skills_available*/ true, + /*extension_metrics*/ None, ) } diff --git a/codex-rs/ext/skills/src/extension.rs b/codex-rs/ext/skills/src/extension.rs index 4fd4670af5..effe7624b8 100644 --- a/codex-rs/ext/skills/src/extension.rs +++ b/codex-rs/ext/skills/src/extension.rs @@ -156,6 +156,7 @@ where input.thread_store.insert(SkillsThreadState::new( (self.config_from_host)(input.config), cloud_skills_available, + input.extension_metrics, )); }) } @@ -167,7 +168,7 @@ where { fn on_config_changed( &self, - _session_store: &ExtensionData, + session_store: &ExtensionData, thread_store: &ExtensionData, _previous_config: &C, new_config: &C, @@ -177,7 +178,13 @@ where state.set_config(next_config); } else { let cloud_skills_available = self.providers.has_cloud_provider(); - thread_store.insert(SkillsThreadState::new(next_config, cloud_skills_available)); + thread_store.insert(SkillsThreadState::new( + next_config, + cloud_skills_available, + session_store + .get::() + .and_then(|state| state.extension_metrics.clone()), + )); } } } diff --git a/codex-rs/ext/skills/src/lib.rs b/codex-rs/ext/skills/src/lib.rs index cca04fa2a9..98ba9de6ea 100644 --- a/codex-rs/ext/skills/src/lib.rs +++ b/codex-rs/ext/skills/src/lib.rs @@ -1,5 +1,6 @@ mod aliases; pub mod catalog; +mod catalog_observability; mod catalog_prompt; mod catalog_state; mod config; diff --git a/codex-rs/ext/skills/src/provider/executor.rs b/codex-rs/ext/skills/src/provider/executor.rs index 3c4a43e93a..ee1278b07a 100644 --- a/codex-rs/ext/skills/src/provider/executor.rs +++ b/codex-rs/ext/skills/src/provider/executor.rs @@ -213,12 +213,14 @@ impl ExecutorSkillProvider { let discovery = match &root.result { Ok(discovery) => discovery.as_ref(), Err(error) => { + catalog.is_partial = true; catalog.warnings.push(format!( "Selected capability root `{selected_root_id}` discovery failed: {error}" )); continue; } }; + catalog.is_partial |= discovery.error.is_some(); let outcome = load_environment_skills_from_discovery(discovery, self.restriction_product); catalog.warnings.extend(outcome.warnings); diff --git a/codex-rs/ext/skills/src/state.rs b/codex-rs/ext/skills/src/state.rs index 20c505d8fe..1f7eb84722 100644 --- a/codex-rs/ext/skills/src/state.rs +++ b/codex-rs/ext/skills/src/state.rs @@ -22,6 +22,7 @@ use crate::catalog::SkillProviderError; use crate::catalog::SkillProviderResult; use crate::catalog::SkillReadResult; use crate::catalog::SkillSourceKind; +use crate::catalog_observability::record_catalog; use crate::catalog_state::CachedExecutorCatalog; use crate::catalog_state::CachedExecutorDiscoveryCatalog; use crate::catalog_state::CachedExecutorV2Catalog; @@ -58,6 +59,7 @@ pub struct SkillsThreadState { config: Mutex, cloud_skills_available: bool, catalog_state: Mutex, + extension_metrics: Option>, aliases: SkillAliasState, shadow_selection_turn: Mutex>, pub(crate) executor_read_snapshot: Mutex>, @@ -66,12 +68,17 @@ pub struct SkillsThreadState { } impl SkillsThreadState { - pub(crate) fn new(config: SkillsExtensionConfig, cloud_skills_available: bool) -> Self { + pub(crate) fn new( + config: SkillsExtensionConfig, + cloud_skills_available: bool, + extension_metrics: Option>, + ) -> Self { Self { cloud_refresh_lock: AsyncMutex::new(false), config: Mutex::new(config), cloud_skills_available, catalog_state: Mutex::new(SkillCatalogState::default()), + extension_metrics, aliases: SkillAliasState::default(), shadow_selection_turn: Mutex::new(None), executor_read_snapshot: Mutex::new(None), @@ -144,18 +151,32 @@ impl SkillsThreadState { ) -> SkillCatalog { if let Some(snapshots) = &query.executor_capability_discovery_v2 { let snapshots = snapshots.clone(); - let cached = self - .catalog_state - .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .executor_v2_cache - .as_ref() - .filter(|cached| cached.matches(&snapshots)) - .map(|cached| cached.catalog.clone()); + let (cached, already_observed) = { + let state = self + .catalog_state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + let cached = state.executor_v2_cache.as_ref(); + let already_observed = + cached.is_some_and(|cached| cached.same_discovery(&snapshots)); + ( + cached + .filter(|_| { + already_observed + && snapshots.iter().all(|snapshot| snapshot.result.is_ok()) + }) + .map(|cached| cached.catalog.clone()), + already_observed, + ) + }; match cached { Some(catalog) => catalog, None => { let catalog = providers.list_executor_for_turn(query).await; + // Failed snapshots are rebuilt for callers, but each is still one observation. + if !already_observed { + record_catalog(self.extension_metrics.as_deref(), "executor", &catalog); + } self.catalog_state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) @@ -191,6 +212,7 @@ impl SkillsThreadState { } else { providers.list_executor_for_turn(query).await }; + record_catalog(self.extension_metrics.as_deref(), "executor", &catalog); self.catalog_state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) @@ -217,27 +239,30 @@ impl SkillsThreadState { .is_none_or(|discovered| discovered.error.is_some()) }) }); - if !discovery_failed - && let Some(cached) = self + let already_observed = { + let state = self .catalog_state .lock() - .unwrap_or_else(std::sync::PoisonError::into_inner) - .executor_discovery_cache - .as_ref() - .filter(|cached| { - cached.matches( - &query.executor_roots, - discovery.as_ref(), - query.include_bundled_skills, - ) - }) - { - return cached.catalog.clone(); - } + .unwrap_or_else(std::sync::PoisonError::into_inner); + let cached = state.executor_discovery_cache.as_ref().filter(|cached| { + cached.matches( + &query.executor_roots, + discovery.as_ref(), + query.include_bundled_skills, + ) + }); + if !discovery_failed && let Some(cached) = cached { + return cached.catalog.clone(); + } + cached.is_some() + }; let roots = query.executor_roots.clone(); let include_bundled_skills = query.include_bundled_skills; let discovered = providers.list_executor_for_turn(query).await; + if !already_observed { + record_catalog(self.extension_metrics.as_deref(), "executor", &discovered); + } let mut state = self .catalog_state .lock() @@ -369,6 +394,7 @@ impl SkillsThreadState { } else { catalogs.cloud_resources = CloudResourceCache::default(); } + record_catalog(self.extension_metrics.as_deref(), "cloud", &catalog); let generation = Arc::new(CloudSkillGeneration { auth_cache_key: cache.auth_cache_key.clone(), resource_cache_key, diff --git a/codex-rs/ext/skills/tests/skills_extension.rs b/codex-rs/ext/skills/tests/skills_extension.rs index 8dc727c410..b0fbf38466 100644 --- a/codex-rs/ext/skills/tests/skills_extension.rs +++ b/codex-rs/ext/skills/tests/skills_extension.rs @@ -454,7 +454,10 @@ async fn host_world_state_records_catalog_metrics_on_publish_and_change() -> Tes let published_snapshot = published_snapshot.unwrap(); assert!(fragment.is_some()); let mut expected = expected_catalog_metric_samples("host_world_state", /*count*/ 1); - assert!(startup_metrics.samples().is_empty()); + let expected_discovery = vec![expected_catalog_entries( + "executor", "complete", /*count*/ 0, + )]; + assert_eq!(startup_metrics.samples(), expected_discovery); assert_eq!(turn_metrics.samples(), expected); let sections = prepared_world_state( @@ -480,7 +483,7 @@ async fn host_world_state_records_catalog_metrics_on_publish_and_change() -> Tes .1 .is_none() ); - assert!(startup_metrics.samples().is_empty()); + assert_eq!(startup_metrics.samples(), expected_discovery); assert_eq!(turn_metrics.samples(), expected); let second_skill_path = @@ -525,7 +528,7 @@ async fn host_world_state_records_catalog_metrics_on_publish_and_change() -> Tes "host_world_state", /*count*/ 2, )); - assert!(startup_metrics.samples().is_empty()); + assert_eq!(startup_metrics.samples(), expected_discovery); assert_eq!(turn_metrics.samples(), expected); Ok(()) @@ -3507,6 +3510,17 @@ impl ExtensionMetrics for RecordingMetrics { } } +fn expected_catalog_entries(stage: &str, status: &str, count: i64) -> RecordedHistogram { + RecordedHistogram { + name: "codex.skills.catalog.entries".to_string(), + value: count, + tags: vec![ + ("stage".to_string(), stage.to_string()), + ("status".to_string(), status.to_string()), + ], + } +} + fn expected_catalog_metric_samples(catalog_surface: &str, count: i64) -> Vec { let tags = vec![("catalog_surface".to_string(), catalog_surface.to_string())]; vec![ diff --git a/codex-rs/ext/skills/tests/skills_extension/discovery_v2_tests.rs b/codex-rs/ext/skills/tests/skills_extension/discovery_v2_tests.rs index 02e65097ce..b7c610f779 100644 --- a/codex-rs/ext/skills/tests/skills_extension/discovery_v2_tests.rs +++ b/codex-rs/ext/skills/tests/skills_extension/discovery_v2_tests.rs @@ -183,6 +183,7 @@ async fn v2_steps_refresh_metadata_and_preserve_separate_environments() -> TestR let mut builder = ExtensionRegistryBuilder::new(); install_with_providers(&mut builder, providers, skills_extension_config); let registry = builder.build(); + let metrics = Arc::new(RecordingMetrics::default()); let session_store = ExtensionData::new("session"); let thread_store = ExtensionData::new("thread"); registry.thread_lifecycle_contributors()[0] @@ -192,73 +193,90 @@ async fn v2_steps_refresh_metadata_and_preserve_separate_environments() -> TestR persistent_thread_state_available: true, environments: &[], mcp_resource_client: None, - extension_metrics: None, + extension_metrics: Some(metrics.clone()), session_store: &session_store, thread_store: &thread_store, }) .await; let turn_store = ExtensionData::new("turn-1"); - for description in [Some("original"), Some("updated"), None] { + let mut expected_metrics = Vec::new(); + for (description, failed_environment, count, status) in [ + (Some("original"), None, 4, "complete"), + (Some("updated"), None, 4, "complete"), + // A replacement executor can return a fresh discovery with identical inventory. + (Some("updated"), None, 4, "complete"), + (Some("updated"), Some("remote-b"), 2, "partial"), + (None, None, 0, "complete"), + ] { if let Some(description) = description { for snapshot in &mut snapshots { - Arc::make_mut(snapshot.result.as_mut().unwrap()).plugins[0].skills[0].description = - description.to_string(); + if Some(snapshot.environment_id.as_str()) == failed_environment { + snapshot.result = Err("discovery unavailable".to_string()); + } else { + let mut response = snapshot.result.as_ref().unwrap().as_ref().clone(); + response.plugins[0].skills[0].description = description.to_string(); + snapshot.result = Ok(Arc::new(response)); + } } } else { snapshots.clear(); } - let step_store = ExtensionData::new("step"); - step_store.insert(snapshots.clone()); - step_store.insert(SelectedPluginSnapshot { - plugins: snapshots + expected_metrics.push(expected_catalog_entries("executor", status, count)); + for _ in 0..2 { + let step_store = ExtensionData::new("step"); + step_store.insert(snapshots.clone()); + step_store.insert(SelectedPluginSnapshot { + plugins: snapshots + .iter() + .map(|snapshot| codex_extension_api::SelectedPluginIdentity { + selected_root_id: Some(format!( + "{}:local-name@marketplace", + snapshot.environment_id + )), + plugin_id: codex_core_plugin_common::PluginIdentity::Remote { + remote_plugin_id: "Plugin_123".into(), + }, + }) + .collect(), + ..Default::default() + }); + let sections = prepared_world_state( + ®istry, + &[], + WorldStateContributionInput { + previous_world_state: None, + step_store: &step_store, + model_info: &catalog_model_info(), + thread_id: codex_protocol::ThreadId::new(), + turn_id: "turn-1", + environments: &[], + extension_metrics: None, + session_store: &session_store, + thread_store: &thread_store, + turn_store: &turn_store, + }, + ) + .await; + let rendered = sections .iter() - .map(|snapshot| codex_extension_api::SelectedPluginIdentity { - selected_root_id: Some(format!( - "{}:local-name@marketplace", - snapshot.environment_id - )), - plugin_id: codex_core_plugin_common::PluginIdentity::Remote { - remote_plugin_id: "Plugin_123".into(), - }, - }) - .collect(), - ..Default::default() - }); - let sections = prepared_world_state( - ®istry, - &[], - WorldStateContributionInput { - previous_world_state: None, - step_store: &step_store, - model_info: &catalog_model_info(), - thread_id: codex_protocol::ThreadId::new(), - turn_id: "turn-1", - environments: &[], - extension_metrics: None, - session_store: &session_store, - thread_store: &thread_store, - turn_store: &turn_store, - }, - ) - .await; - let rendered = sections - .iter() - .filter_map(|section| section.render_diff(PreviousWorldStateSection::Absent).1) - .map(|fragment| fragment.body().clone().expect_text()) - .collect::>() - .join("\n"); - if let Some(description) = description { - assert!( - rendered.contains("skill://local-name@marketplace/"), - "{rendered}" - ); - assert!( - rendered.contains(&format!("- local-name:work: {description}")), - "{rendered}" - ); - assert!(!rendered.contains("- dashboard:"), "{rendered}"); - } else { - assert!(!rendered.contains("- local-name:work:"), "{rendered}"); + .filter_map(|section| section.render_diff(PreviousWorldStateSection::Absent).1) + .map(|fragment| fragment.body().clone().expect_text()) + .collect::>() + .join("\n"); + if let Some(description) = description { + assert!( + rendered.contains("skill://local-name@marketplace/"), + "{rendered}" + ); + assert!( + rendered.contains(&format!("- local-name:work: {description}")), + "{rendered}" + ); + assert!(!rendered.contains("- dashboard:"), "{rendered}"); + } else { + assert!(!rendered.contains("- local-name:work:"), "{rendered}"); + } + assert_eq!(metrics.samples(), expected_metrics); } } Ok(()) diff --git a/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs b/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs index 83515a44ca..403dbe1033 100644 --- a/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs +++ b/codex-rs/ext/skills/tests/skills_extension/merge_catalog_tests.rs @@ -64,6 +64,7 @@ fn provider(entries: Vec) -> StaticSkillProvider { struct MergeHarness { registry: codex_extension_api::ExtensionRegistry, + catalog_metrics: Arc, session_store: ExtensionData, thread_store: ExtensionData, turn_store: ExtensionData, @@ -100,6 +101,7 @@ impl MergeHarness { ); let harness = Self { registry: builder.build(), + catalog_metrics: Arc::default(), session_store: ExtensionData::new("session"), thread_store: ExtensionData::new("thread"), turn_store: ExtensionData::new("turn-1"), @@ -124,7 +126,7 @@ impl MergeHarness { persistent_thread_state_available: true, environments: &[], mcp_resource_client: None, - extension_metrics: None, + extension_metrics: Some(harness.catalog_metrics.clone()), session_store: &harness.session_store, thread_store: &harness.thread_store, }) @@ -405,6 +407,13 @@ async fn v2_merged_skills_preserve_cloud_metadata_and_accepted_plugin_routes() - assert_matches!(result, Err(FunctionCallError::RespondToModel(message)) if message == "skill package is not available"); } } + assert_eq!( + harness.catalog_metrics.samples(), + vec![ + expected_catalog_entries("cloud", "warning", /*count*/ 2), + expected_catalog_entries("executor", "warning", /*count*/ 6), + ] + ); Ok(()) } @@ -646,7 +655,7 @@ async fn v2_only_complete_cloud_catalog_excludes_unmatched_remote_skills() -> Te unavailable.fail_first_list = true; let mut partial = provider(Vec::new()); partial.catalog.is_partial = true; - for (cloud, config, keeps_remote) in [ + for (cloud, config, keeps_remote, cloud_status) in [ ( provider(vec![cloud_skill("work")]), TestConfig { @@ -654,10 +663,16 @@ async fn v2_only_complete_cloud_catalog_excludes_unmatched_remote_skills() -> Te ..default_config() }, true, + None, + ), + (unavailable, default_config(), true, None), + (partial, default_config(), true, Some("partial")), + ( + provider(Vec::new()), + default_config(), + false, + Some("complete"), ), - (unavailable, default_config(), true), - (partial, default_config(), true), - (provider(Vec::new()), default_config(), false), ] { let harness = MergeHarness::new( provider(vec![ @@ -677,6 +692,14 @@ async fn v2_only_complete_cloud_catalog_excludes_unmatched_remote_skills() -> Te assert_eq!(rendered.contains("- deploy:work:"), keeps_remote); assert!(rendered.contains("- deploy:local:")); assert!(!rendered.contains("Cloud work description")); + let mut expected = Vec::new(); + if let Some(status) = cloud_status { + expected.push(expected_catalog_entries("cloud", status, /*count*/ 0)); + } + expected.push(expected_catalog_entries( + "executor", "complete", /*count*/ 2, + )); + assert_eq!(harness.catalog_metrics.samples(), expected); } Ok(()) }