openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-08 20:40:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

[codex-analytics] feature plumbing and emittance

Browse Source
This commit is contained in:
rhan-oai
2026-04-10 13:37:25 -07:00
parent 5779be314a
commit bc876c3c10
28 changed files with 1462 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/bespoke_event_handling.rs
+33
View File
@@ -12,6 +12,7 @@ use crate::thread_state::TurnSummary;
use crate::thread_state::resolve_server_request_on_thread_listener;
use crate::thread_status::ThreadWatchActiveGuard;
use crate::thread_status::ThreadWatchManager;
use codex_analytics::AnalyticsEventsClient;
use codex_app_server_protocol::AccountRateLimitsUpdatedNotification;
use codex_app_server_protocol::AddCreditsNudgeEmailNotification;
use codex_app_server_protocol::AddCreditsNudgeEmailResult;
@@ -170,6 +171,7 @@ pub(crate) async fn apply_bespoke_event_handling(
conversation_id: ThreadId,
conversation: Arc<CodexThread>,
thread_manager: Arc<ThreadManager>,
analytics_events_client: Option<AnalyticsEventsClient>,
outgoing: ThreadScopedOutgoingMessageSender,
thread_state: Arc<tokio::sync::Mutex<ThreadState>>,
thread_watch_manager: ThreadWatchManager,
@@ -205,6 +207,10 @@ pub(crate) async fn apply_bespoke_event_handling(
thread_id: conversation_id.to_string(),
turn,
};
if let Some(analytics_events_client) = analytics_events_client.as_ref() {
analytics_events_client
.track_notification(ServerNotification::TurnStarted(notification.clone()));
}
outgoing
.send_server_notification(ServerNotification::TurnStarted(notification))
.await;
@@ -221,6 +227,7 @@ pub(crate) async fn apply_bespoke_event_handling(
conversation_id,
event_turn_id,
turn_complete_event,
analytics_events_client.as_ref(),
&outgoing,
&thread_state,
)
@@ -1756,6 +1763,7 @@ pub(crate) async fn apply_bespoke_event_handling(
conversation_id,
event_turn_id,
turn_aborted_event,
analytics_events_client.as_ref(),
&outgoing,
&thread_state,
)
@@ -1933,6 +1941,7 @@ async fn emit_turn_completed_with_status(
conversation_id: ThreadId,
event_turn_id: String,
turn_completion_metadata: TurnCompletionMetadata,
analytics_events_client: Option<&AnalyticsEventsClient>,
outgoing: &ThreadScopedOutgoingMessageSender,
) {
let notification = TurnCompletedNotification {
@@ -1947,6 +1956,10 @@ async fn emit_turn_completed_with_status(
duration_ms: turn_completion_metadata.duration_ms,
},
};
if let Some(analytics_events_client) = analytics_events_client {
analytics_events_client
.track_notification(ServerNotification::TurnCompleted(notification.clone()));
}
outgoing
.send_server_notification(ServerNotification::TurnCompleted(notification))
.await;
@@ -2139,6 +2152,7 @@ async fn handle_turn_complete(
conversation_id: ThreadId,
event_turn_id: String,
turn_complete_event: TurnCompleteEvent,
analytics_events_client: Option<&AnalyticsEventsClient>,
outgoing: &ThreadScopedOutgoingMessageSender,
thread_state: &Arc<Mutex<ThreadState>>,
) {
@@ -2159,6 +2173,7 @@ async fn handle_turn_complete(
completed_at: turn_complete_event.completed_at,
duration_ms: turn_complete_event.duration_ms,
},
analytics_events_client,
outgoing,
)
.await;
@@ -2168,6 +2183,7 @@ async fn handle_turn_interrupted(
conversation_id: ThreadId,
event_turn_id: String,
turn_aborted_event: TurnAbortedEvent,
analytics_events_client: Option<&AnalyticsEventsClient>,
outgoing: &ThreadScopedOutgoingMessageSender,
thread_state: &Arc<Mutex<ThreadState>>,
) {
@@ -2183,6 +2199,7 @@ async fn handle_turn_interrupted(
completed_at: turn_aborted_event.completed_at,
duration_ms: turn_aborted_event.duration_ms,
},
analytics_events_client,
outgoing,
)
.await;
@@ -2922,6 +2939,7 @@ mod tests {
use codex_app_server_protocol::GuardianApprovalReviewStatus;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::TurnPlanStepStatus;
use codex_login::AuthManager;
use codex_login::CodexAuth;
use codex_protocol::items::HookPromptFragment;
use codex_protocol::items::build_hook_prompt_message;
@@ -3043,6 +3061,7 @@ mod tests {
outgoing: ThreadScopedOutgoingMessageSender,
thread_state: Arc<Mutex<ThreadState>>,
thread_watch_manager: ThreadWatchManager,
analytics_events_client: AnalyticsEventsClient,
codex_home: PathBuf,
}
@@ -3057,6 +3076,7 @@ mod tests {
self.conversation_id,
self.conversation.clone(),
self.thread_manager.clone(),
Some(self.analytics_events_client.clone()),
self.outgoing.clone(),
self.thread_state.clone(),
self.thread_watch_manager.clone(),
@@ -3368,6 +3388,13 @@ mod tests {
outgoing: outgoing.clone(),
thread_state: thread_state.clone(),
thread_watch_manager: thread_watch_manager.clone(),
analytics_events_client: AnalyticsEventsClient::new(
AuthManager::from_auth_for_testing(
CodexAuth::create_dummy_chatgpt_auth_for_testing(),
),
"http://localhost".to_string(),
Some(false),
),
codex_home: codex_home.path().to_path_buf(),
};
@@ -3776,6 +3803,7 @@ mod tests {
conversation_id,
event_turn_id.clone(),
turn_complete_event(&event_turn_id),
/*analytics_events_client*/ None,
&outgoing,
&thread_state,
)
@@ -3824,6 +3852,7 @@ mod tests {
conversation_id,
event_turn_id.clone(),
turn_aborted_event(&event_turn_id),
/*analytics_events_client*/ None,
&outgoing,
&thread_state,
)
@@ -3871,6 +3900,7 @@ mod tests {
conversation_id,
event_turn_id.clone(),
turn_complete_event(&event_turn_id),
/*analytics_events_client*/ None,
&outgoing,
&thread_state,
)
@@ -4138,6 +4168,7 @@ mod tests {
conversation_a,
a_turn1.clone(),
turn_complete_event(&a_turn1),
/*analytics_events_client*/ None,
&outgoing,
&thread_state,
)
@@ -4159,6 +4190,7 @@ mod tests {
conversation_b,
b_turn1.clone(),
turn_complete_event(&b_turn1),
/*analytics_events_client*/ None,
&outgoing,
&thread_state,
)
@@ -4170,6 +4202,7 @@ mod tests {
conversation_a,
a_turn2.clone(),
turn_complete_event(&a_turn2),
/*analytics_events_client*/ None,
&outgoing,
&thread_state,
)
codex-rs/app-server/src/codex_message_processor.rs
+13
View File
@@ -6836,6 +6836,15 @@ impl CodexMessageProcessor {
};
let response = TurnStartResponse { turn };
if self.config.features.enabled(Feature::GeneralAnalytics) {
self.analytics_events_client.track_response(
request_id.connection_id.0,
ClientResponse::TurnStart {
request_id: request_id.request_id.clone(),
response: response.clone(),
},
);
}
self.outgoing.send_response(request_id, response).await;
}
Err(err) => {
@@ -7647,6 +7656,9 @@ impl CodexMessageProcessor {
conversation_id,
conversation.clone(),
thread_manager.clone(),
listener_task_context
.general_analytics_enabled
.then(|| listener_task_context.analytics_events_client.clone()),
thread_outgoing,
thread_state.clone(),
thread_watch_manager.clone(),
@@ -9514,6 +9526,7 @@ mod tests {
reasoning_effort: None,
personality: None,
session_source: SessionSource::Cli,
initialization_mode: codex_analytics::ThreadInitializationMode::New,
};
assert_eq!(
codex-rs/app-server/src/main.rs
+19 -4
View File
@@ -12,6 +12,7 @@ use std::path::PathBuf;
// Debug-only test hook: lets integration tests point the server at a temporary
// managed config file without writing to /etc.
const MANAGED_CONFIG_PATH_ENV_VAR: &str = "CODEX_APP_SERVER_MANAGED_CONFIG_PATH";
const DISABLE_MANAGED_CONFIG_ENV_VAR: &str = "CODEX_APP_SERVER_DISABLE_MANAGED_CONFIG";
#[derive(Debug, Parser)]
struct AppServerArgs {
@@ -40,10 +41,13 @@ struct AppServerArgs {
fn main() -> anyhow::Result<()> {
arg0_dispatch_or_else(|arg0_paths: Arg0DispatchPaths| async move {
let args = AppServerArgs::parse();
let managed_config_path = managed_config_path_from_debug_env();
let loader_overrides = LoaderOverrides {
managed_config_path,
..Default::default()
let loader_overrides = if disable_managed_config_from_debug_env() {
LoaderOverrides::without_managed_config_for_tests()
} else {
LoaderOverrides {
managed_config_path: managed_config_path_from_debug_env(),
..Default::default()
}
};
let transport = args.listen;
let session_source = args.session_source;
@@ -63,6 +67,17 @@ fn main() -> anyhow::Result<()> {
})
}
fn disable_managed_config_from_debug_env() -> bool {
#[cfg(debug_assertions)]
{
if let Ok(value) = std::env::var(DISABLE_MANAGED_CONFIG_ENV_VAR) {
return matches!(value.as_str(), "1" | "true" | "TRUE" | "yes" | "YES");
}
}
false
}
fn managed_config_path_from_debug_env() -> Option<PathBuf> {
#[cfg(debug_assertions)]
{
codex-rs/app-server/src/message_processor.rs
+9
View File
@@ -677,6 +677,15 @@ impl MessageProcessor {
self.outgoing.send_error(connection_request_id, error).await;
return;
}
if self.config.features.enabled(Feature::GeneralAnalytics)
&& let ClientRequest::TurnStart { request_id, .. } = &codex_request
{
self.analytics_events_client.track_request(
connection_id.0,
request_id.clone(),
codex_request.clone(),
);
}
match codex_request {
ClientRequest::ConfigRead { request_id, params } => {