openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-26 18:38:47 +00:00
Code Issues Packages Projects Releases Wiki Activity

rebase

Browse Source
This commit is contained in:
Roy Han
2026-03-26 14:52:40 -07:00
parent 5a449cd2d0 9e6d9eb198
commit c278d0ab61
16 changed files with 555 additions and 289 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+1
View File
@@ -1388,6 +1388,7 @@ checksum = "e9b18233253483ce2f65329a24072ec414db782531bdbb7d0bbc4bd2ce6b7e21"
name = "codex-analytics"
version = "0.0.0"
dependencies = [
"codex-app-server-protocol",
"codex-git-utils",
"codex-login",
"codex-plugin",
codex-rs/analytics/Cargo.toml
+1
View File
@@ -13,6 +13,7 @@ path = "src/lib.rs"
workspace = true
[dependencies]
codex-app-server-protocol = { workspace = true }
codex-git-utils = { workspace = true }
codex-login = { workspace = true }
codex-plugin = { workspace = true }
codex-rs/analytics/src/analytics_client.rs
+172 -96
View File
@@ -1,3 +1,7 @@
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::InitializeParams;
use codex_app_server_protocol::RequestId;
use codex_app_server_protocol::ServerNotification;
use codex_git_utils::collect_git_info;
use codex_git_utils::get_git_repo_root;
use codex_login::AuthManager;
@@ -87,21 +91,13 @@ pub enum TurnStatus {
}
#[derive(Clone)]
pub struct CodexThreadInitializedInput {
pub struct ThreadInitializeInput {
pub connection_id: u64,
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)]
@@ -145,8 +141,29 @@ pub struct AppInvocation {
pub invocation_type: Option<InvocationType>,
}
pub enum AnalyticsInput {
CodexThreadInitialized(CodexThreadInitializedInput),
pub enum AnalyticsFact {
Initialize {
connection_id: u64,
params: InitializeParams,
},
Request {
connection_id: u64,
request_id: RequestId,
request: ClientRequest,
},
Notification {
connection_id: u64,
notification: ServerNotification,
},
// Facts that do not naturally exist on the app-server protocol surface, or
// would require non-trivial protocol reshaping on this branch.
Custom(CustomAnalyticsFact),
}
pub enum CustomAnalyticsFact {
// This remains custom on this branch because app-server-protocol does not
// yet expose a generic client response enum we can reduce over directly.
ThreadInitialized(ThreadInitializeInput),
TurnStarted(TurnStartedInput),
TurnCompleted(TurnCompletedInput),
SkillInvoked(SkillInvokedInput),
@@ -200,12 +217,12 @@ pub enum PluginState {
#[derive(Default)]
pub struct AnalyticsReducer {
threads: HashMap<String, ThreadState>,
turns: HashMap<String, TurnState>,
clients: HashMap<u64, ClientState>,
}
struct ThreadState {
_initialized_input: CodexThreadInitializedInput,
struct ClientState {
product_client_id: String,
}
struct TurnState {
@@ -214,7 +231,7 @@ struct TurnState {
#[derive(Clone)]
pub(crate) struct AnalyticsEventsQueue {
sender: mpsc::Sender<AnalyticsInput>,
sender: mpsc::Sender<AnalyticsFact>,
app_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
plugin_used_emitted_keys: Arc<Mutex<HashSet<(String, String)>>>,
}
@@ -243,7 +260,7 @@ impl AnalyticsEventsQueue {
}
}
fn try_send(&self, input: AnalyticsInput) {
fn try_send(&self, input: AnalyticsFact) {
if self.sender.try_send(input).is_err() {
//TODO: add a metric for this
tracing::warn!("dropping analytics events: queue is full");
@@ -300,93 +317,106 @@ impl AnalyticsEventsClient {
if invocations.is_empty() {
return;
}
self.record(AnalyticsInput::SkillInvoked(SkillInvokedInput {
tracking,
invocations,
}));
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::SkillInvoked(
SkillInvokedInput {
tracking,
invocations,
},
)));
}
pub fn track_thread_initialized(&self, input: CodexThreadInitializedInput) {
self.record(AnalyticsInput::CodexThreadInitialized(input));
pub fn track_initialize(&self, connection_id: u64, params: InitializeParams) {
self.record_fact(AnalyticsFact::Initialize {
connection_id,
params,
});
}
pub fn track_thread_initialized(&self, input: ThreadInitializeInput) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::ThreadInitialized(input),
));
}
pub fn track_app_mentioned(&self, tracking: TrackEventsContext, mentions: Vec<AppInvocation>) {
if mentions.is_empty() {
return;
}
self.record(AnalyticsInput::AppMentioned(AppMentionedInput {
tracking,
mentions,
}));
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppMentioned(
AppMentionedInput { tracking, mentions },
)));
}
pub fn track_app_used(&self, tracking: TrackEventsContext, app: AppInvocation) {
if !self.queue.should_enqueue_app_used(&tracking, &app) {
return;
}
self.record(AnalyticsInput::AppUsed(AppUsedInput { tracking, app }));
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::AppUsed(
AppUsedInput { tracking, app },
)));
}
pub fn track_plugin_used(&self, tracking: TrackEventsContext, plugin: PluginTelemetryMetadata) {
if !self.queue.should_enqueue_plugin_used(&tracking, &plugin) {
return;
}
self.record(AnalyticsInput::PluginUsed(PluginUsedInput {
tracking,
plugin,
}));
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::PluginUsed(
PluginUsedInput { tracking, plugin },
)));
}
pub fn track_turn_started(&self, tracking: TrackEventsContext, turn_event: CodexTurnEvent) {
self.record(AnalyticsInput::TurnStarted(TurnStartedInput {
tracking,
turn_event,
}));
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::TurnStarted(
TurnStartedInput {
tracking,
turn_event,
},
)));
}
pub fn track_turn_completed(&self, turn_id: String) {
self.record(AnalyticsInput::TurnCompleted(TurnCompletedInput {
turn_id,
}));
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::TurnCompleted(
TurnCompletedInput { turn_id },
)));
}
pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) {
self.record(AnalyticsInput::PluginStateChanged(
PluginStateChangedInput {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Installed,
},
}),
));
}
pub fn track_plugin_uninstalled(&self, plugin: PluginTelemetryMetadata) {
self.record(AnalyticsInput::PluginStateChanged(
PluginStateChangedInput {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Uninstalled,
},
}),
));
}
pub fn track_plugin_enabled(&self, plugin: PluginTelemetryMetadata) {
self.record(AnalyticsInput::PluginStateChanged(
PluginStateChangedInput {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Enabled,
},
}),
));
}
pub fn track_plugin_disabled(&self, plugin: PluginTelemetryMetadata) {
self.record(AnalyticsInput::PluginStateChanged(
PluginStateChangedInput {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {
plugin,
state: PluginState::Disabled,
},
}),
));
}
pub fn record(&self, input: AnalyticsInput) {
pub fn record_fact(&self, input: AnalyticsFact) {
if self.analytics_enabled == Some(false) {
return;
}
@@ -557,48 +587,71 @@ struct CodexPluginUsedEventRequest {
}
impl AnalyticsReducer {
async fn ingest(&mut self, input: AnalyticsInput, out: &mut Vec<TrackEventRequest>) {
async fn ingest(&mut self, input: AnalyticsFact, out: &mut Vec<TrackEventRequest>) {
match input {
AnalyticsInput::CodexThreadInitialized(input) => {
self.ingest_thread_initialized(input, out);
}
AnalyticsInput::TurnStarted(input) => {
self.ingest_turn_started(input);
}
AnalyticsInput::TurnCompleted(input) => {
self.ingest_turn_completed(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);
AnalyticsFact::Initialize {
connection_id,
params,
} => {
self.ingest_initialize(connection_id, params);
}
AnalyticsFact::Request {
connection_id: _connection_id,
request_id: _request_id,
request: _request,
} => {}
AnalyticsFact::Notification {
connection_id: _connection_id,
notification: _notification,
} => {}
AnalyticsFact::Custom(input) => match input {
CustomAnalyticsFact::ThreadInitialized(input) => {
self.ingest_thread_initialized(input, out);
}
CustomAnalyticsFact::TurnStarted(input) => {
self.ingest_turn_started(input);
}
CustomAnalyticsFact::TurnCompleted(input) => {
self.ingest_turn_completed(input, out);
}
CustomAnalyticsFact::SkillInvoked(input) => {
self.ingest_skill_invoked(input, out).await;
}
CustomAnalyticsFact::AppMentioned(input) => {
self.ingest_app_mentioned(input, out);
}
CustomAnalyticsFact::AppUsed(input) => {
self.ingest_app_used(input, out);
}
CustomAnalyticsFact::PluginUsed(input) => {
self.ingest_plugin_used(input, out);
}
CustomAnalyticsFact::PluginStateChanged(input) => {
self.ingest_plugin_state_changed(input, out);
}
},
}
}
fn ingest_initialize(&mut self, connection_id: u64, params: InitializeParams) {
self.clients.insert(
connection_id,
ClientState {
product_client_id: params.client_info.name,
},
);
}
fn ingest_thread_initialized(
&mut self,
input: CodexThreadInitializedInput,
input: ThreadInitializeInput,
out: &mut Vec<TrackEventRequest>,
) {
self.threads.insert(
input.thread_id.clone(),
ThreadState {
_initialized_input: input.clone(),
},
);
let Some(client_state) = self.clients.get(&input.connection_id) else {
return;
};
out.push(TrackEventRequest::CodexThreadInitialized(
codex_thread_initialized_event_request(input),
codex_thread_initialized_event_request(client_state.product_client_id.clone(), input),
));
}
@@ -824,30 +877,30 @@ fn personality_mode(personality: Option<Personality>) -> Option<String> {
}
fn codex_thread_initialized_event_request(
input: CodexThreadInitializedInput,
product_client_id: String,
input: ThreadInitializeInput,
) -> CodexThreadInitializedEvent {
CodexThreadInitializedEvent {
event_type: "codex_thread_initialized",
event_params: codex_thread_initialized_event_params(input),
event_params: codex_thread_initialized_event_params(product_client_id, input),
}
}
fn codex_thread_initialized_event_params(
input: CodexThreadInitializedInput,
product_client_id: String,
input: ThreadInitializeInput,
) -> CodexThreadInitializedEventParams {
CodexThreadInitializedEventParams {
thread_id: input.thread_id,
product_client_id: input.product_client_id,
product_client_id,
model: input.model,
ephemeral: input.thread_context.ephemeral,
session_source: session_source_name(&input.thread_context.session_source),
initialization_mode: input.thread_context.initialization_mode,
subagent_source: input
.thread_context
.subagent_source
ephemeral: input.ephemeral,
session_source: session_source_name(&input.session_source),
initialization_mode: input.initialization_mode,
subagent_source: session_source_subagent_source(&input.session_source)
.map(subagent_source_name),
parent_thread_id: input.thread_context.parent_thread_id,
created_at: input.created_at,
parent_thread_id: session_source_parent_thread_id(&input.session_source),
created_at: now_unix_timestamp_secs(),
}
}
@@ -904,6 +957,29 @@ fn subagent_source_name(subagent_source: SubAgentSource) -> String {
}
}
fn session_source_subagent_source(session_source: &SessionSource) -> Option<SubAgentSource> {
match session_source {
SessionSource::SubAgent(subagent_source) => Some(subagent_source.clone()),
_ => None,
}
}
fn session_source_parent_thread_id(session_source: &SessionSource) -> Option<String> {
match session_source {
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id, ..
}) => Some(parent_thread_id.to_string()),
_ => None,
}
}
fn now_unix_timestamp_secs() -> u64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
}
async fn send_track_events(
auth_manager: &AuthManager,
base_url: &str,
codex-rs/analytics/src/analytics_client_tests.rs
+100 -49
View File
@@ -1,17 +1,17 @@
use super::AnalyticsEventsQueue;
use super::AnalyticsInput;
use super::AnalyticsFact;
use super::AnalyticsReducer;
use super::AppInvocation;
use super::CodexAppMentionedEventRequest;
use super::CodexAppUsedEventRequest;
use super::CodexPluginEventRequest;
use super::CodexPluginUsedEventRequest;
use super::CodexThreadContext;
use super::CodexThreadInitializedInput;
use super::CodexTurnEvent;
use super::CodexTurnEventRequest;
use super::CustomAnalyticsFact;
use super::InitializationMode;
use super::InvocationType;
use super::ThreadInitializeInput;
use super::TrackEventRequest;
use super::TrackEventsContext;
use super::TurnCompletedInput;
@@ -22,6 +22,8 @@ use super::codex_plugin_used_metadata;
use super::codex_thread_initialized_event_request;
use super::codex_turn_event_params;
use super::normalize_path_for_skill_id;
use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::InitializeParams;
use codex_login::default_client::originator;
use codex_plugin::AppConnectorId;
use codex_plugin::PluginCapabilitySummary;
@@ -315,7 +317,7 @@ async fn turn_started_then_completed_emits_turn_event() {
reducer
.ingest(
AnalyticsInput::TurnStarted(TurnStartedInput {
AnalyticsFact::Custom(CustomAnalyticsFact::TurnStarted(TurnStartedInput {
tracking: tracking.clone(),
turn_event: CodexTurnEvent {
submission_type: None,
@@ -351,7 +353,7 @@ async fn turn_started_then_completed_emits_turn_event() {
started_at: None,
completed_at: None,
},
}),
})),
&mut out,
)
.await;
@@ -360,9 +362,9 @@ async fn turn_started_then_completed_emits_turn_event() {
reducer
.ingest(
AnalyticsInput::TurnCompleted(TurnCompletedInput {
AnalyticsFact::Custom(CustomAnalyticsFact::TurnCompleted(TurnCompletedInput {
turn_id: tracking.turn_id.clone(),
}),
})),
&mut out,
)
.await;
@@ -377,18 +379,14 @@ async fn turn_started_then_completed_emits_turn_event() {
#[test]
fn thread_initialized_event_serializes_expected_shape() {
let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request(
CodexThreadInitializedInput {
"codex-tui".to_string(),
ThreadInitializeInput {
connection_id: 1,
thread_id: "thread-0".to_string(),
model: "gpt-5".to_string(),
product_client_id: originator().value,
created_at: 1_716_000_000,
thread_context: CodexThreadContext {
ephemeral: true,
session_source: SessionSource::Exec,
initialization_mode: InitializationMode::New,
subagent_source: None,
parent_thread_id: None,
},
ephemeral: true,
session_source: SessionSource::Exec,
initialization_mode: InitializationMode::New,
},
));
@@ -400,34 +398,31 @@ fn thread_initialized_event_serializes_expected_shape() {
"event_type": "codex_thread_initialized",
"event_params": {
"thread_id": "thread-0",
"product_client_id": originator().value,
"product_client_id": "codex-tui",
"model": "gpt-5",
"ephemeral": true,
"session_source": "user",
"initialization_mode": "new",
"subagent_source": null,
"parent_thread_id": null,
"created_at": 1716000000
"created_at": payload["event_params"]["created_at"]
}
})
);
assert!(payload["event_params"]["created_at"].as_u64().is_some());
}
#[test]
fn thread_initialized_event_serializes_subagent_source() {
let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request(
CodexThreadInitializedInput {
"codex-tui".to_string(),
ThreadInitializeInput {
connection_id: 1,
thread_id: "thread-1".to_string(),
model: "gpt-5".to_string(),
product_client_id: originator().value,
created_at: 1,
thread_context: CodexThreadContext {
ephemeral: false,
session_source: SessionSource::SubAgent(SubAgentSource::Review),
initialization_mode: InitializationMode::New,
subagent_source: Some(SubAgentSource::Review),
parent_thread_id: None,
},
ephemeral: false,
session_source: SessionSource::SubAgent(SubAgentSource::Review),
initialization_mode: InitializationMode::New,
},
));
@@ -437,28 +432,84 @@ fn thread_initialized_event_serializes_subagent_source() {
assert_eq!(payload["event_params"]["subagent_source"], "review");
}
#[test]
fn thread_initialized_event_omits_non_user_non_subagent_session_source() {
let event = TrackEventRequest::CodexThreadInitialized(codex_thread_initialized_event_request(
CodexThreadInitializedInput {
thread_id: "thread-2".to_string(),
model: "gpt-5".to_string(),
product_client_id: originator().value,
created_at: 1,
thread_context: CodexThreadContext {
ephemeral: false,
session_source: SessionSource::Mcp,
initialization_mode: InitializationMode::New,
subagent_source: None,
parent_thread_id: None,
},
},
));
#[tokio::test]
async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialized() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
let payload = serde_json::to_value(&event).expect("serialize mcp thread initialized event");
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitialized(
ThreadInitializeInput {
connection_id: 7,
thread_id: "thread-no-client".to_string(),
model: "gpt-5".to_string(),
ephemeral: false,
session_source: SessionSource::Exec,
initialization_mode: InitializationMode::New,
},
)),
&mut events,
)
.await;
assert!(events.is_empty(), "thread events should require initialize");
reducer
.ingest(
AnalyticsFact::Initialize {
connection_id: 7,
params: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "1.0.0".to_string(),
},
capabilities: None,
},
},
&mut events,
)
.await;
assert!(events.is_empty(), "initialize should not publish by itself");
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::ThreadInitialized(
ThreadInitializeInput {
connection_id: 7,
thread_id: "thread-1".to_string(),
model: "gpt-5".to_string(),
ephemeral: true,
session_source: SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id: codex_protocol::ThreadId::from_string(
"11111111-1111-1111-1111-111111111111",
)
.expect("valid thread id"),
depth: 1,
agent_path: None,
agent_nickname: None,
agent_role: None,
}),
initialization_mode: InitializationMode::Resumed,
},
)),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload.as_array().expect("events array").len(), 1);
assert_eq!(payload[0]["event_type"], "codex_thread_initialized");
assert_eq!(payload[0]["event_params"]["product_client_id"], "codex-tui");
assert_eq!(payload[0]["event_params"]["initialization_mode"], "resumed");
assert_eq!(payload[0]["event_params"]["session_source"], "subagent");
assert_eq!(
payload["event_params"]["session_source"],
serde_json::Value::Null
payload[0]["event_params"]["subagent_source"],
"thread_spawn"
);
assert_eq!(
payload[0]["event_params"]["parent_thread_id"],
"11111111-1111-1111-1111-111111111111"
);
}
codex-rs/analytics/src/lib.rs
+3 -3
View File
@@ -1,14 +1,13 @@
mod analytics_client;
pub use analytics_client::AnalyticsEventsClient;
pub use analytics_client::AnalyticsInput;
pub use analytics_client::AnalyticsFact;
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::CodexThreadInitializedInput;
pub use analytics_client::CodexTurnEvent;
pub use analytics_client::CustomAnalyticsFact;
pub use analytics_client::InitializationMode;
pub use analytics_client::InvocationType;
pub use analytics_client::PluginState;
@@ -16,6 +15,7 @@ pub use analytics_client::PluginStateChangedInput;
pub use analytics_client::PluginUsedInput;
pub use analytics_client::SkillInvocation;
pub use analytics_client::SkillInvokedInput;
pub use analytics_client::ThreadInitializeInput;
pub use analytics_client::TrackEventsContext;
pub use analytics_client::TurnCompletedInput;
pub use analytics_client::TurnStartedInput;
codex-rs/app-server/src/codex_message_processor.rs
+50
View File
@@ -179,16 +179,19 @@ use codex_arg0::Arg0DispatchPaths;
use codex_backend_client::Client as BackendClient;
use codex_chatgpt::connectors;
use codex_cloud_requirements::cloud_requirements_loader;
use codex_core::AnalyticsEventsClient;
use codex_core::AuthManager;
use codex_core::CodexAuth;
use codex_core::CodexThread;
use codex_core::Cursor as RolloutCursor;
use codex_core::ForkSnapshot;
use codex_core::InitializationMode;
use codex_core::NewThread;
use codex_core::RolloutRecorder;
use codex_core::SessionMeta;
use codex_core::SteerInputError;
use codex_core::ThreadConfigSnapshot;
use codex_core::ThreadInitializeInput;
use codex_core::ThreadManager;
use codex_core::ThreadSortKey as CoreThreadSortKey;
use codex_core::auth::AuthMode as CoreAuthMode;
@@ -374,6 +377,7 @@ pub(crate) struct CodexMessageProcessor {
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
arg0_paths: Arg0DispatchPaths,
config: Arc<Config>,
cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
@@ -404,6 +408,7 @@ struct ListenerTaskContext {
thread_manager: Arc<ThreadManager>,
thread_state_manager: ThreadStateManager,
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
thread_watch_manager: ThreadWatchManager,
fallback_model_provider: String,
codex_home: PathBuf,
@@ -426,6 +431,7 @@ pub(crate) struct CodexMessageProcessorArgs {
pub(crate) auth_manager: Arc<AuthManager>,
pub(crate) thread_manager: Arc<ThreadManager>,
pub(crate) outgoing: Arc<OutgoingMessageSender>,
pub(crate) analytics_events_client: AnalyticsEventsClient,
pub(crate) arg0_paths: Arg0DispatchPaths,
pub(crate) config: Arc<Config>,
pub(crate) cli_overrides: Arc<RwLock<Vec<(String, TomlValue)>>>,
@@ -490,6 +496,7 @@ impl CodexMessageProcessor {
auth_manager,
thread_manager,
outgoing,
analytics_events_client,
arg0_paths,
config,
cli_overrides,
@@ -502,6 +509,7 @@ impl CodexMessageProcessor {
auth_manager,
thread_manager,
outgoing: outgoing.clone(),
analytics_events_client,
arg0_paths,
config,
cli_overrides,
@@ -1921,6 +1929,7 @@ impl CodexMessageProcessor {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -2153,6 +2162,14 @@ impl CodexMessageProcessor {
sandbox: config_snapshot.sandbox_policy.into(),
reasoning_effort: config_snapshot.reasoning_effort,
};
listener_task_context
.analytics_events_client
.track_thread_initialized(thread_initialize_input(
request_id.connection_id,
&thread,
response.model.clone(),
InitializationMode::New,
));
listener_task_context
.outgoing
@@ -3631,6 +3648,13 @@ impl CodexMessageProcessor {
sandbox: session_configured.sandbox_policy.into(),
reasoning_effort: session_configured.reasoning_effort,
};
self.analytics_events_client
.track_thread_initialized(thread_initialize_input(
request_id.connection_id,
&response.thread,
response.model.clone(),
InitializationMode::Resumed,
));
self.outgoing.send_response(request_id, response).await;
}
@@ -4238,6 +4262,13 @@ impl CodexMessageProcessor {
sandbox: session_configured.sandbox_policy.into(),
reasoning_effort: session_configured.reasoning_effort,
};
self.analytics_events_client
.track_thread_initialized(thread_initialize_input(
request_id.connection_id,
&thread,
response.model.clone(),
InitializationMode::Forked,
));
self.outgoing.send_response(request_id, response).await;
@@ -6819,6 +6850,7 @@ impl CodexMessageProcessor {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -6906,6 +6938,7 @@ impl CodexMessageProcessor {
thread_manager: Arc::clone(&self.thread_manager),
thread_state_manager: self.thread_state_manager.clone(),
outgoing: Arc::clone(&self.outgoing),
analytics_events_client: self.analytics_events_client.clone(),
thread_watch_manager: self.thread_watch_manager.clone(),
fallback_model_provider: self.config.model_provider_id.clone(),
codex_home: self.config.codex_home.clone(),
@@ -6937,6 +6970,7 @@ impl CodexMessageProcessor {
outgoing,
thread_manager,
thread_state_manager,
analytics_events_client: _,
thread_watch_manager,
fallback_model_provider,
codex_home,
@@ -7854,6 +7888,22 @@ fn cloud_requirements_load_error(err: &std::io::Error) -> Option<&CloudRequireme
None
}
fn thread_initialize_input(
connection_id: ConnectionId,
thread: &Thread,
model: String,
initialization_mode: InitializationMode,
) -> ThreadInitializeInput {
ThreadInitializeInput {
connection_id: connection_id.0,
thread_id: thread.id.clone(),
model,
ephemeral: thread.ephemeral,
session_source: thread.source.clone().into(),
initialization_mode,
}
}
fn config_load_error(err: &std::io::Error) -> JSONRPCErrorError {
let data = cloud_requirements_load_error(err).map(|cloud_error| {
let mut data = serde_json::json!({
codex-rs/app-server/src/message_processor.rs
+8 -2
View File
@@ -158,6 +158,7 @@ pub(crate) struct MessageProcessor {
external_agent_config_api: ExternalAgentConfigApi,
fs_api: FsApi,
auth_manager: Arc<AuthManager>,
analytics_events_client: AnalyticsEventsClient,
fs_watch_manager: FsWatchManager,
config: Arc<Config>,
config_warnings: Arc<Vec<ConfigWarningNotification>>,
@@ -238,6 +239,7 @@ impl MessageProcessor {
auth_manager: auth_manager.clone(),
thread_manager: Arc::clone(&thread_manager),
outgoing: outgoing.clone(),
analytics_events_client: analytics_events_client.clone(),
arg0_paths,
config: Arc::clone(&config),
cli_overrides: cli_overrides.clone(),
@@ -258,7 +260,7 @@ impl MessageProcessor {
loader_overrides,
cloud_requirements,
thread_manager,
analytics_events_client,
analytics_events_client.clone(),
);
let external_agent_config_api = ExternalAgentConfigApi::new(config.codex_home.clone());
let fs_api = FsApi::default();
@@ -271,6 +273,7 @@ impl MessageProcessor {
external_agent_config_api,
fs_api,
auth_manager,
analytics_events_client,
fs_watch_manager,
config,
config_warnings: Arc::new(config_warnings),
@@ -543,6 +546,7 @@ impl MessageProcessor {
// shared thread when another connected client did not opt into
// experimental API). Proposed direction is instance-global first-write-wins
// with initialize-time mismatch rejection.
let analytics_initialize_params = params.clone();
let (experimental_api_enabled, opt_out_notification_methods) =
match params.capabilities {
Some(capabilities) => (
@@ -570,7 +574,7 @@ impl MessageProcessor {
} else {
name.clone()
};
if let Err(error) = set_default_originator(originator) {
if let Err(error) = set_default_originator(originator.clone()) {
match error {
SetOriginatorError::InvalidHeaderValue => {
let error = JSONRPCErrorError {
@@ -593,6 +597,8 @@ impl MessageProcessor {
}
}
}
self.analytics_events_client
.track_initialize(connection_id.0, analytics_initialize_params);
set_default_client_residency_requirement(self.config.enforce_residency.value());
let user_agent_suffix = format!("{name}; {version}");
if let Ok(mut suffix) = USER_AGENT_SUFFIX.lock() {
codex-rs/app-server/tests/suite/v2/analytics.rs
+93
View File
@@ -1,10 +1,23 @@
use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::DEFAULT_CLIENT_NAME;
use app_test_support::write_chatgpt_auth;
use codex_core::auth::AuthCredentialsStoreMode;
use codex_core::config::ConfigBuilder;
use codex_core::config::types::OtelExporterKind;
use codex_core::config::types::OtelHttpProtocol;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::collections::HashMap;
use std::path::Path;
use std::time::Duration;
use tempfile::TempDir;
use tokio::time::timeout;
use wiremock::Mock;
use wiremock::MockServer;
use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
const SERVICE_VERSION: &str = "0.0.0-test";
@@ -65,3 +78,83 @@ async fn app_server_default_analytics_enabled_with_flag() -> Result<()> {
assert_eq!(has_metrics, true);
Ok(())
}
pub(crate) async fn enable_analytics_capture(server: &MockServer, codex_home: &Path) -> Result<()> {
Mock::given(method("POST"))
.and(path("/codex/analytics-events/events"))
.respond_with(ResponseTemplate::new(200))
.mount(server)
.await;
write_chatgpt_auth(
codex_home,
ChatGptAuthFixture::new("chatgpt-token")
.account_id("account-123")
.chatgpt_user_id("user-123")
.chatgpt_account_id("account-123"),
AuthCredentialsStoreMode::File,
)?;
Ok(())
}
pub(crate) async fn wait_for_analytics_payload(
server: &MockServer,
read_timeout: Duration,
) -> Result<Value> {
let body = timeout(read_timeout, async {
loop {
let Some(requests) = server.received_requests().await else {
tokio::time::sleep(Duration::from_millis(25)).await;
continue;
};
if let Some(request) = requests.iter().find(|request| {
request.method == "POST" && request.url.path() == "/codex/analytics-events/events"
}) {
break request.body.clone();
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await?;
serde_json::from_slice(&body).map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"))
}
pub(crate) fn thread_initialized_event(payload: &Value) -> Result<&Value> {
let events = payload["events"]
.as_array()
.ok_or_else(|| anyhow::anyhow!("analytics payload missing events array"))?;
events
.iter()
.find(|event| event["event_type"] == "codex_thread_initialized")
.ok_or_else(|| anyhow::anyhow!("codex_thread_initialized event should be present"))
}
pub(crate) fn assert_basic_thread_initialized_event(
event: &Value,
thread_id: &str,
expected_model: &str,
initialization_mode: &str,
) {
assert_eq!(event["event_params"]["thread_id"], thread_id);
assert_eq!(
event["event_params"]["product_client_id"],
DEFAULT_CLIENT_NAME
);
assert_eq!(event["event_params"]["model"], expected_model);
assert_eq!(event["event_params"]["ephemeral"], false);
assert_eq!(event["event_params"]["session_source"], "user");
assert_eq!(
event["event_params"]["subagent_source"],
serde_json::Value::Null
);
assert_eq!(
event["event_params"]["parent_thread_id"],
serde_json::Value::Null
);
assert_eq!(
event["event_params"]["initialization_mode"],
initialization_mode
);
assert!(event["event_params"]["created_at"].as_u64().is_some());
}
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+44
View File
@@ -38,6 +38,11 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
@@ -177,6 +182,45 @@ async fn thread_fork_creates_new_thread_and_emits_started() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
enable_analytics_capture(&server, codex_home.path()).await?;
let conversation_id = create_fake_rollout(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Some("mock_provider"),
None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let fork_id = mcp
.send_thread_fork_request(ThreadForkParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let fork_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(fork_id)),
)
.await??;
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "forked");
Ok(())
}
#[tokio::test]
async fn thread_fork_rejects_unmaterialized_thread() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+45
View File
@@ -70,6 +70,11 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const CODEX_5_2_INSTRUCTIONS_TEMPLATE_DEFAULT: &str = "You are Codex, a coding agent based on GPT-5. You and the user share the same workspace and collaborate to achieve the user's goals.";
@@ -150,6 +155,46 @@ async fn thread_resume_rejects_unmaterialized_thread() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
enable_analytics_capture(&server, codex_home.path()).await?;
let conversation_id = create_fake_rollout_with_text_elements(
codex_home.path(),
"2025-01-05T12-00-00",
"2025-01-05T12:00:00Z",
"Saved user message",
Vec::new(),
Some("mock_provider"),
None,
)?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let resume_id = mcp
.send_thread_resume_request(ThreadResumeParams {
thread_id: conversation_id,
..Default::default()
})
.await?;
let resume_resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(resume_id)),
)
.await??;
let ThreadResumeResponse { thread, .. } = to_response::<ThreadResumeResponse>(resume_resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(event, &thread.id, "gpt-5.2-codex", "resumed");
Ok(())
}
#[tokio::test]
async fn thread_resume_returns_rollout_history() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
codex-rs/app-server/tests/suite/v2/thread_start.rs
+33
View File
@@ -34,6 +34,11 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::enable_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
#[tokio::test]
@@ -153,6 +158,34 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
let codex_home = TempDir::new()?;
create_config_toml_with_chatgpt_base_url(codex_home.path(), &server.uri(), &server.uri())?;
enable_analytics_capture(&server, codex_home.path()).await?;
let mut mcp = McpProcess::new(codex_home.path()).await?;
timeout(DEFAULT_READ_TIMEOUT, mcp.initialize()).await??;
let req_id = mcp
.send_thread_start_request(ThreadStartParams::default())
.await?;
let resp: JSONRPCResponse = timeout(
DEFAULT_READ_TIMEOUT,
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
)
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
assert_eq!(payload["events"].as_array().expect("events array").len(), 1);
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(event, &thread.id, "mock-model", "new");
Ok(())
}
#[tokio::test]
async fn thread_start_respects_project_config_from_cwd() -> Result<()> {
let server = create_mock_responses_server_repeating_assistant("Done").await;
codex-rs/core/src/codex.rs
-59
View File
@@ -5,8 +5,6 @@ use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::time::SystemTime;
use std::time::UNIX_EPOCH;
use crate::AuthManager;
use crate::CodexAuth;
@@ -351,8 +349,6 @@ use crate::util::backoff;
use crate::windows_sandbox::WindowsSandboxLevelExt;
use codex_analytics::AnalyticsEventsClient;
use codex_analytics::AppInvocation;
use codex_analytics::CodexThreadContext;
use codex_analytics::CodexThreadInitializedInput;
use codex_analytics::CodexTurnEvent;
use codex_analytics::InitializationMode;
use codex_analytics::InvocationType;
@@ -639,12 +635,6 @@ impl Codex {
user_shell_override,
};
let initialization_mode = initialization_mode(&conversation_history);
let thread_session_source = session_configuration.session_source.clone();
let subagent_source = session_source_subagent_source(&thread_session_source);
let parent_thread_id = session_source_parent_thread_id(&thread_session_source);
let thread_initialized_configuration = session_configuration.clone();
// Generate a unique ID for the lifetime of this Codex session.
let session_source_clone = session_configuration.session_source.clone();
let (agent_status_tx, agent_status_rx) = watch::channel(AgentStatus::PendingInit);
@@ -672,31 +662,6 @@ impl Codex {
})?;
let thread_id = session.conversation_id;
session
.services
.analytics_events_client
.track_thread_initialized(CodexThreadInitializedInput {
thread_id: thread_id.to_string(),
model: thread_initialized_configuration
.collaboration_mode
.model()
.to_string(),
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,
parent_thread_id,
},
});
// This task will run until Op::Shutdown is received.
let session_for_loop = Arc::clone(&session);
let session_loop_handle = tokio::spawn(async move {
@@ -811,30 +776,6 @@ impl Codex {
}
}
fn initialization_mode(conversation_history: &InitialHistory) -> InitializationMode {
match conversation_history {
InitialHistory::New => InitializationMode::New,
InitialHistory::Forked(_) => InitializationMode::Forked,
InitialHistory::Resumed(_) => InitializationMode::Resumed,
}
}
fn session_source_subagent_source(session_source: &SessionSource) -> Option<SubAgentSource> {
match session_source {
SessionSource::SubAgent(subagent_source) => Some(subagent_source.clone()),
_ => None,
}
}
fn session_source_parent_thread_id(session_source: &SessionSource) -> Option<String> {
match session_source {
SessionSource::SubAgent(SubAgentSource::ThreadSpawn {
parent_thread_id, ..
}) => Some(parent_thread_id.to_string()),
_ => None,
}
}
#[cfg(test)]
pub(crate) fn completed_session_loop_termination() -> SessionLoopTermination {
futures::future::ready(()).boxed().shared()
codex-rs/core/src/lib.rs
+4
View File
@@ -136,6 +136,10 @@ pub type CodexConversation = CodexThread;
pub use auth::AuthManager;
pub use auth::CodexAuth;
pub use codex_analytics::AnalyticsEventsClient;
pub use codex_analytics::AnalyticsFact;
pub use codex_analytics::CustomAnalyticsFact;
pub use codex_analytics::InitializationMode;
pub use codex_analytics::ThreadInitializeInput;
mod default_client_forwarding;
/// Default Codex HTTP client headers and reqwest construction.
codex-rs/core/tests/suite/analytics.rs
-78
View File
@@ -1,78 +0,0 @@
use anyhow::Result;
use codex_core::CodexAuth;
use codex_core::config::Constrained;
use core_test_support::responses::start_mock_server;
use core_test_support::test_codex::test_codex;
use std::time::Duration;
use std::time::Instant;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn thread_initialization_tracks_thread_initialized_analytics() -> Result<()> {
let server = start_mock_server().await;
let chatgpt_base_url = server.uri();
let test = 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.permissions.sandbox_policy = Constrained::allow_any(
codex_protocol::protocol::SandboxPolicy::new_workspace_write_policy(),
);
config.ephemeral = true;
})
.build(&server)
.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| request.url.path() == "/codex/analytics-events/events")
{
break request;
}
if Instant::now() >= deadline {
panic!("timed out waiting for 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")
.expect("codex_thread_initialized event should be present");
assert_eq!(
event["event_params"]["thread_id"],
serde_json::json!(test.session_configured.session_id.to_string())
);
assert_eq!(
event["event_params"]["product_client_id"],
serde_json::json!(codex_core::default_client::originator().value)
);
assert_eq!(event["event_params"]["model"], "gpt-5");
assert_eq!(event["event_params"]["ephemeral"], true);
assert_eq!(event["event_params"]["session_source"], "user");
assert_eq!(event["event_params"]["initialization_mode"], "new");
assert_eq!(
event["event_params"]["subagent_source"],
serde_json::Value::Null
);
assert_eq!(
event["event_params"]["parent_thread_id"],
serde_json::Value::Null
);
assert!(
event["event_params"]["created_at"]
.as_u64()
.is_some_and(|timestamp| timestamp > 0)
);
Ok(())
}
codex-rs/core/tests/suite/mod.rs
-1
View File
@@ -58,7 +58,6 @@ pub static CODEX_ALIASES_TEMP_DIR: TestCodexAliasesGuard = unsafe {
mod abort_tasks;
mod agent_jobs;
mod agent_websocket;
mod analytics;
mod apply_patch_cli;
#[cfg(not(target_os = "windows"))]
mod approvals;
codex-rs/exec/src/lib.rs
+1 -1
View File
@@ -427,7 +427,7 @@ pub async fn run_main(cli: Cli, arg0_paths: Arg0DispatchPaths) -> anyhow::Result
config_warnings,
session_source: SessionSource::Exec,
enable_codex_api_key_env: true,
client_name: "codex-exec".to_string(),
client_name: "codex_exec".to_string(),
client_version: env!("CARGO_PKG_VERSION").to_string(),
experimental_api: true,
opt_out_notification_methods: Vec::new(),