diff --git a/codex-rs/core/src/analytics_client.rs b/codex-rs/core/src/analytics_client.rs index bfb90bc4da..b38fab2402 100644 --- a/codex-rs/core/src/analytics_client.rs +++ b/codex-rs/core/src/analytics_client.rs @@ -15,6 +15,7 @@ use std::sync::Arc; use std::sync::Mutex; use std::time::Duration; use tokio::sync::mpsc; +use tokio::sync::mpsc::error::TrySendError; #[derive(Clone)] pub(crate) struct TrackEventsContext { @@ -50,6 +51,15 @@ pub(crate) enum InvocationType { Implicit, } +impl InvocationType { + fn tag_value(self) -> &'static str { + match self { + Self::Explicit => "explicit", + Self::Implicit => "implicit", + } + } +} + pub(crate) struct AppInvocation { pub(crate) connector_id: Option, pub(crate) app_name: Option, @@ -110,9 +120,14 @@ impl AnalyticsEventsQueue { } fn try_send(&self, job: TrackEventsJob) { - if self.sender.try_send(job).is_err() { - //TODO: add a metric for this - tracing::warn!("dropping analytics events: queue is full"); + if let Err(err) = self.sender.try_send(job) { + let (reason, job) = match &err { + TrySendError::Full(job) => ("queue_full", job), + TrySendError::Closed(job) => ("queue_closed", job), + }; + let job_type = job.job_type(); + emit_analytics_events_failure_metric(reason, job_type, job.invoke_type_tag(), &[]); + tracing::warn!("dropping analytics events job: reason={reason} job_type={job_type}"); } } @@ -245,6 +260,56 @@ enum TrackEventsJob { PluginDisabled(TrackPluginManagementJob), } +impl TrackEventsJob { + fn job_type(&self) -> &'static str { + match self { + Self::SkillInvocations(_) => "skill_invocations", + Self::AppMentioned(_) => "app_mentioned", + Self::AppUsed(_) => "app_used", + Self::PluginUsed(_) => "plugin_used", + Self::PluginInstalled(_) => "plugin_installed", + Self::PluginUninstalled(_) => "plugin_uninstalled", + Self::PluginEnabled(_) => "plugin_enabled", + Self::PluginDisabled(_) => "plugin_disabled", + } + } + + fn invoke_type_tag(&self) -> Option<&'static str> { + match self { + Self::SkillInvocations(job) => job + .invocations + .first() + .map(|invocation| invocation.invocation_type.tag_value()), + Self::AppMentioned(job) => job + .mentions + .first() + .and_then(|mention| mention.invocation_type.map(InvocationType::tag_value)), + Self::AppUsed(job) => job.app.invocation_type.map(InvocationType::tag_value), + Self::PluginUsed(_) + | Self::PluginInstalled(_) + | Self::PluginUninstalled(_) + | Self::PluginEnabled(_) + | Self::PluginDisabled(_) => None, + } + } +} + +fn emit_analytics_events_failure_metric<'a>( + reason: &'static str, + job_type: &'static str, + invoke_type: Option<&'static str>, + extra_tags: &[(&'a str, &'a str)], +) { + if let Some(metrics) = codex_otel::metrics::global() { + let mut tags = vec![("reason", reason), ("job_type", job_type)]; + if let Some(invoke_type) = invoke_type { + tags.push(("invoke_type", invoke_type)); + } + tags.extend(extra_tags.iter().copied()); + let _ = metrics.counter("codex.analytics_events.emit.failure", /*inc*/ 1, &tags); + } +} + struct TrackSkillInvocationsJob { config: Arc, tracking: TrackEventsContext, @@ -678,17 +743,52 @@ async fn send_track_events( if events.is_empty() { return; } + let (job_type, invoke_type) = match events.first() { + Some(TrackEventRequest::SkillInvocation(event)) => ( + "skill_invocations", + event + .event_params + .invoke_type + .map(InvocationType::tag_value), + ), + Some(TrackEventRequest::AppMentioned(event)) => ( + "app_mentioned", + event + .event_params + .invoke_type + .map(InvocationType::tag_value), + ), + Some(TrackEventRequest::AppUsed(event)) => ( + "app_used", + event + .event_params + .invoke_type + .map(InvocationType::tag_value), + ), + Some(TrackEventRequest::PluginUsed(_)) => ("plugin_used", None), + Some(TrackEventRequest::PluginInstalled(_)) => ("plugin_installed", None), + Some(TrackEventRequest::PluginUninstalled(_)) => ("plugin_uninstalled", None), + Some(TrackEventRequest::PluginEnabled(_)) => ("plugin_enabled", None), + Some(TrackEventRequest::PluginDisabled(_)) => ("plugin_disabled", None), + None => unreachable!("events should be non-empty"), + }; let Some(auth) = auth_manager.auth().await else { + emit_analytics_events_failure_metric("auth_missing", job_type, invoke_type, &[]); return; }; if !auth.is_chatgpt_auth() { + emit_analytics_events_failure_metric("non_chatgpt_auth", job_type, invoke_type, &[]); return; } let access_token = match auth.get_token() { Ok(token) => token, - Err(_) => return, + Err(_) => { + emit_analytics_events_failure_metric("token_error", job_type, invoke_type, &[]); + return; + } }; let Some(account_id) = auth.get_account_id() else { + emit_analytics_events_failure_metric("account_id_missing", job_type, invoke_type, &[]); return; }; @@ -710,10 +810,17 @@ async fn send_track_events( Ok(response) if response.status().is_success() => {} Ok(response) => { let status = response.status(); + emit_analytics_events_failure_metric( + "http_status", + job_type, + invoke_type, + &[("status_code", status.as_str())], + ); let body = response.text().await.unwrap_or_default(); tracing::warn!("events failed with status {status}: {body}"); } Err(err) => { + emit_analytics_events_failure_metric("request_error", job_type, invoke_type, &[]); tracing::warn!("failed to send events request: {err}"); } } @@ -725,6 +832,13 @@ pub(crate) fn skill_id_for_local_skill( skill_path: &Path, skill_name: &str, ) -> String { + tracing::info!( + ?repo_url, + ?repo_root, + skill_path = %skill_path.display(), + skill_name, + "building analytics skill id for local skill" + ); let path = normalize_path_for_skill_id(repo_url, repo_root, skill_path); let prefix = if let Some(url) = repo_url { format!("repo_{url}") @@ -734,7 +848,15 @@ pub(crate) fn skill_id_for_local_skill( let raw_id = format!("{prefix}_{path}_{skill_name}"); let mut hasher = Sha1::new(); hasher.update(raw_id.as_bytes()); - format!("{:x}", hasher.finalize()) + let skill_id = format!("{:x}", hasher.finalize()); + tracing::info!( + normalized_path = path, + prefix, + raw_id, + skill_id, + "built analytics skill id for local skill" + ); + skill_id } /// Returns a normalized path for skill ID construction. diff --git a/codex-rs/core/src/analytics_client_tests.rs b/codex-rs/core/src/analytics_client_tests.rs index a038aae047..6037f66563 100644 --- a/codex-rs/core/src/analytics_client_tests.rs +++ b/codex-rs/core/src/analytics_client_tests.rs @@ -5,12 +5,19 @@ use super::CodexAppUsedEventRequest; use super::CodexPluginEventRequest; use super::CodexPluginUsedEventRequest; use super::InvocationType; +use super::TrackAppMentionedJob; +use super::TrackAppUsedJob; use super::TrackEventRequest; use super::TrackEventsContext; +use super::TrackEventsJob; +use super::TrackPluginManagementJob; +use super::TrackPluginUsedJob; +use super::TrackSkillInvocationsJob; use super::codex_app_metadata; use super::codex_plugin_metadata; use super::codex_plugin_used_metadata; use super::normalize_path_for_skill_id; +use crate::config::ConfigBuilder; use crate::plugins::AppConnectorId; use crate::plugins::PluginCapabilitySummary; use crate::plugins::PluginId; @@ -21,6 +28,7 @@ use std::collections::HashSet; use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; +use tempfile::TempDir; use tokio::sync::mpsc; fn expected_absolute_path(path: &PathBuf) -> String { @@ -271,6 +279,152 @@ fn plugin_used_dedupe_is_keyed_by_turn_and_plugin() { assert_eq!(queue.should_enqueue_plugin_used(&turn_2, &plugin), true); } +#[test] +fn track_events_job_type_uses_expected_tag_values() { + let codex_home = TempDir::new().expect("tempdir should create"); + let config = Arc::new(load_test_config(codex_home.path())); + let tracking = TrackEventsContext { + model_slug: "gpt-5".to_string(), + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + }; + let cases = vec![ + ( + TrackEventsJob::SkillInvocations(TrackSkillInvocationsJob { + config: Arc::clone(&config), + tracking: tracking.clone(), + invocations: Vec::new(), + }), + "skill_invocations", + ), + ( + TrackEventsJob::AppMentioned(TrackAppMentionedJob { + config: Arc::clone(&config), + tracking: tracking.clone(), + mentions: Vec::new(), + }), + "app_mentioned", + ), + ( + TrackEventsJob::AppUsed(TrackAppUsedJob { + config: Arc::clone(&config), + tracking: tracking.clone(), + app: AppInvocation { + connector_id: None, + app_name: None, + invocation_type: None, + }, + }), + "app_used", + ), + ( + TrackEventsJob::PluginUsed(TrackPluginUsedJob { + config: Arc::clone(&config), + tracking: tracking.clone(), + plugin: sample_plugin_metadata(), + }), + "plugin_used", + ), + ( + TrackEventsJob::PluginInstalled(TrackPluginManagementJob { + config: Arc::clone(&config), + plugin: sample_plugin_metadata(), + }), + "plugin_installed", + ), + ( + TrackEventsJob::PluginUninstalled(TrackPluginManagementJob { + config: Arc::clone(&config), + plugin: sample_plugin_metadata(), + }), + "plugin_uninstalled", + ), + ( + TrackEventsJob::PluginEnabled(TrackPluginManagementJob { + config: Arc::clone(&config), + plugin: sample_plugin_metadata(), + }), + "plugin_enabled", + ), + ( + TrackEventsJob::PluginDisabled(TrackPluginManagementJob { + config, + plugin: sample_plugin_metadata(), + }), + "plugin_disabled", + ), + ]; + + for (job, expected) in cases { + assert_eq!(job.job_type(), expected); + } +} + +#[test] +fn track_events_job_invoke_type_tag_uses_first_available_value() { + let codex_home = TempDir::new().expect("tempdir should create"); + let config = Arc::new(load_test_config(codex_home.path())); + let tracking = TrackEventsContext { + model_slug: "gpt-5".to_string(), + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), + }; + let explicit_skill_job = TrackEventsJob::SkillInvocations(TrackSkillInvocationsJob { + config: Arc::clone(&config), + tracking: tracking.clone(), + invocations: vec![super::SkillInvocation { + skill_name: "doc".to_string(), + skill_scope: codex_protocol::protocol::SkillScope::Repo, + skill_path: codex_home.path().join("SKILL.md"), + invocation_type: InvocationType::Explicit, + }], + }); + let implicit_app_job = TrackEventsJob::AppUsed(TrackAppUsedJob { + config: Arc::clone(&config), + tracking: tracking.clone(), + app: AppInvocation { + connector_id: Some("drive".to_string()), + app_name: Some("Google Drive".to_string()), + invocation_type: Some(InvocationType::Implicit), + }, + }); + let first_value_app_mentioned_job = TrackEventsJob::AppMentioned(TrackAppMentionedJob { + config: Arc::clone(&config), + tracking: tracking.clone(), + mentions: vec![ + AppInvocation { + connector_id: Some("drive".to_string()), + app_name: Some("Google Drive".to_string()), + invocation_type: Some(InvocationType::Explicit), + }, + AppInvocation { + connector_id: Some("calendar".to_string()), + app_name: Some("Calendar".to_string()), + invocation_type: Some(InvocationType::Implicit), + }, + ], + }); + let empty_skill_job = TrackEventsJob::SkillInvocations(TrackSkillInvocationsJob { + config: Arc::clone(&config), + tracking: tracking.clone(), + invocations: Vec::new(), + }); + let no_invoke_type_job = TrackEventsJob::PluginUsed(TrackPluginUsedJob { + config, + tracking, + plugin: sample_plugin_metadata(), + }); + + assert_eq!(explicit_skill_job.invoke_type_tag(), Some("explicit")); + assert_eq!(implicit_app_job.invoke_type_tag(), Some("implicit")); + assert_eq!( + first_value_app_mentioned_job.invoke_type_tag(), + Some("explicit") + ); + assert_eq!(empty_skill_job.invoke_type_tag(), None); + assert_eq!(no_invoke_type_job.invoke_type_tag(), None); +} + fn sample_plugin_metadata() -> PluginTelemetryMetadata { PluginTelemetryMetadata { plugin_id: PluginId::parse("sample@test").expect("valid plugin id"), @@ -287,3 +441,17 @@ fn sample_plugin_metadata() -> PluginTelemetryMetadata { }), } } + +fn load_test_config(codex_home: &std::path::Path) -> crate::config::Config { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("tokio runtime should build") + .block_on( + ConfigBuilder::default() + .codex_home(codex_home.to_path_buf()) + .fallback_cwd(Some(codex_home.to_path_buf())) + .build(), + ) + .expect("config should load") +}