From 280d56b1d823a7e9dfd85f796da38bd825d0052d Mon Sep 17 00:00:00 2001 From: rphilizaire-openai Date: Tue, 18 Aug 2026 22:47:47 +0000 Subject: [PATCH] Attribute executor skill invocations to plugins (#39309) ## What changed - Carry selected plugin identities from MCP discovery into per-turn extension data, limited to capability roots that are ready for the turn. - Annotate executor skill catalog entries with the matching plugin ID and `user` scope. - Include that attribution in `skill_invocation` analytics for both explicit and implicit executor skill invocations. ## Testing - Extend the executor skills app-server test to verify plugin ID, skill scope, and invocation type for explicit and implicit analytics events. GitOrigin-RevId: 9345600ae29bcb8eb894fffd2d6f7b05f22ebf74 --- .../tests/suite/v2/executor_skills.rs | 23 +++++++++++++++++++ codex-rs/core/src/mcp.rs | 12 ++++++++++ codex-rs/core/src/session/mcp.rs | 5 +++- codex-rs/core/src/session/mcp_runtime.rs | 3 +++ codex-rs/core/src/session/mod.rs | 13 +++++++++++ .../ext/extension-api/src/contributors.rs | 2 ++ .../ext/extension-api/src/contributors/mcp.rs | 14 +++++++++++ codex-rs/ext/extension-api/src/lib.rs | 2 ++ codex-rs/ext/mcp/src/executor_plugin.rs | 4 ++++ codex-rs/ext/mcp/tests/executor_plugin_mcp.rs | 1 + codex-rs/ext/skills/src/catalog.rs | 2 ++ codex-rs/ext/skills/src/extension.rs | 15 ++++-------- codex-rs/ext/skills/src/invocation.rs | 2 +- codex-rs/ext/skills/src/provider.rs | 1 + codex-rs/ext/skills/src/provider/executor.rs | 18 +++++++++++++++ codex-rs/ext/skills/src/tools/mod.rs | 23 +++++++++++++++---- .../ext/skills/src/world_state_catalogs.rs | 7 +++++- 17 files changed, 129 insertions(+), 18 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/executor_skills.rs b/codex-rs/app-server/tests/suite/v2/executor_skills.rs index 943b1c615a..018a0585da 100644 --- a/codex-rs/app-server/tests/suite/v2/executor_skills.rs +++ b/codex-rs/app-server/tests/suite/v2/executor_skills.rs @@ -28,6 +28,9 @@ use serde_json::json; use tempfile::TempDir; use tokio::time::timeout; +use super::analytics::mount_analytics_capture; +use super::analytics::wait_for_matching_analytics_event; + #[cfg(target_os = "macos")] const READ_TIMEOUT: Duration = Duration::from_secs(60); #[cfg(not(target_os = "macos"))] @@ -109,11 +112,17 @@ async fn exercise_executor_skill(scenario: ExecutorSkillScenario) -> Result<()> } else { ("never", "") }; + let analytics_config = if scenario == ExecutorSkillScenario::ExplicitOnly { + format!("chatgpt_base_url = \"{}\"", server.uri()) + } else { + String::new() + }; std::fs::write( codex_home.path().join("config.toml"), format!( r#" model = "mock-model" +{analytics_config} approval_policy = "{approval_policy}" {sandbox_config} model_provider = "mock_provider" @@ -133,6 +142,9 @@ stream_max_retries = 0 server.uri() ), )?; + if scenario == ExecutorSkillScenario::ExplicitOnly { + mount_analytics_capture(&server, codex_home.path()).await?; + } let local_skill_dir = codex_home.path().join("skills/local-deploy"); std::fs::create_dir_all(&local_skill_dir)?; std::fs::write( @@ -439,6 +451,17 @@ stream_max_retries = 0 app_server.read_stream_until_notification_message("turn/completed"), ) .await??; + if scenario == ExecutorSkillScenario::ExplicitOnly { + for invocation_type in ["explicit", "implicit"] { + let event = wait_for_matching_analytics_event(&server, READ_TIMEOUT, |event| { + event["event_type"] == "skill_invocation" + && event["event_params"]["invoke_type"] == invocation_type + }) + .await?; + assert_eq!(event["event_params"]["plugin_id"], authority_id); + assert_eq!(event["event_params"]["skill_scope"], "user"); + } + } let requests = response_mock.requests(); let request = &requests[0]; diff --git a/codex-rs/core/src/mcp.rs b/codex-rs/core/src/mcp.rs index 84b3859084..6d3d2c6e7b 100644 --- a/codex-rs/core/src/mcp.rs +++ b/codex-rs/core/src/mcp.rs @@ -13,6 +13,8 @@ use codex_extension_api::ExtensionDataInit; use codex_extension_api::ExtensionRegistry; use codex_extension_api::McpServerContribution; use codex_extension_api::McpServerContributionContext; +use codex_extension_api::SelectedPluginIdentity; +use codex_extension_api::SelectedPluginSnapshot; use codex_login::CodexAuth; use codex_mcp::CODEX_APPS_MCP_SERVER_NAME; use codex_mcp::EffectiveMcpServer; @@ -35,6 +37,7 @@ const LEGACY_CODEX_APPS_REGISTRATION_ID: &str = "legacy_codex_apps"; pub(crate) struct McpRuntimeProjection { pub(crate) config: McpConfig, pub(crate) plugins_available: bool, + pub(crate) selected_plugins: SelectedPluginSnapshot, } pub(crate) struct McpThreadIdentity<'a> { @@ -143,6 +146,7 @@ impl McpManager { let mut selected_plugin_available = false; let mut selected_plugin_connector_sources = Vec::new(); let mut selected_plugin_registrations = Vec::new(); + let mut selected_plugins = Vec::new(); let mut overlays = Vec::new(); // A contributor can emit multiple ordered actions, so order each action globally rather // than enumerating contributors. @@ -173,11 +177,16 @@ impl McpManager { ), ), McpServerContribution::SelectedPluginPackage { + selected_root_id, plugin_id, plugin_display_name, connector_ids, } => { selected_plugin_available = true; + selected_plugins.push(SelectedPluginIdentity { + selected_root_id, + plugin_id: plugin_id.clone(), + }); if !connector_ids.is_empty() { selected_plugin_connector_sources.push( PluginConnectorSource::from_connector_ids( @@ -265,6 +274,9 @@ impl McpManager { McpRuntimeProjection { config: mcp_config, plugins_available, + selected_plugins: SelectedPluginSnapshot { + plugins: selected_plugins, + }, } } diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index 954a6d975f..f88d385392 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -259,6 +259,7 @@ impl Session { executor_capability_discovery.as_deref(), ) .await; + let selected_plugins = mcp_projection.selected_plugins.clone(); let input = self.build_mcp_runtime_input( &desired, mcp_projection, @@ -269,7 +270,9 @@ impl Session { input.mcp_servers.contains_key(CODEX_APPS_MCP_SERVER_NAME), "unknown MCP server '{CODEX_APPS_MCP_SERVER_NAME}'" ); - self.services.mcp_runtime.replace_fresh(input).await + let refreshed = self.services.mcp_runtime.replace_fresh(input).await; + self.services.thread_extension_data.insert(selected_plugins); + refreshed } pub(super) fn mark_mcp_runtime_dirty(&self) { diff --git a/codex-rs/core/src/session/mcp_runtime.rs b/codex-rs/core/src/session/mcp_runtime.rs index 580c83d040..3c1bebb4db 100644 --- a/codex-rs/core/src/session/mcp_runtime.rs +++ b/codex-rs/core/src/session/mcp_runtime.rs @@ -141,6 +141,7 @@ impl Session { ready_selected_capability_roots: &[SelectedCapabilityRoot], elicitation_reviewer: Option, ) { + let selected_plugins = mcp_projection.selected_plugins.clone(); let input = self.build_mcp_runtime_input( desired, mcp_projection, @@ -148,6 +149,7 @@ impl Session { elicitation_reviewer, ); self.services.mcp_runtime.replace(input).await; + self.services.thread_extension_data.insert(selected_plugins); } pub(super) fn build_mcp_runtime_input( @@ -161,6 +163,7 @@ impl Session { let McpRuntimeProjection { mut config, plugins_available, + selected_plugins: _, } = mcp_projection; config.approval_policy = desired.config.permissions.approval_policy.clone(); config.permission_profile = desired.config.permissions.effective_permission_profile(); diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index bde2fda59a..6ebdd02778 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3193,6 +3193,19 @@ impl Session { } .or_cancel(cancellation_token) .await?; + let mut selected_plugins = self + .services + .thread_extension_data + .get::() + .map(|snapshot| snapshot.as_ref().clone()) + .unwrap_or_default(); + selected_plugins.plugins.retain(|plugin| { + ready_selected_capability_roots + .iter() + .any(|root| root.id == plugin.selected_root_id) + }); + extension_data.insert(selected_plugins.clone()); + turn_context.extension_data.insert(selected_plugins); let tool_router = turn::built_tools( self.as_ref(), turn_context.as_ref(), diff --git a/codex-rs/ext/extension-api/src/contributors.rs b/codex-rs/ext/extension-api/src/contributors.rs index 4c4c3d4522..d51ee6597e 100644 --- a/codex-rs/ext/extension-api/src/contributors.rs +++ b/codex-rs/ext/extension-api/src/contributors.rs @@ -25,6 +25,8 @@ mod world_state; pub use context::TurnContextContributionInput; pub use mcp::McpServerContribution; pub use mcp::McpServerContributionContext; +pub use mcp::SelectedPluginIdentity; +pub use mcp::SelectedPluginSnapshot; pub use prompt::PromptFragment; pub use prompt::PromptSlot; pub use skill_invocation::SkillInvocationInput; diff --git a/codex-rs/ext/extension-api/src/contributors/mcp.rs b/codex-rs/ext/extension-api/src/contributors/mcp.rs index d32715a403..1968af0734 100644 --- a/codex-rs/ext/extension-api/src/contributors/mcp.rs +++ b/codex-rs/ext/extension-api/src/contributors/mcp.rs @@ -112,6 +112,19 @@ impl<'a, C> McpServerContributionContext<'a, C> { } } +/// Validated plugin identities projected for the current set of selected roots. +#[derive(Clone, Debug, Default)] +pub struct SelectedPluginSnapshot { + pub plugins: Vec, +} + +/// The configured identity of a plugin resolved from one selected root. +#[derive(Clone, Debug)] +pub struct SelectedPluginIdentity { + pub selected_root_id: String, + pub plugin_id: String, +} + /// One extension-owned overlay for the runtime MCP server configuration. #[derive(Clone, Debug)] pub enum McpServerContribution { @@ -130,6 +143,7 @@ pub enum McpServerContribution { }, /// Records a plugin selected for this thread and any connector IDs it declares. SelectedPluginPackage { + selected_root_id: String, plugin_id: String, plugin_display_name: String, connector_ids: Vec, diff --git a/codex-rs/ext/extension-api/src/lib.rs b/codex-rs/ext/extension-api/src/lib.rs index 69b4903af2..54c329ae22 100644 --- a/codex-rs/ext/extension-api/src/lib.rs +++ b/codex-rs/ext/extension-api/src/lib.rs @@ -46,6 +46,8 @@ pub use contributors::PreviousWorldStateSection; pub use contributors::PromptFragment; pub use contributors::PromptSlot; pub use contributors::RenderedWorldStateFragment; +pub use contributors::SelectedPluginIdentity; +pub use contributors::SelectedPluginSnapshot; pub use contributors::SkillInvocationContributor; pub use contributors::SkillInvocationInput; pub use contributors::SkillInvocationKind; diff --git a/codex-rs/ext/mcp/src/executor_plugin.rs b/codex-rs/ext/mcp/src/executor_plugin.rs index aa394b8ebd..45ce045a03 100644 --- a/codex-rs/ext/mcp/src/executor_plugin.rs +++ b/codex-rs/ext/mcp/src/executor_plugin.rs @@ -183,6 +183,7 @@ impl McpServerContributor for SelectedExecutorPluginMcpContributor { context.config(), plugin_policies.get(&plugin.plugin_id), selection_order, + &root.selected_root.id, plugin, )); } @@ -196,6 +197,7 @@ impl McpServerContributor for SelectedExecutorPluginMcpContributor { context.config(), plugin_policies.get(&plugin.plugin_id), selection_order, + &selected_root.id, plugin, )); } @@ -210,6 +212,7 @@ fn project_metadata( config: &Config, plugin_policy: Option<&HashMap>, selection_order: usize, + selected_root_id: &str, plugin: SelectedPluginMetadata, ) -> Vec { let mut servers = plugin.servers.iter().cloned().collect::>(); @@ -231,6 +234,7 @@ fn project_metadata( .collect::>(); // Keep the package visible even when it contributes only skills. contributions.push(McpServerContribution::SelectedPluginPackage { + selected_root_id: selected_root_id.to_owned(), plugin_id: plugin.plugin_id, plugin_display_name: plugin.plugin_display_name, connector_ids: plugin.connector_ids, diff --git a/codex-rs/ext/mcp/tests/executor_plugin_mcp.rs b/codex-rs/ext/mcp/tests/executor_plugin_mcp.rs index 13f8c1a4ff..0e6e489866 100644 --- a/codex-rs/ext/mcp/tests/executor_plugin_mcp.rs +++ b/codex-rs/ext/mcp/tests/executor_plugin_mcp.rs @@ -134,6 +134,7 @@ async fn selected_plugin_package_is_contributed_without_servers_or_connectors() plugin_id, plugin_display_name, connector_ids, + .. } = contribution else { return None; diff --git a/codex-rs/ext/skills/src/catalog.rs b/codex-rs/ext/skills/src/catalog.rs index 9a7300b61a..1877f8fc21 100644 --- a/codex-rs/ext/skills/src/catalog.rs +++ b/codex-rs/ext/skills/src/catalog.rs @@ -179,6 +179,7 @@ pub struct SkillCatalogEntry { pub main_prompt: SkillResourceId, pub display_path: Option, pub(crate) canonical_skill_id: Option, + pub(crate) plugin_id: Option, pub(crate) analytics_scope: Option, alias_root: Option, alias_root_order: Option, @@ -205,6 +206,7 @@ impl SkillCatalogEntry { main_prompt, display_path: None, canonical_skill_id: None, + plugin_id: None, analytics_scope: None, alias_root: None, alias_root_order: None, diff --git a/codex-rs/ext/skills/src/extension.rs b/codex-rs/ext/skills/src/extension.rs index 149ef04ce1..0ac722182d 100644 --- a/codex-rs/ext/skills/src/extension.rs +++ b/codex-rs/ext/skills/src/extension.rs @@ -18,6 +18,7 @@ use codex_extension_api::ExtensionMetrics; use codex_extension_api::ExtensionRegistryBuilder; use codex_extension_api::ExtensionWarning; use codex_extension_api::PromptFragment; +use codex_extension_api::SelectedPluginSnapshot; use codex_extension_api::SkillInvocationContributor; use codex_extension_api::SkillInvocationInput; use codex_extension_api::SkillInvocationKind; @@ -276,6 +277,7 @@ where session_store, thread_store, /*executor_query*/ None, + /*selected_plugins*/ None, /*sandbox_contexts*/ None, ) } @@ -310,6 +312,7 @@ where session_store, thread_store, executor_query, + step_store.get::(), step_store.get::>(), ) } @@ -550,23 +553,15 @@ impl SkillsExtension { session_store: &ExtensionData, thread_store: &ExtensionData, executor_query: Option, + selected_plugins: Option>, sandbox_contexts: Option>>, ) -> Vec>> { - let Some(thread_state) = thread_store.get::() else { - return Vec::new(); - }; - let orchestrator_available = self.providers.has_orchestrator_provider() - && thread_state.orchestrator_skills_enabled(); - if !orchestrator_available && executor_query.is_none() { - return Vec::new(); - } - skill_tools( self.providers.clone(), session_store, thread_store, - orchestrator_available, executor_query, + selected_plugins, sandbox_contexts, Arc::clone(&self.shadow_selection), ) diff --git a/codex-rs/ext/skills/src/invocation.rs b/codex-rs/ext/skills/src/invocation.rs index 8f97cca02f..75774c1379 100644 --- a/codex-rs/ext/skills/src/invocation.rs +++ b/codex-rs/ext/skills/src/invocation.rs @@ -71,7 +71,7 @@ pub fn detect_implicit_skill_invocation( skill_id: entry.canonical_skill_id.clone(), scope: entry.analytics_scope, }, - plugin_id: None, + plugin_id: entry.plugin_id.clone(), remote_plugin_id: None, invocation_type: InvocationType::Implicit, }); diff --git a/codex-rs/ext/skills/src/provider.rs b/codex-rs/ext/skills/src/provider.rs index cd42bde5e3..afd7d7a5a5 100644 --- a/codex-rs/ext/skills/src/provider.rs +++ b/codex-rs/ext/skills/src/provider.rs @@ -22,6 +22,7 @@ use crate::catalog::SkillResourceId; use crate::catalog::SkillSearchResult; pub use executor::ExecutorSkillProvider; +pub(crate) use executor::attribute_executor_plugins; pub use host::HostSkillProvider; pub use orchestrator::OrchestratorSkillProvider; diff --git a/codex-rs/ext/skills/src/provider/executor.rs b/codex-rs/ext/skills/src/provider/executor.rs index 0f3f75fe60..0745d839a1 100644 --- a/codex-rs/ext/skills/src/provider/executor.rs +++ b/codex-rs/ext/skills/src/provider/executor.rs @@ -2,8 +2,10 @@ use std::sync::Arc; use codex_exec_server::EnvironmentManager; use codex_exec_server::FileSystemSandboxContext; +use codex_extension_api::SelectedPluginSnapshot; use codex_protocol::capabilities::CapabilityRootLocation; use codex_protocol::protocol::Product; +use codex_protocol::protocol::SkillScope; use codex_skills::EnvironmentSkillMetadata; use codex_utils_path_uri::PathConvention; use codex_utils_path_uri::PathUri; @@ -46,6 +48,22 @@ impl ExecutorSkillProvider { } } +pub(crate) fn attribute_executor_plugins( + catalog: &mut SkillCatalog, + snapshot: &SelectedPluginSnapshot, +) { + for skill in &mut catalog.entries { + if let Some(plugin) = snapshot + .plugins + .iter() + .find(|plugin| plugin.selected_root_id == skill.authority.id) + { + skill.plugin_id = Some(plugin.plugin_id.clone()); + skill.analytics_scope = Some(SkillScope::User); + } + } +} + impl SkillProvider for ExecutorSkillProvider { fn list(&self, query: SkillListQuery) -> SkillProviderFuture<'_, SkillCatalog> { Box::pin(async move { diff --git a/codex-rs/ext/skills/src/tools/mod.rs b/codex-rs/ext/skills/src/tools/mod.rs index 2887cb5936..62913d230b 100644 --- a/codex-rs/ext/skills/src/tools/mod.rs +++ b/codex-rs/ext/skills/src/tools/mod.rs @@ -14,6 +14,7 @@ use codex_extension_api::ExtensionData; use codex_extension_api::FunctionCallError; use codex_extension_api::JsonToolOutput; use codex_extension_api::ResponsesApiTool; +use codex_extension_api::SelectedPluginSnapshot; use codex_extension_api::ThreadOriginator; use codex_extension_api::ToolCall; use codex_extension_api::ToolExecutor; @@ -37,6 +38,7 @@ use crate::catalog::SkillCatalog; use crate::catalog::SkillCatalogEntry; use crate::catalog::SkillSourceKind; use crate::provider::SkillListQuery; +use crate::provider::attribute_executor_plugins; use crate::shadow_selection_experiment::ShadowSelectionExperiment; use crate::sources::SkillProviders; use crate::state::SkillsSessionState; @@ -53,14 +55,19 @@ pub(crate) fn skill_tools( providers: SkillProviders, session_store: &ExtensionData, thread_store: &ExtensionData, - orchestrator_available: bool, executor_query: Option, + selected_plugins: Option>, sandbox_contexts: Option>>, shadow_selection: Arc, ) -> Vec>> { let Some(thread_state) = thread_store.get::() else { return Vec::new(); }; + let orchestrator_available = + providers.has_orchestrator_provider() && thread_state.orchestrator_skills_enabled(); + if !orchestrator_available && executor_query.is_none() { + return Vec::new(); + } let mcp_resources = session_store .get::() .and_then(|state| state.mcp_resources.clone()); @@ -72,6 +79,7 @@ pub(crate) fn skill_tools( analytics, orchestrator_available, executor_query, + selected_plugins, sandbox_contexts, executor_catalog: Arc::new(OnceCell::new()), shadow_selection, @@ -127,8 +135,7 @@ impl SkillAnalytics { skill_id: skill.canonical_skill_id.clone(), scope: skill.analytics_scope, }, - // TODO: Include plugin identifiers once skills can be attributed to their plugin. - plugin_id: None, + plugin_id: skill.plugin_id.clone(), remote_plugin_id: None, invocation_type, }], @@ -144,6 +151,7 @@ struct SkillToolContext { analytics: Option, orchestrator_available: bool, executor_query: Option, + selected_plugins: Option>, sandbox_contexts: Option>>, executor_catalog: Arc>, shadow_selection: Arc, @@ -178,10 +186,15 @@ impl SkillToolContext { return SkillCatalog::default(); }; query.turn_id = turn_id.to_string(); - self.executor_catalog + let mut catalog = self + .executor_catalog .get_or_init(|| self.providers.list_executor_for_turn(query)) .await - .clone() + .clone(); + if let Some(selected_plugins) = &self.selected_plugins { + attribute_executor_plugins(&mut catalog, selected_plugins); + } + catalog } } } diff --git a/codex-rs/ext/skills/src/world_state_catalogs.rs b/codex-rs/ext/skills/src/world_state_catalogs.rs index 2e4941db08..c30266f5ae 100644 --- a/codex-rs/ext/skills/src/world_state_catalogs.rs +++ b/codex-rs/ext/skills/src/world_state_catalogs.rs @@ -3,6 +3,7 @@ use std::sync::Arc; use codex_extension_api::ContextualUserFragment; use codex_extension_api::ExtensionEventSink; use codex_extension_api::ExtensionWarning; +use codex_extension_api::SelectedPluginSnapshot; use codex_extension_api::WorldStateContributionInput; use codex_extension_api::WorldStateSectionContribution; use codex_protocol::openai_models::ModelInfo; @@ -11,6 +12,7 @@ use crate::HostSkillsSnapshot; use crate::SkillsExtensionConfig; use crate::catalog::SkillCatalog; use crate::provider::SkillListQuery; +use crate::provider::attribute_executor_plugins; use crate::render::AvailableSkillsRender; use crate::render::RenderedSkillCatalogs; use crate::render::SkillMetadataBudget; @@ -167,10 +169,13 @@ impl<'a> CatalogContext<'a> { } async fn discover_executor_catalog(&self, query: SkillListQuery) -> CatalogContribution { - let catalog = self + let mut catalog = self .thread_state .executor_catalog_snapshot(self.providers, query) .await; + if let Some(selected_plugins) = self.input.turn_store.get::() { + attribute_executor_plugins(&mut catalog, &selected_plugins); + } self.input .turn_store .insert(ExecutorSkillsStepState(catalog.clone()));