openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-17 12:23:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Attribute executor skill invocations to plugins (#39309)

Browse Source 
## 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
This commit is contained in:
rphilizaire-openai
2026-08-18 22:47:47 +00:00
committed by copyberry
parent c97bd2dcb5
commit 280d56b1d8
17 changed files with 129 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
codex-rs/app-server/tests/suite/v2/executor_skills.rs
View File

@@ -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];

12
codex-rs/core/src/mcp.rs
View File

@@ -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,
},
}
}

5
codex-rs/core/src/session/mcp.rs
View File

@@ -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) {

3
codex-rs/core/src/session/mcp_runtime.rs
View File

@@ -141,6 +141,7 @@ impl Session {
ready_selected_capability_roots: &[SelectedCapabilityRoot],
elicitation_reviewer: Option<ElicitationReviewerHandle>,
) {
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();

13
codex-rs/core/src/session/mod.rs
View File

@@ -3193,6 +3193,19 @@ impl Session {
}
.or_cancel(cancellation_token)
.await?;
let mut selected_plugins = self
.services
.thread_extension_data
.get::<codex_extension_api::SelectedPluginSnapshot>()
.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(),

2
codex-rs/ext/extension-api/src/contributors.rs
View File

@@ -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;

14
codex-rs/ext/extension-api/src/contributors/mcp.rs
View File

@@ -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<SelectedPluginIdentity>,
}
/// 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<String>,

2
codex-rs/ext/extension-api/src/lib.rs
View File

@@ -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;

4
codex-rs/ext/mcp/src/executor_plugin.rs
View File

@@ -183,6 +183,7 @@ impl McpServerContributor<Config> for SelectedExecutorPluginMcpContributor {
context.config(),
plugin_policies.get(&plugin.plugin_id),
selection_order,
&root.selected_root.id,
plugin,
));
}
@@ -196,6 +197,7 @@ impl McpServerContributor<Config> 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<String, PluginMcpServerConfig>>,
selection_order: usize,
selected_root_id: &str,
plugin: SelectedPluginMetadata,
) -> Vec<McpServerContribution> {
let mut servers = plugin.servers.iter().cloned().collect::<HashMap<_, _>>();
@@ -231,6 +234,7 @@ fn project_metadata(
.collect::<Vec<_>>();
// 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,

1
codex-rs/ext/mcp/tests/executor_plugin_mcp.rs
View File

@@ -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;

2
codex-rs/ext/skills/src/catalog.rs
View File

@@ -179,6 +179,7 @@ pub struct SkillCatalogEntry {
pub main_prompt: SkillResourceId,
pub display_path: Option<String>,
pub(crate) canonical_skill_id: Option<String>,
pub(crate) plugin_id: Option<String>,
pub(crate) analytics_scope: Option<SkillScope>,
alias_root: Option<String>,
alias_root_order: Option<usize>,
@@ -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,

15
codex-rs/ext/skills/src/extension.rs
View File

@@ -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::<SelectedPluginSnapshot>(),
step_store.get::<HashMap<String, FileSystemSandboxContext>>(),
)
}
@@ -550,23 +553,15 @@ impl<C> SkillsExtension<C> {
session_store: &ExtensionData,
thread_store: &ExtensionData,
executor_query: Option<SkillListQuery>,
selected_plugins: Option<Arc<SelectedPluginSnapshot>>,
sandbox_contexts: Option<Arc<HashMap<String, FileSystemSandboxContext>>>,
) -> Vec<Arc<dyn ToolExecutor<ToolCall>>> {
let Some(thread_state) = thread_store.get::<SkillsThreadState>() 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),
)

2
codex-rs/ext/skills/src/invocation.rs
View File

@@ -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,
});

1
codex-rs/ext/skills/src/provider.rs
View File

@@ -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;

18
codex-rs/ext/skills/src/provider/executor.rs
View File

@@ -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 {

23
codex-rs/ext/skills/src/tools/mod.rs
View File

@@ -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<SkillListQuery>,
selected_plugins: Option<Arc<SelectedPluginSnapshot>>,
sandbox_contexts: Option<Arc<HashMap<String, FileSystemSandboxContext>>>,
shadow_selection: Arc<ShadowSelectionExperiment>,
) -> Vec<Arc<dyn ToolExecutor<ToolCall>>> {
let Some(thread_state) = thread_store.get::<SkillsThreadState>() 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::<SkillsSessionState>()
.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<SkillAnalytics>,
orchestrator_available: bool,
executor_query: Option<SkillListQuery>,
selected_plugins: Option<Arc<SelectedPluginSnapshot>>,
sandbox_contexts: Option<Arc<HashMap<String, FileSystemSandboxContext>>>,
executor_catalog: Arc<OnceCell<SkillCatalog>>,
shadow_selection: Arc<ShadowSelectionExperiment>,
@@ -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
}
}
}

7
codex-rs/ext/skills/src/world_state_catalogs.rs
View File

@@ -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::<SelectedPluginSnapshot>() {
attribute_executor_plugins(&mut catalog, &selected_plugins);
}
self.input
.turn_store
.insert(ExecutorSkillsStepState(catalog.clone()));