introduce sink abstraction

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<AuthManager>,
-        base_url: String,
-        backend_analytics_enabled: bool,
-        local_sink: Option<SharedLocalAnalyticsSink>,
-    ) -> Self {
+    pub(crate) fn new(sinks: Vec<AnalyticsEventSink>) -> 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::<Vec<_>>();
+                for sink in &sinks {
+                    sink.write(&events).await;
                 }
             }
         });
@@ -175,16 +165,18 @@ impl AnalyticsEventsClient {
         analytics_enabled: Option<bool>,
         local_sink: Option<SharedLocalAnalyticsSink>,
     ) -> 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<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(events: Vec<TrackEventRequest>) -> Vec<Vec<TrackEventRequest>> {
-    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<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 = "client_tests.rs"]
 mod tests;

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<AnalyticsFact>) {
     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<LocalAnalyticsRecord> {
     for _ in 0..100 {
         if let Ok(contents) = fs::read_to_string(path) {

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<TrackEventRequest> for AnalyticsEvent {
+    fn from(event: TrackEventRequest) -> Self {
+        Self::CodexAnalytics(event)
+    }
+}
+
+pub(crate) enum AnalyticsEventSink {
+    CodexBackend {
+        auth_manager: Arc<AuthManager>,
+        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::<Vec<_>>();
+                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<Item = &AnalyticsEvent> {
+    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;

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::<Vec<_>>();
+    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()),
+        },
+    })
+}

codex-rs/analytics/src/events.rs

@@ -49,8 +49,8 @@ pub enum AppServerRpcTransport {
 }
 
 #[derive(Serialize)]
-pub(crate) struct TrackEventsRequest {
-    pub(crate) events: Vec<TrackEventRequest>,
+pub(crate) struct TrackEventsRequest<'a> {
+    pub(crate) events: &'a [&'a TrackEventRequest],
 }
 
 #[derive(Serialize)]

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;

codex-rs/analytics/src/local_sink.rs

@@ -50,16 +50,14 @@ pub(crate) fn local_analytics_sink_from_env() -> Option<SharedLocalAnalyticsSink
     local_analytics_sink_for_path(PathBuf::from(path))
 }
 
-pub(crate) fn append_codex_analytics_events_best_effort(
+pub(crate) fn append_codex_analytics_event_best_effort(
     sink: &SharedLocalAnalyticsSink,
-    events: &[TrackEventRequest],
+    event: &TrackEventRequest,
 ) {
-    for event in events {
-        let Some(record) = LocalAnalyticsRecord::from_codex_analytics_event(event) else {
-            continue;
-        };
-        append_local_analytics_record_best_effort(sink, &record);
-    }
+    let Some(record) = LocalAnalyticsRecord::from_codex_analytics_event(event) else {
+        return;
+    };
+    append_local_analytics_record_best_effort(sink, &record);
 }
 
 pub(crate) fn append_local_analytics_record_best_effort(