From 5567bf4aed8a98aa97de0bb70156d650abd8be8c Mon Sep 17 00:00:00 2001 From: Roy Han Date: Thu, 26 Mar 2026 10:56:26 -0700 Subject: [PATCH] pivot to consuming from app-event stream --- codex-rs/analytics/src/analytics_client.rs | 139 +++++++++++++----- .../analytics/src/analytics_client_tests.rs | 128 +++++++++++----- codex-rs/analytics/src/lib.rs | 4 +- .../app-server/src/codex_message_processor.rs | 44 ++++++ codex-rs/app-server/src/message_processor.rs | 13 +- .../app-server/tests/suite/v2/analytics.rs | 91 ++++++++++++ .../app-server/tests/suite/v2/thread_fork.rs | 44 ++++++ .../tests/suite/v2/thread_resume.rs | 45 ++++++ .../app-server/tests/suite/v2/thread_start.rs | 33 +++++ codex-rs/core/src/codex.rs | 60 -------- codex-rs/core/src/lib.rs | 2 + codex-rs/core/tests/suite/analytics.rs | 78 ---------- codex-rs/core/tests/suite/mod.rs | 1 - 13 files changed, 467 insertions(+), 215 deletions(-) delete mode 100644 codex-rs/core/tests/suite/analytics.rs diff --git a/codex-rs/analytics/src/analytics_client.rs b/codex-rs/analytics/src/analytics_client.rs index 5df326fb85..3dd4d14f80 100644 --- a/codex-rs/analytics/src/analytics_client.rs +++ b/codex-rs/analytics/src/analytics_client.rs @@ -27,21 +27,18 @@ pub struct TrackEventsContext { } #[derive(Clone)] -pub struct CodexThreadInitializedInput { - pub thread_id: String, - pub model: String, +pub struct InitializeInput { + pub connection_id: u64, pub product_client_id: String, - pub created_at: u64, - pub thread_context: CodexThreadContext, } #[derive(Clone)] -pub struct CodexThreadContext { +pub struct ThreadInitializeInput { + pub connection_id: u64, + pub thread_id: String, + pub model: String, pub ephemeral: bool, pub session_source: SessionSource, - pub initialization_mode: InitializationMode, - pub subagent_source: Option, - pub parent_thread_id: Option, } #[derive(Clone, Copy, Debug, Serialize)] @@ -86,7 +83,10 @@ pub struct AppInvocation { } pub enum AnalyticsInput { - CodexThreadInitialized(CodexThreadInitializedInput), + Initialize(InitializeInput), + ThreadStart(ThreadInitializeInput), + ThreadFork(ThreadInitializeInput), + ThreadResume(ThreadInitializeInput), SkillInvoked(SkillInvokedInput), AppMentioned(AppMentionedInput), AppUsed(AppUsedInput), @@ -129,11 +129,11 @@ pub enum PluginState { #[derive(Default)] pub struct AnalyticsReducer { - threads: HashMap, + clients: HashMap, } -struct ThreadState { - _initialized_input: CodexThreadInitializedInput, +struct ClientState { + product_client_id: String, } #[derive(Clone)] @@ -230,8 +230,20 @@ impl AnalyticsEventsClient { })); } - pub fn track_thread_initialized(&self, input: CodexThreadInitializedInput) { - self.record(AnalyticsInput::CodexThreadInitialized(input)); + pub fn track_initialize(&self, input: InitializeInput) { + self.record(AnalyticsInput::Initialize(input)); + } + + pub fn track_thread_start(&self, input: ThreadInitializeInput) { + self.record(AnalyticsInput::ThreadStart(input)); + } + + pub fn track_thread_fork(&self, input: ThreadInitializeInput) { + self.record(AnalyticsInput::ThreadFork(input)); + } + + pub fn track_thread_resume(&self, input: ThreadInitializeInput) { + self.record(AnalyticsInput::ThreadResume(input)); } pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec) { @@ -423,8 +435,17 @@ struct CodexPluginUsedEventRequest { impl AnalyticsReducer { async fn ingest(&mut self, input: AnalyticsInput, out: &mut Vec) { match input { - AnalyticsInput::CodexThreadInitialized(input) => { - self.ingest_thread_initialized(input, out); + AnalyticsInput::Initialize(input) => { + self.ingest_initialize(input); + } + AnalyticsInput::ThreadStart(input) => { + self.ingest_thread_initialized(input, InitializationMode::New, out); + } + AnalyticsInput::ThreadFork(input) => { + self.ingest_thread_initialized(input, InitializationMode::Forked, out); + } + AnalyticsInput::ThreadResume(input) => { + self.ingest_thread_initialized(input, InitializationMode::Resumed, out); } AnalyticsInput::SkillInvoked(input) => { self.ingest_skill_invoked(input, out).await; @@ -444,19 +465,30 @@ impl AnalyticsReducer { } } - fn ingest_thread_initialized( - &mut self, - input: CodexThreadInitializedInput, - out: &mut Vec, - ) { - self.threads.insert( - input.thread_id.clone(), - ThreadState { - _initialized_input: input.clone(), + fn ingest_initialize(&mut self, input: InitializeInput) { + self.clients.insert( + input.connection_id, + ClientState { + product_client_id: input.product_client_id, }, ); + } + + fn ingest_thread_initialized( + &mut self, + input: ThreadInitializeInput, + initialization_mode: InitializationMode, + out: &mut Vec, + ) { + let Some(client_state) = self.clients.get(&input.connection_id) else { + return; + }; out.push(TrackEventRequest::CodexThreadInitialized( - codex_thread_initialized_event_request(input), + codex_thread_initialized_event_request( + client_state.product_client_id.clone(), + input, + initialization_mode, + ), )); } @@ -577,30 +609,36 @@ fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> Code } fn codex_thread_initialized_event_request( - input: CodexThreadInitializedInput, + product_client_id: String, + input: ThreadInitializeInput, + initialization_mode: InitializationMode, ) -> CodexThreadInitializedEvent { CodexThreadInitializedEvent { event_type: "codex_thread_initialized", - event_params: codex_thread_initialized_event_params(input), + event_params: codex_thread_initialized_event_params( + product_client_id, + input, + initialization_mode, + ), } } fn codex_thread_initialized_event_params( - input: CodexThreadInitializedInput, + product_client_id: String, + input: ThreadInitializeInput, + initialization_mode: InitializationMode, ) -> CodexThreadInitializedEventParams { CodexThreadInitializedEventParams { thread_id: input.thread_id, - product_client_id: input.product_client_id, + product_client_id, model: input.model, - ephemeral: input.thread_context.ephemeral, - session_source: session_source_name(&input.thread_context.session_source), - initialization_mode: input.thread_context.initialization_mode, - subagent_source: input - .thread_context - .subagent_source + ephemeral: input.ephemeral, + session_source: session_source_name(&input.session_source), + initialization_mode, + subagent_source: session_source_subagent_source(&input.session_source) .map(subagent_source_name), - parent_thread_id: input.thread_context.parent_thread_id, - created_at: input.created_at, + parent_thread_id: session_source_parent_thread_id(&input.session_source), + created_at: now_unix_timestamp_secs(), } } @@ -657,6 +695,29 @@ fn subagent_source_name(subagent_source: SubAgentSource) -> String { } } +fn session_source_subagent_source(session_source: &SessionSource) -> Option { + match session_source { + SessionSource::SubAgent(subagent_source) => Some(subagent_source.clone()), + _ => None, + } +} + +fn session_source_parent_thread_id(session_source: &SessionSource) -> Option { + match session_source { + SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, .. + }) => Some(parent_thread_id.to_string()), + _ => None, + } +} + +fn now_unix_timestamp_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() +} + async fn send_track_events( auth_manager: &AuthManager, base_url: &str, diff --git a/codex-rs/analytics/src/analytics_client_tests.rs b/codex-rs/analytics/src/analytics_client_tests.rs index 5e83c2aaf5..28e3860098 100644 --- a/codex-rs/analytics/src/analytics_client_tests.rs +++ b/codex-rs/analytics/src/analytics_client_tests.rs @@ -1,13 +1,15 @@ use super::AnalyticsEventsQueue; +use super::AnalyticsInput; +use super::AnalyticsReducer; use super::AppInvocation; use super::CodexAppMentionedEventRequest; use super::CodexAppUsedEventRequest; use super::CodexPluginEventRequest; use super::CodexPluginUsedEventRequest; -use super::CodexThreadContext; -use super::CodexThreadInitializedInput; use super::InitializationMode; +use super::InitializeInput; use super::InvocationType; +use super::ThreadInitializeInput; use super::TrackEventRequest; use super::TrackEventsContext; use super::codex_app_metadata; @@ -195,19 +197,15 @@ 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( - CodexThreadInitializedInput { + originator().value, + ThreadInitializeInput { + connection_id: 1, thread_id: "thread-0".to_string(), model: "gpt-5".to_string(), - product_client_id: originator().value, - created_at: 1_716_000_000, - thread_context: CodexThreadContext { - ephemeral: true, - session_source: SessionSource::Exec, - initialization_mode: InitializationMode::New, - subagent_source: None, - parent_thread_id: None, - }, + ephemeral: true, + session_source: SessionSource::Exec, }, + InitializationMode::New, )); let payload = serde_json::to_value(&event).expect("serialize thread initialized event"); @@ -225,28 +223,25 @@ fn thread_initialized_event_serializes_expected_shape() { "initialization_mode": "new", "subagent_source": null, "parent_thread_id": null, - "created_at": 1716000000 + "created_at": payload["event_params"]["created_at"] } }) ); + assert!(payload["event_params"]["created_at"].as_u64().is_some()); } #[test] fn thread_initialized_event_serializes_subagent_source() { let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request( - CodexThreadInitializedInput { + originator().value, + ThreadInitializeInput { + connection_id: 1, thread_id: "thread-1".to_string(), model: "gpt-5".to_string(), - product_client_id: originator().value, - created_at: 1, - thread_context: CodexThreadContext { - ephemeral: false, - session_source: SessionSource::SubAgent(SubAgentSource::Review), - initialization_mode: InitializationMode::New, - subagent_source: Some(SubAgentSource::Review), - parent_thread_id: None, - }, + ephemeral: false, + session_source: SessionSource::SubAgent(SubAgentSource::Review), }, + InitializationMode::New, )); let payload = @@ -258,19 +253,15 @@ fn thread_initialized_event_serializes_subagent_source() { #[test] fn thread_initialized_event_omits_non_user_non_subagent_session_source() { let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request( - CodexThreadInitializedInput { + originator().value, + ThreadInitializeInput { + connection_id: 1, thread_id: "thread-2".to_string(), model: "gpt-5".to_string(), - product_client_id: originator().value, - created_at: 1, - thread_context: CodexThreadContext { - ephemeral: false, - session_source: SessionSource::Mcp, - initialization_mode: InitializationMode::New, - subagent_source: None, - parent_thread_id: None, - }, + ephemeral: false, + session_source: SessionSource::Mcp, }, + InitializationMode::New, )); let payload = serde_json::to_value(&event).expect("serialize mcp thread initialized event"); @@ -280,6 +271,77 @@ fn thread_initialized_event_omits_non_user_non_subagent_session_source() { ); } +#[tokio::test] +async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialized() { + let mut reducer = AnalyticsReducer::default(); + let mut events = Vec::new(); + + reducer + .ingest( + AnalyticsInput::ThreadStart(ThreadInitializeInput { + connection_id: 7, + thread_id: "thread-no-client".to_string(), + model: "gpt-5".to_string(), + ephemeral: false, + session_source: SessionSource::Exec, + }), + &mut events, + ) + .await; + assert!(events.is_empty(), "thread events should require initialize"); + + reducer + .ingest( + AnalyticsInput::Initialize(InitializeInput { + connection_id: 7, + product_client_id: "codex-app-server-tests".to_string(), + }), + &mut events, + ) + .await; + assert!(events.is_empty(), "initialize should not publish by itself"); + + reducer + .ingest( + AnalyticsInput::ThreadResume(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, + }), + }), + &mut events, + ) + .await; + + 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"]["initialization_mode"], "resumed"); + assert_eq!(payload[0]["event_params"]["session_source"], "subagent"); + assert_eq!( + payload[0]["event_params"]["subagent_source"], + "thread_spawn" + ); + assert_eq!( + payload[0]["event_params"]["parent_thread_id"], + "11111111-1111-1111-1111-111111111111" + ); +} + #[test] fn plugin_used_event_serializes_expected_shape() { let tracking = TrackEventsContext { diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 16f3d75ac6..df336a241d 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -6,14 +6,14 @@ pub use analytics_client::AnalyticsReducer; pub use analytics_client::AppInvocation; pub use analytics_client::AppMentionedInput; pub use analytics_client::AppUsedInput; -pub use analytics_client::CodexThreadContext; -pub use analytics_client::CodexThreadInitializedInput; 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; pub use analytics_client::PluginUsedInput; pub use analytics_client::SkillInvocation; pub use analytics_client::SkillInvokedInput; +pub use analytics_client::ThreadInitializeInput; pub use analytics_client::TrackEventsContext; pub use analytics_client::build_track_events_context; diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 22122a7991..553e32bc95 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -179,6 +179,7 @@ use codex_arg0::Arg0DispatchPaths; use codex_backend_client::Client as BackendClient; use codex_chatgpt::connectors; use codex_cloud_requirements::cloud_requirements_loader; +use codex_core::AnalyticsEventsClient; use codex_core::AuthManager; use codex_core::CodexAuth; use codex_core::CodexThread; @@ -189,6 +190,7 @@ use codex_core::RolloutRecorder; use codex_core::SessionMeta; use codex_core::SteerInputError; use codex_core::ThreadConfigSnapshot; +use codex_core::ThreadInitializeInput; use codex_core::ThreadManager; use codex_core::ThreadSortKey as CoreThreadSortKey; use codex_core::auth::AuthMode as CoreAuthMode; @@ -374,6 +376,7 @@ pub(crate) struct CodexMessageProcessor { auth_manager: Arc, thread_manager: Arc, outgoing: Arc, + analytics_events_client: AnalyticsEventsClient, arg0_paths: Arg0DispatchPaths, config: Arc, cli_overrides: Arc>>, @@ -404,6 +407,7 @@ struct ListenerTaskContext { thread_manager: Arc, thread_state_manager: ThreadStateManager, outgoing: Arc, + analytics_events_client: AnalyticsEventsClient, thread_watch_manager: ThreadWatchManager, fallback_model_provider: String, codex_home: PathBuf, @@ -426,6 +430,7 @@ pub(crate) struct CodexMessageProcessorArgs { pub(crate) auth_manager: Arc, pub(crate) thread_manager: Arc, pub(crate) outgoing: Arc, + pub(crate) analytics_events_client: AnalyticsEventsClient, pub(crate) arg0_paths: Arg0DispatchPaths, pub(crate) config: Arc, pub(crate) cli_overrides: Arc>>, @@ -490,6 +495,7 @@ impl CodexMessageProcessor { auth_manager, thread_manager, outgoing, + analytics_events_client, arg0_paths, config, cli_overrides, @@ -502,6 +508,7 @@ impl CodexMessageProcessor { auth_manager, thread_manager, outgoing: outgoing.clone(), + analytics_events_client, arg0_paths, config, cli_overrides, @@ -1921,6 +1928,7 @@ impl CodexMessageProcessor { thread_manager: Arc::clone(&self.thread_manager), thread_state_manager: self.thread_state_manager.clone(), outgoing: Arc::clone(&self.outgoing), + analytics_events_client: self.analytics_events_client.clone(), thread_watch_manager: self.thread_watch_manager.clone(), fallback_model_provider: self.config.model_provider_id.clone(), codex_home: self.config.codex_home.clone(), @@ -2153,6 +2161,13 @@ impl CodexMessageProcessor { sandbox: config_snapshot.sandbox_policy.into(), reasoning_effort: config_snapshot.reasoning_effort, }; + listener_task_context + .analytics_events_client + .track_thread_start(thread_lifecycle_input( + request_id.connection_id, + &thread, + response.model.clone(), + )); listener_task_context .outgoing @@ -3631,6 +3646,12 @@ impl CodexMessageProcessor { sandbox: session_configured.sandbox_policy.into(), reasoning_effort: session_configured.reasoning_effort, }; + self.analytics_events_client + .track_thread_resume(thread_lifecycle_input( + request_id.connection_id, + &response.thread, + response.model.clone(), + )); self.outgoing.send_response(request_id, response).await; } @@ -4238,6 +4259,12 @@ impl CodexMessageProcessor { sandbox: session_configured.sandbox_policy.into(), reasoning_effort: session_configured.reasoning_effort, }; + self.analytics_events_client + .track_thread_fork(thread_lifecycle_input( + request_id.connection_id, + &thread, + response.model.clone(), + )); self.outgoing.send_response(request_id, response).await; @@ -6819,6 +6846,7 @@ impl CodexMessageProcessor { thread_manager: Arc::clone(&self.thread_manager), thread_state_manager: self.thread_state_manager.clone(), outgoing: Arc::clone(&self.outgoing), + analytics_events_client: self.analytics_events_client.clone(), thread_watch_manager: self.thread_watch_manager.clone(), fallback_model_provider: self.config.model_provider_id.clone(), codex_home: self.config.codex_home.clone(), @@ -6906,6 +6934,7 @@ impl CodexMessageProcessor { thread_manager: Arc::clone(&self.thread_manager), thread_state_manager: self.thread_state_manager.clone(), outgoing: Arc::clone(&self.outgoing), + analytics_events_client: self.analytics_events_client.clone(), thread_watch_manager: self.thread_watch_manager.clone(), fallback_model_provider: self.config.model_provider_id.clone(), codex_home: self.config.codex_home.clone(), @@ -6937,6 +6966,7 @@ impl CodexMessageProcessor { outgoing, thread_manager, thread_state_manager, + analytics_events_client: _, thread_watch_manager, fallback_model_provider, codex_home, @@ -7854,6 +7884,20 @@ fn cloud_requirements_load_error(err: &std::io::Error) -> Option<&CloudRequireme None } +fn thread_lifecycle_input( + connection_id: ConnectionId, + thread: &Thread, + model: String, +) -> ThreadInitializeInput { + ThreadInitializeInput { + connection_id: connection_id.0, + thread_id: thread.id.clone(), + model, + ephemeral: thread.ephemeral, + session_source: thread.source.clone().into(), + } +} + fn config_load_error(err: &std::io::Error) -> JSONRPCErrorError { let data = cloud_requirements_load_error(err).map(|cloud_error| { let mut data = serde_json::json!({ diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index feca709513..db408aa65d 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -55,6 +55,7 @@ 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; @@ -158,6 +159,7 @@ pub(crate) struct MessageProcessor { external_agent_config_api: ExternalAgentConfigApi, fs_api: FsApi, auth_manager: Arc, + analytics_events_client: AnalyticsEventsClient, fs_watch_manager: FsWatchManager, config: Arc, config_warnings: Arc>, @@ -238,6 +240,7 @@ impl MessageProcessor { auth_manager: auth_manager.clone(), thread_manager: Arc::clone(&thread_manager), outgoing: outgoing.clone(), + analytics_events_client: analytics_events_client.clone(), arg0_paths, config: Arc::clone(&config), cli_overrides: cli_overrides.clone(), @@ -258,7 +261,7 @@ impl MessageProcessor { loader_overrides, cloud_requirements, thread_manager, - analytics_events_client, + analytics_events_client.clone(), ); let external_agent_config_api = ExternalAgentConfigApi::new(config.codex_home.clone()); let fs_api = FsApi::default(); @@ -271,6 +274,7 @@ impl MessageProcessor { external_agent_config_api, fs_api, auth_manager, + analytics_events_client, fs_watch_manager, config, config_warnings: Arc::new(config_warnings), @@ -570,7 +574,7 @@ impl MessageProcessor { } else { name.clone() }; - if let Err(error) = set_default_originator(originator) { + if let Err(error) = set_default_originator(originator.clone()) { match error { SetOriginatorError::InvalidHeaderValue => { let error = JSONRPCErrorError { @@ -593,6 +597,11 @@ impl MessageProcessor { } } } + self.analytics_events_client + .track_initialize(InitializeInput { + connection_id: connection_id.0, + product_client_id: originator.clone(), + }); 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/app-server/tests/suite/v2/analytics.rs b/codex-rs/app-server/tests/suite/v2/analytics.rs index e64287cb22..b656889478 100644 --- a/codex-rs/app-server/tests/suite/v2/analytics.rs +++ b/codex-rs/app-server/tests/suite/v2/analytics.rs @@ -1,10 +1,23 @@ use anyhow::Result; +use app_test_support::ChatGptAuthFixture; +use app_test_support::DEFAULT_CLIENT_NAME; +use app_test_support::write_chatgpt_auth; +use codex_core::auth::AuthCredentialsStoreMode; use codex_core::config::ConfigBuilder; use codex_core::config::types::OtelExporterKind; use codex_core::config::types::OtelHttpProtocol; use pretty_assertions::assert_eq; +use serde_json::Value; use std::collections::HashMap; +use std::path::Path; +use std::time::Duration; use tempfile::TempDir; +use tokio::time::timeout; +use wiremock::Mock; +use wiremock::MockServer; +use wiremock::ResponseTemplate; +use wiremock::matchers::method; +use wiremock::matchers::path; const SERVICE_VERSION: &str = "0.0.0-test"; @@ -65,3 +78,81 @@ async fn app_server_default_analytics_enabled_with_flag() -> Result<()> { assert_eq!(has_metrics, true); Ok(()) } + +pub(crate) async fn enable_analytics_capture(server: &MockServer, codex_home: &Path) -> Result<()> { + Mock::given(method("POST")) + .and(path("/codex/analytics-events/events")) + .respond_with(ResponseTemplate::new(200)) + .mount(server) + .await; + + write_chatgpt_auth( + codex_home, + ChatGptAuthFixture::new("chatgpt-token") + .account_id("account-123") + .chatgpt_user_id("user-123") + .chatgpt_account_id("account-123"), + AuthCredentialsStoreMode::File, + )?; + + Ok(()) +} + +pub(crate) async fn wait_for_analytics_payload( + server: &MockServer, + read_timeout: Duration, +) -> Result { + let body = timeout(read_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() == "/codex/analytics-events/events" + }) { + break request.body.clone(); + } + tokio::time::sleep(Duration::from_millis(25)).await; + } + }) + .await?; + Ok(serde_json::from_slice(&body).expect("analytics payload")) +} + +pub(crate) fn thread_initialized_event(payload: &Value) -> &Value { + payload["events"] + .as_array() + .expect("events array") + .iter() + .find(|event| event["event_type"] == "codex_thread_initialized") + .expect("codex_thread_initialized event should be present") +} + +pub(crate) fn assert_basic_thread_initialized_event( + event: &Value, + thread_id: &str, + initialization_mode: &str, +) { + assert_eq!(event["event_params"]["thread_id"], thread_id); + assert_eq!( + event["event_params"]["product_client_id"], + DEFAULT_CLIENT_NAME + ); + assert_eq!(event["event_params"]["model"], "mock-model"); + assert_eq!(event["event_params"]["ephemeral"], false); + assert_eq!(event["event_params"]["session_source"], "user"); + assert_eq!( + event["event_params"]["subagent_source"], + serde_json::Value::Null + ); + assert_eq!( + event["event_params"]["parent_thread_id"], + serde_json::Value::Null + ); + assert_eq!( + event["event_params"]["initialization_mode"], + initialization_mode + ); + assert!(event["event_params"]["created_at"].as_u64().is_some()); +} diff --git a/codex-rs/app-server/tests/suite/v2/thread_fork.rs b/codex-rs/app-server/tests/suite/v2/thread_fork.rs index a81c87bd4e..141c7e7aed 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_fork.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_fork.rs @@ -38,6 +38,11 @@ use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; +use super::analytics::assert_basic_thread_initialized_event; +use super::analytics::enable_analytics_capture; +use super::analytics::thread_initialized_event; +use super::analytics::wait_for_analytics_payload; + const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] @@ -177,6 +182,45 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + + let codex_home = TempDir::new()?; + create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?; + enable_analytics_capture(&server, codex_home.path()).await?; + + let conversation_id = create_fake_rollout( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "Saved user message", + Some("mock_provider"), + None, + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let fork_id = mcp + .send_thread_fork_request(ThreadForkParams { + thread_id: conversation_id, + ..Default::default() + }) + .await?; + let fork_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(fork_id)), + ) + .await??; + let ThreadForkResponse { thread, .. } = to_response::(fork_resp)?; + + let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?; + let event = thread_initialized_event(&payload); + assert_basic_thread_initialized_event(event, &thread.id, "forked"); + Ok(()) +} + #[tokio::test] async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/app-server/tests/suite/v2/thread_resume.rs b/codex-rs/app-server/tests/suite/v2/thread_resume.rs index 4443abd6ec..5092faee9d 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_resume.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_resume.rs @@ -70,6 +70,11 @@ use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; +use super::analytics::assert_basic_thread_initialized_event; +use super::analytics::enable_analytics_capture; +use super::analytics::thread_initialized_event; +use super::analytics::wait_for_analytics_payload; + const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals."; @@ -150,6 +155,46 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + + let codex_home = TempDir::new()?; + create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?; + enable_analytics_capture(&server, codex_home.path()).await?; + + let conversation_id = create_fake_rollout_with_text_elements( + codex_home.path(), + "2025-01-05T12-00-00", + "2025-01-05T12:00:00Z", + "Saved user message", + Vec::new(), + Some("mock_provider"), + None, + )?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let resume_id = mcp + .send_thread_resume_request(ThreadResumeParams { + thread_id: conversation_id, + ..Default::default() + }) + .await?; + let resume_resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(resume_id)), + ) + .await??; + let ThreadResumeResponse { thread, .. } = to_response::(resume_resp)?; + + let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?; + let event = thread_initialized_event(&payload); + assert_basic_thread_initialized_event(event, &thread.id, "resumed"); + Ok(()) +} + #[tokio::test] async fn thread_resume_returns_rollout_history() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/app-server/tests/suite/v2/thread_start.rs b/codex-rs/app-server/tests/suite/v2/thread_start.rs index 1a73444bda..2a6c1644d6 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -34,6 +34,11 @@ use wiremock::ResponseTemplate; use wiremock::matchers::method; use wiremock::matchers::path; +use super::analytics::assert_basic_thread_initialized_event; +use super::analytics::enable_analytics_capture; +use super::analytics::thread_initialized_event; +use super::analytics::wait_for_analytics_payload; + const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] @@ -153,6 +158,34 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> { Ok(()) } +#[tokio::test] +async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> { + let server = create_mock_responses_server_repeating_assistant("Done").await; + + let codex_home = TempDir::new()?; + create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?; + enable_analytics_capture(&server, codex_home.path()).await?; + + let mut mcp = McpProcess::new(codex_home.path()).await?; + timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??; + + let req_id = mcp + .send_thread_start_request(ThreadStartParams::default()) + .await?; + let resp: JSONRPCResponse = timeout( + DEFAULT_READ_TIMEOUT, + mcp.read_stream_until_response_message(RequestId::Integer(req_id)), + ) + .await??; + let ThreadStartResponse { thread, .. } = to_response::(resp)?; + + let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?; + assert_eq!(payload["events"].as_array().expect("events array").len(), 1); + let event = thread_initialized_event(&payload); + assert_basic_thread_initialized_event(event, &thread.id, "new"); + Ok(()) +} + #[tokio::test] async fn thread_start_respects_project_config_from_cwd() -> Result<()> { let server = create_mock_responses_server_repeating_assistant("Done").await; diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 46472e9c8e..4cc1804b57 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -5,8 +5,6 @@ use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::sync::atomic::AtomicU64; -use std::time::SystemTime; -use std::time::UNIX_EPOCH; use crate::AuthManager; use crate::CodexAuth; @@ -350,9 +348,6 @@ use crate::util::backoff; use crate::windows_sandbox::WindowsSandboxLevelExt; use codex_analytics::AnalyticsEventsClient; use codex_analytics::AppInvocation; -use codex_analytics::CodexThreadContext; -use codex_analytics::CodexThreadInitializedInput; -use codex_analytics::InitializationMode; use codex_analytics::InvocationType; use codex_analytics::build_track_events_context; use codex_async_utils::OrCancelExt; @@ -637,12 +632,6 @@ impl Codex { user_shell_override, }; - let initialization_mode = initialization_mode(&conversation_history); - let thread_session_source = session_configuration.session_source.clone(); - let subagent_source = session_source_subagent_source(&thread_session_source); - let parent_thread_id = session_source_parent_thread_id(&thread_session_source); - let thread_initialized_configuration = session_configuration.clone(); - // Generate a unique ID for the lifetime of this Codex session. let session_source_clone = session_configuration.session_source.clone(); let (agent_status_tx, agent_status_rx) = watch::channel(AgentStatus::PendingInit); @@ -670,31 +659,6 @@ impl Codex { })?; let thread_id = session.conversation_id; - session - .services - .analytics_events_client - .track_thread_initialized(CodexThreadInitializedInput { - thread_id: thread_id.to_string(), - model: thread_initialized_configuration - .collaboration_mode - .model() - .to_string(), - product_client_id: crate::default_client::originator().value, - created_at: SystemTime::now() - .duration_since(UNIX_EPOCH) - .unwrap_or_default() - .as_secs(), - thread_context: CodexThreadContext { - ephemeral: thread_initialized_configuration - .original_config_do_not_use - .ephemeral, - session_source: thread_session_source, - initialization_mode, - subagent_source, - parent_thread_id, - }, - }); - // This task will run until Op::Shutdown is received. let session_for_loop = Arc::clone(&session); let session_loop_handle = tokio::spawn(async move { @@ -809,30 +773,6 @@ impl Codex { } } -fn initialization_mode(conversation_history: &InitialHistory) -> InitializationMode { - match conversation_history { - InitialHistory::New => InitializationMode::New, - InitialHistory::Forked(_) => InitializationMode::Forked, - InitialHistory::Resumed(_) => InitializationMode::Resumed, - } -} - -fn session_source_subagent_source(session_source: &SessionSource) -> Option { - match session_source { - SessionSource::SubAgent(subagent_source) => Some(subagent_source.clone()), - _ => None, - } -} - -fn session_source_parent_thread_id(session_source: &SessionSource) -> Option { - match session_source { - SessionSource::SubAgent(SubAgentSource::ThreadSpawn { - parent_thread_id, .. - }) => Some(parent_thread_id.to_string()), - _ => None, - } -} - #[cfg(test)] pub(crate) fn completed_session_loop_termination() -> SessionLoopTermination { futures::future::ready(()).boxed().shared() diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index 446a1a27e5..b5b094c5a4 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -136,6 +136,8 @@ pub type CodexConversation = CodexThread; pub use auth::AuthManager; pub use auth::CodexAuth; pub use codex_analytics::AnalyticsEventsClient; +pub use codex_analytics::InitializeInput; +pub use codex_analytics::ThreadInitializeInput; mod default_client_forwarding; /// Default Codex HTTP client headers and reqwest construction. diff --git a/codex-rs/core/tests/suite/analytics.rs b/codex-rs/core/tests/suite/analytics.rs deleted file mode 100644 index 3cf0e22ecd..0000000000 --- a/codex-rs/core/tests/suite/analytics.rs +++ /dev/null @@ -1,78 +0,0 @@ -use anyhow::Result; -use codex_core::CodexAuth; -use codex_core::config::Constrained; -use core_test_support::responses::start_mock_server; -use core_test_support::test_codex::test_codex; -use std::time::Duration; -use std::time::Instant; - -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn thread_initialization_tracks_thread_initialized_analytics() -> Result<()> { - let server = start_mock_server().await; - let chatgpt_base_url = server.uri(); - - let test = test_codex() - .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) - .with_config(move |config| { - config.chatgpt_base_url = chatgpt_base_url; - config.model = Some("gpt-5".to_string()); - config.permissions.sandbox_policy = Constrained::allow_any( - codex_protocol::protocol::SandboxPolicy::new_workspace_write_policy(), - ); - config.ephemeral = true; - }) - .build(&server) - .await?; - - let deadline = Instant::now() + Duration::from_secs(10); - let analytics_request = loop { - let requests = server.received_requests().await.unwrap_or_default(); - if let Some(request) = requests - .into_iter() - .find(|request| request.url.path() == "/codex/analytics-events/events") - { - break request; - } - if Instant::now() >= deadline { - panic!("timed out waiting for thread analytics request"); - } - tokio::time::sleep(Duration::from_millis(50)).await; - }; - - let payload: serde_json::Value = - serde_json::from_slice(&analytics_request.body).expect("analytics payload"); - let event = payload["events"] - .as_array() - .expect("events array") - .iter() - .find(|event| event["event_type"] == "codex_thread_initialized") - .expect("codex_thread_initialized event should be present"); - - assert_eq!( - event["event_params"]["thread_id"], - serde_json::json!(test.session_configured.session_id.to_string()) - ); - assert_eq!( - event["event_params"]["product_client_id"], - serde_json::json!(codex_core::default_client::originator().value) - ); - assert_eq!(event["event_params"]["model"], "gpt-5"); - assert_eq!(event["event_params"]["ephemeral"], true); - assert_eq!(event["event_params"]["session_source"], "user"); - assert_eq!(event["event_params"]["initialization_mode"], "new"); - assert_eq!( - event["event_params"]["subagent_source"], - serde_json::Value::Null - ); - assert_eq!( - event["event_params"]["parent_thread_id"], - serde_json::Value::Null - ); - assert!( - event["event_params"]["created_at"] - .as_u64() - .is_some_and(|timestamp| timestamp > 0) - ); - - Ok(()) -} diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index ff2da246fc..d540b01bbf 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -58,7 +58,6 @@ pub static CODEX_ALIASES_TEMP_DIR: TestCodexAliasesGuard = unsafe { mod abort_tasks; mod agent_jobs; mod agent_websocket; -mod analytics; mod apply_patch_cli; #[cfg(not(target_os = "windows"))] mod approvals;