Track app server startup analytics by transport

This commit is contained in:
Adam Perry
2026-06-01 19:29:36 +00:00
parent 53ac02356e
commit 8c3b21f23d
13 changed files with 225 additions and 21 deletions
+16
View File
@@ -8,6 +8,7 @@ use crate::facts::AnalyticsFact;
use crate::facts::AnalyticsJsonRpcError;
use crate::facts::AppInvocation;
use crate::facts::AppMentionedInput;
use crate::facts::AppServerStartedInput;
use crate::facts::AppUsedInput;
use crate::facts::CustomAnalyticsFact;
use crate::facts::HookRunFact;
@@ -169,6 +170,21 @@ impl AnalyticsEventsClient {
));
}
pub fn track_app_server_started(
&self,
rpc_transport: AppServerRpcTransport,
duration: Duration,
) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::AppServerStarted(AppServerStartedInput {
runtime: current_runtime_metadata(),
rpc_transport,
duration_ms: u64::try_from(duration.as_millis()).unwrap_or(u64::MAX),
created_at: crate::now_unix_seconds(),
}),
));
}
pub fn track_guardian_review(
&self,
tracking: &GuardianReviewTrackContext,
+19
View File
@@ -56,6 +56,7 @@ pub(crate) struct TrackEventsRequest {
#[serde(untagged)]
pub(crate) enum TrackEventRequest {
SkillInvocation(SkillInvocationEventRequest),
AppServerStarted(CodexAppServerStartedEventRequest),
ThreadInitialized(ThreadInitializedEvent),
GuardianReview(Box<GuardianReviewEventRequest>),
AppMentioned(CodexAppMentionedEventRequest),
@@ -144,6 +145,24 @@ pub(crate) struct CodexRuntimeMetadata {
pub(crate) runtime_arch: String,
}
/// Analytics parameters emitted when an app-server runtime starts.
#[derive(Serialize)]
pub(crate) struct CodexAppServerStartedEventParams {
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) rpc_transport: AppServerRpcTransport,
/// Elapsed measured startup duration, in milliseconds from a monotonic clock.
pub(crate) duration_ms: u64,
/// Time at which the event was recorded, in seconds since the Unix epoch.
pub(crate) created_at: u64,
}
/// Analytics track-event envelope for an app-server startup event.
#[derive(Serialize)]
pub(crate) struct CodexAppServerStartedEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexAppServerStartedEventParams,
}
#[derive(Serialize)]
pub(crate) struct ThreadInitializedEventParams {
pub(crate) thread_id: String,
+11
View File
@@ -324,6 +324,7 @@ pub(crate) enum AnalyticsFact {
}
pub(crate) enum CustomAnalyticsFact {
AppServerStarted(AppServerStartedInput),
SubAgentThreadStarted(SubAgentThreadStartedInput),
Compaction(Box<CodexCompactionEvent>),
GuardianReview(Box<GuardianReviewEventParams>),
@@ -337,6 +338,16 @@ pub(crate) enum CustomAnalyticsFact {
PluginStateChanged(PluginStateChangedInput),
}
/// Analytics input captured when an app-server runtime starts.
pub(crate) struct AppServerStartedInput {
pub runtime: CodexRuntimeMetadata,
pub rpc_transport: AppServerRpcTransport,
/// Elapsed measured startup duration, in milliseconds from a monotonic clock.
pub duration_ms: u64,
/// Time at which the event was recorded, in seconds since the Unix epoch.
pub created_at: u64,
}
pub(crate) struct SkillInvokedInput {
pub tracking: TrackEventsContext,
pub invocations: Vec<SkillInvocation>,
+24
View File
@@ -5,6 +5,8 @@ use crate::accepted_lines::accepted_line_repo_hash_for_cwd;
use crate::events::AppServerRpcTransport;
use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppServerStartedEventParams;
use crate::events::CodexAppServerStartedEventRequest;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexCollabAgentToolCallEventParams;
use crate::events::CodexCollabAgentToolCallEventRequest;
@@ -60,6 +62,7 @@ use crate::events::subagent_thread_started_event_request;
use crate::facts::AnalyticsFact;
use crate::facts::AnalyticsJsonRpcError;
use crate::facts::AppMentionedInput;
use crate::facts::AppServerStartedInput;
use crate::facts::AppUsedInput;
use crate::facts::CodexCompactionEvent;
use crate::facts::CustomAnalyticsFact;
@@ -446,6 +449,9 @@ impl AnalyticsReducer {
self.ingest_server_request_aborted(completed_at_ms, request_id, out);
}
AnalyticsFact::Custom(input) => match input {
CustomAnalyticsFact::AppServerStarted(input) => {
self.ingest_app_server_started(input, out);
}
CustomAnalyticsFact::SubAgentThreadStarted(input) => {
self.ingest_subagent_thread_started(input, out);
}
@@ -483,6 +489,24 @@ impl AnalyticsReducer {
}
}
fn ingest_app_server_started(
&mut self,
input: AppServerStartedInput,
out: &mut Vec<TrackEventRequest>,
) {
out.push(TrackEventRequest::AppServerStarted(
CodexAppServerStartedEventRequest {
event_type: "codex_app_server_started",
event_params: CodexAppServerStartedEventParams {
runtime: input.runtime,
rpc_transport: input.rpc_transport,
duration_ms: input.duration_ms,
created_at: input.created_at,
},
},
));
}
fn ingest_initialize(
&mut self,
connection_id: u64,
+4
View File
@@ -49,6 +49,7 @@ use std::sync::RwLock;
use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use std::time::Duration;
use std::time::Instant;
use crate::analytics_utils::analytics_events_client_from_config;
use crate::config_manager::ConfigManager;
@@ -370,6 +371,7 @@ pub async fn start(args: InProcessStartArgs) -> IoResult<InProcessClientHandle>
}
async fn start_uninitialized(args: InProcessStartArgs) -> IoResult<InProcessClientHandle> {
let startup_started = Instant::now();
let channel_capacity = args.channel_capacity.max(1);
let installation_id = resolve_installation_id(&args.config.codex_home).await?;
let (client_tx, mut client_rx) = mpsc::channel::<InProcessClientMessage>(channel_capacity);
@@ -420,6 +422,8 @@ async fn start_uninitialized(args: InProcessStartArgs) -> IoResult<InProcessClie
args.thread_config_loader,
);
let (processor_tx, mut processor_rx) = mpsc::channel::<ProcessorCommand>(channel_capacity);
analytics_events_client
.track_app_server_started(AppServerRpcTransport::InProcess, startup_started.elapsed());
let mut processor_handle = tokio::spawn(async move {
let processor = Arc::new(MessageProcessor::new(MessageProcessorArgs {
outgoing: Arc::clone(&processor_outgoing),
+5 -1
View File
@@ -17,6 +17,7 @@ use std::io::Result as IoResult;
use std::sync::Arc;
use std::sync::RwLock;
use std::sync::atomic::AtomicBool;
use std::time::Instant;
use crate::analytics_utils::analytics_events_client_from_config;
use crate::config_manager::ConfigManager;
@@ -428,6 +429,7 @@ pub async fn run_main_with_transport_options(
auth: AppServerWebsocketAuthSettings,
runtime_options: AppServerRuntimeOptions,
) -> IoResult<()> {
let startup_started = Instant::now();
let (transport_event_tx, mut transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<OutgoingEnvelope>(CHANNEL_CAPACITY);
@@ -789,12 +791,14 @@ pub async fn run_main_with_transport_options(
let auth_manager = Arc::clone(&auth_manager);
let analytics_events_client =
analytics_events_client_from_config(Arc::clone(&auth_manager), &config);
let rpc_transport = analytics_rpc_transport(&transport);
let outgoing_message_sender = Arc::new(OutgoingMessageSender::new(
outgoing_tx,
analytics_events_client.clone(),
));
let initialize_notification_sender = outgoing_message_sender.clone();
let outbound_control_tx = outbound_control_tx;
analytics_events_client.track_app_server_started(rpc_transport, startup_started.elapsed());
let processor = Arc::new(MessageProcessor::new(MessageProcessorArgs {
outgoing: outgoing_message_sender,
analytics_events_client,
@@ -809,7 +813,7 @@ pub async fn run_main_with_transport_options(
session_source,
auth_manager,
installation_id,
rpc_transport: analytics_rpc_transport(&transport),
rpc_transport,
remote_control_handle: Some(remote_control_handle.clone()),
plugin_startup_tasks: runtime_options.plugin_startup_tasks,
}));
+112 -8
View File
@@ -1,15 +1,28 @@
use anyhow::Result;
use app_test_support::ChatGptAuthFixture;
use app_test_support::DEFAULT_CLIENT_NAME;
use app_test_support::McpProcess;
use app_test_support::write_chatgpt_auth;
use app_test_support::write_mock_responses_config_toml_with_chatgpt_base_url;
use codex_app_server::in_process;
use codex_app_server::in_process::InProcessStartArgs;
use codex_app_server_protocol::ClientInfo;
use codex_app_server_protocol::InitializeParams;
use codex_arg0::Arg0DispatchPaths;
use codex_config::CloudRequirementsLoader;
use codex_config::LoaderOverrides;
use codex_config::types::AuthCredentialsStoreMode;
use codex_config::types::OtelExporterKind;
use codex_config::types::OtelHttpProtocol;
use codex_core::config::ConfigBuilder;
use codex_exec_server::EnvironmentManager;
use codex_feedback::CodexFeedback;
use codex_protocol::protocol::SessionSource;
use pretty_assertions::assert_eq;
use serde_json::Value;
use std::collections::HashMap;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;
use tempfile::TempDir;
use tokio::time::timeout;
@@ -79,6 +92,91 @@ async fn app_server_default_analytics_enabled_with_flag() -> Result<()> {
Ok(())
}
#[tokio::test]
async fn standalone_app_server_startup_tracks_analytics_event() -> Result<()> {
let server = MockServer::start().await;
let codex_home = TempDir::new()?;
write_mock_responses_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
)?;
mount_analytics_capture(&server, codex_home.path()).await?;
let _mcp = McpProcess::new_without_managed_config(codex_home.path()).await?;
let event =
wait_for_analytics_event(&server, Duration::from_secs(10), "codex_app_server_started")
.await?;
assert_app_server_started_event(&event, "stdio");
Ok(())
}
#[tokio::test]
async fn embedded_app_server_startup_tracks_analytics_event() -> Result<()> {
let server = MockServer::start().await;
let codex_home = TempDir::new()?;
write_mock_responses_config_toml_with_chatgpt_base_url(
codex_home.path(),
&server.uri(),
&server.uri(),
)?;
mount_analytics_capture(&server, codex_home.path()).await?;
let loader_overrides = LoaderOverrides::without_managed_config_for_tests();
let config = ConfigBuilder::default()
.codex_home(codex_home.path().to_path_buf())
.fallback_cwd(Some(codex_home.path().to_path_buf()))
.loader_overrides(loader_overrides.clone())
.build()
.await?;
let client = in_process::start(InProcessStartArgs {
arg0_paths: Arg0DispatchPaths::default(),
config: Arc::new(config),
cli_overrides: Vec::new(),
loader_overrides,
strict_config: false,
cloud_requirements: CloudRequirementsLoader::default(),
thread_config_loader: Arc::new(codex_config::NoopThreadConfigLoader),
feedback: CodexFeedback::new(),
log_db: None,
state_db: None,
environment_manager: Arc::new(EnvironmentManager::default_for_tests()),
config_warnings: Vec::new(),
session_source: SessionSource::Cli,
enable_codex_api_key_env: false,
initialize: InitializeParams {
client_info: ClientInfo {
name: "codex-tui".to_string(),
title: None,
version: "0.1.0".to_string(),
},
capabilities: None,
},
channel_capacity: in_process::DEFAULT_IN_PROCESS_CHANNEL_CAPACITY,
})
.await?;
let event =
wait_for_analytics_event(&server, Duration::from_secs(10), "codex_app_server_started")
.await?;
assert_app_server_started_event(&event, "in_process");
client.shutdown().await?;
Ok(())
}
fn assert_app_server_started_event(event: &Value, rpc_transport: &str) {
assert_eq!(event["event_params"]["rpc_transport"], rpc_transport);
assert!(event["event_params"]["duration_ms"].as_u64().is_some());
assert!(event["event_params"]["created_at"].as_u64().is_some());
assert!(
event["event_params"]["runtime"]["codex_rs_version"]
.as_str()
.is_some()
);
}
pub(crate) async fn mount_analytics_capture(server: &MockServer, codex_home: &Path) -> Result<()> {
Mock::given(method("POST"))
.and(path("/codex/analytics-events/events"))
@@ -98,26 +196,32 @@ pub(crate) async fn mount_analytics_capture(server: &MockServer, codex_home: &Pa
Ok(())
}
pub(crate) async fn wait_for_analytics_payload(
pub(crate) async fn wait_for_thread_initialized_payload(
server: &MockServer,
read_timeout: Duration,
) -> Result<Value> {
let body = timeout(read_timeout, async {
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();
for request in &requests {
if request.method != "POST"
|| request.url.path() != "/codex/analytics-events/events"
{
continue;
}
let payload: Value = serde_json::from_slice(&request.body)
.map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"))?;
if thread_initialized_event(&payload).is_ok() {
return Ok::<Value, anyhow::Error>(payload);
}
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await?;
serde_json::from_slice(&body).map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"))
.await?
}
pub(crate) async fn wait_for_analytics_event(
@@ -1352,15 +1352,22 @@ async fn wait_for_plugin_analytics_payload(server: &MockServer) -> Result<serde_
tokio::time::sleep(Duration::from_millis(25)).await;
continue;
};
if let Some(request) = requests.iter().find(|request| {
for request in requests.iter().filter(|request| {
request.method == "POST"
&& request
.url
.path()
.ends_with("/codex/analytics-events/events")
}) {
return serde_json::from_slice(&request.body)
.map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"));
let payload: serde_json::Value = serde_json::from_slice(&request.body)
.map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"))?;
if payload["events"].as_array().is_some_and(|events| {
events
.iter()
.any(|event| event["event_type"] == "codex_plugin_installed")
}) {
return Ok::<serde_json::Value, anyhow::Error>(payload);
}
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
@@ -122,16 +122,23 @@ async fn plugin_uninstall_tracks_analytics_event() -> Result<()> {
tokio::time::sleep(Duration::from_millis(25)).await;
continue;
};
if let Some(request) = requests.iter().find(|request| {
for request in requests.iter().filter(|request| {
request.method == "POST" && request.url.path() == "/codex/analytics-events/events"
}) {
break request.body.clone();
let payload: serde_json::Value =
serde_json::from_slice(&request.body).expect("analytics payload");
if payload["events"].as_array().is_some_and(|events| {
events
.iter()
.any(|event| event["event_type"] == "codex_plugin_uninstalled")
}) {
return payload;
}
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await?;
let payload: serde_json::Value = serde_json::from_slice(&payload).expect("analytics payload");
assert_eq!(
payload,
json!({
@@ -139,6 +139,14 @@ impl BlockingRemoteControlBackend {
&remote_control_url,
&remote_control_url,
)?;
// This fixture implements remote-control enrollment only; prevent the
// startup event from becoming the first request to its blocking server.
let config_path = codex_home.join("config.toml");
let config = std::fs::read_to_string(&config_path)?;
std::fs::write(
config_path,
format!("{config}\n[analytics]\nenabled = false\n"),
)?;
write_chatgpt_auth(
codex_home,
ChatGptAuthFixture::new("chatgpt-token")
@@ -44,7 +44,7 @@ use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::mount_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
use super::analytics::wait_for_thread_initialized_payload;
#[cfg(windows)]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
@@ -403,7 +403,7 @@ async fn thread_fork_tracks_thread_initialized_analytics() -> Result<()> {
.await??;
let ThreadForkResponse { thread, .. } = to_response::<ThreadForkResponse>(fork_resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let payload = wait_for_thread_initialized_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(
event,
@@ -98,7 +98,7 @@ use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::mount_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
use super::analytics::wait_for_thread_initialized_payload;
#[cfg(windows)]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
@@ -424,7 +424,7 @@ async fn thread_resume_tracks_thread_initialized_analytics() -> Result<()> {
);
assert_eq!(thread.thread_source, Some(ThreadSource::User));
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let payload = wait_for_thread_initialized_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let event = thread_initialized_event(&payload)?;
assert_basic_thread_initialized_event(
event,
@@ -47,7 +47,7 @@ use wiremock::matchers::path;
use super::analytics::assert_basic_thread_initialized_event;
use super::analytics::mount_analytics_capture;
use super::analytics::thread_initialized_event;
use super::analytics::wait_for_analytics_payload;
use super::analytics::wait_for_thread_initialized_payload;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
@@ -436,7 +436,7 @@ async fn thread_start_tracks_thread_initialized_analytics() -> Result<()> {
.await??;
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(resp)?;
let payload = wait_for_analytics_payload(&server, DEFAULT_READ_TIMEOUT).await?;
let payload = wait_for_thread_initialized_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(