diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 9288e12097..65e8b9ed14 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -1,8 +1,8 @@ +use crate::event_sink::AnalyticsEvent; +use crate::event_sink::AnalyticsEventSink; use crate::events::AppServerRpcTransport; use crate::events::GuardianReviewAnalyticsResult; use crate::events::GuardianReviewTrackContext; -use crate::events::TrackEventRequest; -use crate::events::TrackEventsRequest; use crate::events::current_runtime_metadata; use crate::facts::AnalyticsFact; use crate::facts::AnalyticsJsonRpcError; @@ -23,7 +23,6 @@ use crate::facts::TurnProfileFact; use crate::facts::TurnResolvedConfigFact; use crate::facts::TurnTokenUsageFact; use crate::local_sink::SharedLocalAnalyticsSink; -use crate::local_sink::append_codex_analytics_events_best_effort; use crate::local_sink::local_analytics_sink_from_env; use crate::reducer::AnalyticsReducer; use codex_app_server_protocol::ClientRequest; @@ -35,14 +34,11 @@ use codex_app_server_protocol::ServerNotification; use codex_app_server_protocol::ServerRequest; use codex_app_server_protocol::ServerResponse; use codex_login::AuthManager; -use codex_login::CodexAuth; -use codex_login::default_client::create_client; use codex_plugin::PluginTelemetryMetadata; use codex_protocol::request_permissions::RequestPermissionsResponse; use std::collections::HashSet; use std::sync::Arc; use std::sync::Mutex; -use std::time::Duration; use tokio::sync::mpsc; #[cfg(test)] @@ -51,7 +47,6 @@ use crate::local_sink::local_analytics_sink_for_path; use std::path::PathBuf; const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256; -const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10); const ANALYTICS_EVENT_DEDUPE_MAX_KEYS: usize = 4096; #[derive(Clone)] @@ -67,24 +62,19 @@ pub struct AnalyticsEventsClient { } impl AnalyticsEventsQueue { - pub(crate) fn new( - auth_manager: Arc, - base_url: String, - backend_analytics_enabled: bool, - local_sink: Option, - ) -> Self { + pub(crate) fn new(sinks: Vec) -> Self { let (sender, mut receiver) = mpsc::channel(ANALYTICS_EVENTS_QUEUE_SIZE); - let worker_local_sink = local_sink.clone(); tokio::spawn(async move { let mut reducer = AnalyticsReducer::default(); while let Some(input) = receiver.recv().await { - let mut events = Vec::new(); - reducer.ingest(input, &mut events).await; - if let Some(local_sink) = worker_local_sink.as_ref() { - append_codex_analytics_events_best_effort(local_sink, &events); - } - if backend_analytics_enabled { - send_track_events(&auth_manager, &base_url, events).await; + let mut codex_analytics_events = Vec::new(); + reducer.ingest(input, &mut codex_analytics_events).await; + let events = codex_analytics_events + .into_iter() + .map(AnalyticsEvent::from) + .collect::>(); + for sink in &sinks { + sink.write(&events).await; } } }); @@ -175,16 +165,18 @@ impl AnalyticsEventsClient { analytics_enabled: Option, local_sink: Option, ) -> Self { - let backend_analytics_enabled = analytics_enabled != Some(false); + let mut sinks = Vec::new(); + if let Some(local_sink) = local_sink { + sinks.push(AnalyticsEventSink::Local(local_sink)); + } + if analytics_enabled != Some(false) { + sinks.push(AnalyticsEventSink::CodexBackend { + auth_manager, + base_url, + }); + } Self { - queue: (backend_analytics_enabled || local_sink.is_some()).then(|| { - AnalyticsEventsQueue::new( - auth_manager, - base_url, - backend_analytics_enabled, - local_sink, - ) - }), + queue: (!sinks.is_empty()).then(|| AnalyticsEventsQueue::new(sinks)), } } @@ -456,81 +448,6 @@ impl AnalyticsEventsClient { } } -async fn send_track_events( - auth_manager: &AuthManager, - base_url: &str, - events: Vec, -) { - if events.is_empty() { - return; - } - - let Some(auth) = auth_manager.auth().await else { - return; - }; - if !auth.uses_codex_backend() { - return; - } - - let base_url = base_url.trim_end_matches('/'); - let url = format!("{base_url}/codex/analytics-events/events"); - for events in track_event_request_batches(events) { - send_track_events_request(&auth, &url, events).await; - } -} - -fn track_event_request_batches(events: Vec) -> Vec> { - let mut batches = Vec::new(); - let mut current_batch = Vec::new(); - - for event in events { - if event.should_send_in_isolated_request() { - if !current_batch.is_empty() { - batches.push(current_batch); - current_batch = Vec::new(); - } - batches.push(vec![event]); - } else { - current_batch.push(event); - } - } - - if !current_batch.is_empty() { - batches.push(current_batch); - } - - batches -} - -async fn send_track_events_request(auth: &CodexAuth, url: &str, events: Vec) { - if events.is_empty() { - return; - } - - let payload = TrackEventsRequest { events }; - - let response = create_client() - .post(url) - .timeout(ANALYTICS_EVENTS_TIMEOUT) - .headers(codex_model_provider::auth_provider_from_auth(auth).to_auth_headers()) - .header("Content-Type", "application/json") - .json(&payload) - .send() - .await; - - match response { - Ok(response) if response.status().is_success() => {} - Ok(response) => { - let status = response.status(); - let body = response.text().await.unwrap_or_default(); - tracing::warn!("events failed with status {status}: {body}"); - } - Err(err) => { - tracing::warn!("failed to send events request: {err}"); - } - } -} - #[cfg(test)] #[path = "client_tests.rs"] mod tests; diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 3f99590531..32488c2ad1 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -1,15 +1,8 @@ use super::AnalyticsEventsClient; use super::AnalyticsEventsQueue; -use super::track_event_request_batches; use crate::LocalAnalyticsRecord; use crate::events::AppServerRpcTransport; -use crate::events::CodexAcceptedLineFingerprintsEventParams; -use crate::events::CodexAcceptedLineFingerprintsEventRequest; -use crate::events::SkillInvocationEventParams; -use crate::events::SkillInvocationEventRequest; -use crate::events::TrackEventRequest; use crate::facts::AnalyticsFact; -use crate::facts::InvocationType; use codex_app_server_protocol::ApprovalsReviewer as AppServerApprovalsReviewer; use codex_app_server_protocol::AskForApproval as AppServerAskForApproval; use codex_app_server_protocol::ClientInfo; @@ -49,44 +42,6 @@ use tokio::sync::mpsc::error::TryRecvError; static NEXT_TEST_PATH_ID: AtomicU64 = AtomicU64::new(0); -fn sample_accepted_line_fingerprint_event(thread_id: &str) -> TrackEventRequest { - TrackEventRequest::AcceptedLineFingerprints(Box::new( - CodexAcceptedLineFingerprintsEventRequest { - event_type: "codex_accepted_line_fingerprints", - event_params: CodexAcceptedLineFingerprintsEventParams { - event_type: "codex.accepted_line_fingerprints", - turn_id: "turn-1".to_string(), - thread_id: thread_id.to_string(), - product_surface: Some("codex".to_string()), - model_slug: Some("gpt-5.1-codex".to_string()), - completed_at: 1, - repo_hash: None, - accepted_added_lines: 1, - accepted_deleted_lines: 0, - line_fingerprints: Vec::new(), - }, - }, - )) -} - -fn sample_regular_track_event(thread_id: &str) -> TrackEventRequest { - TrackEventRequest::SkillInvocation(SkillInvocationEventRequest { - event_type: "skill_invocation", - skill_id: format!("skill-{thread_id}"), - skill_name: "doc".to_string(), - event_params: SkillInvocationEventParams { - product_client_id: None, - skill_scope: None, - plugin_id: None, - repo_url: None, - thread_id: Some(thread_id.to_string()), - turn_id: Some("turn-1".to_string()), - invoke_type: Some(InvocationType::Explicit), - model_slug: Some("gpt-5.1-codex".to_string()), - }, - }) -} - fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver) { let (sender, receiver) = mpsc::channel(8); let queue = AnalyticsEventsQueue { @@ -320,26 +275,6 @@ fn track_response_only_enqueues_analytics_relevant_responses() { assert!(matches!(receiver.try_recv(), Err(TryRecvError::Empty))); } -#[test] -fn track_event_request_batches_only_isolates_accepted_line_fingerprint_events() { - let batches = track_event_request_batches(vec![ - sample_regular_track_event("thread-1"), - sample_regular_track_event("thread-2"), - sample_accepted_line_fingerprint_event("thread-3"), - sample_accepted_line_fingerprint_event("thread-4"), - sample_regular_track_event("thread-5"), - sample_regular_track_event("thread-6"), - ]); - - assert_eq!(batches.len(), 4); - assert_eq!(batches[0].len(), 2); - assert_eq!(batches[1].len(), 1); - assert_eq!(batches[2].len(), 1); - assert_eq!(batches[3].len(), 2); - assert!(batches[1][0].should_send_in_isolated_request()); - assert!(batches[2][0].should_send_in_isolated_request()); -} - async fn wait_for_local_records(path: &PathBuf, min_records: usize) -> Vec { for _ in 0..100 { if let Ok(contents) = fs::read_to_string(path) { diff --git a/codex-rs/analytics/src/event_sink.rs b/codex-rs/analytics/src/event_sink.rs new file mode 100644 index 0000000000..874e811816 --- /dev/null +++ b/codex-rs/analytics/src/event_sink.rs @@ -0,0 +1,192 @@ +use crate::events::TrackEventRequest; +use crate::events::TrackEventsRequest; +use crate::local_sink::SharedLocalAnalyticsSink; +use crate::local_sink::append_codex_analytics_event_best_effort; +use codex_login::AuthManager; +use codex_login::CodexAuth; +use codex_login::default_client::create_client; +use std::sync::Arc; +use std::time::Duration; + +const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10); +const CODEX_ANALYTICS_EVENT_SINKS: AnalyticsEventSinkSet = + AnalyticsEventSinkSet::CODEX_BACKEND.union(AnalyticsEventSinkSet::LOCAL); + +pub(crate) enum AnalyticsEvent { + CodexAnalytics(TrackEventRequest), +} + +impl AnalyticsEvent { + fn writable_sinks(&self) -> AnalyticsEventSinkSet { + match self { + Self::CodexAnalytics(_) => CODEX_ANALYTICS_EVENT_SINKS, + } + } +} + +impl From for AnalyticsEvent { + fn from(event: TrackEventRequest) -> Self { + Self::CodexAnalytics(event) + } +} + +pub(crate) enum AnalyticsEventSink { + CodexBackend { + auth_manager: Arc, + base_url: String, + }, + Local(SharedLocalAnalyticsSink), +} + +impl AnalyticsEventSink { + pub(crate) async fn write(&self, events: &[AnalyticsEvent]) { + match self { + Self::CodexBackend { + auth_manager, + base_url, + } => { + let events = writable_events(events, self.kind()) + .map(|event| match event { + AnalyticsEvent::CodexAnalytics(event) => event, + }) + .collect::>(); + send_track_events(auth_manager, base_url, &events).await; + } + Self::Local(sink) => { + for event in writable_events(events, self.kind()) { + match event { + AnalyticsEvent::CodexAnalytics(event) => { + append_codex_analytics_event_best_effort(sink, event); + } + } + } + } + } + } + + fn kind(&self) -> AnalyticsEventSinkKind { + match self { + Self::CodexBackend { .. } => AnalyticsEventSinkKind::CodexBackend, + Self::Local(_) => AnalyticsEventSinkKind::Local, + } + } +} + +#[derive(Clone, Copy, Eq, PartialEq)] +enum AnalyticsEventSinkKind { + CodexBackend, + Local, +} + +impl AnalyticsEventSinkKind { + const fn flag(self) -> u8 { + match self { + Self::CodexBackend => 1 << 0, + Self::Local => 1 << 1, + } + } +} + +#[derive(Clone, Copy)] +struct AnalyticsEventSinkSet(u8); + +impl AnalyticsEventSinkSet { + const CODEX_BACKEND: Self = Self(AnalyticsEventSinkKind::CodexBackend.flag()); + const LOCAL: Self = Self(AnalyticsEventSinkKind::Local.flag()); + + const fn union(self, other: Self) -> Self { + Self(self.0 | other.0) + } + + fn contains(self, sink: AnalyticsEventSinkKind) -> bool { + self.0 & sink.flag() != 0 + } +} + +fn writable_events( + events: &[AnalyticsEvent], + sink: AnalyticsEventSinkKind, +) -> impl Iterator { + events + .iter() + .filter(move |event| event.writable_sinks().contains(sink)) +} + +async fn send_track_events( + auth_manager: &AuthManager, + base_url: &str, + events: &[&TrackEventRequest], +) { + if events.is_empty() { + return; + } + + let Some(auth) = auth_manager.auth().await else { + return; + }; + if !auth.uses_codex_backend() { + return; + } + + let base_url = base_url.trim_end_matches('/'); + let url = format!("{base_url}/codex/analytics-events/events"); + for events in track_event_request_batches(events) { + send_track_events_request(&auth, &url, events).await; + } +} + +fn track_event_request_batches<'a>( + events: &'a [&'a TrackEventRequest], +) -> Vec<&'a [&'a TrackEventRequest]> { + let mut batches = Vec::new(); + let mut current_batch_start = 0; + + for (index, event) in events.iter().enumerate() { + if event.should_send_in_isolated_request() { + if current_batch_start < index { + batches.push(&events[current_batch_start..index]); + } + batches.push(&events[index..=index]); + current_batch_start = index + 1; + } + } + + if current_batch_start < events.len() { + batches.push(&events[current_batch_start..]); + } + + batches +} + +async fn send_track_events_request(auth: &CodexAuth, url: &str, events: &[&TrackEventRequest]) { + if events.is_empty() { + return; + } + + let payload = TrackEventsRequest { events }; + + let response = create_client() + .post(url) + .timeout(ANALYTICS_EVENTS_TIMEOUT) + .headers(codex_model_provider::auth_provider_from_auth(auth).to_auth_headers()) + .header("Content-Type", "application/json") + .json(&payload) + .send() + .await; + + match response { + Ok(response) if response.status().is_success() => {} + Ok(response) => { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + tracing::warn!("events failed with status {status}: {body}"); + } + Err(err) => { + tracing::warn!("failed to send events request: {err}"); + } + } +} + +#[cfg(test)] +#[path = "event_sink_tests.rs"] +mod tests; diff --git a/codex-rs/analytics/src/event_sink_tests.rs b/codex-rs/analytics/src/event_sink_tests.rs new file mode 100644 index 0000000000..775fd6c38e --- /dev/null +++ b/codex-rs/analytics/src/event_sink_tests.rs @@ -0,0 +1,85 @@ +use super::track_event_request_batches; +use crate::events::CodexAcceptedLineFingerprintsEventParams; +use crate::events::CodexAcceptedLineFingerprintsEventRequest; +use crate::events::SkillInvocationEventParams; +use crate::events::SkillInvocationEventRequest; +use crate::events::TrackEventRequest; +use crate::events::TrackEventsRequest; +use crate::facts::InvocationType; +use serde_json::json; + +#[test] +fn track_event_request_batches_only_isolates_accepted_line_fingerprint_events() { + let events = [ + sample_regular_track_event("thread-1"), + sample_regular_track_event("thread-2"), + sample_accepted_line_fingerprint_event("thread-3"), + sample_accepted_line_fingerprint_event("thread-4"), + sample_regular_track_event("thread-5"), + sample_regular_track_event("thread-6"), + ]; + let events = events.iter().collect::>(); + let batches = track_event_request_batches(&events); + + assert_eq!(batches.len(), 4); + assert_eq!(batches[0].len(), 2); + assert_eq!(batches[1].len(), 1); + assert_eq!(batches[2].len(), 1); + assert_eq!(batches[3].len(), 2); + assert!(batches[1][0].should_send_in_isolated_request()); + assert!(batches[2][0].should_send_in_isolated_request()); +} + +#[test] +fn track_events_request_serializes_borrowed_events() { + let event = sample_regular_track_event("thread-1"); + let events = [&event]; + + let payload = serde_json::to_value(TrackEventsRequest { events: &events }) + .expect("serialize track events request"); + + assert_eq!( + payload, + json!({ + "events": [serde_json::to_value(&event).expect("serialize track event")] + }) + ); +} + +fn sample_accepted_line_fingerprint_event(thread_id: &str) -> TrackEventRequest { + TrackEventRequest::AcceptedLineFingerprints(Box::new( + CodexAcceptedLineFingerprintsEventRequest { + event_type: "codex_accepted_line_fingerprints", + event_params: CodexAcceptedLineFingerprintsEventParams { + event_type: "codex.accepted_line_fingerprints", + turn_id: "turn-1".to_string(), + thread_id: thread_id.to_string(), + product_surface: Some("codex".to_string()), + model_slug: Some("gpt-5.1-codex".to_string()), + completed_at: 1, + repo_hash: None, + accepted_added_lines: 1, + accepted_deleted_lines: 0, + line_fingerprints: Vec::new(), + }, + }, + )) +} + +fn sample_regular_track_event(thread_id: &str) -> TrackEventRequest { + TrackEventRequest::SkillInvocation(SkillInvocationEventRequest { + event_type: "skill_invocation", + skill_id: format!("skill-{thread_id}"), + skill_name: "doc".to_string(), + event_params: SkillInvocationEventParams { + product_client_id: None, + skill_scope: None, + plugin_id: None, + repo_url: None, + thread_id: Some(thread_id.to_string()), + turn_id: Some("turn-1".to_string()), + invoke_type: Some(InvocationType::Explicit), + model_slug: Some("gpt-5.1-codex".to_string()), + }, + }) +} diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index 0b6c9a0122..9693592aa2 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -49,8 +49,8 @@ pub enum AppServerRpcTransport { } #[derive(Serialize)] -pub(crate) struct TrackEventsRequest { - pub(crate) events: Vec, +pub(crate) struct TrackEventsRequest<'a> { + pub(crate) events: &'a [&'a TrackEventRequest], } #[derive(Serialize)] diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 586828635b..233b0442e2 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -1,5 +1,6 @@ mod accepted_lines; mod client; +mod event_sink; mod events; mod facts; mod local_sink; diff --git a/codex-rs/analytics/src/local_sink.rs b/codex-rs/analytics/src/local_sink.rs index 5ab324d925..2cb51feb0d 100644 --- a/codex-rs/analytics/src/local_sink.rs +++ b/codex-rs/analytics/src/local_sink.rs @@ -50,16 +50,14 @@ pub(crate) fn local_analytics_sink_from_env() -> Option