Emit turn timing as a dedicated event

This commit is contained in:
Ahmed Ibrahim
2026-05-28 11:46:56 -07:00
parent 3f40c4eabf
commit c7b6f807f1
8 changed files with 71 additions and 274 deletions
+17 -164
View File
@@ -71,7 +71,7 @@ use crate::facts::TrackEventsContext;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnStatus;
use crate::facts::TurnSteerRequestError;
use crate::facts::TurnTimingBreakdownFact;
use crate::facts::TurnTimingFact;
use crate::facts::TurnTokenUsageFact;
use crate::reducer::AnalyticsReducer;
use crate::reducer::normalize_path_for_skill_id;
@@ -338,8 +338,8 @@ 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 {
fn sample_turn_timing_fact(thread_id: &str, turn_id: &str) -> TurnTimingFact {
TurnTimingFact {
turn_id: turn_id.to_string(),
thread_id: thread_id.to_string(),
request_start_delay_ms: Some(120),
@@ -3446,11 +3446,6 @@ 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),
},
@@ -3513,11 +3508,6 @@ 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
}
@@ -3841,172 +3831,35 @@ async fn turn_lifecycle_emits_turn_event() {
json!(13)
);
assert_eq!(payload["event_params"]["total_tokens"], json!(321));
assert_eq!(
json!({
"request_start_delay_ms": payload["event_params"]["request_start_delay_ms"],
"sampling_duration_ms": payload["event_params"]["sampling_duration_ms"],
"blocking_tool_critical_path_duration_ms": payload["event_params"]["blocking_tool_critical_path_duration_ms"],
"approval_wait_duration_ms": payload["event_params"]["approval_wait_duration_ms"],
"finalize_duration_ms": payload["event_params"]["finalize_duration_ms"],
}),
json!({
"request_start_delay_ms": null,
"sampling_duration_ms": null,
"blocking_tool_critical_path_duration_ms": null,
"approval_wait_duration_ms": null,
"finalize_duration_ms": null,
})
);
}
#[tokio::test]
async fn turn_timing_breakdown_fact_enriches_turn_event() {
async fn turn_timing_fact_emits_dedicated_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,
AnalyticsFact::Custom(CustomAnalyticsFact::TurnTiming(Box::new(
sample_turn_timing_fact("thread-2", "turn-2"),
))),
&mut out,
)
.await;
let payload = serde_json::to_value(&out[0]).expect("serialize turn event");
let payload = serde_json::to_value(&out).expect("serialize timing event");
assert_eq!(
json!({
"request_start_delay_ms": payload["event_params"]["request_start_delay_ms"],
"sampling_duration_ms": payload["event_params"]["sampling_duration_ms"],
"blocking_tool_critical_path_duration_ms": payload["event_params"]["blocking_tool_critical_path_duration_ms"],
"approval_wait_duration_ms": payload["event_params"]["approval_wait_duration_ms"],
"finalize_duration_ms": payload["event_params"]["finalize_duration_ms"],
}),
json!({
"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,
})
);
}
#[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,
},
}),
payload,
json!([{
"event_type": "turn_timing",
"event_params": {
"thread_id": "thread-2",
"turn_id": "turn-2",
"request_start_delay_ms": 120,
"sampling_duration_ms": 900,
"blocking_tool_critical_path_duration_ms": 140,
},
&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::Accept,
},
}),
},
&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!(
json!({
"approval_wait_duration_ms": payload["event_params"]["approval_wait_duration_ms"],
"finalize_duration_ms": payload["event_params"]["finalize_duration_ms"],
}),
json!({
"approval_wait_duration_ms": 600,
"finalize_duration_ms": 34,
})
}])
);
}
+5 -5
View File
@@ -21,7 +21,7 @@ use crate::facts::SubAgentThreadStartedInput;
use crate::facts::ThreadInitializationTimingFact;
use crate::facts::TrackEventsContext;
use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnTimingBreakdownFact;
use crate::facts::TurnTimingFact;
use crate::facts::TurnTokenUsageFact;
use crate::reducer::AnalyticsReducer;
use codex_app_server_protocol::ClientRequest;
@@ -188,10 +188,10 @@ impl AnalyticsEventsClient {
));
}
pub fn track_turn_timing(&self, fact: TurnTimingBreakdownFact) {
self.record_fact(AnalyticsFact::Custom(
CustomAnalyticsFact::TurnTimingBreakdown(Box::new(fact)),
));
pub fn track_turn_timing(&self, fact: TurnTimingFact) {
self.record_fact(AnalyticsFact::Custom(CustomAnalyticsFact::TurnTiming(
Box::new(fact),
)));
}
pub fn track_guardian_review(
+16 -5
View File
@@ -64,6 +64,7 @@ pub(crate) enum TrackEventRequest {
HookRun(CodexHookRunEventRequest),
Compaction(Box<CodexCompactionEventRequest>),
TurnEvent(Box<CodexTurnEventRequest>),
TurnTiming(TurnTimingEventRequest),
TurnSteer(CodexTurnSteerEventRequest),
CommandExecution(CodexCommandExecutionEventRequest),
FileChange(CodexFileChangeEventRequest),
@@ -836,11 +837,6 @@ 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>,
}
@@ -851,6 +847,21 @@ pub(crate) struct CodexTurnEventRequest {
pub(crate) event_params: CodexTurnEventParams,
}
#[derive(Serialize)]
pub(crate) struct TurnTimingEventParams {
pub(crate) thread_id: String,
pub(crate) turn_id: String,
pub(crate) request_start_delay_ms: Option<u64>,
pub(crate) sampling_duration_ms: u64,
pub(crate) blocking_tool_critical_path_duration_ms: u64,
}
#[derive(Serialize)]
pub(crate) struct TurnTimingEventRequest {
pub(crate) event_type: &'static str,
pub(crate) event_params: TurnTimingEventParams,
}
#[derive(Serialize)]
pub(crate) struct CodexTurnSteerEventParams {
pub(crate) thread_id: String,
+2 -2
View File
@@ -100,7 +100,7 @@ pub struct TurnTokenUsageFact {
}
#[derive(Clone)]
pub struct TurnTimingBreakdownFact {
pub struct TurnTimingFact {
pub turn_id: String,
pub thread_id: String,
pub request_start_delay_ms: Option<u64>,
@@ -353,7 +353,7 @@ pub(crate) enum CustomAnalyticsFact {
Compaction(Box<CodexCompactionEvent>),
GuardianReview(Box<GuardianReviewEventParams>),
TurnResolvedConfig(Box<TurnResolvedConfigFact>),
TurnTimingBreakdown(Box<TurnTimingBreakdownFact>),
TurnTiming(Box<TurnTimingFact>),
TurnTokenUsage(Box<TurnTokenUsageFact>),
SkillInvoked(SkillInvokedInput),
AppMentioned(AppMentionedInput),
+1 -1
View File
@@ -44,7 +44,7 @@ pub use facts::TurnStatus;
pub use facts::TurnSteerRejectionReason;
pub use facts::TurnSteerRequestError;
pub use facts::TurnSteerResult;
pub use facts::TurnTimingBreakdownFact;
pub use facts::TurnTimingFact;
pub use facts::TurnTokenUsageFact;
pub use facts::build_track_events_context;
+18 -92
View File
@@ -51,6 +51,8 @@ use crate::events::ThreadInitializedEventParams;
use crate::events::ToolItemFailureKind;
use crate::events::ToolItemTerminalStatus;
use crate::events::TrackEventRequest;
use crate::events::TurnTimingEventParams;
use crate::events::TurnTimingEventRequest;
use crate::events::WebSearchActionKind;
use crate::events::codex_app_metadata;
use crate::events::codex_compaction_event_params;
@@ -80,7 +82,7 @@ use crate::facts::TurnResolvedConfigFact;
use crate::facts::TurnStatus;
use crate::facts::TurnSteerRejectionReason;
use crate::facts::TurnSteerResult;
use crate::facts::TurnTimingBreakdownFact;
use crate::facts::TurnTimingFact;
use crate::facts::TurnTokenUsageFact;
use crate::now_unix_seconds;
use crate::option_i64_to_u64;
@@ -324,14 +326,6 @@ 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,
}
struct TurnState {
connection_id: Option<u64>,
thread_id: Option<String>,
@@ -343,7 +337,6 @@ struct TurnState {
latest_diff: Option<String>,
steer_count: usize,
tool_counts: TurnToolCounts,
timing: TurnTimingState,
}
#[derive(Hash, Eq, PartialEq)]
@@ -483,8 +476,8 @@ impl AnalyticsReducer {
CustomAnalyticsFact::TurnResolvedConfig(input) => {
self.ingest_turn_resolved_config(*input, out).await;
}
CustomAnalyticsFact::TurnTimingBreakdown(input) => {
self.ingest_turn_timing_breakdown(*input, out).await;
CustomAnalyticsFact::TurnTiming(input) => {
Self::ingest_turn_timing(*input, out);
}
CustomAnalyticsFact::TurnTokenUsage(input) => {
self.ingest_turn_token_usage(*input, out).await;
@@ -671,7 +664,6 @@ impl AnalyticsReducer {
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
timing: TurnTimingState::default(),
});
turn_state.thread_id = Some(thread_id);
turn_state.num_input_images = Some(num_input_images);
@@ -696,38 +688,24 @@ impl AnalyticsReducer {
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
timing: TurnTimingState::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_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(),
timing: TurnTimingState::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;
fn ingest_turn_timing(input: TurnTimingFact, out: &mut Vec<TrackEventRequest>) {
out.push(TrackEventRequest::TurnTiming(TurnTimingEventRequest {
event_type: "turn_timing",
event_params: TurnTimingEventParams {
thread_id: input.thread_id,
turn_id: input.turn_id,
request_start_delay_ms: input.request_start_delay_ms,
sampling_duration_ms: input.sampling_duration_ms,
blocking_tool_critical_path_duration_ms: input
.blocking_tool_critical_path_duration_ms,
},
}));
}
async fn ingest_skill_invoked(
@@ -889,7 +867,6 @@ impl AnalyticsReducer {
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
timing: TurnTimingState::default(),
});
turn_state.connection_id = Some(connection_id);
turn_state.thread_id = Some(pending_request.thread_id);
@@ -1249,7 +1226,6 @@ impl AnalyticsReducer {
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
timing: TurnTimingState::default(),
});
turn_state.started_at = notification
.turn
@@ -1271,7 +1247,6 @@ impl AnalyticsReducer {
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
timing: TurnTimingState::default(),
});
turn_state.thread_id = Some(notification.thread_id);
turn_state.latest_diff = Some(notification.diff);
@@ -1291,7 +1266,6 @@ impl AnalyticsReducer {
latest_diff: None,
steer_count: 0,
tool_counts: TurnToolCounts::default(),
timing: TurnTimingState::default(),
});
turn_state.completed = Some(CompletedTurnState {
status: analytics_turn_status(notification.turn.status),
@@ -1499,7 +1473,6 @@ 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,
@@ -1509,28 +1482,6 @@ impl AnalyticsReducer {
&pending_review,
);
}
if let Some(duration_ms) = duration_ms {
let turn_state =
self.turns
.entry(pending_review.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(),
timing: TurnTimingState::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 {
@@ -1556,7 +1507,7 @@ impl AnalyticsReducer {
resolution,
started_at_ms: pending_review.started_at_ms,
completed_at_ms,
duration_ms,
duration_ms: observed_duration_ms(pending_review.started_at_ms, completed_at_ms),
},
}));
}
@@ -2564,26 +2515,6 @@ 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,
session_id: thread_metadata.session_id.clone(),
@@ -2641,11 +2572,6 @@ 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),
}
@@ -133,10 +133,11 @@ async fn turn_interrupt_aborts_running_turn() -> Result<()> {
)?;
assert_eq!(completed.thread_id, thread_id);
assert_eq!(completed.turn.status, TurnStatus::Interrupted);
let event = wait_for_analytics_event(&server, DEFAULT_READ_TIMEOUT, "codex_turn_event").await?;
let event = wait_for_analytics_event(&server, DEFAULT_READ_TIMEOUT, "turn_timing").await?;
assert_eq!(
(
event["event_params"]["status"].as_str(),
event["event_params"]["thread_id"].as_str(),
event["event_params"]["turn_id"].as_str(),
event["event_params"]["request_start_delay_ms"]
.as_u64()
.is_some(),
@@ -147,7 +148,13 @@ async fn turn_interrupt_aborts_running_turn() -> Result<()> {
.as_u64()
.is_some_and(|duration_ms| duration_ms > 0),
),
(Some("interrupted"), true, true, true)
(
Some(thread_id.as_str()),
Some(turn_id.as_str()),
true,
true,
true
)
);
Ok(())
+2 -2
View File
@@ -7,7 +7,7 @@ mod user_shell;
use std::sync::Arc;
use std::time::Duration;
use codex_analytics::TurnTimingBreakdownFact;
use codex_analytics::TurnTimingFact;
use codex_analytics::TurnTokenUsageFact;
use codex_extension_api::ExtensionData;
use futures::future::BoxFuture;
@@ -824,7 +824,7 @@ impl Session {
let timing = turn_context.turn_timing_state.timing_breakdown().await;
self.services
.analytics_events_client
.track_turn_timing(TurnTimingBreakdownFact {
.track_turn_timing(TurnTimingFact {
turn_id: turn_context.sub_id.clone(),
thread_id: self.conversation_id.to_string(),
request_start_delay_ms: timing.request_start_delay_ms,