diff --git a/codex-rs/core/src/analytics_client.rs b/codex-rs/core/src/analytics_client.rs index bfb90bc4da..86f96e62de 100644 --- a/codex-rs/core/src/analytics_client.rs +++ b/codex-rs/core/src/analytics_client.rs @@ -4,6 +4,15 @@ use crate::default_client::create_client; use crate::git_info::collect_git_info; use crate::git_info::get_git_repo_root; use crate::plugins::PluginTelemetryMetadata; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::config_types::ModeKind; +use codex_protocol::config_types::Personality; +use codex_protocol::config_types::ReasoningSummary; +use codex_protocol::config_types::ServiceTier; +use codex_protocol::openai_models::ReasoningEffort; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::SessionSource; use codex_protocol::protocol::SkillScope; use serde::Serialize; use sha1::Digest; @@ -23,6 +32,35 @@ pub(crate) struct TrackEventsContext { pub(crate) turn_id: String, } +#[derive(Clone)] +pub(crate) struct CodexThreadStartedEvent { + pub(crate) thread_id: String, + pub(crate) model: String, + pub(crate) model_provider: String, + pub(crate) reasoning_effort: Option, + pub(crate) reasoning_summary: Option, + pub(crate) service_tier: Option, + pub(crate) approval_policy: AskForApproval, + pub(crate) approvals_reviewer: ApprovalsReviewer, + pub(crate) sandbox_policy: SandboxPolicy, + pub(crate) sandbox_network_access: bool, + pub(crate) collaboration_mode: ModeKind, + pub(crate) personality: Option, + pub(crate) ephemeral: bool, + pub(crate) session_source: SessionSource, + pub(crate) parent_thread_id: Option, + pub(crate) create_source: ThreadCreateSource, + pub(crate) created_at: u64, +} + +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum ThreadCreateSource { + ThreadStart, + ThreadFork, + DetachedReview, +} + pub(crate) fn build_track_events_context( model_slug: String, thread_id: String, @@ -78,6 +116,9 @@ impl AnalyticsEventsQueue { TrackEventsJob::SkillInvocations(job) => { send_track_skill_invocations(&auth_manager, job).await; } + TrackEventsJob::ThreadStarted(job) => { + send_track_thread_started(&auth_manager, job).await; + } TrackEventsJob::AppMentioned(job) => { send_track_app_mentioned(&auth_manager, job).await; } @@ -167,6 +208,10 @@ impl AnalyticsEventsClient { ); } + pub(crate) fn track_thread_started(&self, thread_event: CodexThreadStartedEvent) { + track_thread_started(&self.queue, Arc::clone(&self.config), thread_event); + } + pub(crate) fn track_app_mentioned( &self, tracking: TrackEventsContext, @@ -236,6 +281,7 @@ impl AnalyticsEventsClient { enum TrackEventsJob { SkillInvocations(TrackSkillInvocationsJob), + ThreadStarted(TrackThreadStartedJob), AppMentioned(TrackAppMentionedJob), AppUsed(TrackAppUsedJob), PluginUsed(TrackPluginUsedJob), @@ -251,6 +297,11 @@ struct TrackSkillInvocationsJob { invocations: Vec, } +struct TrackThreadStartedJob { + config: Arc, + thread_event: CodexThreadStartedEvent, +} + struct TrackAppMentionedJob { config: Arc, tracking: TrackEventsContext, @@ -295,6 +346,7 @@ struct TrackEventsRequest { #[serde(untagged)] enum TrackEventRequest { SkillInvocation(SkillInvocationEventRequest), + ThreadStarted(CodexThreadStartedEventRequest), AppMentioned(CodexAppMentionedEventRequest), AppUsed(CodexAppUsedEventRequest), PluginUsed(CodexPluginUsedEventRequest), @@ -322,6 +374,34 @@ struct SkillInvocationEventParams { model_slug: Option, } +#[derive(Serialize)] +struct CodexThreadStartedEventParams { + thread_id: String, + product_client_id: String, + model: String, + model_provider: String, + reasoning_effort: Option, + reasoning_summary: Option, + service_tier: String, + approval_policy: String, + approvals_reviewer: String, + sandbox_policy: &'static str, + sandbox_network_access: bool, + collaboration_mode: &'static str, + personality: Option, + ephemeral: bool, + session_source: SessionSource, + parent_thread_id: Option, + create_source: ThreadCreateSource, + created_at: u64, +} + +#[derive(Serialize)] +struct CodexThreadStartedEventRequest { + event_type: &'static str, + event_params: CodexThreadStartedEventParams, +} + #[derive(Serialize)] struct CodexAppMetadata { connector_id: Option, @@ -400,6 +480,21 @@ pub(crate) fn track_skill_invocations( queue.try_send(job); } +pub(crate) fn track_thread_started( + queue: &AnalyticsEventsQueue, + config: Arc, + thread_event: CodexThreadStartedEvent, +) { + if config.analytics_enabled == Some(false) { + return; + } + let job = TrackEventsJob::ThreadStarted(TrackThreadStartedJob { + config, + thread_event, + }); + queue.try_send(job); +} + pub(crate) fn track_app_mentioned( queue: &AnalyticsEventsQueue, config: Arc, @@ -536,6 +631,21 @@ async fn send_track_skill_invocations(auth_manager: &AuthManager, job: TrackSkil send_track_events(auth_manager, config, events).await; } +async fn send_track_thread_started(auth_manager: &AuthManager, job: TrackThreadStartedJob) { + let TrackThreadStartedJob { + config, + thread_event, + } = job; + let events = vec![TrackEventRequest::ThreadStarted( + CodexThreadStartedEventRequest { + event_type: "codex_thread_started", + event_params: codex_thread_started_event_params(thread_event), + }, + )]; + + send_track_events(auth_manager, config, events).await; +} + async fn send_track_app_mentioned(auth_manager: &AuthManager, job: TrackAppMentionedJob) { let TrackAppMentionedJob { config, @@ -635,6 +745,36 @@ fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> Code } } +fn codex_thread_started_event_params( + thread_event: CodexThreadStartedEvent, +) -> CodexThreadStartedEventParams { + CodexThreadStartedEventParams { + thread_id: thread_event.thread_id, + product_client_id: crate::default_client::originator().value, + model: thread_event.model, + model_provider: thread_event.model_provider, + reasoning_effort: thread_event.reasoning_effort.map(|value| value.to_string()), + reasoning_summary: thread_event + .reasoning_summary + .map(|value| value.to_string()), + service_tier: thread_event + .service_tier + .map(|value| value.to_string()) + .unwrap_or_else(|| "default".to_string()), + approval_policy: thread_event.approval_policy.to_string(), + approvals_reviewer: thread_event.approvals_reviewer.to_string(), + sandbox_policy: sandbox_policy_mode(&thread_event.sandbox_policy), + sandbox_network_access: thread_event.sandbox_network_access, + collaboration_mode: collaboration_mode_mode(thread_event.collaboration_mode), + personality: thread_event.personality.map(|value| value.to_string()), + ephemeral: thread_event.ephemeral, + session_source: thread_event.session_source, + parent_thread_id: thread_event.parent_thread_id, + create_source: thread_event.create_source, + created_at: thread_event.created_at, + } +} + fn codex_plugin_metadata(plugin: PluginTelemetryMetadata) -> CodexPluginMetadata { let capability_summary = plugin.capability_summary; CodexPluginMetadata { @@ -670,6 +810,24 @@ fn codex_plugin_used_metadata( } } +fn sandbox_policy_mode(sandbox_policy: &SandboxPolicy) -> &'static str { + match sandbox_policy { + SandboxPolicy::DangerFullAccess => "full_access", + SandboxPolicy::ReadOnly { .. } => "read_only", + SandboxPolicy::WorkspaceWrite { .. } => "workspace_write", + SandboxPolicy::ExternalSandbox { .. } => "external_sandbox", + } +} + +fn collaboration_mode_mode(mode: ModeKind) -> &'static str { + match mode { + ModeKind::Plan => "plan", + ModeKind::Default => "default", + ModeKind::PairProgramming => "pair_programming", + ModeKind::Execute => "execute", + } +} + async fn send_track_events( auth_manager: &AuthManager, config: Arc, diff --git a/codex-rs/core/src/analytics_client_tests.rs b/codex-rs/core/src/analytics_client_tests.rs index a038aae047..60266937f3 100644 --- a/codex-rs/core/src/analytics_client_tests.rs +++ b/codex-rs/core/src/analytics_client_tests.rs @@ -4,17 +4,30 @@ use super::CodexAppMentionedEventRequest; use super::CodexAppUsedEventRequest; use super::CodexPluginEventRequest; use super::CodexPluginUsedEventRequest; +use super::CodexThreadStartedEvent; +use super::CodexThreadStartedEventRequest; use super::InvocationType; +use super::ThreadCreateSource; use super::TrackEventRequest; use super::TrackEventsContext; use super::codex_app_metadata; use super::codex_plugin_metadata; use super::codex_plugin_used_metadata; +use super::codex_thread_started_event_params; use super::normalize_path_for_skill_id; use crate::plugins::AppConnectorId; use crate::plugins::PluginCapabilitySummary; use crate::plugins::PluginId; use crate::plugins::PluginTelemetryMetadata; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::config_types::ModeKind; +use codex_protocol::config_types::Personality; +use codex_protocol::config_types::ReasoningSummary; +use codex_protocol::config_types::ServiceTier; +use codex_protocol::openai_models::ReasoningEffort; +use codex_protocol::protocol::AskForApproval; +use codex_protocol::protocol::SandboxPolicy; +use codex_protocol::protocol::SessionSource; use pretty_assertions::assert_eq; use serde_json::json; use std::collections::HashSet; @@ -185,6 +198,61 @@ fn app_used_dedupe_is_keyed_by_turn_and_connector() { assert_eq!(queue.should_enqueue_app_used(&turn_2, &app), true); } +#[test] +fn thread_started_event_serializes_expected_shape() { + let event = TrackEventRequest::ThreadStarted(CodexThreadStartedEventRequest { + event_type: "codex_thread_started", + event_params: codex_thread_started_event_params(CodexThreadStartedEvent { + thread_id: "thread-0".to_string(), + model: "gpt-5".to_string(), + model_provider: "openai".to_string(), + reasoning_effort: Some(ReasoningEffort::High), + reasoning_summary: Some(ReasoningSummary::Detailed), + service_tier: Some(ServiceTier::Flex), + approval_policy: AskForApproval::OnRequest, + approvals_reviewer: ApprovalsReviewer::GuardianSubagent, + sandbox_policy: SandboxPolicy::new_read_only_policy(), + sandbox_network_access: false, + collaboration_mode: ModeKind::Plan, + personality: Some(Personality::Friendly), + ephemeral: true, + session_source: SessionSource::Exec, + parent_thread_id: None, + create_source: ThreadCreateSource::ThreadStart, + created_at: 1_716_000_000, + }), + }); + + let payload = serde_json::to_value(&event).expect("serialize thread started event"); + + assert_eq!( + payload, + json!({ + "event_type": "codex_thread_started", + "event_params": { + "thread_id": "thread-0", + "product_client_id": crate::default_client::originator().value, + "model": "gpt-5", + "model_provider": "openai", + "reasoning_effort": "high", + "reasoning_summary": "detailed", + "service_tier": "flex", + "approval_policy": "on-request", + "approvals_reviewer": "guardian_subagent", + "sandbox_policy": "read_only", + "sandbox_network_access": false, + "collaboration_mode": "plan", + "personality": "friendly", + "ephemeral": true, + "session_source": "exec", + "parent_thread_id": null, + "create_source": "thread_start", + "created_at": 1716000000 + } + }) + ); +} + #[test] fn plugin_used_event_serializes_expected_shape() { let tracking = TrackEventsContext { diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 5e428362f2..ed36649f8d 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -5,6 +5,8 @@ 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; @@ -14,7 +16,9 @@ use crate::agent::AgentStatus; use crate::agent::agent_status_from_event; use crate::analytics_client::AnalyticsEventsClient; use crate::analytics_client::AppInvocation; +use crate::analytics_client::CodexThreadStartedEvent; use crate::analytics_client::InvocationType; +use crate::analytics_client::ThreadCreateSource; use crate::analytics_client::build_track_events_context; use crate::apps::render_apps_section; use crate::auth_env_telemetry::collect_auth_env_telemetry; @@ -627,6 +631,13 @@ impl Codex { user_shell_override, }; + let thread_create_source = classify_thread_create_source( + &conversation_history, + &session_configuration.session_source, + ); + let thread_session_source = session_configuration.session_source.clone(); + let thread_started_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); @@ -654,6 +665,46 @@ impl Codex { })?; let thread_id = session.conversation_id; + if let Some(create_source) = thread_create_source { + session + .services + .analytics_events_client + .track_thread_started(CodexThreadStartedEvent { + thread_id: thread_id.to_string(), + model: thread_started_configuration + .collaboration_mode + .model() + .to_string(), + model_provider: thread_started_configuration + .original_config_do_not_use + .model_provider_id + .clone(), + reasoning_effort: thread_started_configuration + .collaboration_mode + .reasoning_effort(), + reasoning_summary: thread_started_configuration.model_reasoning_summary, + service_tier: thread_started_configuration.service_tier, + approval_policy: thread_started_configuration.approval_policy.value(), + approvals_reviewer: thread_started_configuration.approvals_reviewer, + sandbox_policy: thread_started_configuration.sandbox_policy.get().clone(), + sandbox_network_access: thread_started_configuration + .network_sandbox_policy + .is_enabled(), + collaboration_mode: thread_started_configuration.collaboration_mode.mode, + personality: thread_started_configuration.personality, + ephemeral: thread_started_configuration + .original_config_do_not_use + .ephemeral, + parent_thread_id: session_source_parent_thread_id(&thread_session_source), + session_source: thread_session_source, + create_source, + created_at: SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_secs(), + }); + } + // This task will run until Op::Shutdown is received. let session_for_loop = Arc::clone(&session); let session_loop_handle = tokio::spawn(async move { @@ -768,6 +819,33 @@ impl Codex { } } +fn classify_thread_create_source( + conversation_history: &InitialHistory, + session_source: &SessionSource, +) -> Option { + if matches!( + session_source, + SessionSource::SubAgent(SubAgentSource::Review) + ) { + return Some(ThreadCreateSource::DetachedReview); + } + + match conversation_history { + InitialHistory::New => Some(ThreadCreateSource::ThreadStart), + InitialHistory::Forked(_) => Some(ThreadCreateSource::ThreadFork), + InitialHistory::Resumed(_) => 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/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index 5f7e50f061..d9a60c6075 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -130,6 +130,7 @@ mod stream_error_allows_next_turn; mod stream_no_completed; mod subagent_notifications; mod text_encoding_fix; +mod thread_metadata; mod tool_harness; mod tool_parallelism; mod tools; diff --git a/codex-rs/core/tests/suite/plugins.rs b/codex-rs/core/tests/suite/plugins.rs index 78df34652a..5448e49161 100644 --- a/codex-rs/core/tests/suite/plugins.rs +++ b/codex-rs/core/tests/suite/plugins.rs @@ -365,13 +365,22 @@ async fn explicit_plugin_mentions_track_plugin_used_analytics() -> Result<()> { wait_for_event(&codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; let deadline = Instant::now() + Duration::from_secs(10); - let analytics_request = loop { + let plugin_event = loop { let requests = server.received_requests().await.unwrap_or_default(); - if let Some(request) = requests + if let Some(event) = requests .into_iter() - .find(|request| request.url.path() == "/codex/analytics-events/events") + .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 request; + break event; } if Instant::now() >= deadline { panic!("timed out waiting for plugin analytics request"); @@ -379,10 +388,7 @@ async fn explicit_plugin_mentions_track_plugin_used_analytics() -> Result<()> { 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"][0]; - assert_eq!(event["event_type"], "codex_plugin_used"); + let event = plugin_event; assert_eq!(event["event_params"]["plugin_id"], "sample@test"); assert_eq!(event["event_params"]["plugin_name"], "sample"); assert_eq!(event["event_params"]["marketplace_name"], "test"); diff --git a/codex-rs/core/tests/suite/thread_metadata.rs b/codex-rs/core/tests/suite/thread_metadata.rs new file mode 100644 index 0000000000..874a486bf2 --- /dev/null +++ b/codex-rs/core/tests/suite/thread_metadata.rs @@ -0,0 +1,93 @@ +use anyhow::Result; +use codex_core::CodexAuth; +use codex_protocol::config_types::ApprovalsReviewer; +use codex_protocol::config_types::Personality; +use codex_protocol::config_types::ReasoningSummary; +use codex_protocol::config_types::ServiceTier; +use codex_protocol::openai_models::ReasoningEffort; +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_start_tracks_thread_started_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.model_reasoning_effort = Some(ReasoningEffort::High); + config.model_reasoning_summary = Some(ReasoningSummary::Detailed); + config.service_tier = Some(ServiceTier::Flex); + config.approvals_reviewer = ApprovalsReviewer::GuardianSubagent; + config.personality = Some(Personality::Friendly); + 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_started") + .expect("codex_thread_started 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"]["model_provider"], "openai"); + assert_eq!(event["event_params"]["reasoning_effort"], "high"); + assert_eq!(event["event_params"]["reasoning_summary"], "detailed"); + assert_eq!(event["event_params"]["service_tier"], "flex"); + assert_eq!(event["event_params"]["approval_policy"], "on-request"); + assert_eq!( + event["event_params"]["approvals_reviewer"], + "guardian_subagent" + ); + assert_eq!(event["event_params"]["sandbox_policy"], "workspace_write"); + assert_eq!(event["event_params"]["sandbox_network_access"], false); + assert_eq!(event["event_params"]["collaboration_mode"], "default"); + assert_eq!(event["event_params"]["personality"], "friendly"); + assert_eq!(event["event_params"]["ephemeral"], true); + assert_eq!(event["event_params"]["session_source"], "exec"); + assert_eq!( + event["event_params"]["parent_thread_id"], + serde_json::Value::Null + ); + assert_eq!(event["event_params"]["create_source"], "thread_start"); + assert!( + event["event_params"]["created_at"] + .as_u64() + .is_some_and(|timestamp| timestamp > 0) + ); + + Ok(()) +}