From fa609cf760029ea763f06026be23b4f6752ab1ec Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Wed, 10 Jun 2026 14:13:45 -0700 Subject: [PATCH] simplify --- codex-rs/analytics/src/client.rs | 9 +- codex-rs/analytics/src/client_tests.rs | 42 +++---- codex-rs/analytics/src/events.rs | 30 +---- codex-rs/analytics/src/sinks/codex_backend.rs | 12 +- .../src/sinks/codex_backend_tests.rs | 19 ---- codex-rs/analytics/src/sinks/local.rs | 105 ++++++++---------- codex-rs/analytics/src/sinks/local_tests.rs | 60 +++------- codex-rs/analytics/src/sinks/mod.rs | 33 +----- 8 files changed, 88 insertions(+), 222 deletions(-) diff --git a/codex-rs/analytics/src/client.rs b/codex-rs/analytics/src/client.rs index f6de1bf37a..d00ef2ee14 100644 --- a/codex-rs/analytics/src/client.rs +++ b/codex-rs/analytics/src/client.rs @@ -1,4 +1,3 @@ -use crate::events::AnalyticsEvent; use crate::events::AppServerRpcTransport; use crate::events::GuardianReviewAnalyticsResult; use crate::events::GuardianReviewTrackContext; @@ -67,12 +66,8 @@ impl AnalyticsEventsQueue { tokio::spawn(async move { let mut reducer = AnalyticsReducer::default(); while let Some(input) = receiver.recv().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::>(); + let mut events = Vec::new(); + reducer.ingest(input, &mut events).await; for sink in &sinks { sink.write(&events).await; } diff --git a/codex-rs/analytics/src/client_tests.rs b/codex-rs/analytics/src/client_tests.rs index 32488c2ad1..dbf5f049bc 100644 --- a/codex-rs/analytics/src/client_tests.rs +++ b/codex-rs/analytics/src/client_tests.rs @@ -1,15 +1,14 @@ use super::AnalyticsEventsClient; use super::AnalyticsEventsQueue; use crate::LocalAnalyticsRecord; -use crate::events::AppServerRpcTransport; use crate::facts::AnalyticsFact; +use crate::facts::InvocationType; +use crate::facts::SkillInvocation; +use crate::facts::TrackEventsContext; 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,6 +27,7 @@ use codex_app_server_protocol::TurnSteerParams; use codex_app_server_protocol::TurnSteerResponse; use codex_login::AuthManager; use codex_login::CodexAuth; +use codex_protocol::protocol::SkillScope; use codex_utils_absolute_path::test_support::PathBufExt; use codex_utils_absolute_path::test_support::test_path_buf; use std::collections::HashSet; @@ -196,32 +196,24 @@ async fn local_sink_reduces_events_when_backend_analytics_are_disabled() { ); 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, - }), + client.track_skill_invocations( + TrackEventsContext { + model_slug: "gpt-5.1-codex".to_string(), + thread_id: "thread-1".to_string(), + turn_id: "turn-1".to_string(), }, - "codex".to_string(), - AppServerRpcTransport::Stdio, - ); - client.track_response( - /*connection_id*/ 7, - RequestId::Integer(1), - sample_thread_start_response(), + vec![SkillInvocation { + skill_name: "doc".to_string(), + skill_scope: SkillScope::User, + skill_path: test_path_buf("/tmp/skills/doc/SKILL.md"), + plugin_id: None, + invocation_type: InvocationType::Explicit, + }], ); 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].payload["event_type"], "skill_invocation"); assert_eq!(records[0].thread_id.as_deref(), Some("thread-1")); } diff --git a/codex-rs/analytics/src/events.rs b/codex-rs/analytics/src/events.rs index bd560436a9..ada647b327 100644 --- a/codex-rs/analytics/src/events.rs +++ b/codex-rs/analytics/src/events.rs @@ -50,7 +50,7 @@ pub enum AppServerRpcTransport { #[derive(Serialize)] pub(crate) struct TrackEventsRequest<'a> { - pub(crate) events: &'a [&'a TrackEventRequest], + pub(crate) events: &'a [TrackEventRequest], } #[derive(Serialize)] @@ -88,34 +88,6 @@ impl TrackEventRequest { } } -pub(crate) enum AnalyticsEvent { - CodexAnalytics(TrackEventRequest), -} - -impl AnalyticsEvent { - pub(crate) fn is_writable_to(&self, sink: AnalyticsEventSinkKind) -> bool { - matches!( - (self, sink), - ( - Self::CodexAnalytics(_), - AnalyticsEventSinkKind::CodexBackend | AnalyticsEventSinkKind::Local, - ) - ) - } -} - -impl From for AnalyticsEvent { - fn from(event: TrackEventRequest) -> Self { - Self::CodexAnalytics(event) - } -} - -#[derive(Clone, Copy)] -pub(crate) enum AnalyticsEventSinkKind { - CodexBackend, - Local, -} - #[derive(Serialize)] pub(crate) struct CodexAcceptedLineFingerprintsEventParams { pub(crate) event_type: &'static str, diff --git a/codex-rs/analytics/src/sinks/codex_backend.rs b/codex-rs/analytics/src/sinks/codex_backend.rs index 1fc587cbd6..b8fb2dfd3d 100644 --- a/codex-rs/analytics/src/sinks/codex_backend.rs +++ b/codex-rs/analytics/src/sinks/codex_backend.rs @@ -10,7 +10,7 @@ const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10); pub(super) async fn write( auth_manager: &AuthManager, base_url: &str, - events: &[&TrackEventRequest], + events: &[TrackEventRequest], ) { if events.is_empty() { return; @@ -30,9 +30,7 @@ pub(super) async fn write( } } -fn track_event_request_batches<'a>( - events: &'a [&'a TrackEventRequest], -) -> Vec<&'a [&'a TrackEventRequest]> { +fn track_event_request_batches(events: &[TrackEventRequest]) -> Vec<&[TrackEventRequest]> { let mut batches = Vec::new(); let mut current_batch_start = 0; @@ -53,11 +51,7 @@ fn track_event_request_batches<'a>( batches } -async fn send_track_events_request(auth: &CodexAuth, url: &str, events: &[&TrackEventRequest]) { - if events.is_empty() { - return; - } - +async fn send_track_events_request(auth: &CodexAuth, url: &str, events: &[TrackEventRequest]) { let payload = TrackEventsRequest { events }; let response = create_client() diff --git a/codex-rs/analytics/src/sinks/codex_backend_tests.rs b/codex-rs/analytics/src/sinks/codex_backend_tests.rs index 775fd6c38e..4e4b304b1c 100644 --- a/codex-rs/analytics/src/sinks/codex_backend_tests.rs +++ b/codex-rs/analytics/src/sinks/codex_backend_tests.rs @@ -4,9 +4,7 @@ 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() { @@ -18,7 +16,6 @@ fn track_event_request_batches_only_isolates_accepted_line_fingerprint_events() 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); @@ -30,22 +27,6 @@ fn track_event_request_batches_only_isolates_accepted_line_fingerprint_events() 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 { diff --git a/codex-rs/analytics/src/sinks/local.rs b/codex-rs/analytics/src/sinks/local.rs index dba10a4eb2..636656fdf2 100644 --- a/codex-rs/analytics/src/sinks/local.rs +++ b/codex-rs/analytics/src/sinks/local.rs @@ -1,4 +1,4 @@ -use crate::events::AnalyticsEvent; +use crate::events::TrackEventRequest; use crate::now_unix_millis; use serde::Deserialize; use serde::Serialize; @@ -8,7 +8,6 @@ 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; @@ -16,20 +15,21 @@ 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"; +const LOCAL_ANALYTICS_SINK_PATH_ENV_VAR: &str = "CODEX_ANALYTICS_LOCAL_SINK_PATH"; -pub(crate) type SharedLocalAnalyticsSink = Arc>; +type LocalAnalyticsWriter = BufWriter; +pub(crate) type SharedLocalAnalyticsSink = Arc>; -static PROCESS_LOCAL_SINKS: OnceLock>>>> = +static PROCESS_LOCAL_SINKS: OnceLock>>>> = OnceLock::new(); -#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[derive(Debug, Deserialize, PartialEq, Serialize)] #[serde(rename_all = "snake_case")] pub enum LocalAnalyticsRecordType { CodexAnalyticsEvent, } -#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[derive(Debug, Deserialize, PartialEq, Serialize)] pub struct LocalAnalyticsRecord { pub schema_version: u32, pub recorded_at_epoch_millis: u64, @@ -49,20 +49,17 @@ pub(crate) fn local_analytics_sink_from_env() -> Option Option { - let sink = Arc::new(Mutex::new(sink)); + match OpenOptions::new().create(true).append(true).open(&path) { + Ok(file) => { + let sink = Arc::new(Mutex::new(BufWriter::new(file))); sinks.insert(path, Arc::downgrade(&sink)); Some(sink) } @@ -94,48 +91,34 @@ pub(crate) fn local_analytics_sink_for_path(path: PathBuf) -> Option Option { - match event { - AnalyticsEvent::CodexAnalytics(event) => { - 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, - }) + fn from_codex_analytics_event(event: &TrackEventRequest) -> Option { + 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; } - } - } -} - -pub(crate) struct LocalAnalyticsSink { - writer: BufWriter, -} - -impl LocalAnalyticsSink { - fn open(path: impl AsRef) -> std::io::Result { - let file = OpenOptions::new().create(true).append(true).open(path)?; - Ok(Self { - writer: BufWriter::new(file), + }; + 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, }) } +} - 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 append_record( + writer: &mut BufWriter, + record: &LocalAnalyticsRecord, +) -> std::io::Result<()> { + serde_json::to_writer(&mut *writer, record)?; + writer.write_all(b"\n")?; + writer.flush() } fn string_field(value: Option<&JsonValue>, field: &str) -> Option { diff --git a/codex-rs/analytics/src/sinks/local_tests.rs b/codex-rs/analytics/src/sinks/local_tests.rs index 32de05ef88..c1d31c74a8 100644 --- a/codex-rs/analytics/src/sinks/local_tests.rs +++ b/codex-rs/analytics/src/sinks/local_tests.rs @@ -1,9 +1,7 @@ use super::LOCAL_ANALYTICS_SCHEMA_VERSION; use super::LocalAnalyticsRecord; use super::LocalAnalyticsRecordType; -use super::append_record_best_effort; use super::local_analytics_sink_for_path; -use crate::events::AnalyticsEvent; use crate::events::SkillInvocationEventParams; use crate::events::SkillInvocationEventRequest; use crate::events::TrackEventRequest; @@ -20,8 +18,23 @@ static NEXT_TEST_PATH_ID: AtomicU64 = AtomicU64::new(0); #[test] fn codex_analytics_record_extracts_generic_envelope_metadata() { - let record = LocalAnalyticsRecord::from_event(&sample_analytics_event()) - .expect("serialize local analytics event"); + let event = 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()), + }, + }); + let record = + LocalAnalyticsRecord::from_codex_analytics_event(&event).expect("serialize local event"); assert_eq!( record, @@ -60,25 +73,6 @@ fn process_global_sink_reuses_writer_for_same_path() { 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_event(&sample_analytics_event()).expect("first record"); - let mut second = first.clone(); - second.turn_id = Some("turn-2".to_string()); - - append_record_best_effort(&sink, &first); - append_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::(line).expect("record")) - .collect::>(); - assert_eq!(records, vec![first, second]); -} - #[test] fn sink_initialization_failure_is_best_effort() { let path = test_sink_path("missing-parent") @@ -88,26 +82,6 @@ fn sink_initialization_failure_is_best_effort() { assert!(local_analytics_sink_for_path(path).is_none()); } -fn sample_analytics_event() -> AnalyticsEvent { - AnalyticsEvent::CodexAnalytics(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(); diff --git a/codex-rs/analytics/src/sinks/mod.rs b/codex-rs/analytics/src/sinks/mod.rs index c8d58d72ea..7ae4d02243 100644 --- a/codex-rs/analytics/src/sinks/mod.rs +++ b/codex-rs/analytics/src/sinks/mod.rs @@ -1,8 +1,7 @@ mod codex_backend; mod local; -use crate::events::AnalyticsEvent; -use crate::events::AnalyticsEventSinkKind; +use crate::events::TrackEventRequest; use codex_login::AuthManager; use std::sync::Arc; @@ -23,37 +22,13 @@ pub(crate) enum AnalyticsEventSink { } impl AnalyticsEventSink { - pub(crate) async fn write(&self, events: &[AnalyticsEvent]) { + pub(crate) async fn write(&self, events: &[TrackEventRequest]) { match self { Self::CodexBackend { auth_manager, base_url, - } => { - let sink = self.kind(); - let events = events - .iter() - .filter(|event| event.is_writable_to(sink)) - .map(|event| match event { - AnalyticsEvent::CodexAnalytics(event) => event, - }) - .collect::>(); - codex_backend::write(auth_manager, base_url, &events).await; - } - Self::Local(sink) => { - let sink_kind = self.kind(); - for event in events { - if event.is_writable_to(sink_kind) { - local::write(sink, event); - } - } - } - } - } - - fn kind(&self) -> AnalyticsEventSinkKind { - match self { - Self::CodexBackend { .. } => AnalyticsEventSinkKind::CodexBackend, - Self::Local(_) => AnalyticsEventSinkKind::Local, + } => codex_backend::write(auth_manager, base_url, events).await, + Self::Local(sink) => local::write(sink, events), } } }