diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index fb1710987e..00fe37ebbe 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -1388,6 +1388,7 @@ checksum = "e9b18233253483ce2f65329a24072ec414db782531bdbb7d0bbc4bd2ce6b7e21" name = "codex-analytics" version = "0.0.0" dependencies = [ + "codex-app-server-protocol", "codex-git-utils", "codex-login", "codex-plugin", diff --git a/codex-rs/analytics/Cargo.toml b/codex-rs/analytics/Cargo.toml index 9633eebb14..63bf63c911 100644 --- a/codex-rs/analytics/Cargo.toml +++ b/codex-rs/analytics/Cargo.toml @@ -13,6 +13,7 @@ path = "src/lib.rs" workspace = true [dependencies] +codex-app-server-protocol = { workspace = true } codex-git-utils = { workspace = true } codex-login = { workspace = true } codex-plugin = { workspace = true } diff --git a/codex-rs/analytics/src/analytics_client.rs b/codex-rs/analytics/src/analytics_client.rs index 6b50f0f341..f75af9bb2b 100644 --- a/codex-rs/analytics/src/analytics_client.rs +++ b/codex-rs/analytics/src/analytics_client.rs @@ -1,3 +1,7 @@ +use codex_app_server_protocol::ClientRequest; +use codex_app_server_protocol::InitializeParams; +use codex_app_server_protocol::RequestId; +use codex_app_server_protocol::ServerNotification; use codex_git_utils::collect_git_info; use codex_git_utils::get_git_repo_root; use codex_login::AuthManager; @@ -26,12 +30,6 @@ pub struct TrackEventsContext { pub turn_id: String, } -#[derive(Clone)] -pub struct InitializeInput { - pub connection_id: u64, - pub product_client_id: String, -} - #[derive(Clone)] pub struct ThreadInitializeInput { pub connection_id: u64, @@ -83,8 +81,28 @@ pub struct AppInvocation { pub invocation_type: Option, } -pub enum AnalyticsInput { - Initialize(InitializeInput), +pub enum AnalyticsFact { + Initialize { + connection_id: u64, + params: InitializeParams, + }, + Request { + connection_id: u64, + request_id: RequestId, + request: ClientRequest, + }, + Notification { + connection_id: u64, + notification: ServerNotification, + }, + // Facts that do not naturally exist on the app-server protocol surface, or + // would require non-trivial protocol reshaping on this branch. + Custom(CustomAnalyticsFact), +} + +pub enum CustomAnalyticsFact { + // This remains custom on this branch because app-server-protocol does not + // yet expose a generic client response enum we can reduce over directly. ThreadInitialized(ThreadInitializeInput), SkillInvoked(SkillInvokedInput), AppMentioned(AppMentionedInput), @@ -137,7 +155,7 @@ struct ClientState { #[derive(Clone)] pub(crate) struct AnalyticsEventsQueue { - sender: mpsc::Sender, + sender: mpsc::Sender, app_used_emitted_keys: Arc>>, plugin_used_emitted_keys: Arc>>, } @@ -166,7 +184,7 @@ impl AnalyticsEventsQueue { } } - fn try_send(&self, input: AnalyticsInput) { + fn try_send(&self, input: AnalyticsFact) { if self.sender.try_send(input).is_err() { //TODO: add a metric for this tracing::warn!("dropping analytics events: queue is full"); @@ -223,84 +241,91 @@ impl AnalyticsEventsClient { if invocations.is_empty() { return; } - self.record(AnalyticsInput::SkillInvoked(SkillInvokedInput { - tracking, - invocations, - })); + self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked( + SkillInvokedInput { + tracking, + invocations, + }, + ))); } - pub fn track_initialize(&self, input: InitializeInput) { - self.record(AnalyticsInput::Initialize(input)); + pub fn track_initialize(&self, connection_id: u64, params: InitializeParams) { + self.record_fact(AnalyticsFact::Initialize { + connection_id, + params, + }); } pub fn track_thread_initialized(&self, input: ThreadInitializeInput) { - self.record(AnalyticsInput::ThreadInitialized(input)); + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::ThreadInitialized(input), + )); } pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec) { if mentions.is_empty() { return; } - self.record(AnalyticsInput::AppMentioned(AppMentionedInput { - tracking, - mentions, - })); + self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppMentioned( + AppMentionedInput { tracking, mentions }, + ))); } pub fn track_app_used(&self, tracking: TrackEventsContext, app: AppInvocation) { if !self.queue.should_enqueue_app_used(&tracking, &app) { return; } - self.record(AnalyticsInput::AppUsed(AppUsedInput { tracking, app })); + self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed( + AppUsedInput { tracking, app }, + ))); } pub fn track_plugin_used(&self, tracking: TrackEventsContext, plugin: PluginTelemetryMetadata) { if !self.queue.should_enqueue_plugin_used(&tracking, &plugin) { return; } - self.record(AnalyticsInput::PluginUsed(PluginUsedInput { - tracking, - plugin, - })); + self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::PluginUsed( + PluginUsedInput { tracking, plugin }, + ))); } pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) { - self.record(AnalyticsInput::PluginStateChanged( - PluginStateChangedInput { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput { plugin, state: PluginState::Installed, - }, + }), )); } pub fn track_plugin_uninstalled(&self, plugin: PluginTelemetryMetadata) { - self.record(AnalyticsInput::PluginStateChanged( - PluginStateChangedInput { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput { plugin, state: PluginState::Uninstalled, - }, + }), )); } pub fn track_plugin_enabled(&self, plugin: PluginTelemetryMetadata) { - self.record(AnalyticsInput::PluginStateChanged( - PluginStateChangedInput { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput { plugin, state: PluginState::Enabled, - }, + }), )); } pub fn track_plugin_disabled(&self, plugin: PluginTelemetryMetadata) { - self.record(AnalyticsInput::PluginStateChanged( - PluginStateChangedInput { + self.record_fact(AnalyticsFact::Custom( + CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput { plugin, state: PluginState::Disabled, - }, + }), )); } - pub fn record(&self, input: AnalyticsInput) { + pub fn record_fact(&self, input: AnalyticsFact) { if self.analytics_enabled == Some(false) { return; } @@ -424,37 +449,51 @@ struct CodexPluginUsedEventRequest { } impl AnalyticsReducer { - async fn ingest(&mut self, input: AnalyticsInput, out: &mut Vec) { + async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec) { match input { - AnalyticsInput::Initialize(input) => { - self.ingest_initialize(input); - } - AnalyticsInput::ThreadInitialized(input) => { - self.ingest_thread_initialized(input, out); - } - AnalyticsInput::SkillInvoked(input) => { - self.ingest_skill_invoked(input, out).await; - } - AnalyticsInput::AppMentioned(input) => { - self.ingest_app_mentioned(input, out); - } - AnalyticsInput::AppUsed(input) => { - self.ingest_app_used(input, out); - } - AnalyticsInput::PluginUsed(input) => { - self.ingest_plugin_used(input, out); - } - AnalyticsInput::PluginStateChanged(input) => { - self.ingest_plugin_state_changed(input, out); + AnalyticsFact::Initialize { + connection_id, + params, + } => { + self.ingest_initialize(connection_id, params); } + AnalyticsFact::Request { + connection_id: _connection_id, + request_id: _request_id, + request: _request, + } => {} + AnalyticsFact::Notification { + connection_id: _connection_id, + notification: _notification, + } => {} + AnalyticsFact::Custom(input) => match input { + CustomAnalyticsFact::ThreadInitialized(input) => { + self.ingest_thread_initialized(input, out); + } + CustomAnalyticsFact::SkillInvoked(input) => { + self.ingest_skill_invoked(input, out).await; + } + CustomAnalyticsFact::AppMentioned(input) => { + self.ingest_app_mentioned(input, out); + } + CustomAnalyticsFact::AppUsed(input) => { + self.ingest_app_used(input, out); + } + CustomAnalyticsFact::PluginUsed(input) => { + self.ingest_plugin_used(input, out); + } + CustomAnalyticsFact::PluginStateChanged(input) => { + self.ingest_plugin_state_changed(input, out); + } + }, } } - fn ingest_initialize(&mut self, input: InitializeInput) { + fn ingest_initialize(&mut self, connection_id: u64, params: InitializeParams) { self.clients.insert( - input.connection_id, + connection_id, ClientState { - product_client_id: input.product_client_id, + product_client_id: params.client_info.name, }, ); } diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 7b6bb6c1e2..4264ce7742 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -1,13 +1,13 @@ use super::AnalyticsEventsQueue; -use super::AnalyticsInput; +use super::AnalyticsFact; use super::AnalyticsReducer; use super::AppInvocation; use super::CodexAppMentionedEventRequest; use super::CodexAppUsedEventRequest; use super::CodexPluginEventRequest; use super::CodexPluginUsedEventRequest; +use super::CustomAnalyticsFact; use super::InitializationMode; -use super::InitializeInput; use super::InvocationType; use super::ThreadInitializeInput; use super::TrackEventRequest; @@ -17,6 +17,8 @@ use super::codex_plugin_metadata; use super::codex_plugin_used_metadata; use super::codex_thread_initialized_event_request; use super::normalize_path_for_skill_id; +use codex_app_server_protocol::ClientInfo; +use codex_app_server_protocol::InitializeParams; use codex_login::default_client::originator; use codex_plugin::AppConnectorId; use codex_plugin::PluginCapabilitySummary; @@ -197,7 +199,7 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() { #[test] fn thread_initialized_event_serializes_expected_shape() { let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request( - originator().value, + "codex-tui".to_string(), ThreadInitializeInput { connection_id: 1, thread_id: "thread-0".to_string(), @@ -216,7 +218,7 @@ fn thread_initialized_event_serializes_expected_shape() { "event_type": "codex_thread_initialized", "event_params": { "thread_id": "thread-0", - "product_client_id": originator().value, + "product_client_id": "codex-tui", "model": "gpt-5", "ephemeral": true, "session_source": "user", @@ -233,7 +235,7 @@ fn thread_initialized_event_serializes_expected_shape() { #[test] fn thread_initialized_event_serializes_subagent_source() { let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request( - originator().value, + "codex-tui".to_string(), ThreadInitializeInput { connection_id: 1, thread_id: "thread-1".to_string(), @@ -257,14 +259,16 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize reducer .ingest( - AnalyticsInput::ThreadInitialized(ThreadInitializeInput { - connection_id: 7, - thread_id: "thread-no-client".to_string(), - model: "gpt-5".to_string(), - ephemeral: false, - session_source: SessionSource::Exec, - initialization_mode: InitializationMode::New, - }), + AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitialized( + ThreadInitializeInput { + connection_id: 7, + thread_id: "thread-no-client".to_string(), + model: "gpt-5".to_string(), + ephemeral: false, + session_source: SessionSource::Exec, + initialization_mode: InitializationMode::New, + }, + )), &mut events, ) .await; @@ -272,10 +276,17 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize reducer .ingest( - AnalyticsInput::Initialize(InitializeInput { + AnalyticsFact::Initialize { connection_id: 7, - product_client_id: "codex-app-server-tests".to_string(), - }), + params: InitializeParams { + client_info: ClientInfo { + name: "codex-tui".to_string(), + title: None, + version: "1.0.0".to_string(), + }, + capabilities: None, + }, + }, &mut events, ) .await; @@ -283,23 +294,25 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize reducer .ingest( - AnalyticsInput::ThreadInitialized(ThreadInitializeInput { - connection_id: 7, - thread_id: "thread-1".to_string(), - model: "gpt-5".to_string(), - ephemeral: true, - session_source: SessionSource::SubAgent(SubAgentSource::ThreadSpawn { - parent_thread_id: codex_protocol::ThreadId::from_string( - "11111111-1111-1111-1111-111111111111", - ) - .expect("valid thread id"), - depth: 1, - agent_path: None, - agent_nickname: None, - agent_role: None, - }), - initialization_mode: InitializationMode::Resumed, - }), + AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitialized( + ThreadInitializeInput { + connection_id: 7, + thread_id: "thread-1".to_string(), + model: "gpt-5".to_string(), + ephemeral: true, + session_source: SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id: codex_protocol::ThreadId::from_string( + "11111111-1111-1111-1111-111111111111", + ) + .expect("valid thread id"), + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + }), + initialization_mode: InitializationMode::Resumed, + }, + )), &mut events, ) .await; @@ -307,10 +320,7 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize let payload = serde_json::to_value(&events).expect("serialize events"); assert_eq!(payload.as_array().expect("events array").len(), 1); assert_eq!(payload[0]["event_type"], "codex_thread_initialized"); - assert_eq!( - payload[0]["event_params"]["product_client_id"], - "codex-app-server-tests" - ); + assert_eq!(payload[0]["event_params"]["product_client_id"], "codex-tui"); assert_eq!(payload[0]["event_params"]["initialization_mode"], "resumed"); assert_eq!(payload[0]["event_params"]["session_source"], "subagent"); assert_eq!( diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index df336a241d..ae841bbb75 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -1,13 +1,13 @@ mod analytics_client; pub use analytics_client::AnalyticsEventsClient; -pub use analytics_client::AnalyticsInput; +pub use analytics_client::AnalyticsFact; pub use analytics_client::AnalyticsReducer; pub use analytics_client::AppInvocation; pub use analytics_client::AppMentionedInput; pub use analytics_client::AppUsedInput; +pub use analytics_client::CustomAnalyticsFact; pub use analytics_client::InitializationMode; -pub use analytics_client::InitializeInput; pub use analytics_client::InvocationType; pub use analytics_client::PluginState; pub use analytics_client::PluginStateChangedInput; diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index db408aa65d..32ecb5ac3f 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -55,7 +55,6 @@ use codex_app_server_protocol::experimental_required_message; use codex_arg0::Arg0DispatchPaths; use codex_core::AnalyticsEventsClient; use codex_core::AuthManager; -use codex_core::InitializeInput; use codex_core::ThreadManager; use codex_core::config::Config; use codex_core::config_loader::CloudRequirementsLoader; @@ -560,6 +559,7 @@ impl MessageProcessor { session.experimental_api_enabled = experimental_api_enabled; session.opted_out_notification_methods = opt_out_notification_methods.into_iter().collect(); + let analytics_initialize_params = params.clone(); let ClientInfo { name, title: _title, @@ -598,10 +598,7 @@ impl MessageProcessor { } } self.analytics_events_client - .track_initialize(InitializeInput { - connection_id: connection_id.0, - product_client_id: originator.clone(), - }); + .track_initialize(connection_id.0, analytics_initialize_params); set_default_client_residency_requirement(self.config.enforce_residency.value()); let user_agent_suffix = format!("{name}; {version}"); if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() { diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 6bad79a676..6ee4f2482e 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -136,8 +136,9 @@ pub type CodexConversation = CodexThread; pub use auth::AuthManager; pub use auth::CodexAuth; pub use codex_analytics::AnalyticsEventsClient; +pub use codex_analytics::AnalyticsFact; +pub use codex_analytics::CustomAnalyticsFact; pub use codex_analytics::InitializationMode; -pub use codex_analytics::InitializeInput; pub use codex_analytics::ThreadInitializeInput; mod default_client_forwarding;