openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-02 19:43:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'rhan/oob-turn-metadata' into rhan/oob-queue-data

Browse Source
This commit is contained in:
rhan-oai
2026-03-25 16:07:58 -07:00
committed by GitHub
parent 4aec99194d 80367abdf1
commit d55054761a
5 changed files with 336 additions and 710 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/analytics_client.rs
+313 -511
View File
@@ -19,6 +19,7 @@ use codex_protocol::protocol::SubmissionType;
use serde::Serialize;
use sha1::Digest;
use sha1::Sha1;
use std::collections::HashMap;
use std::collections::HashSet;
use std::path::Path;
use std::path::PathBuf;
@@ -50,19 +51,11 @@ pub struct CodexTurnEvent {
pub num_input_images: usize,
pub is_first_turn: bool,
}
#[derive(Clone)]
pub struct CodexThreadInitializedEvent {
pub thread_id: String,
pub model: String,
pub model_provider: String,
pub reasoning_effort: Option<ReasoningEffort>,
pub reasoning_summary: Option<ReasoningSummary>,
pub service_tier: Option<ServiceTier>,
pub approval_policy: AskForApproval,
pub approvals_reviewer: ApprovalsReviewer,
pub sandbox_policy: SandboxPolicy,
pub sandbox_network_access: bool,
pub collaboration_mode: ModeKind,
pub personality: Option<Personality>,
pub ephemeral: bool,
pub session_source: SessionSource,
pub initialization_mode: InitializationMode,
@@ -71,6 +64,24 @@ pub struct CodexThreadInitializedEvent {
pub created_at: u64,
}
#[derive(Clone)]
pub struct CodexThreadInitializedInput {
pub thread_id: String,
pub model: String,
pub product_client_id: String,
pub created_at: u64,
pub thread_context: CodexThreadContext,
}
#[derive(Clone)]
pub struct CodexThreadContext {
pub ephemeral: bool,
pub session_source: SessionSource,
pub initialization_mode: InitializationMode,
pub subagent_source: Option<SubAgentSource>,
pub parent_thread_id: Option<String>,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum InitializationMode {
@@ -112,9 +123,65 @@ pub struct AppInvocation {
pub invocation_type: Option<InvocationType>,
}
pub enum AnalyticsInput {
ThreadInitialized(CodexThreadInitializedInput),
TurnEvent(TurnEventInput),
SkillInvoked(SkillInvokedInput),
AppMentioned(AppMentionedInput),
AppUsed(AppUsedInput),
PluginUsed(PluginUsedInput),
PluginStateChanged(PluginStateChangedInput),
}
pub struct TurnEventInput {
pub tracking: TrackEventsContext,
pub turn_event: CodexTurnEvent,
}
pub struct SkillInvokedInput {
pub tracking: TrackEventsContext,
pub invocations: Vec<SkillInvocation>,
}
pub struct AppMentionedInput {
pub tracking: TrackEventsContext,
pub mentions: Vec<AppInvocation>,
}
pub struct AppUsedInput {
pub tracking: TrackEventsContext,
pub app: AppInvocation,
}
pub struct PluginUsedInput {
pub tracking: TrackEventsContext,
pub plugin: PluginTelemetryMetadata,
}
pub struct PluginStateChangedInput {
pub plugin: PluginTelemetryMetadata,
pub state: PluginState,
}
#[derive(Clone, Copy)]
pub enum PluginState {
Installed,
Uninstalled,
Enabled,
Disabled,
}
#[derive(Default)]
pub struct AnalyticsReducer {
threads: HashMap<String, ThreadState>,
}
struct ThreadState {
_initialized_event: CodexThreadInitializedEvent,
}
#[derive(Clone)]
pub(crate) struct AnalyticsEventsQueue {
sender: mpsc::Sender<TrackEventsJob>,
sender: mpsc::Sender<AnalyticsInput>,
app_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
plugin_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
}
@@ -129,39 +196,11 @@ impl AnalyticsEventsQueue {
pub(crate) fn new(auth_manager: Arc<AuthManager>, base_url: String) -> Self {
let (sender, mut receiver) = mpsc::channel(ANALYTICS_EVENTS_QUEUE_SIZE);
tokio::spawn(async move {
let mut reducer = AnalyticsReducer::default();
while let Some(job) = receiver.recv().await {
match job {
TrackEventsJob::SkillInvocations(job) => {
send_track_skill_invocations(&auth_manager, &base_url, job).await;
}
TrackEventsJob::ThreadInitialized(job) => {
send_track_thread_initialized(&auth_manager, &base_url, job).await;
}
TrackEventsJob::AppMentioned(job) => {
send_track_app_mentioned(&auth_manager, &base_url, job).await;
}
TrackEventsJob::AppUsed(job) => {
send_track_app_used(&auth_manager, &base_url, job).await;
}
TrackEventsJob::TurnEvent(job) => {
send_track_turn_event(&auth_manager, &base_url, job).await;
}
TrackEventsJob::PluginUsed(job) => {
send_track_plugin_used(&auth_manager, &base_url, job).await;
}
TrackEventsJob::PluginInstalled(job) => {
send_track_plugin_installed(&auth_manager, &base_url, job).await;
}
TrackEventsJob::PluginUninstalled(job) => {
send_track_plugin_uninstalled(&auth_manager, &base_url, job).await;
}
TrackEventsJob::PluginEnabled(job) => {
send_track_plugin_enabled(&auth_manager, &base_url, job).await;
}
TrackEventsJob::PluginDisabled(job) => {
send_track_plugin_disabled(&auth_manager, &base_url, job).await;
}
}
let mut events = Vec::new();
reducer.ingest(job, &mut events).await;
send_track_events(&auth_manager, &base_url, events).await;
}
});
Self {
@@ -171,8 +210,8 @@ impl AnalyticsEventsQueue {
}
}
fn try_send(&self, job: TrackEventsJob) {
if self.sender.try_send(job).is_err() {
fn try_send(&self, input: AnalyticsInput) {
if self.sender.try_send(input).is_err() {
//TODO: add a metric for this
tracing::warn!("dropping analytics events: queue is full");
}
@@ -225,140 +264,95 @@ impl AnalyticsEventsClient {
tracking: TrackEventsContext,
invocations: Vec<SkillInvocation>,
) {
track_skill_invocations(
&self.queue,
self.analytics_enabled,
Some(tracking),
if invocations.is_empty() {
return;
}
self.record(AnalyticsInput::SkillInvoked(SkillInvokedInput {
tracking,
invocations,
);
}));
}
pub fn track_thread_initialized(&self, thread_event: CodexThreadInitializedEvent) {
track_thread_initialized(&self.queue, self.analytics_enabled, thread_event);
pub fn track_thread_initialized(&self, input: CodexThreadInitializedInput) {
self.record(AnalyticsInput::ThreadInitialized(input));
}
pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec<AppInvocation>) {
track_app_mentioned(
&self.queue,
self.analytics_enabled,
Some(tracking),
if mentions.is_empty() {
return;
}
self.record(AnalyticsInput::AppMentioned(AppMentionedInput {
tracking,
mentions,
);
}));
}
pub fn track_app_used(&self, tracking: TrackEventsContext, app: AppInvocation) {
track_app_used(&self.queue, self.analytics_enabled, Some(tracking), app);
if !self.queue.should_enqueue_app_used(&tracking, &app) {
return;
}
self.record(AnalyticsInput::AppUsed(AppUsedInput { tracking, app }));
}
pub fn track_plugin_used(&self, tracking: TrackEventsContext, plugin: PluginTelemetryMetadata) {
track_plugin_used(&self.queue, self.analytics_enabled, Some(tracking), plugin);
if !self.queue.should_enqueue_plugin_used(&tracking, &plugin) {
return;
}
self.record(AnalyticsInput::PluginUsed(PluginUsedInput {
tracking,
plugin,
}));
}
pub fn track_turn_event(&self, tracking: TrackEventsContext, turn_event: CodexTurnEvent) {
track_turn_event(
&self.queue,
self.analytics_enabled,
Some(tracking),
self.record(AnalyticsInput::TurnEvent(TurnEventInput {
tracking,
turn_event,
);
}));
}
pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) {
track_plugin_management(
&self.queue,
self.analytics_enabled,
PluginManagementEventType::Installed,
plugin,
);
self.record(AnalyticsInput::PluginStateChanged(
PluginStateChangedInput {
plugin,
state: PluginState::Installed,
},
));
}
pub fn track_plugin_uninstalled(&self, plugin: PluginTelemetryMetadata) {
track_plugin_management(
&self.queue,
self.analytics_enabled,
PluginManagementEventType::Uninstalled,
plugin,
);
self.record(AnalyticsInput::PluginStateChanged(
PluginStateChangedInput {
plugin,
state: PluginState::Uninstalled,
},
));
}
pub fn track_plugin_enabled(&self, plugin: PluginTelemetryMetadata) {
track_plugin_management(
&self.queue,
self.analytics_enabled,
PluginManagementEventType::Enabled,
plugin,
);
self.record(AnalyticsInput::PluginStateChanged(
PluginStateChangedInput {
plugin,
state: PluginState::Enabled,
},
));
}
pub fn track_plugin_disabled(&self, plugin: PluginTelemetryMetadata) {
track_plugin_management(
&self.queue,
self.analytics_enabled,
PluginManagementEventType::Disabled,
plugin,
);
self.record(AnalyticsInput::PluginStateChanged(
PluginStateChangedInput {
plugin,
state: PluginState::Disabled,
},
));
}
}
enum TrackEventsJob {
SkillInvocations(TrackSkillInvocationsJob),
ThreadInitialized(TrackThreadInitializedJob),
AppMentioned(TrackAppMentionedJob),
AppUsed(TrackAppUsedJob),
TurnEvent(TrackTurnEventJob),
PluginUsed(TrackPluginUsedJob),
PluginInstalled(TrackPluginManagementJob),
PluginUninstalled(TrackPluginManagementJob),
PluginEnabled(TrackPluginManagementJob),
PluginDisabled(TrackPluginManagementJob),
}
struct TrackSkillInvocationsJob {
analytics_enabled: Option<bool>,
tracking: TrackEventsContext,
invocations: Vec<SkillInvocation>,
}
struct TrackThreadInitializedJob {
analytics_enabled: Option<bool>,
thread_event: CodexThreadInitializedEvent,
}
struct TrackAppMentionedJob {
analytics_enabled: Option<bool>,
tracking: TrackEventsContext,
mentions: Vec<AppInvocation>,
}
struct TrackAppUsedJob {
analytics_enabled: Option<bool>,
tracking: TrackEventsContext,
app: AppInvocation,
}
struct TrackTurnEventJob {
analytics_enabled: Option<bool>,
tracking: TrackEventsContext,
turn_event: CodexTurnEvent,
}
struct TrackPluginUsedJob {
analytics_enabled: Option<bool>,
tracking: TrackEventsContext,
plugin: PluginTelemetryMetadata,
}
struct TrackPluginManagementJob {
analytics_enabled: Option<bool>,
plugin: PluginTelemetryMetadata,
}
#[derive(Clone, Copy)]
enum PluginManagementEventType {
Installed,
Uninstalled,
Enabled,
Disabled,
pub fn record(&self, input: AnalyticsInput) {
if self.analytics_enabled == Some(false) {
return;
}
self.queue.try_send(input);
}
}
const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256;
@@ -408,16 +402,6 @@ struct CodexThreadInitializedEventParams {
thread_id: String,
product_client_id: String,
model: String,
model_provider: String,
reasoning_effort: Option<String>,
reasoning_summary: Option<String>,
service_tier: String,
approval_policy: String,
approvals_reviewer: String,
sandbox_policy: &'static str,
sandbox_network_access: bool,
collaboration_mode: &'static str,
personality: Option<String>,
ephemeral: bool,
session_source: Option<&'static str>,
initialization_mode: InitializationMode,
@@ -514,354 +498,172 @@ struct CodexPluginUsedEventRequest {
event_params: CodexPluginUsedMetadata,
}
pub(crate) fn track_skill_invocations(
queue: &AnalyticsEventsQueue,
analytics_enabled: Option<bool>,
tracking: Option<TrackEventsContext>,
invocations: Vec<SkillInvocation>,
) {
if analytics_enabled == Some(false) {
return;
impl AnalyticsReducer {
async fn ingest(&mut self, input: AnalyticsInput, out: &mut Vec<TrackEventRequest>) {
match input {
AnalyticsInput::ThreadInitialized(input) => {
self.ingest_thread_initialized(input, out);
}
AnalyticsInput::TurnEvent(input) => {
self.ingest_turn_event(input, out);
}
AnalyticsInput::SkillInvoked(input) => {
self.ingest_skill_invoked(input, out).await;
}
AnalyticsInput::AppMentioned(input) => {
self.ingest_app_mentioned(input, out);
}
AnalyticsInput::AppUsed(input) => {
self.ingest_app_used(input, out);
}
AnalyticsInput::PluginUsed(input) => {
self.ingest_plugin_used(input, out);
}
AnalyticsInput::PluginStateChanged(input) => {
self.ingest_plugin_state_changed(input, out);
}
}
}
let Some(tracking) = tracking else {
return;
};
if invocations.is_empty() {
return;
}
let job = TrackEventsJob::SkillInvocations(TrackSkillInvocationsJob {
analytics_enabled,
tracking,
invocations,
});
queue.try_send(job);
}
pub(crate) fn track_thread_initialized(
queue: &AnalyticsEventsQueue,
analytics_enabled: Option<bool>,
thread_event: CodexThreadInitializedEvent,
) {
if analytics_enabled == Some(false) {
return;
}
let job = TrackEventsJob::ThreadInitialized(TrackThreadInitializedJob {
analytics_enabled,
thread_event,
});
queue.try_send(job);
}
pub(crate) fn track_app_mentioned(
queue: &AnalyticsEventsQueue,
analytics_enabled: Option<bool>,
tracking: Option<TrackEventsContext>,
mentions: Vec<AppInvocation>,
) {
if analytics_enabled == Some(false) {
return;
}
let Some(tracking) = tracking else {
return;
};
if mentions.is_empty() {
return;
}
let job = TrackEventsJob::AppMentioned(TrackAppMentionedJob {
analytics_enabled,
tracking,
mentions,
});
queue.try_send(job);
}
pub(crate) fn track_app_used(
queue: &AnalyticsEventsQueue,
analytics_enabled: Option<bool>,
tracking: Option<TrackEventsContext>,
app: AppInvocation,
) {
if analytics_enabled == Some(false) {
return;
}
let Some(tracking) = tracking else {
return;
};
if !queue.should_enqueue_app_used(&tracking, &app) {
return;
}
let job = TrackEventsJob::AppUsed(TrackAppUsedJob {
analytics_enabled,
tracking,
app,
});
queue.try_send(job);
}
pub(crate) fn track_turn_event(
queue: &AnalyticsEventsQueue,
analytics_enabled: Option<bool>,
tracking: Option<TrackEventsContext>,
turn_event: CodexTurnEvent,
) {
if analytics_enabled == Some(false) {
return;
}
let Some(tracking) = tracking else {
return;
};
let job = TrackEventsJob::TurnEvent(TrackTurnEventJob {
analytics_enabled,
tracking,
turn_event,
});
queue.try_send(job);
}
pub(crate) fn track_plugin_used(
queue: &AnalyticsEventsQueue,
analytics_enabled: Option<bool>,
tracking: Option<TrackEventsContext>,
plugin: PluginTelemetryMetadata,
) {
if analytics_enabled == Some(false) {
return;
}
let Some(tracking) = tracking else {
return;
};
if !queue.should_enqueue_plugin_used(&tracking, &plugin) {
return;
}
let job = TrackEventsJob::PluginUsed(TrackPluginUsedJob {
analytics_enabled,
tracking,
plugin,
});
queue.try_send(job);
}
fn track_plugin_management(
queue: &AnalyticsEventsQueue,
analytics_enabled: Option<bool>,
event_type: PluginManagementEventType,
plugin: PluginTelemetryMetadata,
) {
if analytics_enabled == Some(false) {
return;
}
let job = TrackPluginManagementJob {
analytics_enabled,
plugin,
};
let job = match event_type {
PluginManagementEventType::Installed => TrackEventsJob::PluginInstalled(job),
PluginManagementEventType::Uninstalled => TrackEventsJob::PluginUninstalled(job),
PluginManagementEventType::Enabled => TrackEventsJob::PluginEnabled(job),
PluginManagementEventType::Disabled => TrackEventsJob::PluginDisabled(job),
};
queue.try_send(job);
}
async fn send_track_skill_invocations(
auth_manager: &AuthManager,
base_url: &str,
job: TrackSkillInvocationsJob,
) {
let TrackSkillInvocationsJob {
analytics_enabled,
tracking,
invocations,
} = job;
let mut events = Vec::with_capacity(invocations.len());
for invocation in invocations {
let skill_scope = match invocation.skill_scope {
SkillScope::User => "user",
SkillScope::Repo => "repo",
SkillScope::System => "system",
SkillScope::Admin => "admin",
fn ingest_thread_initialized(
&mut self,
input: CodexThreadInitializedInput,
out: &mut Vec<TrackEventRequest>,
) {
let product_client_id = input.product_client_id.clone();
let event = CodexThreadInitializedEvent {
thread_id: input.thread_id,
model: input.model,
ephemeral: input.thread_context.ephemeral,
session_source: input.thread_context.session_source,
initialization_mode: input.thread_context.initialization_mode,
subagent_source: input.thread_context.subagent_source,
parent_thread_id: input.thread_context.parent_thread_id,
created_at: input.created_at,
};
let repo_root = get_git_repo_root(invocation.skill_path.as_path());
let repo_url = if let Some(root) = repo_root.as_ref() {
collect_git_info(root)
.await
.and_then(|info| info.repository_url)
} else {
None
};
let skill_id = skill_id_for_local_skill(
repo_url.as_deref(),
repo_root.as_deref(),
invocation.skill_path.as_path(),
invocation.skill_name.as_str(),
);
events.push(TrackEventRequest::SkillInvocation(
SkillInvocationEventRequest {
event_type: "skill_invocation",
skill_id,
skill_name: invocation.skill_name.clone(),
event_params: SkillInvocationEventParams {
thread_id: Some(tracking.thread_id.clone()),
invoke_type: Some(invocation.invocation_type),
model_slug: Some(tracking.model_slug.clone()),
product_client_id: Some(originator().value),
repo_url,
skill_scope: Some(skill_scope.to_string()),
},
self.threads.insert(
event.thread_id.clone(),
ThreadState {
_initialized_event: event.clone(),
},
);
out.push(TrackEventRequest::ThreadInitialized(
codex_thread_initialized_event_request(product_client_id, event),
));
}
send_track_events(auth_manager, analytics_enabled, base_url, events).await;
}
fn ingest_turn_event(&mut self, input: TurnEventInput, out: &mut Vec<TrackEventRequest>) {
let TurnEventInput {
tracking,
turn_event,
} = input;
out.push(TrackEventRequest::TurnEvent(CodexTurnEventRequest {
event_type: "codex_turn_event",
event_params: codex_turn_event_params(&tracking, turn_event),
}));
}
async fn send_track_thread_initialized(
auth_manager: &AuthManager,
base_url: &str,
job: TrackThreadInitializedJob,
) {
let TrackThreadInitializedJob {
analytics_enabled,
thread_event,
} = job;
let events = vec![TrackEventRequest::ThreadInitialized(
CodexThreadInitializedEventRequest {
event_type: "codex_thread_initialized",
event_params: codex_thread_initialized_event_params(thread_event),
},
)];
send_track_events(auth_manager, analytics_enabled, base_url, events).await;
}
async fn send_track_app_mentioned(
auth_manager: &AuthManager,
base_url: &str,
job: TrackAppMentionedJob,
) {
let TrackAppMentionedJob {
analytics_enabled,
tracking,
mentions,
} = job;
let events = mentions
.into_iter()
.map(|mention| {
async fn ingest_skill_invoked(
&mut self,
input: SkillInvokedInput,
out: &mut Vec<TrackEventRequest>,
) {
let SkillInvokedInput {
tracking,
invocations,
} = input;
for invocation in invocations {
let skill_scope = match invocation.skill_scope {
SkillScope::User => "user",
SkillScope::Repo => "repo",
SkillScope::System => "system",
SkillScope::Admin => "admin",
};
let repo_root = get_git_repo_root(invocation.skill_path.as_path());
let repo_url = if let Some(root) = repo_root.as_ref() {
collect_git_info(root)
.await
.and_then(|info| info.repository_url)
} else {
None
};
let skill_id = skill_id_for_local_skill(
repo_url.as_deref(),
repo_root.as_deref(),
invocation.skill_path.as_path(),
invocation.skill_name.as_str(),
);
out.push(TrackEventRequest::SkillInvocation(
SkillInvocationEventRequest {
event_type: "skill_invocation",
skill_id,
skill_name: invocation.skill_name.clone(),
event_params: SkillInvocationEventParams {
thread_id: Some(tracking.thread_id.clone()),
invoke_type: Some(invocation.invocation_type),
model_slug: Some(tracking.model_slug.clone()),
product_client_id: Some(originator().value),
repo_url,
skill_scope: Some(skill_scope.to_string()),
},
},
));
}
}
fn ingest_app_mentioned(&mut self, input: AppMentionedInput, out: &mut Vec<TrackEventRequest>) {
let AppMentionedInput { tracking, mentions } = input;
out.extend(mentions.into_iter().map(|mention| {
let event_params = codex_app_metadata(&tracking, mention);
TrackEventRequest::AppMentioned(CodexAppMentionedEventRequest {
event_type: "codex_app_mentioned",
event_params,
})
})
.collect::<Vec<_>>();
}));
}
send_track_events(auth_manager, analytics_enabled, base_url, events).await;
fn ingest_app_used(&mut self, input: AppUsedInput, out: &mut Vec<TrackEventRequest>) {
let AppUsedInput { tracking, app } = input;
let event_params = codex_app_metadata(&tracking, app);
out.push(TrackEventRequest::AppUsed(CodexAppUsedEventRequest {
event_type: "codex_app_used",
event_params,
}));
}
fn ingest_plugin_used(&mut self, input: PluginUsedInput, out: &mut Vec<TrackEventRequest>) {
let PluginUsedInput { tracking, plugin } = input;
out.push(TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest {
event_type: "codex_plugin_used",
event_params: codex_plugin_used_metadata(&tracking, plugin),
}));
}
fn ingest_plugin_state_changed(
&mut self,
input: PluginStateChangedInput,
out: &mut Vec<TrackEventRequest>,
) {
let PluginStateChangedInput { plugin, state } = input;
let event = CodexPluginEventRequest {
event_type: plugin_state_event_type(state),
event_params: codex_plugin_metadata(plugin),
};
out.push(match state {
PluginState::Installed => TrackEventRequest::PluginInstalled(event),
PluginState::Uninstalled => TrackEventRequest::PluginUninstalled(event),
PluginState::Enabled => TrackEventRequest::PluginEnabled(event),
PluginState::Disabled => TrackEventRequest::PluginDisabled(event),
});
}
}
async fn send_track_app_used(auth_manager: &AuthManager, base_url: &str, job: TrackAppUsedJob) {
let TrackAppUsedJob {
analytics_enabled,
tracking,
app,
} = job;
let event_params = codex_app_metadata(&tracking, app);
let events = vec![TrackEventRequest::AppUsed(CodexAppUsedEventRequest {
event_type: "codex_app_used",
event_params,
})];
send_track_events(auth_manager, analytics_enabled, base_url, events).await;
}
async fn send_track_turn_event(auth_manager: &AuthManager, base_url: &str, job: TrackTurnEventJob) {
let TrackTurnEventJob {
analytics_enabled,
tracking,
turn_event,
} = job;
let events = vec![TrackEventRequest::TurnEvent(CodexTurnEventRequest {
event_type: "codex_turn_event",
event_params: codex_turn_event_params(&tracking, turn_event),
})];
send_track_events(auth_manager, analytics_enabled, base_url, events).await;
}
async fn send_track_plugin_used(
auth_manager: &AuthManager,
base_url: &str,
job: TrackPluginUsedJob,
) {
let TrackPluginUsedJob {
analytics_enabled,
tracking,
plugin,
} = job;
let events = vec![TrackEventRequest::PluginUsed(CodexPluginUsedEventRequest {
event_type: "codex_plugin_used",
event_params: codex_plugin_used_metadata(&tracking, plugin),
})];
send_track_events(auth_manager, analytics_enabled, base_url, events).await;
}
async fn send_track_plugin_installed(
auth_manager: &AuthManager,
base_url: &str,
job: TrackPluginManagementJob,
) {
send_track_plugin_management_event(auth_manager, base_url, job, "codex_plugin_installed").await;
}
async fn send_track_plugin_uninstalled(
auth_manager: &AuthManager,
base_url: &str,
job: TrackPluginManagementJob,
) {
send_track_plugin_management_event(auth_manager, base_url, job, "codex_plugin_uninstalled")
.await;
}
async fn send_track_plugin_enabled(
auth_manager: &AuthManager,
base_url: &str,
job: TrackPluginManagementJob,
) {
send_track_plugin_management_event(auth_manager, base_url, job, "codex_plugin_enabled").await;
}
async fn send_track_plugin_disabled(
auth_manager: &AuthManager,
base_url: &str,
job: TrackPluginManagementJob,
) {
send_track_plugin_management_event(auth_manager, base_url, job, "codex_plugin_disabled").await;
}
async fn send_track_plugin_management_event(
auth_manager: &AuthManager,
base_url: &str,
job: TrackPluginManagementJob,
event_type: &'static str,
) {
let TrackPluginManagementJob {
analytics_enabled,
plugin,
} = job;
let event_params = codex_plugin_metadata(plugin);
let event = CodexPluginEventRequest {
event_type,
event_params,
};
let events = vec![match event_type {
"codex_plugin_installed" => TrackEventRequest::PluginInstalled(event),
"codex_plugin_uninstalled" => TrackEventRequest::PluginUninstalled(event),
"codex_plugin_enabled" => TrackEventRequest::PluginEnabled(event),
"codex_plugin_disabled" => TrackEventRequest::PluginDisabled(event),
_ => unreachable!("unknown plugin management event type"),
}];
send_track_events(auth_manager, analytics_enabled, base_url, events).await;
fn plugin_state_event_type(state: PluginState) -> &'static str {
match state {
PluginState::Installed => "codex_plugin_installed",
PluginState::Uninstalled => "codex_plugin_uninstalled",
PluginState::Enabled => "codex_plugin_enabled",
PluginState::Disabled => "codex_plugin_disabled",
}
}
fn codex_app_metadata(tracking: &TrackEventsContext, app: AppInvocation) -> CodexAppMetadata {
@@ -933,29 +735,33 @@ fn personality_mode(personality: Option<Personality>) -> Option<String> {
Some(personality) => Some(personality.to_string()),
}
}
fn codex_thread_initialized_event_params(
thread_event: CodexThreadInitializedEvent,
) -> CodexThreadInitializedEventParams {
codex_thread_initialized_event_params_with_product_client_id(originator().value, thread_event)
}
fn codex_thread_initialized_event_request(
product_client_id: String,
thread_event: CodexThreadInitializedEvent,
) -> CodexThreadInitializedEventRequest {
CodexThreadInitializedEventRequest {
event_type: "codex_thread_initialized",
event_params: codex_thread_initialized_event_params_with_product_client_id(
product_client_id,
thread_event,
),
}
}
fn codex_thread_initialized_event_params_with_product_client_id(
product_client_id: String,
thread_event: CodexThreadInitializedEvent,
) -> CodexThreadInitializedEventParams {
CodexThreadInitializedEventParams {
thread_id: thread_event.thread_id,
product_client_id: originator().value,
product_client_id,
model: thread_event.model,
model_provider: thread_event.model_provider,
reasoning_effort: thread_event.reasoning_effort.map(|value| value.to_string()),
reasoning_summary: thread_event
.reasoning_summary
.map(|value| value.to_string()),
service_tier: thread_event
.service_tier
.map(|value| value.to_string())
.unwrap_or_else(|| "default".to_string()),
approval_policy: thread_event.approval_policy.to_string(),
approvals_reviewer: thread_event.approvals_reviewer.to_string(),
sandbox_policy: sandbox_policy_mode(&thread_event.sandbox_policy),
sandbox_network_access: thread_event.sandbox_network_access,
collaboration_mode: collaboration_mode_mode(thread_event.collaboration_mode),
personality: thread_event.personality.map(|value| value.to_string()),
ephemeral: thread_event.ephemeral,
session_source: session_source_name(&thread_event.session_source),
initialization_mode: thread_event.initialization_mode,
@@ -1020,13 +826,9 @@ fn subagent_source_name(subagent_source: SubAgentSource) -> String {
async fn send_track_events(
auth_manager: &AuthManager,
analytics_enabled: Option<bool>,
base_url: &str,
events: Vec<TrackEventRequest>,
) {
if analytics_enabled == Some(false) {
return;
}
if events.is_empty() {
return;
}
codex-rs/analytics/src/analytics_client_tests.rs
-40
View File
@@ -269,16 +269,6 @@ fn thread_initialized_event_serializes_expected_shape() {
event_params: codex_thread_initialized_event_params(CodexThreadInitializedEvent {
thread_id: "thread-0".to_string(),
model: "gpt-5".to_string(),
model_provider: "openai".to_string(),
reasoning_effort: Some(ReasoningEffort::High),
reasoning_summary: Some(ReasoningSummary::Detailed),
service_tier: Some(ServiceTier::Flex),
approval_policy: AskForApproval::OnRequest,
approvals_reviewer: ApprovalsReviewer::GuardianSubagent,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
sandbox_network_access: false,
collaboration_mode: ModeKind::Plan,
personality: Some(Personality::Friendly),
ephemeral: true,
session_source: SessionSource::Exec,
initialization_mode: InitializationMode::New,
@@ -298,16 +288,6 @@ fn thread_initialized_event_serializes_expected_shape() {
"thread_id": "thread-0",
"product_client_id": originator().value,
"model": "gpt-5",
"model_provider": "openai",
"reasoning_effort": "high",
"reasoning_summary": "detailed",
"service_tier": "flex",
"approval_policy": "on-request",
"approvals_reviewer": "guardian_subagent",
"sandbox_policy": "read_only",
"sandbox_network_access": false,
"collaboration_mode": "plan",
"personality": "friendly",
"ephemeral": true,
"session_source": "user",
"initialization_mode": "new",
@@ -326,16 +306,6 @@ fn thread_initialized_event_serializes_subagent_source() {
event_params: codex_thread_initialized_event_params(CodexThreadInitializedEvent {
thread_id: "thread-1".to_string(),
model: "gpt-5".to_string(),
model_provider: "openai".to_string(),
reasoning_effort: None,
reasoning_summary: None,
service_tier: None,
approval_policy: AskForApproval::OnRequest,
approvals_reviewer: ApprovalsReviewer::User,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
sandbox_network_access: false,
collaboration_mode: ModeKind::Default,
personality: None,
ephemeral: false,
session_source: SessionSource::SubAgent(SubAgentSource::Review),
initialization_mode: InitializationMode::New,
@@ -358,16 +328,6 @@ fn thread_initialized_event_omits_non_user_non_subagent_session_source() {
event_params: codex_thread_initialized_event_params(CodexThreadInitializedEvent {
thread_id: "thread-2".to_string(),
model: "gpt-5".to_string(),
model_provider: "openai".to_string(),
reasoning_effort: None,
reasoning_summary: None,
service_tier: None,
approval_policy: AskForApproval::OnRequest,
approvals_reviewer: ApprovalsReviewer::User,
sandbox_policy: SandboxPolicy::new_read_only_policy(),
sandbox_network_access: false,
collaboration_mode: ModeKind::Default,
personality: None,
ephemeral: false,
session_source: SessionSource::Mcp,
initialization_mode: InitializationMode::New,
codex-rs/analytics/src/lib.rs
+10
View File
@@ -1,11 +1,21 @@
mod analytics_client;
pub use analytics_client::AnalyticsEventsClient;
pub use analytics_client::AnalyticsInput;
pub use analytics_client::AnalyticsReducer;
pub use analytics_client::AppInvocation;
pub use analytics_client::AppMentionedInput;
pub use analytics_client::AppUsedInput;
pub use analytics_client::CodexThreadContext;
pub use analytics_client::CodexThreadInitializedEvent;
pub use analytics_client::CodexThreadInitializedInput;
pub use analytics_client::CodexTurnEvent;
pub use analytics_client::InitializationMode;
pub use analytics_client::InvocationType;
pub use analytics_client::PluginState;
pub use analytics_client::PluginStateChangedInput;
pub use analytics_client::PluginUsedInput;
pub use analytics_client::SkillInvocation;
pub use analytics_client::SkillInvokedInput;
pub use analytics_client::TrackEventsContext;
pub use analytics_client::build_track_events_context;
codex-rs/core/src/codex.rs
+13 -29
View File
@@ -352,7 +352,8 @@ use crate::util::backoff;
use crate::windows_sandbox::WindowsSandboxLevelExt;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::AppInvocation;
use codex_analytics::CodexThreadInitializedEvent;
use codex_analytics::CodexThreadContext;
use codex_analytics::CodexThreadInitializedInput;
use codex_analytics::CodexTurnEvent;
use codex_analytics::InitializationMode;
use codex_analytics::InvocationType;
@@ -673,43 +674,26 @@ impl Codex {
session
.services
.analytics_events_client
.track_thread_initialized(CodexThreadInitializedEvent {
.track_thread_initialized(CodexThreadInitializedInput {
thread_id: thread_id.to_string(),
model: thread_initialized_configuration
.collaboration_mode
.model()
.to_string(),
model_provider: thread_initialized_configuration
.original_config_do_not_use
.model_provider_id
.clone(),
reasoning_effort: thread_initialized_configuration
.collaboration_mode
.reasoning_effort(),
reasoning_summary: thread_initialized_configuration.model_reasoning_summary,
service_tier: thread_initialized_configuration.service_tier,
approval_policy: thread_initialized_configuration.approval_policy.value(),
approvals_reviewer: thread_initialized_configuration.approvals_reviewer,
sandbox_policy: thread_initialized_configuration
.sandbox_policy
.get()
.clone(),
sandbox_network_access: thread_initialized_configuration
.network_sandbox_policy
.is_enabled(),
collaboration_mode: thread_initialized_configuration.collaboration_mode.mode,
personality: thread_initialized_configuration.personality,
ephemeral: thread_initialized_configuration
.original_config_do_not_use
.ephemeral,
initialization_mode,
subagent_source: session_source_subagent_source(&thread_session_source),
parent_thread_id: session_source_parent_thread_id(&thread_session_source),
session_source: thread_session_source,
product_client_id: crate::default_client::originator().value,
created_at: SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs(),
thread_context: CodexThreadContext {
ephemeral: thread_initialized_configuration
.original_config_do_not_use
.ephemeral,
session_source: thread_session_source,
initialization_mode,
subagent_source: session_source_subagent_source(&thread_session_source),
parent_thread_id: session_source_parent_thread_id(&thread_session_source),
},
});
// This task will run until Op::Shutdown is received.
codex-rs/core/tests/suite/thread_metadata.rs
-130
View File
@@ -1,13 +1,6 @@
use anyhow::Result;
use codex_core::CodexAuth;
use codex_core::RolloutRecorder;
use codex_core::config::Constrained;
use codex_protocol::config_types::ApprovalsReviewer;
use codex_protocol::config_types::Personality;
use codex_protocol::config_types::ReasoningSummary;
use codex_protocol::config_types::ServiceTier;
use codex_protocol::openai_models::ReasoningEffort;
use codex_protocol::protocol::InitialHistory;
use core_test_support::responses::start_mock_server;
use core_test_support::test_codex::test_codex;
use std::time::Duration;
@@ -23,14 +16,9 @@ async fn thread_initialization_tracks_thread_initialized_analytics() -> Result<(
.with_config(move |config| {
config.chatgpt_base_url = chatgpt_base_url;
config.model = Some("gpt-5".to_string());
config.model_reasoning_effort = Some(ReasoningEffort::High);
config.model_reasoning_summary = Some(ReasoningSummary::Detailed);
config.service_tier = Some(ServiceTier::Flex);
config.approvals_reviewer = ApprovalsReviewer::GuardianSubagent;
config.permissions.sandbox_policy = Constrained::allow_any(
codex_protocol::protocol::SandboxPolicy::new_workspace_write_policy(),
);
config.personality = Some(Personality::Friendly);
config.ephemeral = true;
})
.build(&server)
@@ -69,19 +57,6 @@ async fn thread_initialization_tracks_thread_initialized_analytics() -> Result<(
serde_json::json!(codex_core::default_client::originator().value)
);
assert_eq!(event["event_params"]["model"], "gpt-5");
assert_eq!(event["event_params"]["model_provider"], "openai");
assert_eq!(event["event_params"]["reasoning_effort"], "high");
assert_eq!(event["event_params"]["reasoning_summary"], "detailed");
assert_eq!(event["event_params"]["service_tier"], "flex");
assert_eq!(event["event_params"]["approval_policy"], "on-request");
assert_eq!(
event["event_params"]["approvals_reviewer"],
"guardian_subagent"
);
assert_eq!(event["event_params"]["sandbox_policy"], "workspace_write");
assert_eq!(event["event_params"]["sandbox_network_access"], false);
assert_eq!(event["event_params"]["collaboration_mode"], "default");
assert_eq!(event["event_params"]["personality"], "friendly");
assert_eq!(event["event_params"]["ephemeral"], true);
assert_eq!(event["event_params"]["session_source"], "user");
assert_eq!(event["event_params"]["initialization_mode"], "new");
@@ -101,108 +76,3 @@ async fn thread_initialization_tracks_thread_initialized_analytics() -> Result<(
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn resumed_thread_emits_thread_initialized_analytics() -> Result<()> {
let server = start_mock_server().await;
let chatgpt_base_url = server.uri();
let initial = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config({
let chatgpt_base_url = chatgpt_base_url.clone();
move |config| {
config.chatgpt_base_url = chatgpt_base_url;
config.model = Some("gpt-5".to_string());
config.model_reasoning_effort = Some(ReasoningEffort::High);
config.model_reasoning_summary = Some(ReasoningSummary::Detailed);
config.service_tier = Some(ServiceTier::Flex);
config.approvals_reviewer = ApprovalsReviewer::GuardianSubagent;
config.permissions.sandbox_policy = Constrained::allow_any(
codex_protocol::protocol::SandboxPolicy::new_workspace_write_policy(),
);
config.personality = Some(Personality::Friendly);
}
})
.build(&server)
.await?;
let rollout_path = initial
.codex
.rollout_path()
.expect("rollout path for initial thread");
let resume_deadline = Instant::now() + Duration::from_secs(10);
loop {
if matches!(
RolloutRecorder::get_rollout_history(&rollout_path).await?,
InitialHistory::Resumed(_)
) {
break;
}
if Instant::now() >= resume_deadline {
panic!("timed out waiting for rollout to become resumable");
}
tokio::time::sleep(Duration::from_millis(50)).await;
}
let _resumed = test_codex()
.with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing())
.with_config(move |config| {
config.chatgpt_base_url = chatgpt_base_url;
config.model = Some("gpt-5".to_string());
config.model_reasoning_effort = Some(ReasoningEffort::High);
config.model_reasoning_summary = Some(ReasoningSummary::Detailed);
config.service_tier = Some(ServiceTier::Flex);
config.approvals_reviewer = ApprovalsReviewer::GuardianSubagent;
config.permissions.sandbox_policy = Constrained::allow_any(
codex_protocol::protocol::SandboxPolicy::new_workspace_write_policy(),
);
config.personality = Some(Personality::Friendly);
})
.resume(&server, initial.home.clone(), rollout_path)
.await?;
let deadline = Instant::now() + Duration::from_secs(10);
let analytics_request = loop {
let requests = server.received_requests().await.unwrap_or_default();
if let Some(request) = requests.into_iter().find(|request| {
if request.url.path() != "/codex/analytics-events/events" {
return false;
}
let Ok(payload) = serde_json::from_slice::<serde_json::Value>(&request.body) else {
return false;
};
payload["events"]
.as_array()
.into_iter()
.flatten()
.any(|event| {
event["event_type"] == "codex_thread_initialized"
&& event["event_params"]["initialization_mode"] == "resumed"
})
}) {
break request;
}
if Instant::now() >= deadline {
panic!("timed out waiting for resumed thread analytics request");
}
tokio::time::sleep(Duration::from_millis(50)).await;
};
let payload: serde_json::Value =
serde_json::from_slice(&analytics_request.body).expect("analytics payload");
let event = payload["events"]
.as_array()
.expect("events array")
.iter()
.find(|event| {
event["event_type"] == "codex_thread_initialized"
&& event["event_params"]["initialization_mode"] == "resumed"
})
.expect("codex_thread_initialized resumed event should be present");
assert_eq!(event["event_params"]["session_source"], "user");
assert_eq!(event["event_params"]["initialization_mode"], "resumed");
Ok(())
}