feat(analytics): add a local sink

codex-rs/analytics/src/client.rs

@@ -22,6 +22,9 @@ use crate::facts::TurnCodexErrorFact;
 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;
 use codex_app_server_protocol::ClientResponsePayload;
@@ -42,6 +45,11 @@ use std::sync::Mutex;
 use std::time::Duration;
 use tokio::sync::mpsc;
 
+#[cfg(test)]
+use crate::local_sink::local_analytics_sink_for_path;
+#[cfg(test)]
+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;
@@ -59,14 +67,25 @@ pub struct AnalyticsEventsClient {
 }
 
 impl AnalyticsEventsQueue {
-    pub(crate) fn new(auth_manager: Arc<AuthManager>, base_url: String) -> Self {
+    pub(crate) fn new(
+        auth_manager: Arc<AuthManager>,
+        base_url: String,
+        backend_analytics_enabled: bool,
+        local_sink: Option<SharedLocalAnalyticsSink>,
+    ) -> 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;
-                send_track_events(&auth_manager, &base_url, 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;
+                }
             }
         });
         Self {
@@ -123,16 +142,52 @@ impl AnalyticsEventsClient {
         base_url: String,
         analytics_enabled: Option<bool>,
     ) -> Self {
-        Self {
-            queue: (analytics_enabled != Some(false))
-                .then(|| AnalyticsEventsQueue::new(Arc::clone(&auth_manager), base_url)),
-        }
+        Self::new_with_local_sink(
+            auth_manager,
+            base_url,
+            analytics_enabled,
+            local_analytics_sink_from_env(),
+        )
     }
 
     pub fn disabled() -> Self {
         Self { queue: None }
     }
 
+    #[cfg(test)]
+    pub(crate) fn new_with_local_sink_path(
+        auth_manager: Arc<AuthManager>,
+        base_url: String,
+        analytics_enabled: Option<bool>,
+        local_sink_path: Option<PathBuf>,
+    ) -> Self {
+        Self::new_with_local_sink(
+            auth_manager,
+            base_url,
+            analytics_enabled,
+            local_sink_path.and_then(local_analytics_sink_for_path),
+        )
+    }
+
+    fn new_with_local_sink(
+        auth_manager: Arc<AuthManager>,
+        base_url: String,
+        analytics_enabled: Option<bool>,
+        local_sink: Option<SharedLocalAnalyticsSink>,
+    ) -> Self {
+        let backend_analytics_enabled = analytics_enabled != Some(false);
+        Self {
+            queue: (backend_analytics_enabled || local_sink.is_some()).then(|| {
+                AnalyticsEventsQueue::new(
+                    auth_manager,
+                    base_url,
+                    backend_analytics_enabled,
+                    local_sink,
+                )
+            }),
+        }
+    }
+
     pub fn track_skill_invocations(
         &self,
         tracking: TrackEventsContext,

codex-rs/analytics/src/client_tests.rs

@@ -1,6 +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;
@@ -10,8 +12,11 @@ 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;
 use codex_app_server_protocol::ClientRequest;
 use codex_app_server_protocol::ClientResponsePayload;
+use codex_app_server_protocol::InitializeCapabilities;
+use codex_app_server_protocol::InitializeParams;
 use codex_app_server_protocol::RequestId;
 use codex_app_server_protocol::SandboxPolicy as AppServerSandboxPolicy;
 use codex_app_server_protocol::SessionSource as AppServerSessionSource;
@@ -28,14 +33,22 @@ use codex_app_server_protocol::TurnStartResponse;
 use codex_app_server_protocol::TurnStatus as AppServerTurnStatus;
 use codex_app_server_protocol::TurnSteerParams;
 use codex_app_server_protocol::TurnSteerResponse;
+use codex_login::AuthManager;
+use codex_login::CodexAuth;
 use codex_utils_absolute_path::test_support::PathBufExt;
 use codex_utils_absolute_path::test_support::test_path_buf;
 use std::collections::HashSet;
+use std::fs;
+use std::path::PathBuf;
 use std::sync::Arc;
 use std::sync::Mutex;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering;
 use tokio::sync::mpsc;
 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 {
@@ -217,6 +230,46 @@ fn sample_turn_steer_response() -> ClientResponsePayload {
     })
 }
 
+#[tokio::test]
+async fn local_sink_reduces_events_when_backend_analytics_are_disabled() {
+    let path = test_sink_path("backend-disabled");
+    let client = AnalyticsEventsClient::new_with_local_sink_path(
+        AuthManager::from_auth_for_testing(CodexAuth::from_api_key("test")),
+        "https://example.invalid".to_string(),
+        /*analytics_enabled*/ Some(false),
+        Some(path.clone()),
+    );
+
+    assert!(client.queue.is_some());
+    client.track_initialize(
+        /*connection_id*/ 7,
+        InitializeParams {
+            client_info: ClientInfo {
+                name: "codex-tui".to_string(),
+                title: None,
+                version: "1.0.0".to_string(),
+            },
+            capabilities: Some(InitializeCapabilities {
+                experimental_api: false,
+                request_attestation: false,
+                opt_out_notification_methods: None,
+            }),
+        },
+        "codex".to_string(),
+        AppServerRpcTransport::Stdio,
+    );
+    client.track_response(
+        /*connection_id*/ 7,
+        RequestId::Integer(1),
+        sample_thread_start_response(),
+    );
+
+    let records = wait_for_local_records(&path, 1).await;
+    assert_eq!(records.len(), 1);
+    assert_eq!(records[0].payload["event_type"], "codex_thread_initialized");
+    assert_eq!(records[0].thread_id.as_deref(), Some("thread-1"));
+}
+
 #[test]
 fn track_request_only_enqueues_analytics_relevant_requests() {
     let (client, mut receiver) = client_with_receiver();
@@ -286,3 +339,29 @@ fn track_event_request_batches_only_isolates_accepted_line_fingerprint_events()
     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<LocalAnalyticsRecord> {
+    for _ in 0..100 {
+        if let Ok(contents) = fs::read_to_string(path) {
+            let records = contents
+                .lines()
+                .map(|line| serde_json::from_str::<LocalAnalyticsRecord>(line).expect("record"))
+                .collect::<Vec<_>>();
+            if records.len() >= min_records {
+                return records;
+            }
+        }
+        tokio::task::yield_now().await;
+    }
+
+    panic!("timed out waiting for {min_records} local analytics records");
+}
+
+fn test_sink_path(label: &str) -> PathBuf {
+    let id = NEXT_TEST_PATH_ID.fetch_add(1, Ordering::Relaxed);
+    let process_id = std::process::id();
+    let dir =
+        std::env::temp_dir().join(format!("codex-analytics-client-{process_id}-{label}-{id}"));
+    fs::create_dir_all(&dir).expect("create temp dir");
+    dir.join("events.jsonl")
+}

codex-rs/analytics/src/lib.rs

@@ -2,6 +2,7 @@ mod accepted_lines;
 mod client;
 mod events;
 mod facts;
+mod local_sink;
 mod reducer;
 
 use std::time::SystemTime;
@@ -48,6 +49,9 @@ pub use facts::TurnSteerRequestError;
 pub use facts::TurnSteerResult;
 pub use facts::TurnTokenUsageFact;
 pub use facts::build_track_events_context;
+pub use local_sink::LOCAL_ANALYTICS_SCHEMA_VERSION;
+pub use local_sink::LocalAnalyticsRecord;
+pub use local_sink::LocalAnalyticsRecordType;
 
 #[cfg(test)]
 mod analytics_client_tests;

codex-rs/analytics/src/local_sink.rs

@@ -0,0 +1,155 @@
+use crate::events::TrackEventRequest;
+use crate::now_unix_millis;
+use serde::Deserialize;
+use serde::Serialize;
+use serde_json::Value as JsonValue;
+use std::collections::HashMap;
+use std::fs::File;
+use std::fs::OpenOptions;
+use std::io::BufWriter;
+use std::io::Write;
+use std::path::Path;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::sync::OnceLock;
+use std::sync::Weak;
+
+pub const LOCAL_ANALYTICS_SCHEMA_VERSION: u32 = 1;
+pub const LOCAL_ANALYTICS_SINK_PATH_ENV_VAR: &str = "CODEX_ANALYTICS_LOCAL_SINK_PATH";
+
+pub(crate) type SharedLocalAnalyticsSink = Arc<Mutex<LocalAnalyticsSink>>;
+
+static PROCESS_LOCAL_SINKS: OnceLock<Mutex<HashMap<PathBuf, Weak<Mutex<LocalAnalyticsSink>>>>> =
+    OnceLock::new();
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum LocalAnalyticsRecordType {
+    CodexAnalyticsEvent,
+    ResponsesApiCall,
+}
+
+#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
+pub struct LocalAnalyticsRecord {
+    pub schema_version: u32,
+    pub recorded_at_epoch_millis: u64,
+    pub record_type: LocalAnalyticsRecordType,
+    pub session_id: Option<String>,
+    pub thread_id: Option<String>,
+    pub turn_id: Option<String>,
+    pub payload: JsonValue,
+}
+
+pub(crate) fn local_analytics_sink_from_env() -> Option<SharedLocalAnalyticsSink> {
+    let path = std::env::var_os(LOCAL_ANALYTICS_SINK_PATH_ENV_VAR)?;
+    if path.is_empty() {
+        return None;
+    }
+
+    local_analytics_sink_for_path(PathBuf::from(path))
+}
+
+pub(crate) fn append_codex_analytics_events_best_effort(
+    sink: &SharedLocalAnalyticsSink,
+    events: &[TrackEventRequest],
+) {
+    for event in events {
+        let Some(record) = LocalAnalyticsRecord::from_codex_analytics_event(event) else {
+            continue;
+        };
+        append_local_analytics_record_best_effort(sink, &record);
+    }
+}
+
+pub(crate) fn append_local_analytics_record_best_effort(
+    sink: &SharedLocalAnalyticsSink,
+    record: &LocalAnalyticsRecord,
+) {
+    let result = sink
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+        .append(record);
+    if let Err(err) = result {
+        tracing::warn!(error = %err, "failed to append local analytics record");
+    }
+}
+
+pub(crate) fn local_analytics_sink_for_path(path: PathBuf) -> Option<SharedLocalAnalyticsSink> {
+    let sinks = PROCESS_LOCAL_SINKS.get_or_init(|| Mutex::new(HashMap::new()));
+    let mut sinks = sinks
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    sinks.retain(|_, sink| sink.strong_count() > 0);
+    if let Some(sink) = sinks.get(&path).and_then(Weak::upgrade) {
+        return Some(sink);
+    }
+
+    match LocalAnalyticsSink::open(path.clone()) {
+        Ok(sink) => {
+            let sink = Arc::new(Mutex::new(sink));
+            sinks.insert(path, Arc::downgrade(&sink));
+            Some(sink)
+        }
+        Err(err) => {
+            tracing::warn!(
+                path = %path.display(),
+                error = %err,
+                "failed to initialize local analytics sink"
+            );
+            None
+        }
+    }
+}
+
+impl LocalAnalyticsRecord {
+    fn from_codex_analytics_event(event: &TrackEventRequest) -> Option<Self> {
+        let payload = match serde_json::to_value(event) {
+            Ok(payload) => payload,
+            Err(err) => {
+                tracing::warn!(error = %err, "failed to serialize local analytics event");
+                return None;
+            }
+        };
+        let event_params = payload.get("event_params");
+        Some(Self {
+            schema_version: LOCAL_ANALYTICS_SCHEMA_VERSION,
+            recorded_at_epoch_millis: now_unix_millis(),
+            record_type: LocalAnalyticsRecordType::CodexAnalyticsEvent,
+            session_id: string_field(event_params, "session_id"),
+            thread_id: string_field(event_params, "thread_id"),
+            turn_id: string_field(event_params, "turn_id"),
+            payload,
+        })
+    }
+}
+
+pub(crate) struct LocalAnalyticsSink {
+    writer: BufWriter<File>,
+}
+
+impl LocalAnalyticsSink {
+    fn open(path: impl AsRef<Path>) -> std::io::Result<Self> {
+        let file = OpenOptions::new().create(true).append(true).open(path)?;
+        Ok(Self {
+            writer: BufWriter::new(file),
+        })
+    }
+
+    fn append(&mut self, record: &LocalAnalyticsRecord) -> std::io::Result<()> {
+        serde_json::to_writer(&mut self.writer, record)?;
+        self.writer.write_all(b"\n")?;
+        self.writer.flush()
+    }
+}
+
+fn string_field(value: Option<&JsonValue>, field: &str) -> Option<String> {
+    value?
+        .get(field)?
+        .as_str()
+        .map(std::string::ToString::to_string)
+}
+
+#[cfg(test)]
+#[path = "local_sink_tests.rs"]
+mod tests;

codex-rs/analytics/src/local_sink_tests.rs

@@ -0,0 +1,115 @@
+use super::LOCAL_ANALYTICS_SCHEMA_VERSION;
+use super::LocalAnalyticsRecord;
+use super::LocalAnalyticsRecordType;
+use super::append_local_analytics_record_best_effort;
+use super::local_analytics_sink_for_path;
+use crate::events::SkillInvocationEventParams;
+use crate::events::SkillInvocationEventRequest;
+use crate::events::TrackEventRequest;
+use crate::facts::InvocationType;
+use pretty_assertions::assert_eq;
+use serde_json::json;
+use std::fs;
+use std::path::PathBuf;
+use std::sync::Arc;
+use std::sync::atomic::AtomicU64;
+use std::sync::atomic::Ordering;
+
+static NEXT_TEST_PATH_ID: AtomicU64 = AtomicU64::new(0);
+
+#[test]
+fn codex_analytics_record_extracts_generic_envelope_metadata() {
+    let record = LocalAnalyticsRecord::from_codex_analytics_event(&sample_track_event())
+        .expect("serialize local analytics event");
+
+    assert_eq!(
+        record,
+        LocalAnalyticsRecord {
+            schema_version: LOCAL_ANALYTICS_SCHEMA_VERSION,
+            recorded_at_epoch_millis: record.recorded_at_epoch_millis,
+            record_type: LocalAnalyticsRecordType::CodexAnalyticsEvent,
+            session_id: None,
+            thread_id: Some("thread-1".to_string()),
+            turn_id: Some("turn-1".to_string()),
+            payload: json!({
+                "event_type": "skill_invocation",
+                "skill_id": "skill-1",
+                "skill_name": "doc",
+                "event_params": {
+                    "product_client_id": null,
+                    "skill_scope": null,
+                    "plugin_id": null,
+                    "repo_url": null,
+                    "thread_id": "thread-1",
+                    "turn_id": "turn-1",
+                    "invoke_type": "explicit",
+                    "model_slug": "gpt-5.1-codex"
+                }
+            }),
+        }
+    );
+}
+
+#[test]
+fn process_global_sink_reuses_writer_for_same_path() {
+    let path = test_sink_path("shared");
+    let first = local_analytics_sink_for_path(path.clone()).expect("first sink");
+    let second = local_analytics_sink_for_path(path).expect("second sink");
+
+    assert!(Arc::ptr_eq(&first, &second));
+}
+
+#[test]
+fn sink_appends_complete_jsonl_records() {
+    let path = test_sink_path("records");
+    let sink = local_analytics_sink_for_path(path.clone()).expect("sink");
+    let first = LocalAnalyticsRecord::from_codex_analytics_event(&sample_track_event())
+        .expect("first record");
+    let mut second = first.clone();
+    second.turn_id = Some("turn-2".to_string());
+
+    append_local_analytics_record_best_effort(&sink, &first);
+    append_local_analytics_record_best_effort(&sink, &second);
+
+    let contents = fs::read_to_string(path).expect("read sink");
+    let records = contents
+        .lines()
+        .map(|line| serde_json::from_str::<LocalAnalyticsRecord>(line).expect("record"))
+        .collect::<Vec<_>>();
+    assert_eq!(records, vec![first, second]);
+}
+
+#[test]
+fn sink_initialization_failure_is_best_effort() {
+    let path = test_sink_path("missing-parent")
+        .join("missing")
+        .join("events.jsonl");
+
+    assert!(local_analytics_sink_for_path(path).is_none());
+}
+
+fn sample_track_event() -> TrackEventRequest {
+    TrackEventRequest::SkillInvocation(SkillInvocationEventRequest {
+        event_type: "skill_invocation",
+        skill_id: "skill-1".to_string(),
+        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-1".to_string()),
+            turn_id: Some("turn-1".to_string()),
+            invoke_type: Some(InvocationType::Explicit),
+            model_slug: Some("gpt-5.1-codex".to_string()),
+        },
+    })
+}
+
+fn test_sink_path(label: &str) -> PathBuf {
+    let id = NEXT_TEST_PATH_ID.fetch_add(1, Ordering::Relaxed);
+    let process_id = std::process::id();
+    let dir = std::env::temp_dir().join(format!("codex-analytics-{process_id}-{label}-{id}"));
+    fs::create_dir_all(&dir).expect("create temp dir");
+    dir.join("events.jsonl")
+}