diff --git a/codex-rs/analytics/src/analytics_capture.rs b/codex-rs/analytics/src/analytics_capture.rs new file mode 100644 index 000000000..7e1dd6eb7 --- /dev/null +++ b/codex-rs/analytics/src/analytics_capture.rs @@ -0,0 +1,34 @@ +use crate::events::TrackEventsRequest; +use std::fs::File; +use std::fs::OpenOptions; +use std::io; +use std::io::Write; +use std::path::Path; + +pub(crate) const ANALYTICS_EVENTS_CAPTURE_FILE_ENV_VAR: &str = + "CODEX_ANALYTICS_EVENTS_CAPTURE_FILE"; + +pub(crate) fn initialize(path: &Path) -> io::Result<()> { + open_capture_file(path).map(drop) +} + +pub(crate) fn append_payload(path: &Path, payload: &TrackEventsRequest) -> io::Result<()> { + let mut line = serde_json::to_vec(payload) + .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?; + line.push(b'\n'); + + let mut file = open_capture_file(path)?; + file.write_all(&line)?; + file.flush() +} + +fn open_capture_file(path: &Path) -> io::Result { + let mut options = OpenOptions::new(); + options.create(true).append(true); + #[cfg(unix)] + { + use std::os::unix::fs::OpenOptionsExt; + options.mode(0o600); + } + options.open(path) +} diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index 193de9ef6..e88c0114a 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -38,6 +38,7 @@ use codex_login::default_client::create_client; use codex_plugin::PluginTelemetryMetadata; use codex_protocol::request_permissions::RequestPermissionsResponse; use std::collections::HashSet; +use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; use std::time::Duration; @@ -59,15 +60,70 @@ pub struct AnalyticsEventsClient { queue: Option, } +#[derive(Clone, Debug, Eq, PartialEq)] +enum AnalyticsEventsDestination { + Http { + url: String, + }, + #[cfg(debug_assertions)] + CaptureFile { + path: PathBuf, + }, +} + +impl AnalyticsEventsDestination { + fn from_base_url(base_url: String) -> Self { + let capture_file = analytics_capture_file_from_env(); + Self::from_base_url_and_capture_file(base_url, capture_file) + } + + fn from_base_url_and_capture_file(base_url: String, capture_file: Option) -> Self { + #[cfg(debug_assertions)] + if let Some(path) = capture_file { + if let Err(err) = crate::analytics_capture::initialize(&path) { + tracing::error!( + path = %path.display(), + "failed to initialize analytics event capture; network delivery remains disabled: {err}" + ); + } + tracing::warn!( + path = %path.display(), + "analytics event capture enabled; network delivery is disabled" + ); + return Self::CaptureFile { path }; + } + + #[cfg(not(debug_assertions))] + let _ = capture_file; + + let base_url = base_url.trim_end_matches('/'); + Self::Http { + url: format!("{base_url}/codex/analytics-events/events"), + } + } +} + +fn analytics_capture_file_from_env() -> Option { + #[cfg(debug_assertions)] + { + std::env::var_os(crate::analytics_capture::ANALYTICS_EVENTS_CAPTURE_FILE_ENV_VAR) + .filter(|value| !value.is_empty()) + .map(PathBuf::from) + } + + #[cfg(not(debug_assertions))] + None +} + impl AnalyticsEventsQueue { - pub(crate) fn new(auth_manager: Arc, base_url: String) -> Self { + fn new(auth_manager: Arc, destination: AnalyticsEventsDestination) -> Self { let (sender, mut receiver) = mpsc::channel(ANALYTICS_EVENTS_QUEUE_SIZE); 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; - send_track_events(&auth_manager, &base_url, events).await; + send_track_events(&auth_manager, &destination, events).await; } }); Self { @@ -124,9 +180,10 @@ impl AnalyticsEventsClient { base_url: String, analytics_enabled: Option, ) -> Self { + let destination = AnalyticsEventsDestination::from_base_url(base_url); Self { queue: (analytics_enabled != Some(false)) - .then(|| AnalyticsEventsQueue::new(Arc::clone(&auth_manager), base_url)), + .then(|| AnalyticsEventsQueue::new(Arc::clone(&auth_manager), destination)), } } @@ -410,7 +467,7 @@ impl AnalyticsEventsClient { async fn send_track_events( auth_manager: &AuthManager, - base_url: &str, + destination: &AnalyticsEventsDestination, events: Vec, ) { if events.is_empty() { @@ -424,10 +481,8 @@ async fn send_track_events( 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; + send_track_events_request(&auth, destination, events).await; } } @@ -454,13 +509,27 @@ fn track_event_request_batches(events: Vec) -> Vec) { +async fn send_track_events_request( + auth: &CodexAuth, + destination: &AnalyticsEventsDestination, + events: Vec, +) { if events.is_empty() { return; } let payload = TrackEventsRequest { events }; + #[cfg(debug_assertions)] + if capture_track_events_request(destination, &payload) { + return; + } + + let url = match destination { + AnalyticsEventsDestination::Http { url } => url, + #[cfg(debug_assertions)] + AnalyticsEventsDestination::CaptureFile { .. } => return, + }; let response = create_client() .post(url) .timeout(ANALYTICS_EVENTS_TIMEOUT) @@ -483,6 +552,24 @@ async fn send_track_events_request(auth: &CodexAuth, url: &str, events: Vec bool { + let AnalyticsEventsDestination::CaptureFile { path } = destination else { + return false; + }; + + if let Err(err) = crate::analytics_capture::append_payload(path, payload) { + tracing::error!( + path = %path.display(), + "failed to capture analytics events; network delivery remains disabled: {err}" + ); + } + true +} + #[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 b7aa9f1c9..1835932e8 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -1,5 +1,10 @@ use super::AnalyticsEventsClient; +use super::AnalyticsEventsDestination; use super::AnalyticsEventsQueue; +#[cfg(debug_assertions)] +use super::capture_track_events_request; +#[cfg(debug_assertions)] +use super::send_track_events_request; use super::track_event_request_batches; use crate::events::CodexAcceptedLineFingerprintsEventParams; use crate::events::CodexAcceptedLineFingerprintsEventRequest; @@ -31,8 +36,14 @@ use codex_app_server_protocol::TurnSteerResponse; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use std::collections::HashSet; +#[cfg(debug_assertions)] +use std::fs; +#[cfg(debug_assertions)] +use std::path::PathBuf; use std::sync::Arc; use std::sync::Mutex; +#[cfg(debug_assertions)] +use std::time::SystemTime; use tokio::sync::mpsc; use tokio::sync::mpsc::error::TryRecvError; @@ -74,6 +85,18 @@ fn sample_regular_track_event(thread_id: &str) -> TrackEventRequest { }) } +#[cfg(debug_assertions)] +fn unique_capture_path(name: &str) -> PathBuf { + let nonce = SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .expect("system clock should be after Unix epoch") + .as_nanos(); + std::env::temp_dir().join(format!( + "codex-analytics-{name}-{}-{nonce}.jsonl", + std::process::id() + )) +} + fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver) { let (sender, receiver) = mpsc::channel(8); let queue = AnalyticsEventsQueue { @@ -84,6 +107,138 @@ fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver>(); + assert_eq!(lines.len(), 1); + let payload: serde_json::Value = + serde_json::from_str(lines[0]).expect("parse captured payload"); + assert_eq!(payload, serde_json::json!({"events": [expected_event]})); + + fs::remove_file(capture_path).expect("remove capture file"); +} + +#[tokio::test] +#[cfg(debug_assertions)] +async fn capture_file_writes_final_batches_as_separate_lines() { + let capture_path = unique_capture_path("batches"); + let destination = AnalyticsEventsDestination::CaptureFile { + path: capture_path.clone(), + }; + let auth = codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing(); + let events = vec![ + sample_regular_track_event("thread-1"), + sample_accepted_line_fingerprint_event("thread-2"), + sample_regular_track_event("thread-3"), + ]; + + for batch in track_event_request_batches(events) { + send_track_events_request(&auth, &destination, batch).await; + } + + let contents = fs::read_to_string(&capture_path).expect("read capture file"); + let payloads = contents + .lines() + .map(|line| serde_json::from_str::(line).expect("parse capture line")) + .collect::>(); + assert_eq!(payloads.len(), 3); + assert_eq!(payloads[0]["events"][0]["skill_id"], "skill-thread-1"); + assert_eq!( + payloads[1]["events"][0]["event_type"], + "codex_accepted_line_fingerprints" + ); + assert_eq!(payloads[2]["events"][0]["skill_id"], "skill-thread-3"); + + fs::remove_file(capture_path).expect("remove capture file"); +} + +#[test] +#[cfg(debug_assertions)] +fn capture_write_failure_still_consumes_delivery() { + let capture_path = unique_capture_path("missing-parent").join("events.jsonl"); + let destination = AnalyticsEventsDestination::CaptureFile { path: capture_path }; + let payload = crate::events::TrackEventsRequest { + events: vec![sample_regular_track_event("thread-1")], + }; + + assert!(capture_track_events_request(&destination, &payload)); +} + fn sample_turn_start_request() -> ClientRequest { ClientRequest::TurnStart { request_id: RequestId::Integer(1), diff --git a/codex-rs/analytics/src/lib.rs b/codex-rs/analytics/src/lib.rs index 6e4723275..687a3a763 100644 --- a/codex-rs/analytics/src/lib.rs +++ b/codex-rs/analytics/src/lib.rs @@ -1,4 +1,6 @@ mod accepted_lines; +#[cfg(debug_assertions)] +mod analytics_capture; mod client; mod events; mod facts;