From d6eefb26a6d3f610372a4ea4b8a59a2e382c731f Mon Sep 17 00:00:00 2001 From: Kyle Brown Date: Wed, 12 Aug 2026 19:03:57 +0000 Subject: [PATCH] Add bounded plugin measurement analytics (#38239) ## What changed - Add `PluginMeasurementsInput` and `PluginMeasurementRow`, plus an analytics client method for recording batches and checking whether analytics are enabled. - Emit each valid row as a `codex_plugin_measurement_event` with thread, turn, item, plugin, execution, and operation context. - Bound batches at 100 rows and reject non-finite values, invalid identifiers, and rows with more than eight dimensions. - Treat measurement events as plugin events when routing API-key analytics. ## Testing - Cover batch limits, row filtering, serialized event payloads, and API-key plugin-event routing. GitOrigin-RevId: f949ed06246ee1e22466df2660076f2d4f556444 --- .../analytics/src/analytics_client_tests.rs | 87 ++++++++++++++++++ codex-rs/analytics/src/client.rs | 24 +++++ codex-rs/analytics/src/client_tests.rs | 92 +++++++++++++++++++ codex-rs/analytics/src/events.rs | 22 +++++ codex-rs/analytics/src/facts.rs | 20 ++++ codex-rs/analytics/src/lib.rs | 2 + codex-rs/analytics/src/reducer.rs | 71 ++++++++++++++ 7 files changed, 318 insertions(+) diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 8895c9cfd6..f609449de3 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -77,6 +77,8 @@ use crate::facts::PluginInstallRequested; use crate::facts::PluginInstallRequestedInput; use crate::facts::PluginInstallRequestedPlugin; use crate::facts::PluginInstallSource; +use crate::facts::PluginMeasurementRow; +use crate::facts::PluginMeasurementsInput; use crate::facts::PluginState; use crate::facts::PluginStateChangedInput; use crate::facts::PluginUsedInput; @@ -182,6 +184,7 @@ use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use pretty_assertions::assert_eq; use serde_json::json; +use std::collections::BTreeMap; use std::collections::HashSet; use std::path::PathBuf; use std::sync::Arc; @@ -813,6 +816,18 @@ async fn ingest_completed_command_execution_item( .await; } +fn plugin_measurements(rows: Vec) -> PluginMeasurementsInput { + PluginMeasurementsInput { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + item_id: "item-1".to_string(), + plugin_id: "sample@openai-curated".to_string(), + execution_id: "execution-1".to_string(), + operation: "security_scan".to_string(), + rows, + } +} + fn sample_initialize_fact(connection_id: u64) -> AnalyticsFact { AnalyticsFact::Initialize { connection_id, @@ -2597,6 +2612,78 @@ async fn item_lifecycle_notifications_publish_command_execution_event() { assert_eq!(payload[0]["event_params"]["thread_source"], "user"); } +#[tokio::test] +async fn plugin_measurement_batch_emits_directly_and_filters_invalid_rows() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + let mut too_many_dimensions = BTreeMap::new(); + for index in 0..9 { + too_many_dimensions.insert(format!("dimension_{index}"), "allowed".to_string()); + } + let measurements = plugin_measurements(vec![ + PluginMeasurementRow { + measurement_name: "finding_count".to_string(), + number_value: 3.0, + dimensions: BTreeMap::from([("severity".to_string(), "high".to_string())]), + }, + PluginMeasurementRow { + measurement_name: "non_finite".to_string(), + number_value: f64::NAN, + dimensions: BTreeMap::new(), + }, + PluginMeasurementRow { + measurement_name: "too_many_dimensions".to_string(), + number_value: 1.0, + dimensions: too_many_dimensions, + }, + PluginMeasurementRow { + measurement_name: "files_scanned".to_string(), + number_value: 17.0, + dimensions: BTreeMap::new(), + }, + ]); + reducer + .ingest( + AnalyticsFact::Custom(CustomAnalyticsFact::PluginMeasurements(measurements)), + &mut events, + ) + .await; + let payload = serde_json::to_value(&events).expect("serialize events"); + assert_eq!( + payload, + json!([ + { + "event_type": "codex_plugin_measurement_event", + "event_params": { + "thread_id": "thread-1", + "turn_id": "turn-1", + "item_id": "item-1", + "plugin_id": "sample@openai-curated", + "execution_id": "execution-1", + "operation": "security_scan", + "measurement_name": "finding_count", + "number_value": 3.0, + "dimensions": {"severity": "high"}, + }, + }, + { + "event_type": "codex_plugin_measurement_event", + "event_params": { + "thread_id": "thread-1", + "turn_id": "turn-1", + "item_id": "item-1", + "plugin_id": "sample@openai-curated", + "execution_id": "execution-1", + "operation": "security_scan", + "measurement_name": "files_scanned", + "number_value": 17.0, + "dimensions": null, + }, + }, + ]) + ); +} + #[tokio::test] async fn command_execution_approval_response_publishes_user_review_event() { let mut reducer = AnalyticsReducer::default(); diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 5b081e53b3..6791aee084 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -22,6 +22,7 @@ use crate::facts::PluginInstallFailedInput; use crate::facts::PluginInstallRequested; use crate::facts::PluginInstallRequestedInput; use crate::facts::PluginInstallSource; +use crate::facts::PluginMeasurementsInput; use crate::facts::PluginState; use crate::facts::PluginStateChangedInput; use crate::facts::SkillInvocation; @@ -34,6 +35,9 @@ use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTokenUsageFact; use crate::now_unix_millis; use crate::reducer::AnalyticsReducer; +use crate::reducer::MAX_PLUGIN_MEASUREMENTS_PER_BATCH; +use crate::reducer::valid_plugin_measurement_identifier; +use crate::reducer::valid_plugin_measurement_row; use codex_app_server_protocol::ClientRequest; use codex_app_server_protocol::ClientResponsePayload; use codex_app_server_protocol::InitializeParams; @@ -255,6 +259,26 @@ impl AnalyticsEventsClient { } } + pub fn is_enabled(&self) -> bool { + self.queue.is_some() + } + + pub fn track_plugin_measurements(&self, mut input: PluginMeasurementsInput) { + if input.rows.is_empty() + || input.rows.len() > MAX_PLUGIN_MEASUREMENTS_PER_BATCH + || !valid_plugin_measurement_identifier(&input.operation) + { + return; + } + input.rows.retain(valid_plugin_measurement_row); + if input.rows.is_empty() { + return; + } + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::PluginMeasurements(input), + )); + } + pub fn track_skill_invocations( &self, tracking: TrackEventsContext, diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index b89a3bf422..c3609045e9 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -20,6 +20,10 @@ use crate::events::CodexMcpToolCallEventParams; #[cfg(debug_assertions)] use crate::events::CodexMcpToolCallEventRequest; #[cfg(debug_assertions)] +use crate::events::CodexPluginMeasurementEventParams; +#[cfg(debug_assertions)] +use crate::events::CodexPluginMeasurementEventRequest; +#[cfg(debug_assertions)] use crate::events::CodexPluginMetadata; #[cfg(debug_assertions)] use crate::events::CodexPluginUsedEventRequest; @@ -49,9 +53,13 @@ use crate::facts::AnalyticsFact; use crate::facts::ArtifactOperation; #[cfg(debug_assertions)] use crate::facts::ArtifactOperationLifecycle; +use crate::facts::CustomAnalyticsFact; use crate::facts::InvocationType; +use crate::facts::PluginMeasurementRow; +use crate::facts::PluginMeasurementsInput; #[cfg(debug_assertions)] use crate::facts::TrackEventsContext; +use crate::reducer::MAX_PLUGIN_MEASUREMENTS_PER_BATCH; use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientRequest; @@ -83,6 +91,7 @@ use codex_login::AuthManager; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use pretty_assertions::assert_eq; +use std::collections::BTreeMap; use std::collections::HashSet; #[cfg(debug_assertions)] use std::fs; @@ -407,6 +416,22 @@ async fn api_key_auth_sends_only_plugin_events_to_codex_backend() { let auth_manager = codex_login::AuthManager::from_auth_for_testing( codex_login::CodexAuth::from_api_key("sk-test"), ); + let plugin_measurement = |thread_id: &str, plugin_id: &str| { + TrackEventRequest::PluginMeasurement(CodexPluginMeasurementEventRequest { + event_type: "codex_plugin_measurement_event", + event_params: CodexPluginMeasurementEventParams { + thread_id: thread_id.to_string(), + turn_id: "turn-1".to_string(), + item_id: "item-1".to_string(), + plugin_id: plugin_id.to_string(), + execution_id: "execution-1".to_string(), + operation: "security_scan".to_string(), + measurement_name: "findings".to_string(), + number_value: 1.0, + dimensions: None, + }, + }) + }; send_track_events( &auth_manager, @@ -415,6 +440,7 @@ async fn api_key_auth_sends_only_plugin_events_to_codex_backend() { sample_regular_track_event("non-plugin-skill"), sample_mcp_tool_call_event("non-plugin-mcp", /*plugin_id*/ None), sample_plugin_used_track_event("non-plugin-used", /*plugin_id*/ None), + plugin_measurement("non-plugin-measurement", /*plugin_id*/ ""), sample_accepted_line_fingerprint_event("other-event"), TrackEventRequest::ThreadArchive(ThreadArchiveEvent { event_type: "codex_thread_archive_event", @@ -428,6 +454,7 @@ async fn api_key_auth_sends_only_plugin_events_to_codex_backend() { sample_skill_track_event("plugin-skill", Some("sample@test")), sample_mcp_tool_call_event("plugin-mcp", Some("sample@test")), sample_artifact_operation_event("plugin-artifact"), + plugin_measurement("plugin-measurement", "sample@test"), ], ) .await; @@ -482,6 +509,11 @@ async fn api_key_auth_sends_only_plugin_events_to_codex_backend() { "plugin_id": "presentations@openai-primary-runtime", "thread_id": "plugin-artifact", }), + serde_json::json!({ + "event_type": "codex_plugin_measurement_event", + "plugin_id": "sample@test", + "thread_id": "plugin-measurement", + }), ] ); @@ -658,6 +690,66 @@ fn sample_turn_steer_response() -> ClientResponsePayload { }) } +#[test] +fn track_plugin_measurements_rejects_unbounded_inputs_before_queueing() { + let (client, mut receiver) = client_with_receiver(); + let measurements = |row_count| PluginMeasurementsInput { + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + item_id: "item-1".to_string(), + plugin_id: "sample@openai-curated".to_string(), + execution_id: "execution-1".to_string(), + operation: "security_scan".to_string(), + rows: vec![ + PluginMeasurementRow { + measurement_name: "finding_count".to_string(), + number_value: 1.0, + dimensions: BTreeMap::new(), + }; + row_count + ], + }; + + client.track_plugin_measurements(measurements(MAX_PLUGIN_MEASUREMENTS_PER_BATCH + 1)); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); + + let mut oversized_operation = measurements(1); + oversized_operation.operation = "o".repeat(65); + client.track_plugin_measurements(oversized_operation); + assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); + + let mut mixed_rows = measurements(4); + mixed_rows.rows[0].measurement_name = "m".repeat(65); + mixed_rows.rows[1] + .dimensions + .insert("d".repeat(65), "valid".to_string()); + mixed_rows.rows[2] + .dimensions + .insert("valid".to_string(), "v".repeat(65)); + client.track_plugin_measurements(mixed_rows); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsEventsQueueMessage::Fact(fact)) + if matches!( + fact.as_ref(), + AnalyticsFact::Custom(CustomAnalyticsFact::PluginMeasurements(input)) + if input.rows.len() == 1 + && input.rows[0].measurement_name == "finding_count" + ) + )); + + client.track_plugin_measurements(measurements(MAX_PLUGIN_MEASUREMENTS_PER_BATCH)); + assert!(matches!( + receiver.try_recv(), + Ok(AnalyticsEventsQueueMessage::Fact(fact)) + if matches!( + fact.as_ref(), + AnalyticsFact::Custom(CustomAnalyticsFact::PluginMeasurements(input)) + if input.rows.len() == MAX_PLUGIN_MEASUREMENTS_PER_BATCH + ) + )); +} + #[test] fn track_request_only_enqueues_analytics_relevant_requests() { let (client, mut receiver) = client_with_receiver(); diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index f3ca378b6b..34c5e35997 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -46,6 +46,7 @@ use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::ThreadSource; use codex_protocol::protocol::TokenUsage; use serde::Serialize; +use std::collections::BTreeMap; #[derive(Clone, Copy, Debug, Serialize)] #[serde(rename_all = "snake_case")] @@ -76,6 +77,7 @@ pub(crate) enum TrackEventRequest { TurnSteer(CodexTurnSteerEventRequest), ArtifactOperation(CodexArtifactOperationEventRequest), CommandExecution(CodexCommandExecutionEventRequest), + PluginMeasurement(CodexPluginMeasurementEventRequest), FileChange(CodexFileChangeEventRequest), McpToolCall(CodexMcpToolCallEventRequest), DynamicToolCall(CodexDynamicToolCallEventRequest), @@ -160,6 +162,7 @@ impl TrackEventRequest { Self::SkillInvocation(event) => event.event_params.plugin_id.is_some(), Self::McpToolCall(event) => event.event_params.plugin_id.is_some(), Self::ArtifactOperation(event) => !event.event_params.plugin_id.is_empty(), + Self::PluginMeasurement(event) => !event.event_params.plugin_id.is_empty(), _ => false, } } @@ -750,6 +753,25 @@ pub(crate) struct CodexCommandExecutionEventRequest { pub(crate) event_params: CodexCommandExecutionEventParams, } +#[derive(Serialize)] +pub(crate) struct CodexPluginMeasurementEventParams { + pub(crate) thread_id: String, + pub(crate) turn_id: String, + pub(crate) item_id: String, + pub(crate) plugin_id: String, + pub(crate) execution_id: String, + pub(crate) operation: String, + pub(crate) measurement_name: String, + pub(crate) number_value: f64, + pub(crate) dimensions: Option>, +} + +#[derive(Serialize)] +pub(crate) struct CodexPluginMeasurementEventRequest { + pub(crate) event_type: &'static str, + pub(crate) event_params: CodexPluginMeasurementEventParams, +} + #[derive(Serialize)] pub(crate) struct CodexFileChangeEventParams { #[serde(flatten)] diff --git a/codex-rs/analytics/src/facts.rs b/codex-rs/analytics/src/facts.rs index 8d2ad0ceef..74c1fe67e8 100644 --- a/codex-rs/analytics/src/facts.rs +++ b/codex-rs/analytics/src/facts.rs @@ -29,6 +29,7 @@ use codex_protocol::protocol::SubAgentSource; use codex_protocol::protocol::TokenUsage; use codex_protocol::request_permissions::RequestPermissionsResponse; use serde::Serialize; +use std::collections::BTreeMap; use std::path::PathBuf; #[derive(Clone, Debug, PartialEq, Eq, Serialize)] @@ -541,6 +542,7 @@ pub(crate) enum CustomAnalyticsFact { PluginInstallRequested(PluginInstallRequestedInput), PluginStateChanged(PluginStateChangedInput), PluginInstallFailed(PluginInstallFailedInput), + PluginMeasurements(PluginMeasurementsInput), ExternalAgentConfigImportCompleted(ExternalAgentConfigImportCompletedInput), ExternalAgentConfigImportFailure(ExternalAgentConfigImportFailureInput), } @@ -550,6 +552,24 @@ pub(crate) struct ArtifactOperationInput { pub operation: ArtifactOperation, } +#[derive(Clone, Debug, PartialEq)] +pub struct PluginMeasurementRow { + pub measurement_name: String, + pub number_value: f64, + pub dimensions: BTreeMap, +} + +#[derive(Clone, Debug, PartialEq)] +pub struct PluginMeasurementsInput { + pub thread_id: String, + pub turn_id: String, + pub item_id: String, + pub plugin_id: String, + pub execution_id: String, + pub operation: String, + pub rows: Vec, +} + pub(crate) struct SkillInvokedInput { pub tracking: TrackEventsContext, pub invocations: Vec, diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 99950a040b..af64fdc1b0 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -53,6 +53,8 @@ pub use facts::PluginInstallRequestSource; pub use facts::PluginInstallRequested; pub use facts::PluginInstallRequestedPlugin; pub use facts::PluginInstallSource; +pub use facts::PluginMeasurementRow; +pub use facts::PluginMeasurementsInput; pub use facts::SkillInvocation; pub use facts::SkillInvocationLocation; pub use facts::SubAgentThreadStartedInput; diff --git a/codex-rs/analytics/src/reducer.rs b/codex-rs/analytics/src/reducer.rs index bb55a94ae5..0b05cfc4e9 100644 --- a/codex-rs/analytics/src/reducer.rs +++ b/codex-rs/analytics/src/reducer.rs @@ -29,6 +29,8 @@ use crate::events::CodexPluginEventRequest; use crate::events::CodexPluginInstallFailedEventRequest; use crate::events::CodexPluginInstallFailedMetadata; use crate::events::CodexPluginInstallRequestedEventRequest; +use crate::events::CodexPluginMeasurementEventParams; +use crate::events::CodexPluginMeasurementEventRequest; use crate::events::CodexPluginUsedEventRequest; use crate::events::CodexReviewEventParams; use crate::events::CodexReviewEventRequest; @@ -89,6 +91,8 @@ use crate::facts::ImagePreparationMetadata; use crate::facts::InvocationType; use crate::facts::PluginInstallFailedInput; use crate::facts::PluginInstallRequestedInput; +use crate::facts::PluginMeasurementRow; +use crate::facts::PluginMeasurementsInput; use crate::facts::PluginState; use crate::facts::PluginStateChangedInput; use crate::facts::PluginUsedInput; @@ -161,6 +165,28 @@ use std::path::Path; use std::path::PathBuf; const MAX_TOOL_RESPONSE_ENTRIES: usize = 256; +pub(crate) const MAX_PLUGIN_MEASUREMENTS_PER_BATCH: usize = 100; +const MAX_PLUGIN_MEASUREMENT_DIMENSIONS: usize = 8; +const MAX_PLUGIN_MEASUREMENT_IDENTIFIER_BYTES: usize = 64; + +pub(crate) fn valid_plugin_measurement_identifier(value: &str) -> bool { + let mut characters = value.chars(); + matches!(characters.next(), Some('a'..='z')) + && value.len() <= MAX_PLUGIN_MEASUREMENT_IDENTIFIER_BYTES + && characters.all(|character| { + character.is_ascii_lowercase() || character.is_ascii_digit() || character == '_' + }) +} + +pub(crate) fn valid_plugin_measurement_row(row: &PluginMeasurementRow) -> bool { + row.number_value.is_finite() + && valid_plugin_measurement_identifier(&row.measurement_name) + && row.dimensions.len() <= MAX_PLUGIN_MEASUREMENT_DIMENSIONS + && row.dimensions.iter().all(|(name, value)| { + valid_plugin_measurement_identifier(name) && valid_plugin_measurement_identifier(value) + }) +} + #[derive(Default)] pub(crate) struct AnalyticsReducer { requests: HashMap<(u64, RequestId), RequestState>, @@ -613,6 +639,9 @@ impl AnalyticsReducer { CustomAnalyticsFact::PluginInstallFailed(input) => { self.ingest_plugin_install_failed(input, out); } + CustomAnalyticsFact::PluginMeasurements(input) => { + self.ingest_plugin_measurements(input, out); + } CustomAnalyticsFact::ExternalAgentConfigImportCompleted(input) => { self.ingest_external_agent_config_import_completed(input, out); } @@ -1784,6 +1813,48 @@ impl AnalyticsReducer { } } + fn ingest_plugin_measurements( + &mut self, + input: PluginMeasurementsInput, + out: &mut Vec, + ) { + if input.rows.is_empty() + || input.rows.len() > MAX_PLUGIN_MEASUREMENTS_PER_BATCH + || !valid_plugin_measurement_identifier(&input.operation) + { + return; + } + let PluginMeasurementsInput { + thread_id, + turn_id, + item_id, + plugin_id, + execution_id, + operation, + rows, + } = input; + out.extend( + rows.into_iter() + .filter(valid_plugin_measurement_row) + .map(|row| { + TrackEventRequest::PluginMeasurement(CodexPluginMeasurementEventRequest { + event_type: "codex_plugin_measurement_event", + event_params: CodexPluginMeasurementEventParams { + thread_id: thread_id.clone(), + turn_id: turn_id.clone(), + item_id: item_id.clone(), + plugin_id: plugin_id.clone(), + execution_id: execution_id.clone(), + operation: operation.clone(), + measurement_name: row.measurement_name, + number_value: row.number_value, + dimensions: (!row.dimensions.is_empty()).then_some(row.dimensions), + }, + }) + }), + ); + } + fn emit_thread_initialized( &mut self, connection_id: u64,