From d761bf87db2cfb6ef3fb5822896b8756c74e8769 Mon Sep 17 00:00:00 2001 From: James Watterson-Tenvooren Date: Wed, 3 Jun 2026 18:17:32 -0700 Subject: [PATCH] Add remote plugin IDs to analytics events --- .../analytics/src/analytics_client_tests.rs | 8 +- codex-rs/analytics/src/events.rs | 5 +- codex-rs/app-server/src/request_processors.rs | 1 - .../request_processors/config_processor.rs | 9 +- .../src/request_processors/plugins.rs | 32 ++- .../tests/suite/v2/plugin_install.rs | 4 +- .../app-server/tests/suite/v2/plugin_list.rs | 137 +++++++++++++ .../tests/suite/v2/plugin_uninstall.rs | 93 ++++++++- codex-rs/core-plugins/src/loader.rs | 55 ++---- codex-rs/core-plugins/src/manager.rs | 66 ++++++- codex-rs/core-plugins/src/manager_tests.rs | 110 ++++++++++- codex-rs/core-plugins/src/remote.rs | 67 ++++--- codex-rs/core/src/session/turn.rs | 17 +- codex-rs/core/tests/suite/plugins.rs | 184 +++++++++++++++--- codex-rs/plugin/src/lib.rs | 26 +-- 15 files changed, 668 insertions(+), 146 deletions(-) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index a3b6dfa949..073df7a455 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -2877,6 +2877,7 @@ fn plugin_used_event_serializes_expected_shape() { "event_type": "codex_plugin_used", "event_params": { "plugin_id": "sample@test", + "remote_plugin_id": null, "plugin_name": "sample", "marketplace_name": "test", "has_skills": true, @@ -2907,6 +2908,7 @@ fn plugin_management_event_serializes_expected_shape() { "event_type": "codex_plugin_installed", "event_params": { "plugin_id": "sample@test", + "remote_plugin_id": null, "plugin_name": "sample", "marketplace_name": "test", "has_skills": true, @@ -2919,7 +2921,7 @@ fn plugin_management_event_serializes_expected_shape() { } #[test] -fn plugin_management_event_can_use_remote_plugin_id_override() { +fn plugin_management_event_serializes_local_and_remote_plugin_ids() { let mut plugin = sample_plugin_metadata(); plugin.remote_plugin_id = Some("plugins~Plugin_remote".to_string()); let event = TrackEventRequest::PluginInstalled(CodexPluginEventRequest { @@ -2929,8 +2931,9 @@ fn plugin_management_event_can_use_remote_plugin_id_override() { let payload = serde_json::to_value(&event).expect("serialize plugin installed event"); + assert_eq!(payload["event_params"]["plugin_id"], "sample@test"); assert_eq!( - payload["event_params"]["plugin_id"], + payload["event_params"]["remote_plugin_id"], "plugins~Plugin_remote" ); assert_eq!(payload["event_params"]["plugin_name"], "sample"); @@ -3274,6 +3277,7 @@ async fn reducer_ingests_plugin_state_changed_fact() { "event_type": "codex_plugin_disabled", "event_params": { "plugin_id": "sample@test", + "remote_plugin_id": null, "plugin_name": "sample", "marketplace_name": "test", "has_skills": true, diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 019d59776a..cbb79f1152 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -863,6 +863,7 @@ pub(crate) struct CodexTurnSteerEventRequest { #[derive(Serialize)] pub(crate) struct CodexPluginMetadata { pub(crate) plugin_id: Option, + pub(crate) remote_plugin_id: Option, pub(crate) plugin_name: Option, pub(crate) marketplace_name: Option, pub(crate) has_skills: Option, @@ -923,9 +924,9 @@ pub(crate) fn codex_plugin_metadata(plugin: PluginTelemetryMetadata) -> CodexPlu remote_plugin_id, capability_summary, } = plugin; - let event_plugin_id = remote_plugin_id.unwrap_or_else(|| plugin_id.as_key()); CodexPluginMetadata { - plugin_id: Some(event_plugin_id), + plugin_id: Some(plugin_id.as_key()), + remote_plugin_id, plugin_name: Some(plugin_id.plugin_name), marketplace_name: Some(plugin_id.marketplace_name), has_skills: capability_summary diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 79da49fba4..92883ec213 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -311,7 +311,6 @@ use codex_core_plugins::PluginReadRequest; use codex_core_plugins::PluginUninstallError as CorePluginUninstallError; use codex_core_plugins::loader::load_plugin_apps; use codex_core_plugins::loader::load_plugin_mcp_servers; -use codex_core_plugins::loader::plugin_telemetry_metadata_from_root; use codex_core_plugins::manifest::PluginManifestInterface; use codex_core_plugins::marketplace::MarketplaceError; use codex_core_plugins::marketplace::MarketplacePluginSource; diff --git a/codex-rs/app-server/src/request_processors/config_processor.rs b/codex-rs/app-server/src/request_processors/config_processor.rs index bdc8013950..9b5eb632ea 100644 --- a/codex-rs/app-server/src/request_processors/config_processor.rs +++ b/codex-rs/app-server/src/request_processors/config_processor.rs @@ -295,15 +295,14 @@ impl ConfigRequestProcessor { &self, pending_changes: std::collections::BTreeMap, ) { + let plugins_manager = self.thread_manager.plugins_manager(); for (plugin_id, enabled) in pending_changes { let Ok(plugin_id) = PluginId::parse(&plugin_id) else { continue; }; - let metadata = codex_core_plugins::loader::installed_plugin_telemetry_metadata( - self.config_manager.codex_home(), - &plugin_id, - ) - .await; + let metadata = plugins_manager + .telemetry_metadata_for_installed_plugin(&plugin_id) + .await; if enabled { self.analytics_events_client.track_plugin_enabled(metadata); } else { diff --git a/codex-rs/app-server/src/request_processors/plugins.rs b/codex-rs/app-server/src/request_processors/plugins.rs index 79791c142a..3afacf95b7 100644 --- a/codex-rs/app-server/src/request_processors/plugins.rs +++ b/codex-rs/app-server/src/request_processors/plugins.rs @@ -1523,9 +1523,14 @@ impl PluginRequestProcessor { Some(self.effective_plugins_changed_callback()), ); - let mut plugin_metadata = - plugin_telemetry_metadata_from_root(&result.plugin_id, &result.installed_path).await; - plugin_metadata.remote_plugin_id = Some(remote_plugin_id); + let plugin_metadata = self + .thread_manager + .plugins_manager() + .telemetry_metadata_for_installed_plugin_with_remote_id( + &result.plugin_id, + &remote_plugin_id, + ) + .await; self.analytics_events_client .track_plugin_installed(plugin_metadata); @@ -1801,11 +1806,27 @@ impl PluginRequestProcessor { let remote_plugin_service_config = RemotePluginServiceConfig { chatgpt_base_url: config.chatgpt_base_url.clone(), }; + let uninstall_target = codex_core_plugins::remote::resolve_remote_plugin_uninstall_target( + &remote_plugin_service_config, + auth.as_ref(), + &plugin_id, + ) + .await + .map_err(|err| { + remote_plugin_catalog_error_to_jsonrpc(err, "resolve remote plugin before uninstall") + })?; + let plugins_manager = self.thread_manager.plugins_manager(); + let plugin_telemetry = plugins_manager + .telemetry_metadata_for_installed_plugin_with_remote_id( + &uninstall_target.plugin_id, + &uninstall_target.remote_plugin_id, + ) + .await; let uninstall_result = codex_core_plugins::remote::uninstall_remote_plugin( &remote_plugin_service_config, auth.as_ref(), config.codex_home.to_path_buf(), - &plugin_id, + uninstall_target, ) .await; @@ -1813,7 +1834,8 @@ impl PluginRequestProcessor { &uninstall_result, Ok(()) | Err(RemotePluginCatalogError::CacheRemove(_)) ) { - let plugins_manager = self.thread_manager.plugins_manager(); + self.analytics_events_client + .track_plugin_uninstalled(plugin_telemetry); if plugins_manager.clear_remote_installed_plugins_cache() { self.on_effective_plugins_changed(); } diff --git a/codex-rs/app-server/tests/suite/v2/plugin_install.rs b/codex-rs/app-server/tests/suite/v2/plugin_install.rs index 2de63f1451..0dd6ce8a64 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_install.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_install.rs @@ -739,6 +739,7 @@ async fn plugin_install_tracks_analytics_event() -> Result<()> { "event_type": "codex_plugin_installed", "event_params": { "plugin_id": "sample-plugin@debug", + "remote_plugin_id": null, "plugin_name": "sample-plugin", "marketplace_name": "debug", "has_skills": false, @@ -791,7 +792,8 @@ async fn plugin_install_tracks_remote_plugin_analytics_event() -> Result<()> { "events": [{ "event_type": "codex_plugin_installed", "event_params": { - "plugin_id": REMOTE_PLUGIN_ID, + "plugin_id": "linear@openai-curated-remote", + "remote_plugin_id": REMOTE_PLUGIN_ID, "plugin_name": "linear", "marketplace_name": "openai-curated-remote", "has_skills": true, diff --git a/codex-rs/app-server/tests/suite/v2/plugin_list.rs b/codex-rs/app-server/tests/suite/v2/plugin_list.rs index d9a0a00493..bfa7be2710 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_list.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_list.rs @@ -3,10 +3,14 @@ use std::time::Duration; use anyhow::Result; use anyhow::bail; use app_test_support::ChatGptAuthFixture; +use app_test_support::DEFAULT_CLIENT_NAME; use app_test_support::TestAppServer; use app_test_support::to_response; use app_test_support::write_chatgpt_auth; +use codex_app_server_protocol::ConfigValueWriteParams; +use codex_app_server_protocol::ConfigWriteResponse; use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::MergeStrategy; use codex_app_server_protocol::PluginAuthPolicy; use codex_app_server_protocol::PluginInstallPolicy; use codex_app_server_protocol::PluginInstalledParams; @@ -19,6 +23,7 @@ use codex_app_server_protocol::PluginShareDiscoverability; use codex_app_server_protocol::PluginSource; use codex_app_server_protocol::PluginSummary; use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::WriteStatus; use codex_config::types::AuthCredentialsStoreMode; use codex_core::config::set_project_trust_level; use codex_protocol::config_types::TrustLevel; @@ -280,6 +285,103 @@ enabled = true Ok(()) } +#[tokio::test] +async fn plugin_toggle_analytics_resolves_remote_plugin_id_from_installed_snapshot() -> Result<()> { + let codex_home = TempDir::new()?; + let server = MockServer::start().await; + write_remote_plugin_catalog_config( + codex_home.path(), + &format!("{}/backend-api/", server.uri()), + )?; + write_chatgpt_auth( + codex_home.path(), + ChatGptAuthFixture::new("chatgpt-token") + .account_id("account-123") + .chatgpt_user_id("user-123") + .chatgpt_account_id("account-123"), + AuthCredentialsStoreMode::File, + )?; + write_installed_plugin_with_version(&codex_home, "openai-curated-remote", "linear", "1.2.3")?; + let global_installed_body = remote_installed_plugin_body("", "1.2.3", /*enabled*/ true); + mount_remote_installed_plugins(&server, "GLOBAL", &global_installed_body).await; + mount_remote_installed_plugins(&server, "WORKSPACE", empty_remote_installed_plugins_body()) + .await; + Mock::given(method("POST")) + .and(path("/backend-api/codex/analytics-events/events")) + .respond_with(ResponseTemplate::new(200).set_body_string(r#"{"status":"ok"}"#)) + .mount(&server) + .await; + + let mut app_server = TestAppServer::new(codex_home.path()).await?; + timeout(DEFAULT_TIMEOUT, app_server.initialize()).await??; + + let installed_request_id = app_server + .send_plugin_installed_request(PluginInstalledParams { + cwds: None, + install_suggestion_plugin_names: None, + }) + .await?; + let installed_response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(installed_request_id)), + ) + .await??; + let _: PluginInstalledResponse = to_response(installed_response)?; + + for enabled in [false, true] { + let request_id = app_server + .send_config_value_write_request(ConfigValueWriteParams { + file_path: None, + key_path: "plugins.linear@openai-curated-remote.enabled".to_string(), + value: serde_json::json!(enabled), + merge_strategy: MergeStrategy::Replace, + expected_version: None, + }) + .await?; + let response: JSONRPCResponse = timeout( + DEFAULT_TIMEOUT, + app_server.read_stream_until_response_message(RequestId::Integer(request_id)), + ) + .await??; + let response: ConfigWriteResponse = to_response(response)?; + assert_eq!(response.status, WriteStatus::Ok); + } + + let events = wait_for_plugin_analytics_events(&server, /*expected_count*/ 2).await?; + assert_eq!( + events, + vec![ + serde_json::json!({ + "event_type": "codex_plugin_disabled", + "event_params": { + "plugin_id": "linear@openai-curated-remote", + "remote_plugin_id": "plugins~Plugin_00000000000000000000000000000000", + "plugin_name": "linear", + "marketplace_name": "openai-curated-remote", + "has_skills": false, + "mcp_server_count": 0, + "connector_ids": [], + "product_client_id": DEFAULT_CLIENT_NAME, + } + }), + serde_json::json!({ + "event_type": "codex_plugin_enabled", + "event_params": { + "plugin_id": "linear@openai-curated-remote", + "remote_plugin_id": "plugins~Plugin_00000000000000000000000000000000", + "plugin_name": "linear", + "marketplace_name": "openai-curated-remote", + "has_skills": false, + "mcp_server_count": 0, + "connector_ids": [], + "product_client_id": DEFAULT_CLIENT_NAME, + } + }), + ] + ); + Ok(()) +} + #[tokio::test] async fn plugin_installed_ignores_local_cache_without_catalog() -> Result<()> { let codex_home = TempDir::new()?; @@ -3415,3 +3517,38 @@ fn write_plugin_share_local_path_mapping( format!("{contents}\n"), ) } + +async fn wait_for_plugin_analytics_events( + server: &MockServer, + expected_count: usize, +) -> Result> { + timeout(DEFAULT_TIMEOUT, async { + loop { + let requests = server.received_requests().await.unwrap_or_default(); + let events = requests + .iter() + .filter(|request| { + request.method == "POST" + && request + .url + .path() + .ends_with("/codex/analytics-events/events") + }) + .filter_map(|request| { + serde_json::from_slice::(&request.body).ok() + }) + .flat_map(|payload| payload["events"].as_array().cloned().unwrap_or_default()) + .filter(|event| { + event["event_type"] + .as_str() + .is_some_and(|event_type| event_type.starts_with("codex_plugin_")) + }) + .collect::>(); + if events.len() >= expected_count { + return Ok::<_, anyhow::Error>(events); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await? +} diff --git a/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs b/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs index 46299d1186..d890ffaf6f 100644 --- a/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs +++ b/codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs @@ -139,6 +139,7 @@ async fn plugin_uninstall_tracks_analytics_event() -> Result<()> { "event_type": "codex_plugin_uninstalled", "event_params": { "plugin_id": "sample-plugin@debug", + "remote_plugin_id": null, "plugin_name": "sample-plugin", "marketplace_name": "debug", "has_skills": false, @@ -216,6 +217,11 @@ async fn plugin_uninstall_writes_remote_plugin_to_cloud_when_remote_plugin_enabl ) .mount(&server) .await; + Mock::given(method("POST")) + .and(path("/backend-api/codex/analytics-events/events")) + .respond_with(ResponseTemplate::new(200).set_body_string(r#"{"status":"ok"}"#)) + .mount(&server) + .await; let remote_plugin_cache_root = codex_home .path() @@ -225,6 +231,11 @@ async fn plugin_uninstall_writes_remote_plugin_to_cloud_when_remote_plugin_enabl remote_plugin_cache_root.join("1.0.0/.codex-plugin/plugin.json"), r#"{"name":"linear","version":"1.0.0"}"#, )?; + std::fs::create_dir_all(remote_plugin_cache_root.join("1.0.0/skills/plan-work"))?; + std::fs::write( + remote_plugin_cache_root.join("1.0.0/skills/plan-work/SKILL.md"), + "---\nname: plan-work\ndescription: Plan work\n---\n", + )?; let legacy_remote_plugin_cache_root = codex_home.path().join(format!( "plugins/cache/openai-curated-remote/{REMOTE_PLUGIN_ID}" )); @@ -255,11 +266,31 @@ async fn plugin_uninstall_writes_remote_plugin_to_cloud_when_remote_plugin_enabl .await?; assert!(!remote_plugin_cache_root.exists()); assert!(!legacy_remote_plugin_cache_root.exists()); + let payload = wait_for_plugin_analytics_payload(&server).await?; + assert_eq!( + payload, + json!({ + "events": [{ + "event_type": "codex_plugin_uninstalled", + "event_params": { + "plugin_id": "linear@openai-curated-remote", + "remote_plugin_id": REMOTE_PLUGIN_ID, + "plugin_name": "linear", + "marketplace_name": "openai-curated-remote", + "has_skills": true, + "mcp_server_count": 0, + "connector_ids": [], + "product_client_id": DEFAULT_CLIENT_NAME, + } + }] + }) + ); Ok(()) } #[tokio::test] -async fn plugin_uninstall_uses_detail_scope_for_cache_namespace() -> Result<()> { +async fn plugin_uninstall_uses_detail_identity_and_scope_for_uninstall_target() -> Result<()> { + const REQUEST_REMOTE_PLUGIN_ID: &str = "plugins~Plugin_linear_alias"; let codex_home = TempDir::new()?; let server = MockServer::start().await; write_remote_plugin_catalog_config( @@ -274,7 +305,15 @@ async fn plugin_uninstall_uses_detail_scope_for_cache_namespace() -> Result<()> .chatgpt_account_id("account-123"), AuthCredentialsStoreMode::File, )?; - mount_remote_plugin_detail(&server, REMOTE_PLUGIN_ID, "1.0.0", "WORKSPACE").await; + mount_remote_plugin_detail_response_with_name( + &server, + REQUEST_REMOTE_PLUGIN_ID, + REMOTE_PLUGIN_ID, + "linear", + "1.0.0", + "WORKSPACE", + ) + .await; Mock::given(method("POST")) .and(path(format!( @@ -307,7 +346,7 @@ async fn plugin_uninstall_uses_detail_scope_for_cache_namespace() -> Result<()> let request_id = mcp .send_plugin_uninstall_request(PluginUninstallParams { - plugin_id: REMOTE_PLUGIN_ID.to_string(), + plugin_id: REQUEST_REMOTE_PLUGIN_ID.to_string(), }) .await?; let response: JSONRPCResponse = timeout( @@ -616,6 +655,25 @@ async fn mount_remote_plugin_detail_with_name( plugin_name: &str, release_version: &str, scope: &str, +) { + mount_remote_plugin_detail_response_with_name( + server, + remote_plugin_id, + remote_plugin_id, + plugin_name, + release_version, + scope, + ) + .await; +} + +async fn mount_remote_plugin_detail_response_with_name( + server: &MockServer, + request_remote_plugin_id: &str, + response_remote_plugin_id: &str, + plugin_name: &str, + release_version: &str, + scope: &str, ) { let discoverability = if scope == "WORKSPACE" { r#" @@ -625,7 +683,7 @@ async fn mount_remote_plugin_detail_with_name( }; let detail_body = format!( r#"{{ - "id": "{remote_plugin_id}", + "id": "{response_remote_plugin_id}", "name": "{plugin_name}", "scope": "{scope}",{discoverability} "installation_policy": "AVAILABLE", @@ -644,7 +702,9 @@ async fn mount_remote_plugin_detail_with_name( ); Mock::given(method("GET")) - .and(path(format!("/backend-api/ps/plugins/{remote_plugin_id}"))) + .and(path(format!( + "/backend-api/ps/plugins/{request_remote_plugin_id}" + ))) .and(header("authorization", "Bearer chatgpt-token")) .and(header("chatgpt-account-id", "account-123")) .respond_with(ResponseTemplate::new(200).set_body_string(detail_body)) @@ -686,3 +746,26 @@ async fn wait_for_remote_plugin_request_count( .await??; Ok(()) } + +async fn wait_for_plugin_analytics_payload(server: &MockServer) -> Result { + timeout(DEFAULT_TIMEOUT, async { + loop { + let Some(requests) = server.received_requests().await else { + tokio::time::sleep(Duration::from_millis(25)).await; + continue; + }; + if let Some(request) = requests.iter().find(|request| { + request.method == "POST" + && request + .url + .path() + .ends_with("/codex/analytics-events/events") + }) { + return serde_json::from_slice(&request.body) + .map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}")); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await? +} diff --git a/codex-rs/core-plugins/src/loader.rs b/codex-rs/core-plugins/src/loader.rs index d6414be6ff..f49e2c07a5 100644 --- a/codex-rs/core-plugins/src/loader.rs +++ b/codex-rs/core-plugins/src/loader.rs @@ -28,7 +28,6 @@ use codex_plugin::PluginHookSource; use codex_plugin::PluginId; use codex_plugin::PluginIdError; use codex_plugin::PluginLoadOutcome; -use codex_plugin::PluginTelemetryMetadata; use codex_protocol::protocol::Product; use codex_protocol::protocol::SkillScope; use codex_utils_absolute_path::AbsolutePathBuf; @@ -1019,13 +1018,11 @@ async fn load_apps_from_paths( connector_ids } -pub async fn plugin_telemetry_metadata_from_root( +pub async fn plugin_capability_summary_from_root( plugin_id: &PluginId, plugin_root: &AbsolutePathBuf, -) -> PluginTelemetryMetadata { - let Some(manifest) = load_plugin_manifest(plugin_root.as_path()) else { - return PluginTelemetryMetadata::from_plugin_id(plugin_id); - }; +) -> Option { + let manifest = load_plugin_manifest(plugin_root.as_path())?; let manifest_paths = &manifest.paths; let has_skills = !plugin_skill_roots(plugin_root, manifest_paths).is_empty(); @@ -1041,22 +1038,18 @@ pub async fn plugin_telemetry_metadata_from_root( mcp_server_names.sort_unstable(); mcp_server_names.dedup(); - PluginTelemetryMetadata { - plugin_id: plugin_id.clone(), - remote_plugin_id: None, - capability_summary: Some(PluginCapabilitySummary { - config_name: plugin_id.as_key(), - display_name: plugin_id.plugin_name.clone(), - description: None, - has_skills, - mcp_server_names, - app_connector_ids: load_apps_from_paths( - plugin_root.as_path(), - plugin_app_config_paths(plugin_root.as_path(), manifest_paths), - ) - .await, - }), - } + Some(PluginCapabilitySummary { + config_name: plugin_id.as_key(), + display_name: plugin_id.plugin_name.clone(), + description: None, + has_skills, + mcp_server_names, + app_connector_ids: load_apps_from_paths( + plugin_root.as_path(), + plugin_app_config_paths(plugin_root.as_path(), manifest_paths), + ) + .await, + }) } pub async fn load_plugin_mcp_servers(plugin_root: &Path) -> HashMap { @@ -1075,24 +1068,6 @@ pub async fn load_plugin_mcp_servers(plugin_root: &Path) -> HashMap PluginTelemetryMetadata { - let store = match PluginStore::try_new(codex_home.to_path_buf()) { - Ok(store) => store, - Err(err) => { - warn!("failed to resolve plugin cache root: {err}"); - return PluginTelemetryMetadata::from_plugin_id(plugin_id); - } - }; - let Some(plugin_root) = store.active_plugin_root(plugin_id) else { - return PluginTelemetryMetadata::from_plugin_id(plugin_id); - }; - - plugin_telemetry_metadata_from_root(plugin_id, &plugin_root).await -} - async fn load_mcp_servers_from_file( plugin_root: &Path, mcp_config_path: &AbsolutePathBuf, diff --git a/codex-rs/core-plugins/src/manager.rs b/codex-rs/core-plugins/src/manager.rs index 1839b06e02..fc8d1c3c5a 100644 --- a/codex-rs/core-plugins/src/manager.rs +++ b/codex-rs/core-plugins/src/manager.rs @@ -4,7 +4,6 @@ use crate::installed_marketplaces::installed_marketplace_roots_from_layer_stack; use crate::loader::PluginHookLoadOutcome; use crate::loader::configured_curated_plugin_ids_from_codex_home; use crate::loader::curated_plugin_cache_version; -use crate::loader::installed_plugin_telemetry_metadata; use crate::loader::load_plugin_apps; use crate::loader::load_plugin_hooks; use crate::loader::load_plugin_hooks_from_layer_stack; @@ -13,7 +12,7 @@ use crate::loader::load_plugin_skills; use crate::loader::load_plugins_from_layer_stack; use crate::loader::log_plugin_load_errors; use crate::loader::materialize_marketplace_plugin_source; -use crate::loader::plugin_telemetry_metadata_from_root; +use crate::loader::plugin_capability_summary_from_root; use crate::loader::refresh_curated_plugin_cache; use crate::loader::refresh_non_curated_plugin_cache; use crate::loader::refresh_non_curated_plugin_cache_force_reinstall; @@ -63,6 +62,7 @@ use codex_plugin::AppConnectorId; use codex_plugin::PluginCapabilitySummary; use codex_plugin::PluginId; use codex_plugin::PluginIdError; +use codex_plugin::PluginTelemetryMetadata; use codex_plugin::prompt_safe_plugin_description; use codex_protocol::protocol::HookEventName; use codex_protocol::protocol::Product; @@ -538,6 +538,61 @@ impl PluginsManager { remote_installed_plugins_to_config(plugins, &self.store) } + fn remote_plugin_id_for(&self, plugin_id: &PluginId) -> Option { + let cache = match self.remote_installed_plugins_cache.read() { + Ok(cache) => cache, + Err(err) => err.into_inner(), + }; + cache.as_ref()?.iter().find_map(|plugin| { + (plugin.name == plugin_id.plugin_name + && plugin.marketplace_name == plugin_id.marketplace_name) + .then(|| plugin.id.clone()) + }) + } + + pub async fn telemetry_metadata_for_installed_plugin( + &self, + plugin_id: &PluginId, + ) -> PluginTelemetryMetadata { + let capability_summary = match self.store.active_plugin_root(plugin_id) { + Some(plugin_root) => plugin_capability_summary_from_root(plugin_id, &plugin_root).await, + None => None, + }; + PluginTelemetryMetadata { + plugin_id: plugin_id.clone(), + remote_plugin_id: self.remote_plugin_id_for(plugin_id), + capability_summary, + } + } + + pub async fn telemetry_metadata_for_installed_plugin_with_remote_id( + &self, + plugin_id: &PluginId, + remote_plugin_id: &str, + ) -> PluginTelemetryMetadata { + let capability_summary = match self.store.active_plugin_root(plugin_id) { + Some(plugin_root) => plugin_capability_summary_from_root(plugin_id, &plugin_root).await, + None => None, + }; + PluginTelemetryMetadata { + plugin_id: plugin_id.clone(), + remote_plugin_id: Some(remote_plugin_id.to_string()), + capability_summary, + } + } + + pub fn telemetry_metadata_for_capability_summary( + &self, + summary: &PluginCapabilitySummary, + ) -> Option { + let plugin_id = PluginId::parse(&summary.config_name).ok()?; + Some(PluginTelemetryMetadata { + remote_plugin_id: self.remote_plugin_id_for(&plugin_id), + plugin_id, + capability_summary: Some(summary.clone()), + }) + } + pub fn build_remote_installed_plugin_marketplaces_from_cache( &self, visible_scopes: &[RemotePluginScope], @@ -875,7 +930,7 @@ impl PluginsManager { }; if let Some(analytics_events_client) = analytics_events_client { analytics_events_client.track_plugin_installed( - plugin_telemetry_metadata_from_root(&result.plugin_id, &result.installed_path) + self.telemetry_metadata_for_installed_plugin(&result.plugin_id) .await, ); } @@ -916,7 +971,10 @@ impl PluginsManager { async fn uninstall_plugin_id(&self, plugin_id: PluginId) -> Result<(), PluginUninstallError> { let plugin_telemetry = if self.store.active_plugin_root(&plugin_id).is_some() { - Some(installed_plugin_telemetry_metadata(self.codex_home.as_path(), &plugin_id).await) + Some( + self.telemetry_metadata_for_installed_plugin(&plugin_id) + .await, + ) } else { None }; diff --git a/codex-rs/core-plugins/src/manager_tests.rs b/codex-rs/core-plugins/src/manager_tests.rs index b1be37655b..96aa291a64 100644 --- a/codex-rs/core-plugins/src/manager_tests.rs +++ b/codex-rs/core-plugins/src/manager_tests.rs @@ -368,6 +368,112 @@ remote_plugin = true assert_eq!(outcome, PluginLoadOutcome::default()); } +#[tokio::test] +async fn installed_plugin_telemetry_metadata_uses_local_identity_without_remote_snapshot() { + let codex_home = TempDir::new().unwrap(); + write_cached_plugin(codex_home.path(), "test", "sample"); + let manager = PluginsManager::new(codex_home.path().to_path_buf()); + let plugin_id = PluginId::parse("sample@test").expect("plugin id should parse"); + + let metadata = manager + .telemetry_metadata_for_installed_plugin(&plugin_id) + .await; + + assert_eq!( + metadata, + PluginTelemetryMetadata { + plugin_id, + remote_plugin_id: None, + capability_summary: Some(PluginCapabilitySummary { + config_name: "sample@test".to_string(), + display_name: "sample".to_string(), + description: None, + has_skills: true, + mcp_server_names: Vec::new(), + app_connector_ids: Vec::new(), + }), + } + ); +} + +#[tokio::test] +async fn installed_plugin_telemetry_metadata_resolves_remote_snapshot_identity() { + let codex_home = TempDir::new().unwrap(); + write_cached_plugin(codex_home.path(), "openai-curated-remote", "linear"); + let manager = PluginsManager::new(codex_home.path().to_path_buf()); + manager.write_remote_installed_plugins_cache(vec![remote_installed_linear_plugin()]); + let plugin_id = + PluginId::parse("linear@openai-curated-remote").expect("plugin id should parse"); + + let metadata = manager + .telemetry_metadata_for_installed_plugin(&plugin_id) + .await; + + assert_eq!( + metadata, + PluginTelemetryMetadata { + plugin_id, + remote_plugin_id: Some("plugins~Plugin_linear".to_string()), + capability_summary: Some(PluginCapabilitySummary { + config_name: "linear@openai-curated-remote".to_string(), + display_name: "linear".to_string(), + description: None, + has_skills: true, + mcp_server_names: Vec::new(), + app_connector_ids: Vec::new(), + }), + } + ); +} + +#[tokio::test] +async fn installed_plugin_telemetry_metadata_accepts_authoritative_remote_identity() { + let codex_home = TempDir::new().unwrap(); + let manager = PluginsManager::new(codex_home.path().to_path_buf()); + let plugin_id = + PluginId::parse("linear@openai-curated-remote").expect("plugin id should parse"); + + let metadata = manager + .telemetry_metadata_for_installed_plugin_with_remote_id(&plugin_id, "plugins~Plugin_linear") + .await; + + assert_eq!( + metadata, + PluginTelemetryMetadata { + plugin_id, + remote_plugin_id: Some("plugins~Plugin_linear".to_string()), + capability_summary: None, + } + ); +} + +#[test] +fn capability_summary_telemetry_metadata_resolves_remote_snapshot_identity() { + let codex_home = TempDir::new().unwrap(); + let manager = PluginsManager::new(codex_home.path().to_path_buf()); + manager.write_remote_installed_plugins_cache(vec![remote_installed_linear_plugin()]); + let summary = PluginCapabilitySummary { + config_name: "linear@openai-curated-remote".to_string(), + display_name: "Linear".to_string(), + description: Some("Track work".to_string()), + has_skills: true, + mcp_server_names: vec!["linear".to_string()], + app_connector_ids: vec![AppConnectorId("linear-app".to_string())], + }; + + let metadata = manager.telemetry_metadata_for_capability_summary(&summary); + + assert_eq!( + metadata, + Some(PluginTelemetryMetadata { + plugin_id: PluginId::parse("linear@openai-curated-remote") + .expect("plugin id should parse"), + remote_plugin_id: Some("plugins~Plugin_linear".to_string()), + capability_summary: Some(summary), + }) + ); +} + #[tokio::test] async fn remote_installed_cache_prefers_local_curated_conflicts_when_remote_plugin_disabled() { let codex_home = TempDir::new().unwrap(); @@ -638,14 +744,14 @@ async fn plugin_telemetry_metadata_uses_default_mcp_config_path() { }"#, ); - let metadata = plugin_telemetry_metadata_from_root( + let summary = plugin_capability_summary_from_root( &PluginId::parse("sample@test").expect("plugin id should parse"), &plugin_root.abs(), ) .await; assert_eq!( - metadata.capability_summary, + summary, Some(PluginCapabilitySummary { config_name: "sample@test".to_string(), display_name: "sample".to_string(), diff --git a/codex-rs/core-plugins/src/remote.rs b/codex-rs/core-plugins/src/remote.rs index b96746b637..701f9b95c0 100644 --- a/codex-rs/core-plugins/src/remote.rs +++ b/codex-rs/core-plugins/src/remote.rs @@ -99,6 +99,12 @@ pub struct RemotePluginServiceConfig { pub chatgpt_base_url: String, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct RemotePluginUninstallTarget { + pub plugin_id: PluginId, + pub remote_plugin_id: String, +} + #[derive(Debug, Clone, PartialEq)] pub struct RemoteMarketplace { pub name: String, @@ -1068,42 +1074,64 @@ pub async fn install_remote_plugin( Ok(()) } +pub async fn resolve_remote_plugin_uninstall_target( + config: &RemotePluginServiceConfig, + auth: Option<&CodexAuth>, + remote_plugin_id: &str, +) -> Result { + let auth = ensure_chatgpt_auth(auth)?; + let plugin = fetch_plugin_detail( + config, + auth, + remote_plugin_id, + /*include_download_urls*/ false, + ) + .await?; + let marketplace_name = remote_plugin_canonical_marketplace_name(&plugin)?.to_string(); + let remote_plugin_id = plugin.id; + let plugin_id = PluginId::new(plugin.name, marketplace_name).map_err(|err| { + RemotePluginCatalogError::UnexpectedResponse(format!( + "invalid local plugin id for remote plugin `{remote_plugin_id}`: {err}" + )) + })?; + Ok(RemotePluginUninstallTarget { + plugin_id, + remote_plugin_id, + }) +} + pub async fn uninstall_remote_plugin( config: &RemotePluginServiceConfig, auth: Option<&CodexAuth>, codex_home: PathBuf, - plugin_id: &str, + target: RemotePluginUninstallTarget, ) -> Result<(), RemotePluginCatalogError> { let auth = ensure_chatgpt_auth(auth)?; - let plugin = fetch_plugin_detail( - config, auth, plugin_id, /*include_download_urls*/ false, - ) - .await?; - let marketplace_name = remote_plugin_canonical_marketplace_name(&plugin)?.to_string(); - let plugin_name = plugin.name; - + let RemotePluginUninstallTarget { + plugin_id, + remote_plugin_id, + } = target; let base_url = config.chatgpt_base_url.trim_end_matches('/'); - let url = format!("{base_url}/plugins/{plugin_id}/uninstall"); + let url = format!("{base_url}/plugins/{remote_plugin_id}/uninstall"); let client = build_reqwest_client(); let request = authenticated_request(client.post(&url), auth)?; let response: RemotePluginMutationResponse = send_and_decode(request, &url).await?; - if response.id != plugin_id { + if response.id != remote_plugin_id { return Err(RemotePluginCatalogError::UnexpectedPluginId { - expected: plugin_id.to_string(), + expected: remote_plugin_id, actual: response.id, }); } if response.enabled { return Err(RemotePluginCatalogError::UnexpectedEnabledState { - plugin_id: plugin_id.to_string(), + plugin_id: response.id, expected_enabled: false, actual_enabled: response.enabled, }); } - let legacy_plugin_id = plugin_id.to_string(); tokio::task::spawn_blocking(move || { - remove_remote_plugin_cache(codex_home, marketplace_name, plugin_name, legacy_plugin_id) + remove_remote_plugin_cache(codex_home, plugin_id, response.id) }) .await .map_err(|err| { @@ -1118,18 +1146,11 @@ pub async fn uninstall_remote_plugin( fn remove_remote_plugin_cache( codex_home: PathBuf, - marketplace_name: String, - plugin_name: String, + plugin_id: PluginId, legacy_plugin_id: String, ) -> Result<(), String> { let store = PluginStore::try_new(codex_home.clone()) .map_err(|err| format!("failed to resolve remote plugin cache root: {err}"))?; - let plugin_id = - PluginId::new(plugin_name.clone(), marketplace_name.clone()).map_err(|err| { - format!( - "invalid remote plugin cache id for `{plugin_name}` in `{marketplace_name}`: {err}" - ) - })?; let plugin_cache_root = store.plugin_base_root(&plugin_id); store.uninstall(&plugin_id).map_err(|err| { format!( @@ -1140,7 +1161,7 @@ fn remove_remote_plugin_cache( let legacy_remote_plugin_cache_root = codex_home .join(PLUGINS_CACHE_DIR) - .join(marketplace_name) + .join(plugin_id.marketplace_name) .join(legacy_plugin_id); if legacy_remote_plugin_cache_root != plugin_cache_root.as_path() && legacy_remote_plugin_cache_root.exists() diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 9d4e7d1e82..a69ba7301e 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -561,13 +561,16 @@ async fn build_skills_and_plugins( sess.services .analytics_events_client .track_app_mentioned(tracking.clone(), mentioned_app_invocations); - for plugin in mentioned_plugins - .iter() - .filter_map(crate::plugins::PluginCapabilitySummary::telemetry_metadata) - { - sess.services - .analytics_events_client - .track_plugin_used(tracking.clone(), plugin); + for summary in &mentioned_plugins { + if let Some(plugin) = sess + .services + .plugins_manager + .telemetry_metadata_for_capability_summary(summary) + { + sess.services + .analytics_events_client + .track_plugin_used(tracking.clone(), plugin); + } } let mut injection_items: Vec = match injected_host_skill_prompts { diff --git a/codex-rs/core/tests/suite/plugins.rs b/codex-rs/core/tests/suite/plugins.rs index 03b1afdfaa..514291c5d5 100644 --- a/codex-rs/core/tests/suite/plugins.rs +++ b/codex-rs/core/tests/suite/plugins.rs @@ -6,6 +6,7 @@ use std::time::Duration; use std::time::Instant; use anyhow::Result; +use codex_core_plugins::remote::RemotePluginScope; use codex_features::Feature; use codex_login::CodexAuth; use codex_protocol::protocol::EventMsg; @@ -23,11 +24,18 @@ use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use core_test_support::wait_for_mcp_server; use tempfile::TempDir; +use wiremock::Mock; use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path; +use wiremock::matchers::query_param; const SAMPLE_PLUGIN_CONFIG_NAME: &str = "sample@test"; const SAMPLE_PLUGIN_DISPLAY_NAME: &str = "sample"; const SAMPLE_PLUGIN_DESCRIPTION: &str = "inspect sample data"; +const REMOTE_SAMPLE_PLUGIN_CONFIG_NAME: &str = "sample@openai-curated-remote"; +const REMOTE_SAMPLE_PLUGIN_ID: &str = "plugins~Plugin_sample"; fn sample_plugin_root(home: &TempDir) -> std::path::PathBuf { home.path().join("plugins/cache/test/sample/local") @@ -65,6 +73,33 @@ fn write_plugin_skill_plugin(home: &TempDir) -> std::path::PathBuf { skill_dir.join("SKILL.md") } +fn write_remote_plugin_skill_plugin(home: &TempDir) { + let plugin_root = home + .path() + .join("plugins/cache/openai-curated-remote/sample/local"); + std::fs::create_dir_all(plugin_root.join(".codex-plugin")) + .expect("create remote plugin manifest dir"); + std::fs::write( + plugin_root.join(".codex-plugin/plugin.json"), + r#"{"name":"sample","description":"inspect sample data"}"#, + ) + .expect("write remote plugin manifest"); + let skill_dir = plugin_root.join("skills/sample-search"); + std::fs::create_dir_all(&skill_dir).expect("create remote plugin skill dir"); + std::fs::write( + skill_dir.join("SKILL.md"), + "---\ndescription: inspect sample data\n---\n\n# body\n", + ) + .expect("write remote plugin skill"); + std::fs::write( + home.path().join("config.toml"), + format!( + "[features]\nplugins = true\nremote_plugin = true\n\n[plugins.\"{REMOTE_SAMPLE_PLUGIN_CONFIG_NAME}\"]\nenabled = true\n" + ), + ) + .expect("write remote plugin config"); +} + fn write_plugin_mcp_plugin(home: &TempDir, command: &str) { let plugin_root = write_sample_plugin_manifest_and_config(home); std::fs::write( @@ -358,32 +393,12 @@ async fn explicit_plugin_mentions_track_plugin_used_analytics() -> Result<()> { .await?; wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; - let deadline = Instant::now() + Duration::from_secs(10); - let plugin_event = loop { - let requests = server.received_requests().await.unwrap_or_default(); - if let Some(event) = requests - .into_iter() - .filter(|request| request.url.path() == "/codex/analytics-events/events") - .find_map(|request| { - let payload: serde_json::Value = serde_json::from_slice(&request.body).ok()?; - payload["events"].as_array().and_then(|events| { - events - .iter() - .find(|event| event["event_type"] == "codex_plugin_used") - .cloned() - }) - }) - { - break event; - } - if Instant::now() >= deadline { - panic!("timed out waiting for plugin analytics request"); - } - tokio::time::sleep(Duration::from_millis(50)).await; - }; - - let event = plugin_event; + let event = wait_for_plugin_used_analytics_event(&server).await; assert_eq!(event["event_params"]["plugin_id"], "sample@test"); + assert_eq!( + event["event_params"]["remote_plugin_id"], + serde_json::Value::Null + ); assert_eq!(event["event_params"]["plugin_name"], "sample"); assert_eq!(event["event_params"]["marketplace_name"], "test"); assert_eq!(event["event_params"]["has_skills"], true); @@ -406,3 +421,122 @@ async fn explicit_plugin_mentions_track_plugin_used_analytics() -> Result<()> { Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn explicit_remote_plugin_mentions_track_remote_plugin_id() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let _resp_mock = mount_sse_once( + &server, + sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), + ) + .await; + let global_installed_body = format!( + r#"{{ + "plugins": [{{ + "id": "{REMOTE_SAMPLE_PLUGIN_ID}", + "name": "sample", + "scope": "GLOBAL", + "installation_policy": "AVAILABLE", + "authentication_policy": "ON_USE", + "release": {{ + "version": "1.0.0", + "display_name": "Sample", + "description": "Inspect sample data", + "app_ids": [], + "interface": {{}}, + "skills": [] + }}, + "enabled": true, + "disabled_skill_names": [] + }}], + "pagination": {{"limit": 50, "next_page_token": null}} +}}"# + ); + Mock::given(method("GET")) + .and(path("/ps/plugins/installed")) + .and(query_param("scope", "GLOBAL")) + .respond_with(ResponseTemplate::new(200).set_body_string(global_installed_body)) + .mount(&server) + .await; + Mock::given(method("GET")) + .and(path("/ps/plugins/installed")) + .and(query_param("scope", "WORKSPACE")) + .respond_with( + ResponseTemplate::new(200).set_body_string( + r#"{"plugins":[],"pagination":{"limit":50,"next_page_token":null}}"#, + ), + ) + .mount(&server) + .await; + + let codex_home = Arc::new(TempDir::new()?); + write_remote_plugin_skill_plugin(codex_home.as_ref()); + let test_codex = build_analytics_plugin_test_codex(&server, codex_home).await?; + let auth = CodexAuth::create_dummy_chatgpt_auth_for_testing(); + test_codex + .thread_manager + .plugins_manager() + .build_and_cache_remote_installed_plugin_marketplaces( + &test_codex.config.plugins_config_input(), + Some(&auth), + &[RemotePluginScope::Global], + /*on_effective_plugins_changed*/ None, + ) + .await?; + let codex = Arc::clone(&test_codex.codex); + + codex + .submit(Op::UserInput { + environments: None, + items: vec![codex_protocol::user_input::UserInput::Mention { + name: "sample".into(), + path: format!("plugin://{REMOTE_SAMPLE_PLUGIN_CONFIG_NAME}"), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + + let event = wait_for_plugin_used_analytics_event(&server).await; + assert_eq!( + event["event_params"]["plugin_id"], + REMOTE_SAMPLE_PLUGIN_CONFIG_NAME + ); + assert_eq!( + event["event_params"]["remote_plugin_id"], + REMOTE_SAMPLE_PLUGIN_ID + ); + assert_eq!(event["event_params"]["has_skills"], true); + + Ok(()) +} + +async fn wait_for_plugin_used_analytics_event(server: &MockServer) -> serde_json::Value { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + let requests = server.received_requests().await.unwrap_or_default(); + if let Some(event) = requests + .into_iter() + .filter(|request| request.url.path() == "/codex/analytics-events/events") + .find_map(|request| { + let payload: serde_json::Value = serde_json::from_slice(&request.body).ok()?; + payload["events"].as_array().and_then(|events| { + events + .iter() + .find(|event| event["event_type"] == "codex_plugin_used") + .cloned() + }) + }) + { + return event; + } + if Instant::now() >= deadline { + panic!("timed out waiting for plugin analytics request"); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } +} diff --git a/codex-rs/plugin/src/lib.rs b/codex-rs/plugin/src/lib.rs index 92a2ace21a..84d5583b14 100644 --- a/codex-rs/plugin/src/lib.rs +++ b/codex-rs/plugin/src/lib.rs @@ -41,31 +41,9 @@ pub struct PluginHookSource { #[derive(Debug, Clone, PartialEq, Eq)] pub struct PluginTelemetryMetadata { + /// Local plugin identifier used by Codex configuration and the plugin cache. pub plugin_id: PluginId, - /// Optional backend identifier for remote plugins, used when analytics - /// should report the remote id instead of the local plugin cache id. + /// Optional backend identifier for remote plugins. pub remote_plugin_id: Option, pub capability_summary: Option, } - -impl PluginTelemetryMetadata { - pub fn from_plugin_id(plugin_id: &PluginId) -> Self { - Self { - plugin_id: plugin_id.clone(), - remote_plugin_id: None, - capability_summary: None, - } - } -} - -impl PluginCapabilitySummary { - pub fn telemetry_metadata(&self) -> Option { - PluginId::parse(&self.config_name) - .ok() - .map(|plugin_id| PluginTelemetryMetadata { - plugin_id, - remote_plugin_id: None, - capability_summary: Some(self.clone()), - }) - } -}