Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 34 additions & 0 deletions codex-rs/analytics/src/analytics_capture.rs
Original file line number Diff line number Diff line change
@@ -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<File> {
let mut options = OpenOptions::new();
options.create(true).append(true);
#[cfg(unix)]
{
use std::os::unix::fs::OpenOptionsExt;
options.mode(0o600);
}
options.open(path)
}
103 changes: 95 additions & 8 deletions codex-rs/analytics/src/client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,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;
Expand All @@ -58,15 +59,70 @@ pub struct AnalyticsEventsClient {
queue: Option<AnalyticsEventsQueue>,
}

#[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<PathBuf>) -> 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<PathBuf> {
#[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<AuthManager>, base_url: String) -> Self {
fn new(auth_manager: Arc<AuthManager>, 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 {
Expand Down Expand Up @@ -123,9 +179,10 @@ impl AnalyticsEventsClient {
base_url: String,
analytics_enabled: Option<bool>,
) -> 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)),
}
}

Expand Down Expand Up @@ -403,7 +460,7 @@ impl AnalyticsEventsClient {

async fn send_track_events(
auth_manager: &AuthManager,
base_url: &str,
destination: &AnalyticsEventsDestination,
events: Vec<TrackEventRequest>,
) {
if events.is_empty() {
Expand All @@ -417,10 +474,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;
}
}

Expand All @@ -447,13 +502,27 @@ fn track_event_request_batches(events: Vec<TrackEventRequest>) -> Vec<Vec<TrackE
batches
}

async fn send_track_events_request(auth: &CodexAuth, url: &str, events: Vec<TrackEventRequest>) {
async fn send_track_events_request(
auth: &CodexAuth,
destination: &AnalyticsEventsDestination,
events: Vec<TrackEventRequest>,
) {
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)
Expand All @@ -476,6 +545,24 @@ async fn send_track_events_request(auth: &CodexAuth, url: &str, events: Vec<Trac
}
}

#[cfg(debug_assertions)]
fn capture_track_events_request(
destination: &AnalyticsEventsDestination,
payload: &TrackEventsRequest,
) -> 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;
155 changes: 155 additions & 0 deletions codex-rs/analytics/src/client_tests.rs
Original file line number Diff line number Diff line change
@@ -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;
Expand Down Expand Up @@ -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;

Expand Down Expand Up @@ -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<AnalyticsFact>) {
let (sender, receiver) = mpsc::channel(8);
let queue = AnalyticsEventsQueue {
Expand All @@ -84,6 +107,138 @@ fn client_with_receiver() -> (AnalyticsEventsClient, mpsc::Receiver<AnalyticsFac
(AnalyticsEventsClient { queue: Some(queue) }, receiver)
}

#[test]
#[cfg(debug_assertions)]
fn analytics_destination_uses_explicit_capture_file() {
let capture_path = unique_capture_path("destination");
let destination = AnalyticsEventsDestination::from_base_url_and_capture_file(
"https://chatgpt.com/backend-api/".to_string(),
Some(capture_path.clone()),
);

assert_eq!(
destination,
AnalyticsEventsDestination::CaptureFile {
path: capture_path.clone()
}
);
assert_eq!(
fs::read_to_string(&capture_path).expect("read capture file"),
""
);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;

let mode = fs::metadata(&capture_path)
.expect("read capture file metadata")
.permissions()
.mode();
assert_eq!(mode & 0o777, 0o600);
}
fs::remove_file(capture_path).expect("remove capture file");
}

#[test]
fn analytics_destination_uses_http_without_capture_file() {
let destination = AnalyticsEventsDestination::from_base_url_and_capture_file(
"https://chatgpt.com/backend-api/".to_string(),
/*capture_file*/ None,
);

assert_eq!(
destination,
AnalyticsEventsDestination::Http {
url: "https://chatgpt.com/backend-api/codex/analytics-events/events".to_string()
}
);
}

#[test]
#[cfg(not(debug_assertions))]
fn analytics_destination_ignores_capture_file_in_release() {
let destination = AnalyticsEventsDestination::from_base_url_and_capture_file(
"https://chatgpt.com/backend-api/".to_string(),
Some(std::path::PathBuf::from("ignored.jsonl")),
);

assert_eq!(
destination,
AnalyticsEventsDestination::Http {
url: "https://chatgpt.com/backend-api/codex/analytics-events/events".to_string()
}
);
}

#[tokio::test]
#[cfg(debug_assertions)]
async fn capture_file_writes_exact_serialized_request() {
let capture_path = unique_capture_path("single");
let destination = AnalyticsEventsDestination::CaptureFile {
path: capture_path.clone(),
};
let event = sample_regular_track_event("thread-1");
let expected_event = serde_json::to_value(&event).expect("serialize expected event");
let auth = codex_login::CodexAuth::create_dummy_chatgpt_auth_for_testing();

send_track_events_request(&auth, &destination, vec![event]).await;

let contents = fs::read_to_string(&capture_path).expect("read capture file");
let lines = contents.lines().collect::<Vec<_>>();
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::<serde_json::Value>(line).expect("parse capture line"))
.collect::<Vec<_>>();
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),
Expand Down
2 changes: 2 additions & 0 deletions codex-rs/analytics/src/lib.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,6 @@
mod accepted_lines;
#[cfg(debug_assertions)]
mod analytics_capture;
mod client;
mod events;
mod facts;
Expand Down
Loading