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

Track app-server and thread start in ChatGPT telemetry

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-05-27 12:11:01 -07:00
parent 2d1ad374a7
commit 2adcc3e32e
18 changed files with 535 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/analytics/src/analytics_client_tests.rs
+156
View File
@@ -4,6 +4,8 @@ use crate::events::CodexAcceptedLineFingerprintsEventParams;
use crate::events::CodexAcceptedLineFingerprintsEventRequest;
use crate::events::CodexAppMentionedEventRequest;
use crate::events::CodexAppServerClientMetadata;
use crate::events::CodexAppServerStartedEventParams;
use crate::events::CodexAppServerStartedEventRequest;
use crate::events::CodexAppUsedEventRequest;
use crate::events::CodexCommandExecutionEventParams;
use crate::events::CodexCommandExecutionEventRequest;
@@ -30,6 +32,7 @@ use crate::events::ReviewTrigger;
use crate::events::Reviewer;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::ThreadStartTimingEventParams;
use crate::events::ToolItemTerminalStatus;
use crate::events::TrackEventRequest;
use crate::events::codex_app_metadata;
@@ -41,6 +44,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::CodexCompactionEvent;
use crate::facts::CompactionImplementation;
@@ -61,6 +65,7 @@ use crate::facts::SkillInvocation;
use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadInitializationMode;
use crate::facts::ThreadStartTimingFact;
use crate::facts::TrackEventsContext;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnStatus;
@@ -1329,6 +1334,12 @@ fn thread_initialized_event_serializes_expected_shape() {
initialization_mode: ThreadInitializationMode::New,
subagent_source: None,
parent_thread_id: None,
thread_start_timing: ThreadStartTimingEventParams {
thread_start_duration_ms: Some(321),
thread_start_prepare_duration_ms: Some(111),
thread_start_spawn_duration_ms: Some(123),
thread_start_finalize_duration_ms: Some(87),
},
created_at: 1,
},
});
@@ -1361,12 +1372,49 @@ fn thread_initialized_event_serializes_expected_shape() {
"initialization_mode": "new",
"subagent_source": null,
"parent_thread_id": null,
"thread_start_duration_ms": 321,
"thread_start_prepare_duration_ms": 111,
"thread_start_spawn_duration_ms": 123,
"thread_start_finalize_duration_ms": 87,
"created_at": 1
}
})
);
}
#[test]
fn app_server_started_event_serializes_expected_shape() {
let event = TrackEventRequest::AppServerStarted(CodexAppServerStartedEventRequest {
event_type: "codex_app_server_started",
event_params: CodexAppServerStartedEventParams {
runtime: sample_runtime_metadata(),
remote_control_enabled: true,
startup_duration_ms: 987,
completed_at: 12,
},
});
let payload = serde_json::to_value(&event).expect("serialize app-server started event");
assert_eq!(
payload,
json!({
"event_type": "codex_app_server_started",
"event_params": {
"runtime": {
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"remote_control_enabled": true,
"startup_duration_ms": 987,
"completed_at": 12
}
})
);
}
#[test]
fn command_execution_event_serializes_expected_shape() {
let event = TrackEventRequest::CommandExecution(CodexCommandExecutionEventRequest {
@@ -1643,6 +1691,114 @@ async fn initialize_caches_client_and_thread_lifecycle_publishes_once_initialize
payload[0]["event_params"]["runtime"]["runtime_arch"],
"x86_64"
);
assert_eq!(
payload[0]["event_params"]["thread_start_duration_ms"],
json!(null)
);
assert_eq!(
payload[0]["event_params"]["thread_start_prepare_duration_ms"],
json!(null)
);
assert_eq!(
payload[0]["event_params"]["thread_start_spawn_duration_ms"],
json!(null)
);
assert_eq!(
payload[0]["event_params"]["thread_start_finalize_duration_ms"],
json!(null)
);
}
#[tokio::test]
async fn app_server_started_fact_emits_event() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::AppServerStarted(
AppServerStartedInput {
runtime: sample_runtime_metadata(),
remote_control_enabled: true,
startup_duration_ms: 456,
completed_at: 12,
},
)),
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(
payload,
json!([{
"event_type": "codex_app_server_started",
"event_params": {
"runtime": {
"codex_rs_version": "0.1.0",
"runtime_os": "macos",
"runtime_os_version": "15.3.1",
"runtime_arch": "aarch64"
},
"remote_control_enabled": true,
"startup_duration_ms": 456,
"completed_at": 12
}
}])
);
}
#[tokio::test]
async fn thread_start_timing_fact_enriches_thread_initialized_event() {
let mut reducer = AnalyticsReducer::default();
let mut events = Vec::new();
ingest_initialize(&mut reducer, &mut events).await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::ThreadStartTiming(
ThreadStartTimingFact {
thread_id: "thread-1".to_string(),
duration_ms: 222,
prepare_duration_ms: 12,
spawn_duration_ms: 123,
finalize_duration_ms: 87,
},
)),
&mut events,
)
.await;
reducer
.ingest(
AnalyticsFact::ClientResponse {
connection_id: 7,
request_id: RequestId::Integer(2),
response: Box::new(sample_thread_start_response(
"thread-1", /*ephemeral*/ true, "gpt-5",
)),
},
&mut events,
)
.await;
let payload = serde_json::to_value(&events).expect("serialize events");
assert_eq!(payload[0]["event_type"], json!("codex_thread_initialized"));
assert_eq!(
payload[0]["event_params"]["thread_start_duration_ms"],
json!(222)
);
assert_eq!(
payload[0]["event_params"]["thread_start_prepare_duration_ms"],
json!(12)
);
assert_eq!(
payload[0]["event_params"]["thread_start_spawn_duration_ms"],
json!(123)
);
assert_eq!(
payload[0]["event_params"]["thread_start_finalize_duration_ms"],
json!(87)
);
}
#[tokio::test]
codex-rs/analytics/src/client.rs
+43
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;
@@ -17,9 +18,11 @@ use crate::facts::PluginStateChangedInput;
use crate::facts::SkillInvocation;
use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadStartTimingFact;
use crate::facts::TrackEventsContext;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnTokenUsageFact;
use crate::now_unix_seconds;
use crate::reducer::AnalyticsReducer;
use codex_app_server_protocol::ClientRequest;
use codex_app_server_protocol::ClientResponsePayload;
@@ -38,12 +41,35 @@ use std::collections::HashSet;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use std::time::Instant;
use tokio::sync::mpsc;
const ANALYTICS_EVENTS_QUEUE_SIZE: usize = 256;
const ANALYTICS_EVENTS_TIMEOUT: Duration = Duration::from_secs(10);
const ANALYTICS_EVENT_DEDUPE_MAX_KEYS: usize = 4096;
#[derive(Clone, Copy, Debug)]
pub struct StartedTimer {
started_at: Instant,
}
impl StartedTimer {
#[must_use]
pub fn start() -> Self {
Self {
started_at: Instant::now(),
}
}
fn elapsed_ms(self) -> u64 {
self.started_at
.elapsed()
.as_millis()
.try_into()
.unwrap_or(u64::MAX)
}
}
#[derive(Clone)]
pub(crate) struct AnalyticsEventsQueue {
pub(crate) sender: mpsc::Sender<AnalyticsFact>,
@@ -163,12 +189,29 @@ impl AnalyticsEventsClient {
});
}
pub fn track_app_server_started(&self, timer: StartedTimer, remote_control_enabled: bool) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::AppServerStarted(AppServerStartedInput {
runtime: current_runtime_metadata(),
remote_control_enabled,
startup_duration_ms: timer.elapsed_ms(),
completed_at: now_unix_seconds(),
}),
));
}
pub fn track_subagent_thread_started(&self, input: SubAgentThreadStartedInput) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::SubAgentThreadStarted(input),
));
}
pub fn track_thread_start_timing(&self, fact: ThreadStartTimingFact) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::ThreadStartTiming(fact),
));
}
pub fn track_guardian_review(
&self,
tracking: &GuardianReviewTrackContext,
codex-rs/analytics/src/events.rs
+26
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,28 @@ pub(crate) struct CodexRuntimeMetadata {
pub(crate) runtime_arch: String,
}
#[derive(Serialize)]
pub(crate) struct CodexAppServerStartedEventParams {
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) remote_control_enabled: bool,
pub(crate) startup_duration_ms: u64,
pub(crate) completed_at: u64,
}
#[derive(Serialize)]
pub(crate) struct CodexAppServerStartedEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: CodexAppServerStartedEventParams,
}
#[derive(Default, Serialize)]
pub(crate) struct ThreadStartTimingEventParams {
pub(crate) thread_start_duration_ms: Option<u64>,
pub(crate) thread_start_prepare_duration_ms: Option<u64>,
pub(crate) thread_start_spawn_duration_ms: Option<u64>,
pub(crate) thread_start_finalize_duration_ms: Option<u64>,
}
#[derive(Serialize)]
pub(crate) struct ThreadInitializedEventParams {
pub(crate) thread_id: String,
@@ -156,6 +179,8 @@ pub(crate) struct ThreadInitializedEventParams {
pub(crate) initialization_mode: ThreadInitializationMode,
pub(crate) subagent_source: Option<String>,
pub(crate) parent_thread_id: Option<String>,
#[serde(flatten)]
pub(crate) thread_start_timing: ThreadStartTimingEventParams,
pub(crate) created_at: u64,
}
@@ -1050,6 +1075,7 @@ pub(crate) fn subagent_thread_started_event_request(
parent_thread_id: input
.parent_thread_id
.or_else(|| subagent_parent_thread_id(&input.subagent_source)),
thread_start_timing: ThreadStartTimingEventParams::default(),
created_at: input.created_at,
};
ThreadInitializedEvent {
codex-rs/analytics/src/facts.rs
+17
View File
@@ -275,6 +275,21 @@ pub struct CodexCompactionEvent {
pub duration_ms: Option<u64>,
}
pub(crate) struct AppServerStartedInput {
pub runtime: CodexRuntimeMetadata,
pub remote_control_enabled: bool,
pub startup_duration_ms: u64,
pub completed_at: u64,
}
pub struct ThreadStartTimingFact {
pub thread_id: String,
pub duration_ms: u64,
pub prepare_duration_ms: u64,
pub spawn_duration_ms: u64,
pub finalize_duration_ms: u64,
}
#[allow(dead_code)]
pub(crate) enum AnalyticsFact {
Initialize {
@@ -324,7 +339,9 @@ pub(crate) enum AnalyticsFact {
}
pub(crate) enum CustomAnalyticsFact {
AppServerStarted(AppServerStartedInput),
SubAgentThreadStarted(SubAgentThreadStartedInput),
ThreadStartTiming(ThreadStartTimingFact),
Compaction(Box<CodexCompactionEvent>),
GuardianReview(Box<GuardianReviewEventParams>),
TurnResolvedConfig(Box<TurnResolvedConfigFact>),
codex-rs/analytics/src/lib.rs
+2
View File
@@ -10,6 +10,7 @@ use std::time::UNIX_EPOCH;
pub use accepted_lines::accepted_line_fingerprints_from_unified_diff;
pub use accepted_lines::fingerprint_hash;
pub use client::AnalyticsEventsClient;
pub use client::StartedTimer;
pub use events::AppServerRpcTransport;
pub use events::GuardianApprovalRequestSource;
pub use events::GuardianReviewAnalyticsResult;
@@ -37,6 +38,7 @@ pub use facts::InvocationType;
pub use facts::SkillInvocation;
pub use facts::SubAgentThreadStartedInput;
pub use facts::ThreadInitializationMode;
pub use facts::ThreadStartTimingFact;
pub use facts::TrackEventsContext;
pub use facts::TurnResolvedConfigFact;
pub use facts::TurnStatus;
codex-rs/analytics/src/reducer.rs
+66 -7
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;
@@ -45,6 +47,7 @@ use crate::events::SkillInvocationEventParams;
use crate::events::SkillInvocationEventRequest;
use crate::events::ThreadInitializedEvent;
use crate::events::ThreadInitializedEventParams;
use crate::events::ThreadStartTimingEventParams;
use crate::events::ToolItemFailureKind;
use crate::events::ToolItemTerminalStatus;
use crate::events::TrackEventRequest;
@@ -61,6 +64,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;
@@ -71,6 +75,7 @@ use crate::facts::PluginUsedInput;
use crate::facts::SkillInvokedInput;
use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadInitializationMode;
use crate::facts::ThreadStartTimingFact;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnStatus;
use crate::facts::TurnSteerRejectionReason;
@@ -147,6 +152,15 @@ struct ConnectionState {
struct ThreadAnalyticsState {
connection_id: Option<u64>,
metadata: Option<ThreadMetadataState>,
thread_start_timing: Option<ThreadStartTimingState>,
}
#[derive(Clone, Copy)]
struct ThreadStartTimingState {
duration_ms: u64,
prepare_duration_ms: u64,
spawn_duration_ms: u64,
finalize_duration_ms: u64,
}
#[derive(Clone, Copy)]
@@ -449,9 +463,15 @@ 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);
}
CustomAnalyticsFact::ThreadStartTiming(input) => {
self.ingest_thread_start_timing(input);
}
CustomAnalyticsFact::Compaction(input) => {
self.ingest_compaction(*input, out);
}
@@ -511,6 +531,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,
remote_control_enabled: input.remote_control_enabled,
startup_duration_ms: input.startup_duration_ms,
completed_at: input.completed_at,
},
},
));
}
fn ingest_subagent_thread_started(
&mut self,
input: SubAgentThreadStartedInput,
@@ -542,6 +580,19 @@ impl AnalyticsReducer {
));
}
fn ingest_thread_start_timing(&mut self, input: ThreadStartTimingFact) {
let thread_start_timing = ThreadStartTimingState {
duration_ms: input.duration_ms,
prepare_duration_ms: input.prepare_duration_ms,
spawn_duration_ms: input.spawn_duration_ms,
finalize_duration_ms: input.finalize_duration_ms,
};
self.threads
.entry(input.thread_id)
.or_default()
.thread_start_timing = Some(thread_start_timing);
}
fn ingest_guardian_review(
&mut self,
input: GuardianReviewEventParams,
@@ -1247,13 +1298,12 @@ impl AnalyticsReducer {
thread.thread_source.map(Into::into),
initialization_mode,
);
self.threads.insert(
thread_id.clone(),
ThreadAnalyticsState {
connection_id: Some(connection_id),
metadata: Some(thread_metadata.clone()),
},
);
let thread_state = self.threads.entry(thread_id.clone()).or_default();
thread_state.connection_id = Some(connection_id);
thread_state.metadata = Some(thread_metadata.clone());
let thread_start_timing = matches!(initialization_mode, ThreadInitializationMode::New)
.then_some(thread_state.thread_start_timing)
.flatten();
out.push(TrackEventRequest::ThreadInitialized(
ThreadInitializedEvent {
event_type: "codex_thread_initialized",
@@ -1268,6 +1318,15 @@ impl AnalyticsReducer {
initialization_mode,
subagent_source: thread_metadata.subagent_source.clone(),
parent_thread_id: thread_metadata.parent_thread_id,
thread_start_timing: match thread_start_timing {
Some(timing) => ThreadStartTimingEventParams {
thread_start_duration_ms: Some(timing.duration_ms),
thread_start_prepare_duration_ms: Some(timing.prepare_duration_ms),
thread_start_spawn_duration_ms: Some(timing.spawn_duration_ms),
thread_start_finalize_duration_ms: Some(timing.finalize_duration_ms),
},
None => ThreadStartTimingEventParams::default(),
},
created_at: u64::try_from(thread.created_at).unwrap_or_default(),
},
},
codex-rs/app-server/src/lib.rs
+9 -3
View File
@@ -41,6 +41,7 @@ use crate::transport::start_remote_control;
use crate::transport::start_stdio_connection;
use crate::transport::start_websocket_acceptor;
use codex_analytics::AppServerRpcTransport;
use codex_analytics::StartedTimer;
use codex_app_server_protocol::ConfigLayerSource;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::JSONRPCMessage;
@@ -428,6 +429,7 @@ pub async fn run_main_with_transport_options(
auth: AppServerWebsocketAuthSettings,
runtime_options: AppServerRuntimeOptions,
) -> IoResult<()> {
let app_server_start_timer = StartedTimer::start();
let (transport_event_tx, mut transport_event_rx) =
mpsc::channel::<TransportEvent>(CHANNEL_CAPACITY);
let (outgoing_tx, mut outgoing_rx) = mpsc::channel::<OutgoingEnvelope>(CHANNEL_CAPACITY);
@@ -698,6 +700,9 @@ pub async fn run_main_with_transport_options(
let auth_manager =
AuthManager::shared_from_config(&config, /*enable_codex_api_key_env*/ false).await;
let analytics_events_client =
analytics_events_client_from_config(Arc::clone(&auth_manager), &config);
let analytics_transport = analytics_rpc_transport(&transport);
let remote_control_requested = runtime_options.remote_control_enabled;
let remote_control_enabled = remote_control_requested && state_db.is_some();
@@ -787,8 +792,7 @@ pub async fn run_main_with_transport_options(
let processor_handle = tokio::spawn({
let auth_manager = Arc::clone(&auth_manager);
let analytics_events_client =
analytics_events_client_from_config(Arc::clone(&auth_manager), &config);
let analytics_events_client = analytics_events_client.clone();
let outgoing_message_sender = Arc::new(OutgoingMessageSender::new(
outgoing_tx,
analytics_events_client.clone(),
@@ -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: analytics_transport,
remote_control_handle: Some(remote_control_handle.clone()),
plugin_startup_tasks: runtime_options.plugin_startup_tasks,
}));
@@ -1065,6 +1069,8 @@ pub async fn run_main_with_transport_options(
info!("processor task exited (channel closed)");
}
});
analytics_events_client
.track_app_server_started(app_server_start_timer, remote_control_enabled);
drop(transport_event_tx);
codex-rs/app-server/tests/suite/v2/analytics.rs
+10 -33
View File
@@ -21,6 +21,12 @@ use wiremock::matchers::path;
const SERVICE_VERSION: &str = "0.0.0-test";
#[cfg(any(target_os = "macos", windows))]
pub(crate) const ANALYTICS_TEST_TIMEOUT: Duration = Duration::from_secs(25);
#[cfg(not(any(target_os = "macos", windows)))]
pub(crate) const ANALYTICS_TEST_TIMEOUT: Duration = Duration::from_secs(10);
fn set_metrics_exporter(config: &mut codex_core::config::Config) {
config.otel.metrics_exporter = OtelExporterKind::OtlpHttp {
endpoint: "http://localhost:4318".to_string(),
@@ -98,28 +104,6 @@ pub(crate) async fn mount_analytics_capture(server: &MockServer, codex_home: &Pa
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) async fn wait_for_analytics_event(
server: &MockServer,
read_timeout: Duration,
@@ -133,7 +117,10 @@ pub(crate) async fn wait_for_analytics_event(
};
for request in &requests {
if request.method != "POST"
|| request.url.path() != "/codex/analytics-events/events"
|| !request
.url
.path()
.ends_with("/codex/analytics-events/events")
{
continue;
}
@@ -155,16 +142,6 @@ pub(crate) async fn wait_for_analytics_event(
.await?
}
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,
codex-rs/app-server/tests/suite/v2/plugin_install.rs
+29 -51
View File
@@ -58,6 +58,8 @@ use wiremock::matchers::method;
use wiremock::matchers::path;
use wiremock::matchers::query_param;
use super::analytics::wait_for_analytics_event;
// Plugin install tests wait on connector discovery after the install response path
// starts, which is noticeably slower on Windows CI.
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(60);
@@ -727,22 +729,22 @@ async fn plugin_install_tracks_analytics_event() -> Result<()> {
let response: PluginInstallResponse = to_response(response)?;
assert_eq!(response.apps_needing_auth, Vec::<AppSummary>::new());
let payload = wait_for_plugin_analytics_payload(&analytics_server).await?;
let event =
wait_for_analytics_event(&analytics_server, DEFAULT_TIMEOUT, "codex_plugin_installed")
.await?;
assert_eq!(
payload,
event,
json!({
"events": [{
"event_type": "codex_plugin_installed",
"event_params": {
"plugin_id": "sample-plugin@debug",
"plugin_name": "sample-plugin",
"marketplace_name": "debug",
"has_skills": false,
"mcp_server_count": 0,
"connector_ids": [],
"product_client_id": DEFAULT_CLIENT_NAME,
}
}]
"event_type": "codex_plugin_installed",
"event_params": {
"plugin_id": "sample-plugin@debug",
"plugin_name": "sample-plugin",
"marketplace_name": "debug",
"has_skills": false,
"mcp_server_count": 0,
"connector_ids": [],
"product_client_id": DEFAULT_CLIENT_NAME,
}
})
);
Ok(())
@@ -780,22 +782,21 @@ async fn plugin_install_tracks_remote_plugin_analytics_event() -> Result<()> {
let response: PluginInstallResponse = to_response(response)?;
assert_eq!(response.apps_needing_auth, Vec::<AppSummary>::new());
let payload = wait_for_plugin_analytics_payload(&server).await?;
let event =
wait_for_analytics_event(&server, DEFAULT_TIMEOUT, "codex_plugin_installed").await?;
assert_eq!(
payload,
event,
json!({
"events": [{
"event_type": "codex_plugin_installed",
"event_params": {
"plugin_id": REMOTE_PLUGIN_ID,
"plugin_name": "linear",
"marketplace_name": "openai-curated-remote",
"has_skills": true,
"mcp_server_count": 0,
"connector_ids": [],
"product_client_id": DEFAULT_CLIENT_NAME,
}
}]
"event_type": "codex_plugin_installed",
"event_params": {
"plugin_id": REMOTE_PLUGIN_ID,
"plugin_name": "linear",
"marketplace_name": "openai-curated-remote",
"has_skills": true,
"mcp_server_count": 0,
"connector_ids": [],
"product_client_id": DEFAULT_CLIENT_NAME,
}
})
);
Ok(())
@@ -1288,29 +1289,6 @@ async fn mount_backend_analytics_events(server: &MockServer) {
.await;
}
async fn wait_for_plugin_analytics_payload(server: &MockServer) -> Result<serde_json::Value> {
timeout(DEFAULT_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()
.ends_with("/codex/analytics-events/events")
}) {
return serde_json::from_slice(&request.body)
.map_err(|err| anyhow::anyhow!("invalid analytics payload: {err}"));
}
tokio::time::sleep(Duration::from_millis(25)).await;
}
})
.await?
}
fn write_remote_plugin_catalog_config(
codex_home: &std::path::Path,
base_url: &str,
codex-rs/app-server/tests/suite/v2/plugin_share.rs
+7 -1
View File
@@ -383,7 +383,13 @@ plugin_sharing = false
.received_requests()
.await
.expect("wiremock should record requests")
.is_empty()
.iter()
.all(|request| {
request
.url
.path()
.ends_with("/codex/analytics-events/events")
})
);
Ok(())
}
codex-rs/app-server/tests/suite/v2/plugin_uninstall.rs
+18 -28
View File
@@ -24,6 +24,8 @@ use wiremock::matchers::header;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::wait_for_analytics_event;
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);
const REMOTE_PLUGIN_ID: &str = "plugins~Plugin_linear";
const WORKSPACE_REMOTE_PLUGIN_ID: &str = "plugins_69f27c3e67848191a45cbaa5f2adb39d";
@@ -116,37 +118,25 @@ async fn plugin_uninstall_tracks_analytics_event() -> Result<()> {
let response: PluginUninstallResponse = to_response(response)?;
assert_eq!(response, PluginUninstallResponse {});
let payload = timeout(DEFAULT_TIMEOUT, async {
loop {
let Some(requests) = analytics_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;
}
})
let event = wait_for_analytics_event(
&analytics_server,
DEFAULT_TIMEOUT,
"codex_plugin_uninstalled",
)
.await?;
let payload: serde_json::Value = serde_json::from_slice(&payload).expect("analytics payload");
assert_eq!(
payload,
event,
json!({
"events": [{
"event_type": "codex_plugin_uninstalled",
"event_params": {
"plugin_id": "sample-plugin@debug",
"plugin_name": "sample-plugin",
"marketplace_name": "debug",
"has_skills": false,
"mcp_server_count": 0,
"connector_ids": [],
"product_client_id": DEFAULT_CLIENT_NAME,
}
}]
"event_type": "codex_plugin_uninstalled",
"event_params": {
"plugin_id": "sample-plugin@debug",
"plugin_name": "sample-plugin",
"marketplace_name": "debug",
"has_skills": false,
"mcp_server_count": 0,
"connector_ids": [],
"product_client_id": DEFAULT_CLIENT_NAME,
}
})
);
Ok(())
codex-rs/app-server/tests/suite/v2/remote_control.rs
+51 -17
View File
@@ -17,6 +17,8 @@ use codex_config::types::AuthCredentialsStoreMode;
use pretty_assertions::assert_eq;
use tempfile::TempDir;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
@@ -150,8 +152,12 @@ impl BlockingRemoteControlBackend {
let (enroll_request_tx, enroll_request_rx) = oneshot::channel();
let server_task = tokio::spawn(async move {
match read_enroll_request(listener).await {
Ok((request_line, _reader)) => {
Ok(PendingHttpRequest {
request_line,
connection,
}) => {
let _ = enroll_request_tx.send(Ok(request_line));
let _connection = connection;
std::future::pending::<()>().await;
}
Err(err) => {
@@ -181,20 +187,48 @@ impl Drop for BlockingRemoteControlBackend {
}
}
async fn read_enroll_request(listener: TcpListener) -> Result<(String, BufReader<TcpStream>)> {
let (stream, _) = listener.accept().await?;
let mut reader = BufReader::new(stream);
let mut request_line = String::new();
reader.read_line(&mut request_line).await?;
loop {
let mut line = String::new();
reader.read_line(&mut line).await?;
if line == "\r\n" {
break;
}
}
Ok((request_line.trim_end().to_string(), reader))
struct PendingHttpRequest {
request_line: String,
connection: BufReader<TcpStream>,
}
async fn read_enroll_request(listener: TcpListener) -> Result<PendingHttpRequest> {
loop {
let (stream, _) = listener.accept().await?;
let mut reader = BufReader::new(stream);
let mut request_line = String::new();
reader.read_line(&mut request_line).await?;
let mut content_length = 0usize;
loop {
let mut line = String::new();
reader.read_line(&mut line).await?;
if let Some(value) = line.strip_prefix("Content-Length: ") {
content_length = value.trim().parse().unwrap_or_default();
}
if line == "\r\n" {
break;
}
}
if content_length > 0 {
let mut body = vec![0; content_length];
reader.read_exact(&mut body).await?;
}
let request_line = request_line.trim_end().to_string();
if request_line.contains("/codex/analytics-events/events ") {
reader
.get_mut()
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")
.await?;
continue;
}
return Ok(PendingHttpRequest {
request_line,
connection: reader,
});
}
}
codex-rs/app-server/tests/suite/v2/thread_fork.rs
+6 -5
View File
@@ -41,10 +41,10 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::ANALYTICS_TEST_TIMEOUT;
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_analytics_event;
#[cfg(windows)]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
@@ -403,10 +403,11 @@ 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 event = thread_initialized_event(&payload)?;
let event =
wait_for_analytics_event(&server, ANALYTICS_TEST_TIMEOUT, "codex_thread_initialized")
.await?;
assert_basic_thread_initialized_event(
event,
&event,
&thread.id,
&thread.session_id,
"mock-model",
codex-rs/app-server/tests/suite/v2/thread_resume.rs
+6 -5
View File
@@ -93,10 +93,10 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::ANALYTICS_TEST_TIMEOUT;
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_analytics_event;
#[cfg(windows)]
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(25);
@@ -420,10 +420,11 @@ 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 event = thread_initialized_event(&payload)?;
let event =
wait_for_analytics_event(&server, ANALYTICS_TEST_TIMEOUT, "codex_thread_initialized")
.await?;
assert_basic_thread_initialized_event(
event,
&event,
&thread.id,
&thread.session_id,
"gpt-5.3-codex",
codex-rs/app-server/tests/suite/v2/thread_start.rs
+14 -6
View File
@@ -44,10 +44,10 @@ use wiremock::ResponseTemplate;
use wiremock::matchers::method;
use wiremock::matchers::path;
use super::analytics::ANALYTICS_TEST_TIMEOUT;
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_analytics_event;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const INVALID_REQUEST_ERROR_CODE: i64 = -32600;
@@ -436,17 +436,25 @@ 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?;
assert_eq!(payload["events"].as_array().expect("events array").len(), 1);
let event = thread_initialized_event(&payload)?;
let event =
wait_for_analytics_event(&server, ANALYTICS_TEST_TIMEOUT, "codex_thread_initialized")
.await?;
assert_basic_thread_initialized_event(
event,
&event,
&thread.id,
&thread.session_id,
"mock-model",
"new",
"user",
);
let thread_start_timings = [
"thread_start_duration_ms",
"thread_start_prepare_duration_ms",
"thread_start_spawn_duration_ms",
"thread_start_finalize_duration_ms",
]
.map(|field| event["event_params"][field].as_u64().is_some());
assert_eq!(thread_start_timings, [true, true, true, true]);
Ok(())
}
codex-rs/core/src/lib.rs
+1
View File
@@ -138,6 +138,7 @@ pub(crate) mod state_db_bridge;
pub use state_db_bridge::StateDbHandle;
pub use state_db_bridge::init_state_db;
mod thread_rollout_truncation;
mod thread_start_timing;
mod tools;
pub(crate) mod turn_diff_tracker;
mod turn_metadata;
codex-rs/core/src/thread_manager.rs
+16
View File
@@ -15,6 +15,7 @@ use crate::session::INITIAL_SUBMIT_ID;
use crate::shell_snapshot::ShellSnapshot;
use crate::tasks::InterruptedTurnHistoryMarker;
use crate::tasks::interrupted_turn_history_marker;
use crate::thread_start_timing::ThreadStartTiming;
use codex_analytics::AnalyticsEventsClient;
use codex_app_server_protocol::ThreadHistoryBuilder;
use codex_app_server_protocol::TurnStatus;
@@ -1219,6 +1220,7 @@ impl ThreadManagerState {
user_shell_override: Option<crate::shell::Shell>,
) -> CodexResult<NewThread> {
let is_resumed_thread = matches!(&initial_history, InitialHistory::Resumed(_));
let mut thread_start_timing = ThreadStartTiming::start();
if let InitialHistory::Resumed(resumed) = &initial_history {
let mut threads = self.threads.write().await;
if let Some(thread) = threads.get(&resumed.conversation_id).cloned() {
@@ -1231,6 +1233,12 @@ impl ThreadManagerState {
resumed.conversation_id
)));
}
thread_start_timing.mark_prepare_completed();
if let Some(analytics_events_client) = self.analytics_events_client.as_ref() {
analytics_events_client.track_thread_start_timing(
thread_start_timing.into_fact(resumed.conversation_id.to_string()),
);
}
return Ok(NewThread {
thread_id: resumed.conversation_id,
session_configured: thread.session_configured(),
@@ -1245,6 +1253,7 @@ impl ThreadManagerState {
let parent_rollout_thread_trace = self
.parent_rollout_thread_trace_for_source(&session_source, &initial_history)
.await;
thread_start_timing.mark_prepare_completed();
let tracked_session_source = session_source.clone();
let CodexSpawnOk {
codex, thread_id, ..
@@ -1277,9 +1286,16 @@ impl ThreadManagerState {
attestation_provider: self.attestation_provider.clone(),
})
.await?;
thread_start_timing.mark_spawn_completed();
let new_thread = self
.finalize_thread_spawn(codex, thread_id, tracked_session_source)
.await?;
thread_start_timing.mark_finalize_completed();
if let Some(analytics_events_client) = self.analytics_events_client.as_ref() {
analytics_events_client.track_thread_start_timing(
thread_start_timing.into_fact(new_thread.thread_id.to_string()),
);
}
if is_resumed_thread {
new_thread.thread.emit_thread_resume_lifecycle().await;
if let Err(err) = new_thread.thread.apply_goal_resume_runtime_effects().await {
codex-rs/core/src/thread_start_timing.rs
+58
View File
@@ -0,0 +1,58 @@
use std::time::Duration;
use std::time::Instant;
use codex_analytics::ThreadStartTimingFact;
#[derive(Debug)]
pub(crate) struct ThreadStartTiming {
phase_started_at: Instant,
prepare_duration: Option<Duration>,
spawn_duration: Option<Duration>,
finalize_duration: Option<Duration>,
}
impl ThreadStartTiming {
pub(crate) fn start() -> Self {
Self {
phase_started_at: Instant::now(),
prepare_duration: None,
spawn_duration: None,
finalize_duration: None,
}
}
pub(crate) fn mark_prepare_completed(&mut self) {
self.prepare_duration = Some(self.finish_phase());
}
pub(crate) fn mark_spawn_completed(&mut self) {
self.spawn_duration = Some(self.finish_phase());
}
pub(crate) fn mark_finalize_completed(&mut self) {
self.finalize_duration = Some(self.finish_phase());
}
pub(crate) fn into_fact(self, thread_id: String) -> ThreadStartTimingFact {
let prepare_duration = self.prepare_duration.unwrap_or_default();
let spawn_duration = self.spawn_duration.unwrap_or_default();
let finalize_duration = self.finalize_duration.unwrap_or_default();
ThreadStartTimingFact {
thread_id,
duration_ms: duration_ms(prepare_duration + spawn_duration + finalize_duration),
prepare_duration_ms: duration_ms(prepare_duration),
spawn_duration_ms: duration_ms(spawn_duration),
finalize_duration_ms: duration_ms(finalize_duration),
}
}
fn finish_phase(&mut self) -> Duration {
let duration = self.phase_started_at.elapsed();
self.phase_started_at = Instant::now();
duration
}
}
fn duration_ms(duration: Duration) -> u64 {
duration.as_millis().try_into().unwrap_or(u64::MAX)
}