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

Track app-server, thread, and turn timing in ChatGPT telemetry

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-05-22 12:21:01 -07:00
parent c20d45cbb9
commit b2a1da0664
13 changed files with 656 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files

302
codex-rs/analytics/src/analytics_client_tests.rs
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;
@@ -15,6 +17,7 @@ use crate::events::CodexReviewEventParams;
use crate::events::CodexReviewEventRequest;
use crate::events::CodexRuntimeMetadata;
use crate::events::CodexToolItemEventBase;
use crate::events::CodexTurnEventParams;
use crate::events::CodexTurnEventRequest;
use crate::events::FinalApprovalOutcome;
use crate::events::GuardianApprovalRequestSource;
@@ -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,10 +65,12 @@ 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;
use crate::facts::TurnSteerRequestError;
use crate::facts::TurnTimingBreakdownFact;
use crate::facts::TurnTokenUsageFact;
use crate::reducer::AnalyticsReducer;
use crate::reducer::normalize_path_for_skill_id;
@@ -331,6 +337,16 @@ fn sample_turn_token_usage_fact(thread_id: &str, turn_id: &str) -> TurnTokenUsag
}
}
fn sample_turn_timing_breakdown_fact(thread_id: &str, turn_id: &str) -> TurnTimingBreakdownFact {
TurnTimingBreakdownFact {
turn_id: turn_id.to_string(),
thread_id: thread_id.to_string(),
request_start_delay_ms: Some(120),
sampling_duration_ms: 900,
blocking_tool_critical_path_duration_ms: 140,
}
}
fn sample_turn_completed_notification(
thread_id: &str,
turn_id: &str,
@@ -1317,6 +1333,7 @@ fn thread_initialized_event_serializes_expected_shape() {
initialization_mode: ThreadInitializationMode::New,
subagent_source: None,
parent_thread_id: None,
thread_start_duration_ms: Some(321),
created_at: 1,
},
});
@@ -1348,12 +1365,46 @@ fn thread_initialized_event_serializes_expected_shape() {
"initialization_mode": "new",
"subagent_source": null,
"parent_thread_id": null,
"thread_start_duration_ms": 321,
"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(),
rpc_transport: AppServerRpcTransport::Websocket,
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"
},
"rpc_transport": "websocket",
"startup_duration_ms": 987,
"completed_at": 12
}
})
);
}
#[test]
fn command_execution_event_serializes_expected_shape() {
let event = TrackEventRequest::CommandExecution(CodexCommandExecutionEventRequest {
@@ -1629,6 +1680,87 @@ 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)
);
}
#[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,
},
)),
&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)
);
}
#[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(),
rpc_transport: AppServerRpcTransport::Stdio,
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"
},
"rpc_transport": "stdio",
"startup_duration_ms": 456,
"completed_at": 12
}
}])
);
}
#[tokio::test]
@@ -3180,7 +3312,7 @@ async fn reducer_ingests_plugin_state_changed_fact() {
fn turn_event_serializes_expected_shape() {
let event = TrackEventRequest::TurnEvent(Box::new(CodexTurnEventRequest {
event_type: "codex_turn_event",
event_params: crate::events::CodexTurnEventParams {
event_params: CodexTurnEventParams {
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
app_server_client: sample_app_server_client_metadata(),
@@ -3221,6 +3353,11 @@ fn turn_event_serializes_expected_shape() {
reasoning_output_tokens: None,
total_tokens: None,
duration_ms: Some(1234),
request_start_delay_ms: Some(120),
sampling_duration_ms: Some(900),
blocking_tool_critical_path_duration_ms: Some(140),
approval_wait_duration_ms: Some(0),
finalize_duration_ms: Some(74),
started_at: Some(455),
completed_at: Some(456),
},
@@ -3282,6 +3419,11 @@ fn turn_event_serializes_expected_shape() {
"reasoning_output_tokens": null,
"total_tokens": null,
"duration_ms": 1234,
"request_start_delay_ms": 120,
"sampling_duration_ms": 900,
"blocking_tool_critical_path_duration_ms": 140,
"approval_wait_duration_ms": 0,
"finalize_duration_ms": 74,
"started_at": 455,
"completed_at": 456
}
@@ -3597,6 +3739,164 @@ async fn turn_lifecycle_emits_turn_event() {
json!(13)
);
assert_eq!(payload["event_params"]["total_tokens"], json!(321));
assert_eq!(
payload["event_params"]["request_start_delay_ms"],
json!(null)
);
assert_eq!(payload["event_params"]["sampling_duration_ms"], json!(null));
assert_eq!(
payload["event_params"]["blocking_tool_critical_path_duration_ms"],
json!(null)
);
assert_eq!(
payload["event_params"]["approval_wait_duration_ms"],
json!(null)
);
assert_eq!(payload["event_params"]["finalize_duration_ms"], json!(null));
}
#[tokio::test]
async fn turn_timing_breakdown_fact_enriches_turn_event() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnTimingBreakdown(Box::new(
sample_turn_timing_breakdown_fact("thread-2", "turn-2"),
))),
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
assert_eq!(
payload["event_params"]["request_start_delay_ms"],
json!(120)
);
assert_eq!(payload["event_params"]["sampling_duration_ms"], json!(900));
assert_eq!(
payload["event_params"]["blocking_tool_critical_path_duration_ms"],
json!(140)
);
assert_eq!(
payload["event_params"]["approval_wait_duration_ms"],
json!(0)
);
assert_eq!(payload["event_params"]["finalize_duration_ms"], json!(74));
}
#[tokio::test]
async fn review_durations_roll_up_into_turn_approval_wait() {
let mut reducer = AnalyticsReducer::default();
let mut out = Vec::new();
ingest_turn_prerequisites(
&mut reducer,
&mut out,
/*include_initialize*/ true,
/*include_resolved_config*/ true,
/*include_started*/ true,
/*include_token_usage*/ false,
)
.await;
reducer
.ingest(
AnalyticsFact::Custom(CustomAnalyticsFact::TurnTimingBreakdown(Box::new(
TurnTimingBreakdownFact {
turn_id: "turn-2".to_string(),
thread_id: "thread-2".to_string(),
request_start_delay_ms: Some(100),
sampling_duration_ms: 400,
blocking_tool_critical_path_duration_ms: 700,
},
))),
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::ServerRequest {
connection_id: 7,
request: Box::new(ServerRequest::CommandExecutionRequestApproval {
request_id: RequestId::Integer(99),
params: CommandExecutionRequestApprovalParams {
thread_id: "thread-2".to_string(),
turn_id: "turn-2".to_string(),
item_id: "item-1".to_string(),
started_at_ms: 1_000,
approval_id: None,
reason: None,
network_approval_context: None,
command: Some("echo hi".to_string()),
cwd: None,
command_actions: None,
additional_permissions: None,
proposed_execpolicy_amendment: None,
proposed_network_policy_amendments: None,
available_decisions: None,
},
}),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::ServerResponse {
completed_at_ms: 1_600,
response: Box::new(ServerResponse::CommandExecutionRequestApproval {
request_id: RequestId::Integer(99),
response: CommandExecutionRequestApprovalResponse {
decision: CommandExecutionApprovalDecision::AllowOnce,
},
}),
},
&mut out,
)
.await;
reducer
.ingest(
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
"thread-2",
"turn-2",
AppServerTurnStatus::Completed,
/*codex_error_info*/ None,
))),
&mut out,
)
.await;
let turn_event = out
.iter()
.find(|event| matches!(event, TrackEventRequest::TurnEvent(_)))
.expect("turn event should be emitted");
let payload = serde_json::to_value(turn_event).expect("serialize turn event");
assert_eq!(
payload["event_params"]["approval_wait_duration_ms"],
json!(600)
);
assert_eq!(payload["event_params"]["finalize_duration_ms"], json!(34));
}
#[tokio::test]

40
codex-rs/analytics/src/client.rs
View File

@@ -8,17 +8,22 @@ 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::CustomAnalyticsFact;
use crate::facts::HookRunFact;
use crate::facts::HookRunInput;
use crate::facts::PluginState;
use crate::facts::PluginStateChangedInput;
use crate::facts::PluginUsedInput;
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::TurnTimingBreakdownFact;
use crate::facts::TurnTokenUsageFact;
use crate::reducer::AnalyticsReducer;
use codex_app_server_protocol::ClientRequest;
@@ -163,12 +168,37 @@ impl AnalyticsEventsClient {
});
}
pub fn track_app_server_started(
&self,
rpc_transport: AppServerRpcTransport,
startup_duration_ms: u64,
completed_at: u64,
) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::AppServerStarted(AppServerStartedInput {
runtime: current_runtime_metadata(),
rpc_transport,
startup_duration_ms,
completed_at,
}),
));
}
pub fn track_subagent_thread_started(&self, input: SubAgentThreadStartedInput) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::SubAgentThreadStarted(input),
));
}
pub fn track_thread_start_timing(&self, thread_id: String, duration_ms: u64) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::ThreadStartTiming(ThreadStartTimingFact {
thread_id,
duration_ms,
}),
));
}
pub fn track_guardian_review(
&self,
tracking: &GuardianReviewTrackContext,
@@ -234,11 +264,11 @@ impl AnalyticsEventsClient {
return;
}
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::PluginUsed(
crate::facts::PluginUsedInput { tracking, plugin },
PluginUsedInput { tracking, plugin },
)));
}
pub fn track_compaction(&self, event: crate::facts::CodexCompactionEvent) {
pub fn track_compaction(&self, event: CodexCompactionEvent) {
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::Compaction(
Box::new(event),
)));
@@ -256,6 +286,12 @@ impl AnalyticsEventsClient {
)));
}
pub fn track_turn_timing(&self, fact: TurnTimingBreakdownFact) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::TurnTimingBreakdown(Box::new(fact)),
));
}
pub fn track_plugin_installed(&self, plugin: PluginTelemetryMetadata) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::PluginStateChanged(PluginStateChangedInput {

22
codex-rs/analytics/src/events.rs
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,20 @@ pub(crate) struct CodexRuntimeMetadata {
pub(crate) runtime_arch: String,
}
#[derive(Serialize)]
pub(crate) struct CodexAppServerStartedEventParams {
pub(crate) runtime: CodexRuntimeMetadata,
pub(crate) rpc_transport: AppServerRpcTransport,
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(Serialize)]
pub(crate) struct ThreadInitializedEventParams {
pub(crate) thread_id: String,
@@ -155,6 +170,7 @@ pub(crate) struct ThreadInitializedEventParams {
pub(crate) initialization_mode: ThreadInitializationMode,
pub(crate) subagent_source: Option<String>,
pub(crate) parent_thread_id: Option<String>,
pub(crate) thread_start_duration_ms: Option<u64>,
pub(crate) created_at: u64,
}
@@ -808,6 +824,11 @@ pub(crate) struct CodexTurnEventParams {
pub(crate) reasoning_output_tokens: Option<i64>,
pub(crate) total_tokens: Option<i64>,
pub(crate) duration_ms: Option<u64>,
pub(crate) request_start_delay_ms: Option<u64>,
pub(crate) sampling_duration_ms: Option<u64>,
pub(crate) blocking_tool_critical_path_duration_ms: Option<u64>,
pub(crate) approval_wait_duration_ms: Option<u64>,
pub(crate) finalize_duration_ms: Option<u64>,
pub(crate) started_at: Option<u64>,
pub(crate) completed_at: Option<u64>,
}
@@ -1041,6 +1062,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_duration_ms: None,
created_at: input.created_at,
};
ThreadInitializedEvent {

24
codex-rs/analytics/src/facts.rs
View File

@@ -99,6 +99,15 @@ pub struct TurnTokenUsageFact {
pub token_usage: TokenUsage,
}
#[derive(Clone)]
pub struct TurnTimingBreakdownFact {
pub turn_id: String,
pub thread_id: String,
pub request_start_delay_ms: Option<u64>,
pub sampling_duration_ms: u64,
pub blocking_tool_critical_path_duration_ms: u64,
}
#[derive(Clone, Copy, Debug, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum TurnStatus {
@@ -273,6 +282,18 @@ pub struct CodexCompactionEvent {
pub duration_ms: Option<u64>,
}
pub(crate) struct AppServerStartedInput {
pub runtime: CodexRuntimeMetadata,
pub rpc_transport: AppServerRpcTransport,
pub startup_duration_ms: u64,
pub completed_at: u64,
}
pub(crate) struct ThreadStartTimingFact {
pub thread_id: String,
pub duration_ms: u64,
}
#[allow(dead_code)]
pub(crate) enum AnalyticsFact {
Initialize {
@@ -322,11 +343,14 @@ pub(crate) enum AnalyticsFact {
}
pub(crate) enum CustomAnalyticsFact {
AppServerStarted(AppServerStartedInput),
SubAgentThreadStarted(SubAgentThreadStartedInput),
ThreadStartTiming(ThreadStartTimingFact),
Compaction(Box<CodexCompactionEvent>),
GuardianReview(Box<GuardianReviewEventParams>),
TurnResolvedConfig(Box<TurnResolvedConfigFact>),
TurnTokenUsage(Box<TurnTokenUsageFact>),
TurnTimingBreakdown(Box<TurnTimingBreakdownFact>),
SkillInvoked(SkillInvokedInput),
AppMentioned(AppMentionedInput),
AppUsed(AppUsedInput),

1
codex-rs/analytics/src/lib.rs
View File

@@ -43,6 +43,7 @@ pub use facts::TurnStatus;
pub use facts::TurnSteerRejectionReason;
pub use facts::TurnSteerRequestError;
pub use facts::TurnSteerResult;
pub use facts::TurnTimingBreakdownFact;
pub use facts::TurnTokenUsageFact;
pub use facts::build_track_events_context;

201
codex-rs/analytics/src/reducer.rs
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;
@@ -61,6 +63,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,10 +74,12 @@ 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;
use crate::facts::TurnSteerResult;
use crate::facts::TurnTimingBreakdownFact;
use crate::facts::TurnTokenUsageFact;
use crate::now_unix_seconds;
use crate::option_i64_to_u64;
@@ -147,6 +152,7 @@ struct ConnectionState {
struct ThreadAnalyticsState {
connection_id: Option<u64>,
metadata: Option<ThreadMetadataState>,
thread_start_duration_ms: Option<u64>,
}
#[derive(Clone, Copy)]
@@ -314,6 +320,15 @@ struct CompletedTurnState {
duration_ms: Option<u64>,
}
#[derive(Clone, Default)]
struct TurnTimingState {
request_start_delay_ms: Option<u64>,
sampling_duration_ms: Option<u64>,
blocking_tool_critical_path_duration_ms: Option<u64>,
approval_wait_duration_ms: u64,
}
#[derive(Default)]
struct TurnState {
connection_id: Option<u64>,
thread_id: Option<String>,
@@ -325,6 +340,7 @@ struct TurnState {
latest_diff: Option<String>,
steer_count: usize,
tool_counts: TurnToolCounts,
timing: TurnTimingState,
}
#[derive(Hash, Eq, PartialEq)]
@@ -446,9 +462,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);
}
@@ -461,6 +483,9 @@ impl AnalyticsReducer {
CustomAnalyticsFact::TurnTokenUsage(input) => {
self.ingest_turn_token_usage(*input, out).await;
}
CustomAnalyticsFact::TurnTimingBreakdown(input) => {
self.ingest_turn_timing_breakdown(*input, out).await;
}
CustomAnalyticsFact::SkillInvoked(input) => {
self.ingest_skill_invoked(input, out).await;
}
@@ -508,6 +533,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,
startup_duration_ms: input.startup_duration_ms,
completed_at: input.completed_at,
},
},
));
}
fn ingest_subagent_thread_started(
&mut self,
input: SubAgentThreadStartedInput,
@@ -538,6 +581,13 @@ impl AnalyticsReducer {
));
}
fn ingest_thread_start_timing(&mut self, input: ThreadStartTimingFact) {
self.threads
.entry(input.thread_id)
.or_default()
.thread_start_duration_ms = Some(input.duration_ms);
}
fn ingest_guardian_review(
&mut self,
input: GuardianReviewEventParams,
@@ -599,18 +649,7 @@ impl AnalyticsReducer {
let turn_id = input.turn_id.clone();
let thread_id = input.thread_id.clone();
let num_input_images = input.num_input_images;
let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
token_usage: None,
completed: None,
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
});
let turn_state = self.turns.entry(turn_id.clone()).or_default();
turn_state.thread_id = Some(thread_id);
turn_state.num_input_images = Some(num_input_images);
turn_state.resolved_config = Some(input);
@@ -623,23 +662,27 @@ impl AnalyticsReducer {
out: &mut Vec<TrackEventRequest>,
) {
let turn_id = input.turn_id.clone();
let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
token_usage: None,
completed: None,
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
});
let turn_state = self.turns.entry(turn_id.clone()).or_default();
turn_state.thread_id = Some(input.thread_id);
turn_state.token_usage = Some(input.token_usage);
self.maybe_emit_turn_event(&turn_id, out).await;
}
async fn ingest_turn_timing_breakdown(
&mut self,
input: TurnTimingBreakdownFact,
out: &mut Vec<TrackEventRequest>,
) {
let turn_id = input.turn_id.clone();
let turn_state = self.turns.entry(turn_id.clone()).or_default();
turn_state.thread_id = Some(input.thread_id);
turn_state.timing.request_start_delay_ms = input.request_start_delay_ms;
turn_state.timing.sampling_duration_ms = Some(input.sampling_duration_ms);
turn_state.timing.blocking_tool_critical_path_duration_ms =
Some(input.blocking_tool_critical_path_duration_ms);
self.maybe_emit_turn_event(&turn_id, out).await;
}
async fn ingest_skill_invoked(
&mut self,
input: SkillInvokedInput,
@@ -788,18 +831,7 @@ impl AnalyticsReducer {
else {
return;
};
let turn_state = self.turns.entry(turn_id.clone()).or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
token_usage: None,
completed: None,
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
});
let turn_state = self.turns.entry(turn_id.clone()).or_default();
turn_state.connection_id = Some(connection_id);
turn_state.thread_id = Some(pending_request.thread_id);
turn_state.num_input_images = Some(pending_request.num_input_images);
@@ -1147,58 +1179,19 @@ impl AnalyticsReducer {
self.ingest_guardian_review_completed(notification, out);
}
ServerNotification::TurnStarted(notification) => {
let turn_state = self.turns.entry(notification.turn.id).or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
token_usage: None,
completed: None,
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
});
let turn_state = self.turns.entry(notification.turn.id).or_default();
turn_state.started_at = notification
.turn
.started_at
.and_then(|started_at| u64::try_from(started_at).ok());
}
ServerNotification::TurnDiffUpdated(notification) => {
let turn_state =
self.turns
.entry(notification.turn_id.clone())
.or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
token_usage: None,
completed: None,
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
});
let turn_state = self.turns.entry(notification.turn_id.clone()).or_default();
turn_state.thread_id = Some(notification.thread_id);
turn_state.latest_diff = Some(notification.diff);
}
ServerNotification::TurnCompleted(notification) => {
let turn_state =
self.turns
.entry(notification.turn.id.clone())
.or_insert(TurnState {
connection_id: None,
thread_id: None,
num_input_images: None,
resolved_config: None,
started_at: None,
token_usage: None,
completed: None,
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
});
let turn_state = self.turns.entry(notification.turn.id.clone()).or_default();
turn_state.completed = Some(CompletedTurnState {
status: analytics_turn_status(notification.turn.status),
turn_error: notification
@@ -1240,13 +1233,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_duration_ms = matches!(initialization_mode, ThreadInitializationMode::New)
.then_some(thread_state.thread_start_duration_ms)
.flatten();
out.push(TrackEventRequest::ThreadInitialized(
ThreadInitializedEvent {
event_type: "codex_thread_initialized",
@@ -1260,6 +1252,7 @@ impl AnalyticsReducer {
initialization_mode,
subagent_source: thread_metadata.subagent_source.clone(),
parent_thread_id: thread_metadata.parent_thread_id,
thread_start_duration_ms,
created_at: u64::try_from(thread.created_at).unwrap_or_default(),
},
},
@@ -1403,6 +1396,7 @@ impl AnalyticsReducer {
completed_at_ms: u64,
out: &mut Vec<TrackEventRequest>,
) {
let duration_ms = observed_duration_ms(pending_review.started_at_ms, completed_at_ms);
if let Some(item_key) = item_review_summary_key(&pending_review) {
self.record_item_review_summary(
item_key,
@@ -1412,6 +1406,16 @@ impl AnalyticsReducer {
&pending_review,
);
}
if let Some(duration_ms) = duration_ms {
let turn_state = self
.turns
.entry(pending_review.turn_id.clone())
.or_default();
turn_state.timing.approval_wait_duration_ms = turn_state
.timing
.approval_wait_duration_ms
.saturating_add(duration_ms);
}
let Some((connection_state, thread_metadata)) =
self.thread_context_or_warn(AnalyticsDropSite::review(&pending_review))
else {
@@ -1437,7 +1441,7 @@ impl AnalyticsReducer {
resolution,
started_at_ms: pending_review.started_at_ms,
completed_at_ms,
duration_ms: observed_duration_ms(pending_review.started_at_ms, completed_at_ms),
duration_ms,
},
}));
}
@@ -2445,6 +2449,26 @@ fn codex_turn_event_params(
is_first_turn,
} = resolved_config;
let token_usage = turn_state.token_usage.clone();
let timing = &turn_state.timing;
let has_turn_timing = timing.request_start_delay_ms.is_some()
|| timing.sampling_duration_ms.is_some()
|| timing.blocking_tool_critical_path_duration_ms.is_some();
let approval_wait_duration_ms = has_turn_timing.then_some(timing.approval_wait_duration_ms);
let finalize_duration_ms = completed.duration_ms.and_then(|duration_ms| {
has_turn_timing.then_some(
duration_ms.saturating_sub(
timing
.request_start_delay_ms
.unwrap_or_default()
.saturating_add(timing.sampling_duration_ms.unwrap_or_default())
.saturating_add(
timing
.blocking_tool_critical_path_duration_ms
.unwrap_or_default(),
),
),
)
});
CodexTurnEventParams {
thread_id,
turn_id,
@@ -2501,6 +2525,11 @@ fn codex_turn_event_params(
.as_ref()
.map(|token_usage| token_usage.total_tokens),
duration_ms: completed.duration_ms,
request_start_delay_ms: timing.request_start_delay_ms,
sampling_duration_ms: timing.sampling_duration_ms,
blocking_tool_critical_path_duration_ms: timing.blocking_tool_critical_path_duration_ms,
approval_wait_duration_ms,
finalize_duration_ms,
started_at,
completed_at: Some(completed.completed_at),
}

23
codex-rs/app-server/src/lib.rs
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;
@@ -41,6 +42,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::now_unix_seconds;
use codex_app_server_protocol::ConfigLayerSource;
use codex_app_server_protocol::ConfigWarningNotification;
use codex_app_server_protocol::JSONRPCMessage;
@@ -428,6 +430,7 @@ pub async fn run_main_with_transport_options(
auth: AppServerWebsocketAuthSettings,
runtime_options: AppServerRuntimeOptions,
) -> IoResult<()> {
let app_server_start_started_at = 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);
@@ -698,6 +701,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();
@@ -786,10 +792,8 @@ pub async fn run_main_with_transport_options(
});
let processor_handle = tokio::spawn({
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 auth_manager = Arc::clone(&auth_manager);
let analytics_events_client = analytics_events_client.clone();
let outgoing_message_sender = Arc::new(OutgoingMessageSender::new(
outgoing_tx,
analytics_events_client.clone(),
@@ -810,7 +814,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,
}));
@@ -1066,6 +1070,15 @@ pub async fn run_main_with_transport_options(
info!("processor task exited (channel closed)");
}
});
analytics_events_client.track_app_server_started(
analytics_transport,
app_server_start_started_at
.elapsed()
.as_millis()
.try_into()
.unwrap_or(u64::MAX),
now_unix_seconds(),
);
drop(transport_event_tx);

1
codex-rs/app-server/src/message_processor.rs
View File

@@ -410,6 +410,7 @@ impl MessageProcessor {
auth_manager.clone(),
Arc::clone(&thread_manager),
outgoing.clone(),
analytics_events_client.clone(),
arg0_paths.clone(),
Arc::clone(&config),
config_manager.clone(),

14
codex-rs/app-server/src/request_processors/thread_processor.rs
View File

@@ -327,6 +327,7 @@ pub(crate) struct ThreadRequestProcessor {
pub(super) auth_manager: Arc<AuthManager>,
pub(super) thread_manager: Arc<ThreadManager>,
pub(super) outgoing: Arc<OutgoingMessageSender>,
pub(super) analytics_events_client: AnalyticsEventsClient,
pub(super) arg0_paths: Arg0DispatchPaths,
pub(super) config: Arc<Config>,
pub(super) config_manager: ConfigManager,
@@ -347,6 +348,7 @@ impl ThreadRequestProcessor {
auth_manager: Arc<AuthManager>,
thread_manager: Arc<ThreadManager>,
outgoing: Arc<OutgoingMessageSender>,
analytics_events_client: AnalyticsEventsClient,
arg0_paths: Arg0DispatchPaths,
config: Arc<Config>,
config_manager: ConfigManager,
@@ -363,6 +365,7 @@ impl ThreadRequestProcessor {
auth_manager,
thread_manager,
outgoing,
analytics_events_client,
arg0_paths,
config,
config_manager,
@@ -880,11 +883,13 @@ impl ThreadRequestProcessor {
};
let request_trace = request_context.request_trace();
let config_manager = self.config_manager.clone();
let analytics_events_client = self.analytics_events_client.clone();
let outgoing = Arc::clone(&listener_task_context.outgoing);
let error_request_id = request_id.clone();
let thread_start_task = async move {
if let Err(error) = Self::thread_start_task(
listener_task_context,
analytics_events_client,
config_manager,
request_id,
app_server_client_name,
@@ -969,6 +974,7 @@ impl ThreadRequestProcessor {
#[allow(clippy::too_many_arguments)]
async fn thread_start_task(
listener_task_context: ListenerTaskContext,
analytics_events_client: AnalyticsEventsClient,
config_manager: ConfigManager,
request_id: ConnectionRequestId,
app_server_client_name: Option<String>,
@@ -1202,6 +1208,14 @@ impl ThreadRequestProcessor {
active_permission_profile,
reasoning_effort: config_snapshot.reasoning_effort,
};
analytics_events_client.track_thread_start_timing(
thread.id.clone(),
thread_start_started_at
.elapsed()
.as_millis()
.try_into()
.unwrap_or(u64::MAX),
);
let notif = thread_started_notification(thread);
listener_task_context
.outgoing

36
codex-rs/core/src/session/turn.rs
View File

@@ -2,6 +2,7 @@ use std::collections::HashMap;
use std::collections::HashSet;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Instant;
use crate::SkillInjections;
use crate::build_skill_injections;
@@ -1733,7 +1734,12 @@ async fn try_run_sampling_request(
turn_context.model_info.slug.as_str(),
turn_context.provider.info().name.as_str(),
);
let mut stream = client_session
turn_context
.turn_timing_state
.mark_model_request_started()
.await;
let sampling_started_at = Instant::now();
let stream_result = client_session
.stream(
prompt,
&turn_context.model_info,
@@ -1746,7 +1752,24 @@ async fn try_run_sampling_request(
)
.instrument(trace_span!("stream_request"))
.or_cancel(&cancellation_token)
.await??;
.await;
let mut stream = match stream_result {
Ok(Ok(stream)) => stream,
Ok(Err(err)) => {
turn_context
.turn_timing_state
.record_sampling_duration(sampling_started_at.elapsed())
.await;
return Err(err);
}
Err(codex_async_utils::CancelErr::Cancelled) => {
turn_context
.turn_timing_state
.record_sampling_duration(sampling_started_at.elapsed())
.await;
return Err(CodexErr::TurnAborted);
}
};
let mut in_flight: FuturesOrdered<BoxFuture<'static, CodexResult<ResponseInputItem>>> =
FuturesOrdered::new();
let mut needs_follow_up = false;
@@ -2151,6 +2174,10 @@ async fn try_run_sampling_request(
&mut assistant_message_stream_parsers,
)
.await;
turn_context
.turn_timing_state
.record_sampling_duration(sampling_started_at.elapsed())
.await;
if sess
.features
@@ -2161,7 +2188,12 @@ async fn try_run_sampling_request(
client_session.send_response_processed(response_id).await;
}
let blocking_tool_started_at = Instant::now();
drain_in_flight(&mut in_flight, sess.clone(), turn_context.clone()).await?;
turn_context
.turn_timing_state
.record_blocking_tool_critical_path_duration(blocking_tool_started_at.elapsed())
.await;
if should_emit_token_count {
// A tool call such as request_user_input can intentionally pause the turn. Emit token

12
codex-rs/core/src/tasks/mod.rs
View File

@@ -33,6 +33,7 @@ use crate::session::turn_context::TurnContext;
use crate::state::ActiveTurn;
use crate::state::RunningTask;
use crate::state::TaskKind;
use codex_analytics::TurnTimingBreakdownFact;
use codex_analytics::TurnTokenUsageFact;
use codex_login::AuthManager;
use codex_models_manager::manager::SharedModelsManager;
@@ -767,10 +768,21 @@ impl Session {
.turn_timing_state
.completed_at_and_duration_ms()
.await;
let turn_timing_breakdown = turn_context.turn_timing_state.timing_breakdown().await;
let time_to_first_token_ms = turn_context
.turn_timing_state
.time_to_first_token_ms()
.await;
self.services
.analytics_events_client
.track_turn_timing(TurnTimingBreakdownFact {
turn_id: turn_context.sub_id.clone(),
thread_id: self.conversation_id.to_string(),
request_start_delay_ms: turn_timing_breakdown.request_start_delay_ms,
sampling_duration_ms: turn_timing_breakdown.sampling_duration_ms,
blocking_tool_critical_path_duration_ms: turn_timing_breakdown
.blocking_tool_critical_path_duration_ms,
});
if should_clear_active_turn {
self.emit_turn_stop_lifecycle(turn_context.extension_data.as_ref())
.await;

54
codex-rs/core/src/turn_timing.rs
View File

@@ -36,6 +36,13 @@ pub(crate) async fn record_turn_ttfm_metric(turn_context: &TurnContext, item: &T
.record_duration(TURN_TTFM_DURATION_METRIC, duration, &[]);
}
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
pub(crate) struct TurnTimingBreakdown {
pub(crate) request_start_delay_ms: Option<u64>,
pub(crate) sampling_duration_ms: u64,
pub(crate) blocking_tool_critical_path_duration_ms: u64,
}
#[derive(Debug, Default)]
pub(crate) struct TurnTimingState {
state: Mutex<TurnTimingStateInner>,
@@ -47,6 +54,9 @@ struct TurnTimingStateInner {
started_at_unix_secs: Option<i64>,
first_token_at: Option<Instant>,
first_message_at: Option<Instant>,
first_request_started_at: Option<Instant>,
sampling_duration: Duration,
blocking_tool_critical_path_duration: Duration,
}
impl TurnTimingState {
@@ -57,6 +67,9 @@ impl TurnTimingState {
state.started_at_unix_secs = Some(started_at_unix_ms / 1000);
state.first_token_at = None;
state.first_message_at = None;
state.first_request_started_at = None;
state.sampling_duration = Duration::default();
state.blocking_tool_critical_path_duration = Duration::default();
started_at_unix_ms
}
@@ -98,6 +111,36 @@ impl TurnTimingState {
let mut state = self.state.lock().await;
state.record_turn_ttfm()
}
pub(crate) async fn mark_model_request_started(&self) {
let mut state = self.state.lock().await;
if state.first_request_started_at.is_none() {
state.first_request_started_at = Some(Instant::now());
}
}
pub(crate) async fn record_sampling_duration(&self, duration: Duration) {
let mut state = self.state.lock().await;
state.sampling_duration = state.sampling_duration.saturating_add(duration);
}
pub(crate) async fn record_blocking_tool_critical_path_duration(&self, duration: Duration) {
let mut state = self.state.lock().await;
state.blocking_tool_critical_path_duration = state
.blocking_tool_critical_path_duration
.saturating_add(duration);
}
pub(crate) async fn timing_breakdown(&self) -> TurnTimingBreakdown {
let state = self.state.lock().await;
TurnTimingBreakdown {
request_start_delay_ms: state.request_start_delay_ms(),
sampling_duration_ms: duration_to_u64_millis(state.sampling_duration),
blocking_tool_critical_path_duration_ms: duration_to_u64_millis(
state.blocking_tool_critical_path_duration,
),
}
}
}
fn now_unix_timestamp_secs() -> i64 {
@@ -112,6 +155,13 @@ pub(crate) fn now_unix_timestamp_ms() -> i64 {
}
impl TurnTimingStateInner {
fn request_start_delay_ms(&self) -> Option<u64> {
let duration = self
.first_request_started_at?
.duration_since(self.started_at?);
Some(duration_to_u64_millis(duration))
}
fn time_to_first_token(&self) -> Option<Duration> {
Some(self.first_token_at?.duration_since(self.started_at?))
}
@@ -136,6 +186,10 @@ impl TurnTimingStateInner {
}
}
fn duration_to_u64_millis(duration: Duration) -> u64 {
u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
}
fn response_event_records_turn_ttft(event: &ResponseEvent) -> bool {
match event {
ResponseEvent::OutputItemDone(item) | ResponseEvent::OutputItemAdded(item) => {

22
codex-rs/core/src/turn_timing_tests.rs
View File

@@ -99,6 +99,28 @@ async fn turn_timing_state_records_turn_started_epoch_millis() {
);
}
#[tokio::test]
async fn turn_timing_state_tracks_request_start_and_duration_breakdown() {
let state = TurnTimingState::default();
state.mark_turn_started(Instant::now()).await;
state.mark_model_request_started().await;
state
.record_sampling_duration(std::time::Duration::from_millis(120))
.await;
state
.record_sampling_duration(std::time::Duration::from_millis(30))
.await;
state
.record_blocking_tool_critical_path_duration(std::time::Duration::from_millis(45))
.await;
let breakdown = state.timing_breakdown().await;
assert_eq!(breakdown.sampling_duration_ms, 150);
assert_eq!(breakdown.blocking_tool_critical_path_duration_ms, 45);
assert!(breakdown.request_start_delay_ms.is_some());
}
#[test]
fn response_item_records_turn_ttft_for_first_output_signals() {
assert!(response_item_records_turn_ttft(