openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-06 15:29:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

simplify

Browse Source
This commit is contained in:
Owen Lin
2026-06-10 14:13:45 -07:00
parent 8205a396a3
commit fa609cf760
8 changed files with 88 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
codex-rs/analytics/src/client.rs
View File

@@ -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::<Vec<_>>();
let mut events = Vec::new();
reducer.ingest(input, &mut events).await;
for sink in &sinks {
sink.write(&events).await;
}

42
codex-rs/analytics/src/client_tests.rs
View File

@@ -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"));
}

30
codex-rs/analytics/src/events.rs
View File

@@ -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<TrackEventRequest> 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,

12
codex-rs/analytics/src/sinks/codex_backend.rs
View File

@@ -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()

19
codex-rs/analytics/src/sinks/codex_backend_tests.rs
View File

@@ -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::<Vec<_>>();
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 {

105
codex-rs/analytics/src/sinks/local.rs
View File

@@ -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<Mutex<LocalAnalyticsSink>>;
type LocalAnalyticsWriter = BufWriter<File>;
pub(crate) type SharedLocalAnalyticsSink = Arc<Mutex<LocalAnalyticsWriter>>;
static PROCESS_LOCAL_SINKS: OnceLock<Mutex<HashMap<PathBuf, Weak<Mutex<LocalAnalyticsSink>>>>> =
static PROCESS_LOCAL_SINKS: OnceLock<Mutex<HashMap<PathBuf, Weak<Mutex<LocalAnalyticsWriter>>>>> =
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<SharedLocalAnalyticsSink
local_analytics_sink_for_path(PathBuf::from(path))
}
pub(super) fn write(sink: &SharedLocalAnalyticsSink, event: &AnalyticsEvent) {
let Some(record) = LocalAnalyticsRecord::from_event(event) else {
return;
};
append_record_best_effort(sink, &record);
}
fn append_record_best_effort(sink: &SharedLocalAnalyticsSink, record: &LocalAnalyticsRecord) {
let result = sink
pub(super) fn write(sink: &SharedLocalAnalyticsSink, events: &[TrackEventRequest]) {
let mut writer = 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");
.unwrap_or_else(std::sync::PoisonError::into_inner);
for event in events {
let Some(record) = LocalAnalyticsRecord::from_codex_analytics_event(event) else {
continue;
};
if let Err(err) = append_record(&mut writer, &record) {
tracing::warn!(error = %err, "failed to append local analytics record");
}
}
}
@@ -76,9 +73,9 @@ pub(crate) fn local_analytics_sink_for_path(path: PathBuf) -> Option<SharedLocal
return Some(sink);
}
match LocalAnalyticsSink::open(path.clone()) {
Ok(sink) => {
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<SharedLocal
}
impl LocalAnalyticsRecord {
fn from_event(event: &AnalyticsEvent) -> Option<Self> {
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<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;
}
}
}
}
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),
};
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<File>,
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<String> {

60
codex-rs/analytics/src/sinks/local_tests.rs
View File

@@ -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::<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")
@@ -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();

33
codex-rs/analytics/src/sinks/mod.rs
View File

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