mirror of
https://github.com/openai/codex.git
synced 2026-10-27 18:48:37 +00:00
Report turn critical-path timing in analytics (#57209)
## What changed Include critical-path capture status and duration totals in `codex_turn_event`, with categories for inference, auto review, user wait, tools, overhead, and unknown time. Count sampling and compaction as inference. Distinguish `disabled`, `incomplete`, and `complete` capture. Emit null durations for disabled or incomplete capture so missing measurements are not reported as zero. Freeze the completion timestamp so repeated terminal notifications preserve the original measurement window. ## Testing Add core and analytics tests for capture status, duration reporting, rounding, and repeated completion. Add app-server integration coverage for completed and interrupted turns with default, enabled, and disabled capture. GitOrigin-RevId: c6c867120ee563ac48584c2f4b12df23acbabe06
This commit is contained in:
@@ -1110,6 +1110,8 @@ pub(crate) struct CodexGoalEventRequest {
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub(crate) struct CodexTurnEventParams {
|
||||
#[serde(flatten)]
|
||||
pub(crate) critical_path: crate::turn_critical_path::CriticalPathEventParams,
|
||||
pub(crate) thread_id: String,
|
||||
pub(crate) session_id: String,
|
||||
pub(crate) turn_id: String,
|
||||
|
||||
@@ -246,6 +246,7 @@ pub struct TurnTokenUsageFact {
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
pub struct TurnProfile {
|
||||
pub critical_path: crate::TurnCriticalPath,
|
||||
pub before_first_sampling_ms: u64,
|
||||
pub sampling_ms: u64,
|
||||
pub compaction_ms: u64,
|
||||
|
||||
@@ -8,6 +8,7 @@ mod guardian_v2;
|
||||
mod product_attribution;
|
||||
mod reducer;
|
||||
mod thread_hint;
|
||||
mod turn_critical_path;
|
||||
|
||||
use std::time::SystemTime;
|
||||
use std::time::UNIX_EPOCH;
|
||||
@@ -82,6 +83,8 @@ pub use guardian_v2::GuardianV2EventKind;
|
||||
pub use product_attribution::ThreadProductUpdate;
|
||||
pub use thread_hint::ThreadHintStatus;
|
||||
pub use thread_hint::ThreadHintStatusEvent;
|
||||
pub use turn_critical_path::TurnCriticalPath;
|
||||
pub use turn_critical_path::TurnCriticalPathDurations;
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
@@ -3958,6 +3958,7 @@ fn codex_turn_event_params(
|
||||
is_first_turn,
|
||||
} = resolved_config;
|
||||
let TurnProfile {
|
||||
critical_path,
|
||||
before_first_sampling_ms,
|
||||
sampling_ms,
|
||||
compaction_ms,
|
||||
@@ -3971,6 +3972,7 @@ fn codex_turn_event_params(
|
||||
let token_usage = turn_state.token_usage.clone();
|
||||
let codex_error = turn_state.codex_error.as_ref();
|
||||
CodexTurnEventParams {
|
||||
critical_path: critical_path.into(),
|
||||
thread_id,
|
||||
session_id: thread_metadata.session_id.clone(),
|
||||
turn_id,
|
||||
|
||||
@@ -102,6 +102,7 @@ fn turn_event_serializes_expected_shape() {
|
||||
let event = TrackEventRequest::TurnEvent(Box::new(CodexTurnEventRequest {
|
||||
event_type: "codex_turn_event",
|
||||
event_params: crate::events::CodexTurnEventParams {
|
||||
critical_path: crate::TurnCriticalPath::Disabled.into(),
|
||||
thread_id: "thread-2".to_string(),
|
||||
session_id: "session-thread-2".to_string(),
|
||||
turn_id: "turn-2".to_string(),
|
||||
@@ -187,6 +188,14 @@ fn turn_event_serializes_expected_shape() {
|
||||
r#"{
|
||||
"event_type": "codex_turn_event",
|
||||
"event_params": {
|
||||
"critical_path_status": "disabled",
|
||||
"critical_path_duration_ms": null,
|
||||
"critical_path_inference_ms": null,
|
||||
"critical_path_auto_review_ms": null,
|
||||
"critical_path_user_wait_ms": null,
|
||||
"critical_path_tools_ms": null,
|
||||
"critical_path_overhead_ms": null,
|
||||
"critical_path_unknown_ms": null,
|
||||
"thread_id": "thread-2",
|
||||
"session_id": "session-thread-2",
|
||||
"turn_id": "turn-2",
|
||||
@@ -958,3 +967,107 @@ async fn turn_completed_without_started_notification_emits_null_started_at() {
|
||||
);
|
||||
assert_eq!(payload["event_params"]["total_tokens"], json!(null));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn critical_path_payload_preserves_capture_status_and_emits_once() {
|
||||
use crate::TurnCriticalPath;
|
||||
use crate::TurnCriticalPathDurations;
|
||||
use crate::TurnProfileFact;
|
||||
use crate::tests::support::sample_turn_profile;
|
||||
|
||||
for status in [
|
||||
AppServerTurnStatus::Completed,
|
||||
AppServerTurnStatus::Interrupted,
|
||||
] {
|
||||
for (critical_path, expected) in [
|
||||
(
|
||||
TurnCriticalPath::Disabled,
|
||||
json!(["disabled", null, null, null, null, null, null, null]),
|
||||
),
|
||||
(
|
||||
TurnCriticalPath::Incomplete,
|
||||
json!(["incomplete", null, null, null, null, null, null, null]),
|
||||
),
|
||||
(
|
||||
TurnCriticalPath::Complete(TurnCriticalPathDurations {
|
||||
duration_ms: 1234,
|
||||
inference_ms: 700,
|
||||
auto_review_ms: 200,
|
||||
user_wait_ms: 100,
|
||||
tools_ms: 150,
|
||||
overhead_ms: 80,
|
||||
unknown_ms: 4,
|
||||
}),
|
||||
json!(["complete", 1234, 700, 200, 100, 150, 80, 4]),
|
||||
),
|
||||
(
|
||||
TurnCriticalPath::Complete(TurnCriticalPathDurations {
|
||||
duration_ms: 1234,
|
||||
inference_ms: 1234,
|
||||
auto_review_ms: 0,
|
||||
user_wait_ms: 0,
|
||||
tools_ms: 0,
|
||||
overhead_ms: 0,
|
||||
unknown_ms: 0,
|
||||
}),
|
||||
json!(["complete", 1234, 1234, 0, 0, 0, 0, 0]),
|
||||
),
|
||||
] {
|
||||
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*/ true,
|
||||
)
|
||||
.await;
|
||||
let mut profile = sample_turn_profile();
|
||||
profile.critical_path = critical_path;
|
||||
reducer
|
||||
.ingest(
|
||||
AnalyticsFact::Custom(CustomAnalyticsFact::TurnProfile(Box::new(
|
||||
TurnProfileFact {
|
||||
turn_id: "turn-2".into(),
|
||||
profile,
|
||||
},
|
||||
))),
|
||||
&mut out,
|
||||
)
|
||||
.await;
|
||||
for _ in 0..2 {
|
||||
reducer
|
||||
.ingest(
|
||||
AnalyticsFact::Notification(Box::new(sample_turn_completed_notification(
|
||||
"thread-2",
|
||||
"turn-2",
|
||||
status.clone(),
|
||||
/*codex_error_info*/ None,
|
||||
))),
|
||||
&mut out,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
assert_eq!(out.len(), 1);
|
||||
let payload = serde_json::to_value(&out[0]).unwrap();
|
||||
let params = &payload["event_params"];
|
||||
assert_eq!(
|
||||
json!([
|
||||
params["critical_path_status"],
|
||||
params["critical_path_duration_ms"],
|
||||
params["critical_path_inference_ms"],
|
||||
params["critical_path_auto_review_ms"],
|
||||
params["critical_path_user_wait_ms"],
|
||||
params["critical_path_tools_ms"],
|
||||
params["critical_path_overhead_ms"],
|
||||
params["critical_path_unknown_ms"],
|
||||
]),
|
||||
expected
|
||||
);
|
||||
assert_eq!(params["tool_blocking_ms"], json!(250));
|
||||
assert_eq!(params["duration_ms"], json!(1234));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -374,6 +374,7 @@ pub(super) fn sample_turn_resolved_config(
|
||||
|
||||
pub(super) fn sample_turn_profile() -> TurnProfile {
|
||||
TurnProfile {
|
||||
critical_path: crate::TurnCriticalPath::Disabled,
|
||||
before_first_sampling_ms: 100,
|
||||
sampling_ms: 700,
|
||||
compaction_ms: 40,
|
||||
|
||||
@@ -0,0 +1,59 @@
|
||||
//! Turn critical-path facts and their nullable analytics fields.
|
||||
//! Missing capture never becomes a zero-duration measurement.
|
||||
|
||||
use serde::Serialize;
|
||||
|
||||
/// Capture coverage for one finalized turn, independent of its completion status.
|
||||
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize)]
|
||||
pub enum TurnCriticalPath {
|
||||
#[default]
|
||||
Disabled,
|
||||
Incomplete,
|
||||
Complete(TurnCriticalPathDurations),
|
||||
}
|
||||
|
||||
/// Category totals over the server turn interval, truncated to milliseconds.
|
||||
/// Sampling and compaction both contribute to inference. Independent truncation
|
||||
/// can leave the category sum up to five milliseconds below duration_ms.
|
||||
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
|
||||
pub struct TurnCriticalPathDurations {
|
||||
pub duration_ms: u64,
|
||||
pub inference_ms: u64,
|
||||
pub auto_review_ms: u64,
|
||||
pub user_wait_ms: u64,
|
||||
pub tools_ms: u64,
|
||||
pub overhead_ms: u64,
|
||||
pub unknown_ms: u64,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub(crate) struct CriticalPathEventParams {
|
||||
critical_path_status: &'static str,
|
||||
critical_path_duration_ms: Option<u64>,
|
||||
critical_path_inference_ms: Option<u64>,
|
||||
critical_path_auto_review_ms: Option<u64>,
|
||||
critical_path_user_wait_ms: Option<u64>,
|
||||
critical_path_tools_ms: Option<u64>,
|
||||
critical_path_overhead_ms: Option<u64>,
|
||||
critical_path_unknown_ms: Option<u64>,
|
||||
}
|
||||
|
||||
impl From<TurnCriticalPath> for CriticalPathEventParams {
|
||||
fn from(path: TurnCriticalPath) -> Self {
|
||||
let (status, durations) = match path {
|
||||
TurnCriticalPath::Disabled => ("disabled", None),
|
||||
TurnCriticalPath::Incomplete => ("incomplete", None),
|
||||
TurnCriticalPath::Complete(durations) => ("complete", Some(durations)),
|
||||
};
|
||||
Self {
|
||||
critical_path_status: status,
|
||||
critical_path_duration_ms: durations.as_ref().map(|value| value.duration_ms),
|
||||
critical_path_inference_ms: durations.as_ref().map(|value| value.inference_ms),
|
||||
critical_path_auto_review_ms: durations.as_ref().map(|value| value.auto_review_ms),
|
||||
critical_path_user_wait_ms: durations.as_ref().map(|value| value.user_wait_ms),
|
||||
critical_path_tools_ms: durations.as_ref().map(|value| value.tools_ms),
|
||||
critical_path_overhead_ms: durations.as_ref().map(|value| value.overhead_ms),
|
||||
critical_path_unknown_ms: durations.as_ref().map(|value| value.unknown_ms),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,179 @@
|
||||
//! Verify Core capture reaches the existing analytics endpoint on terminal turns.
|
||||
|
||||
use super::analytics::mount_analytics_capture;
|
||||
use super::analytics::wait_for_matching_analytics_event;
|
||||
use anyhow::Result;
|
||||
use app_test_support::MockResponsesConfig;
|
||||
use app_test_support::TestAppServer;
|
||||
use codex_app_server_protocol::ClientRequest;
|
||||
use codex_app_server_protocol::ThreadStartParams;
|
||||
use codex_app_server_protocol::TurnStartParams;
|
||||
use codex_app_server_protocol::TurnStartResponse;
|
||||
use codex_app_server_protocol::UserInput;
|
||||
use codex_features::Feature;
|
||||
use core_test_support::responses;
|
||||
use core_test_support::skip_if_no_network;
|
||||
use pretty_assertions::assert_eq;
|
||||
use serde_json::json;
|
||||
use std::collections::BTreeMap;
|
||||
use std::time::Duration;
|
||||
use tempfile::TempDir;
|
||||
use tokio::time::timeout;
|
||||
use wiremock::Mock;
|
||||
use wiremock::ResponseTemplate;
|
||||
use wiremock::matchers::method;
|
||||
use wiremock::matchers::path_regex;
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum Finish {
|
||||
Complete,
|
||||
Interrupt,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy)]
|
||||
enum Capture {
|
||||
Default,
|
||||
Enabled,
|
||||
Disabled,
|
||||
}
|
||||
|
||||
#[test_case::test_case(Finish::Complete, Capture::Default; "complete_default")]
|
||||
#[test_case::test_case(Finish::Complete, Capture::Enabled; "complete_enabled")]
|
||||
#[test_case::test_case(Finish::Complete, Capture::Disabled; "complete_disabled")]
|
||||
#[test_case::test_case(Finish::Interrupt, Capture::Default; "interrupt_default")]
|
||||
#[test_case::test_case(Finish::Interrupt, Capture::Enabled; "interrupt_enabled")]
|
||||
#[test_case::test_case(Finish::Interrupt, Capture::Disabled; "interrupt_disabled")]
|
||||
#[tokio::test]
|
||||
async fn critical_path_reaches_turn_event(finish: Finish, capture: Capture) -> Result<()> {
|
||||
skip_if_no_network!(Ok(()));
|
||||
let server = responses::start_mock_server().await;
|
||||
let body = responses::sse(vec![
|
||||
responses::ev_response_created("response-1"),
|
||||
responses::ev_assistant_message("answer", "Done"),
|
||||
responses::ev_completed("response-1"),
|
||||
]);
|
||||
let mut response = ResponseTemplate::new(200).set_body_raw(body, "text/event-stream");
|
||||
if matches!(finish, Finish::Interrupt) {
|
||||
response = response.set_delay(Duration::from_secs(60));
|
||||
}
|
||||
Mock::given(method("POST"))
|
||||
.and(path_regex(".*/responses$"))
|
||||
.respond_with(response)
|
||||
.mount(&server)
|
||||
.await;
|
||||
let codex_home = TempDir::new()?;
|
||||
let mut features = BTreeMap::from([(Feature::RuntimeMetrics, true)]);
|
||||
match capture {
|
||||
Capture::Default => {}
|
||||
Capture::Enabled => {
|
||||
features.insert(Feature::CriticalPathTiming, true);
|
||||
}
|
||||
Capture::Disabled => {
|
||||
features.insert(Feature::CriticalPathTiming, false);
|
||||
}
|
||||
}
|
||||
MockResponsesConfig::new(&server.uri())
|
||||
.with_root_config(&format!("chatgpt_base_url = {:?}", server.uri()))
|
||||
.with_provider_config("supports_websockets = false")
|
||||
.with_features(&features)
|
||||
.write(codex_home.path())?;
|
||||
mount_analytics_capture(&server, codex_home.path()).await?;
|
||||
let mut app = TestAppServer::builder()
|
||||
.with_codex_home(codex_home.path())
|
||||
.without_managed_config()
|
||||
.build_initialized()
|
||||
.await?;
|
||||
let thread = app.start_thread(ThreadStartParams::default()).await?.thread;
|
||||
let params = TurnStartParams {
|
||||
thread_id: thread.id.clone(),
|
||||
input: vec![UserInput::Text {
|
||||
text: "Hello".into(),
|
||||
text_elements: Vec::new(),
|
||||
}],
|
||||
..Default::default()
|
||||
};
|
||||
let (turn_id, expected_status) = match finish {
|
||||
Finish::Complete => {
|
||||
let completed = app.start_turn_and_wait_for_completion(params).await?;
|
||||
(completed.turn.id, "completed")
|
||||
}
|
||||
Finish::Interrupt => {
|
||||
let started: TurnStartResponse = app
|
||||
.request(|request_id| ClientRequest::TurnStart { request_id, params })
|
||||
.await?;
|
||||
// Wait for sampling to begin before interrupting, without a timing-dependent sleep.
|
||||
timeout(Duration::from_secs(10), async {
|
||||
loop {
|
||||
if server
|
||||
.received_requests()
|
||||
.await
|
||||
.unwrap_or_default()
|
||||
.iter()
|
||||
.any(|r| r.url.path().ends_with("/responses"))
|
||||
{
|
||||
break;
|
||||
}
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await?;
|
||||
app.interrupt_turn_and_wait_for_aborted(
|
||||
thread.id.clone(),
|
||||
started.turn.id.clone(),
|
||||
Duration::from_secs(10),
|
||||
)
|
||||
.await?;
|
||||
(started.turn.id, "interrupted")
|
||||
}
|
||||
};
|
||||
let event = wait_for_matching_analytics_event(&server, Duration::from_secs(30), |event| {
|
||||
event["event_type"] == "codex_turn_event" && event["event_params"]["turn_id"] == turn_id
|
||||
})
|
||||
.await?;
|
||||
let p = &event["event_params"];
|
||||
assert_eq!(
|
||||
json!([p["thread_id"], p["status"]]),
|
||||
json!([thread.id, expected_status])
|
||||
);
|
||||
let fields = [
|
||||
"critical_path_duration_ms",
|
||||
"critical_path_inference_ms",
|
||||
"critical_path_auto_review_ms",
|
||||
"critical_path_user_wait_ms",
|
||||
"critical_path_tools_ms",
|
||||
"critical_path_overhead_ms",
|
||||
"critical_path_unknown_ms",
|
||||
];
|
||||
if matches!(capture, Capture::Enabled) {
|
||||
assert_eq!(p["critical_path_status"], "complete");
|
||||
assert_eq!(p[fields[0]], p["duration_ms"]);
|
||||
let total = p[fields[0]]
|
||||
.as_u64()
|
||||
.expect("complete capture has a numeric total");
|
||||
let sum: u64 = fields[1..]
|
||||
.iter()
|
||||
.map(|key| {
|
||||
p[key]
|
||||
.as_u64()
|
||||
.expect("complete capture has numeric categories")
|
||||
})
|
||||
.sum();
|
||||
assert!(sum <= total && total - sum <= 5);
|
||||
assert_eq!(
|
||||
json!([
|
||||
p["critical_path_user_wait_ms"],
|
||||
p["critical_path_tools_ms"],
|
||||
p["critical_path_auto_review_ms"],
|
||||
p["critical_path_unknown_ms"],
|
||||
]),
|
||||
json!([0, 0, 0, 0])
|
||||
);
|
||||
} else {
|
||||
assert_eq!(p["critical_path_status"], "disabled");
|
||||
assert_eq!(
|
||||
fields.map(|key| p.get(key)),
|
||||
[Some(&serde_json::Value::Null); 7]
|
||||
);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -48,6 +48,8 @@ mod connection_handling_websocket;
|
||||
mod connection_handling_websocket_compression;
|
||||
#[cfg(unix)]
|
||||
mod connection_handling_websocket_unix;
|
||||
#[path = "critical_path_analytics_tests.rs"]
|
||||
mod critical_path_analytics;
|
||||
#[cfg(unix)]
|
||||
mod curated_mcp_sync;
|
||||
mod current_time;
|
||||
|
||||
@@ -72,6 +72,7 @@ pub(crate) struct TurnTimingState {
|
||||
struct TurnTimingStateInner {
|
||||
started_at: Option<Instant>,
|
||||
started_at_unix_secs: Option<i64>,
|
||||
completed_at: Option<(Instant, i64)>,
|
||||
item_started_at_ms: HashMap<String, i64>,
|
||||
first_token_at: Option<Instant>,
|
||||
first_message_at: Option<Instant>,
|
||||
@@ -79,6 +80,7 @@ struct TurnTimingStateInner {
|
||||
|
||||
#[derive(Debug, Default)]
|
||||
struct TurnProfileState {
|
||||
capture_enabled: bool,
|
||||
path: Option<CriticalPath>,
|
||||
tool_dependency: Option<CriticalPath>,
|
||||
started_at: Option<Instant>,
|
||||
@@ -116,6 +118,7 @@ impl TurnTimingState {
|
||||
pub(crate) fn enable_critical_path(&self) {
|
||||
let mut state = self.profile_state();
|
||||
if state.completed_profile.is_none() {
|
||||
state.capture_enabled = true;
|
||||
state.path = Some(CriticalPath::default());
|
||||
}
|
||||
}
|
||||
@@ -140,6 +143,7 @@ impl TurnTimingState {
|
||||
let started_at_unix_ms = now_unix_timestamp_ms();
|
||||
let mut state = self.state.lock().await;
|
||||
state.started_at = Some(started_at);
|
||||
state.completed_at = None;
|
||||
state.started_at_unix_secs = Some(started_at_unix_ms / 1000);
|
||||
state.item_started_at_ms.clear();
|
||||
state.first_token_at = None;
|
||||
@@ -169,9 +173,11 @@ impl TurnTimingState {
|
||||
pub(crate) async fn complete_profile_and_duration_ms(
|
||||
&self,
|
||||
) -> (Option<i64>, Option<i64>, TurnProfile) {
|
||||
let completed_at_instant = Instant::now();
|
||||
let state = self.state.lock().await;
|
||||
let completed_at = Some(now_unix_timestamp_secs());
|
||||
let mut state = self.state.lock().await;
|
||||
// Repeated terminal notifications must retain the original measurement window.
|
||||
let (completed_at_instant, completed_at) = *state
|
||||
.completed_at
|
||||
.get_or_insert_with(|| (Instant::now(), now_unix_timestamp_secs()));
|
||||
let duration_ms = state.started_at.map(|started_at| {
|
||||
i64::try_from(
|
||||
completed_at_instant
|
||||
@@ -181,7 +187,7 @@ impl TurnTimingState {
|
||||
.unwrap_or(i64::MAX)
|
||||
});
|
||||
let profile = self.profile_state().complete(completed_at_instant);
|
||||
(completed_at, duration_ms, profile)
|
||||
(Some(completed_at), duration_ms, profile)
|
||||
}
|
||||
|
||||
pub(crate) async fn time_to_first_token_ms(&self) -> Option<i64> {
|
||||
@@ -296,6 +302,7 @@ impl TurnProfileState {
|
||||
let path = self.path.clone();
|
||||
*self = Self {
|
||||
path,
|
||||
capture_enabled: self.capture_enabled,
|
||||
started_at: Some(started_at),
|
||||
last_transition_at: Some(started_at),
|
||||
..Self::default()
|
||||
@@ -412,6 +419,7 @@ impl TurnProfileState {
|
||||
};
|
||||
|
||||
let mut profile = TurnProfile {
|
||||
critical_path: report::completed_critical_path(self, now),
|
||||
before_first_sampling_ms: duration_to_u64_ms(self.before_first_sampling),
|
||||
sampling_ms: duration_to_u64_ms(self.sampling),
|
||||
compaction_ms: duration_to_u64_ms(self.compaction),
|
||||
@@ -535,6 +543,8 @@ fn response_item_records_turn_ttft(item: &ResponseItem) -> bool {
|
||||
}
|
||||
}
|
||||
|
||||
mod report;
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "turn_timing_tests.rs"]
|
||||
mod tests;
|
||||
|
||||
@@ -0,0 +1,26 @@
|
||||
//! Report the frozen critical path over the same interval as the turn profile.
|
||||
|
||||
use super::TurnProfileState;
|
||||
use super::duration_to_u64_ms;
|
||||
use codex_analytics::TurnCriticalPath;
|
||||
use codex_analytics::TurnCriticalPathDurations;
|
||||
use std::time::Instant;
|
||||
|
||||
pub(super) fn completed_critical_path(state: &TurnProfileState, end: Instant) -> TurnCriticalPath {
|
||||
if !state.capture_enabled {
|
||||
return TurnCriticalPath::Disabled;
|
||||
}
|
||||
let (Some(path), Some(start)) = (&state.path, state.started_at) else {
|
||||
return TurnCriticalPath::Incomplete;
|
||||
};
|
||||
let totals = path.durations(start, end);
|
||||
TurnCriticalPath::Complete(TurnCriticalPathDurations {
|
||||
duration_ms: duration_to_u64_ms(end.saturating_duration_since(start)),
|
||||
inference_ms: duration_to_u64_ms(totals.inference),
|
||||
auto_review_ms: duration_to_u64_ms(totals.auto_review),
|
||||
user_wait_ms: duration_to_u64_ms(totals.user_wait),
|
||||
tools_ms: duration_to_u64_ms(totals.tools),
|
||||
overhead_ms: duration_to_u64_ms(totals.overhead),
|
||||
unknown_ms: duration_to_u64_ms(totals.unknown),
|
||||
})
|
||||
}
|
||||
@@ -245,6 +245,7 @@ fn turn_profile_breaks_down_sampling_blocking_and_retry_overhead() {
|
||||
assert_eq!(
|
||||
state.complete(started_at + Duration::from_millis(1_300)),
|
||||
TurnProfile {
|
||||
critical_path: codex_analytics::TurnCriticalPath::Disabled,
|
||||
before_first_sampling_ms: 100,
|
||||
sampling_ms: 700,
|
||||
compaction_ms: 0,
|
||||
@@ -263,6 +264,7 @@ fn turn_profile_counts_compaction_as_an_exclusive_phase() {
|
||||
let started_at = Instant::now();
|
||||
let mut state = TurnProfileState {
|
||||
path: Some(CriticalPath::default()),
|
||||
capture_enabled: true,
|
||||
..Default::default()
|
||||
};
|
||||
state.start(started_at);
|
||||
@@ -282,6 +284,17 @@ fn turn_profile_counts_compaction_as_an_exclusive_phase() {
|
||||
assert_eq!(
|
||||
state.complete(started_at + Duration::from_millis(1_100)),
|
||||
TurnProfile {
|
||||
critical_path: codex_analytics::TurnCriticalPath::Complete(
|
||||
codex_analytics::TurnCriticalPathDurations {
|
||||
duration_ms: 1100,
|
||||
inference_ms: 800,
|
||||
auto_review_ms: 0,
|
||||
user_wait_ms: 0,
|
||||
tools_ms: 0,
|
||||
overhead_ms: 300,
|
||||
unknown_ms: 0,
|
||||
}
|
||||
),
|
||||
before_first_sampling_ms: 200,
|
||||
sampling_ms: 300,
|
||||
compaction_ms: 500,
|
||||
@@ -310,7 +323,22 @@ async fn turn_profile_and_duration_share_a_completion_instant() {
|
||||
.mark_turn_started(Instant::now() - Duration::from_millis(100))
|
||||
.await;
|
||||
|
||||
let (_, duration_ms, profile) = state.complete_profile_and_duration_ms().await;
|
||||
state.enable_critical_path();
|
||||
let result = state.complete_profile_and_duration_ms().await;
|
||||
assert_eq!(state.complete_profile_and_duration_ms().await, result);
|
||||
let (_, duration_ms, profile) = result;
|
||||
assert_eq!(
|
||||
profile.critical_path,
|
||||
codex_analytics::TurnCriticalPath::Complete(codex_analytics::TurnCriticalPathDurations {
|
||||
duration_ms: duration_ms.unwrap() as u64,
|
||||
inference_ms: 0,
|
||||
auto_review_ms: 0,
|
||||
user_wait_ms: 0,
|
||||
tools_ms: 0,
|
||||
overhead_ms: duration_ms.unwrap() as u64,
|
||||
unknown_ms: 0,
|
||||
})
|
||||
);
|
||||
let classified_ms = profile.before_first_sampling_ms
|
||||
+ profile.sampling_ms
|
||||
+ profile.compaction_ms
|
||||
@@ -331,6 +359,7 @@ fn capture_preserves_phases_and_freezes_missing_dependencies() {
|
||||
let mut state = TurnProfileState::default();
|
||||
state.start(started_at);
|
||||
assert!(state.path_at(time_after_turn_start(10)).is_none());
|
||||
state.capture_enabled = true;
|
||||
state.path = Some(CriticalPath::default());
|
||||
assert!(state.begin_sampling(time_after_turn_start(10)));
|
||||
state.end_phase(time_after_turn_start(20), TurnProfilePhase::Sampling);
|
||||
@@ -350,6 +379,18 @@ fn capture_preserves_phases_and_freezes_missing_dependencies() {
|
||||
..PathDurations::default()
|
||||
}
|
||||
);
|
||||
assert_eq!(
|
||||
profile.critical_path,
|
||||
codex_analytics::TurnCriticalPath::Complete(codex_analytics::TurnCriticalPathDurations {
|
||||
duration_ms: 60,
|
||||
inference_ms: 10,
|
||||
auto_review_ms: 0,
|
||||
user_wait_ms: 0,
|
||||
tools_ms: 0,
|
||||
overhead_ms: 20,
|
||||
unknown_ms: 30,
|
||||
})
|
||||
);
|
||||
let frozen_path = state.path.clone();
|
||||
state.end_phase(time_after_turn_start(70), TurnProfilePhase::ToolBlocking);
|
||||
assert_eq!(state.complete(time_after_turn_start(80)), profile);
|
||||
@@ -363,7 +404,11 @@ async fn invalidation_and_late_guard_drop_cannot_restore_a_frozen_path() {
|
||||
timing.mark_turn_started(Instant::now()).await;
|
||||
let guard = timing.begin_tool_blocking(Some(ToolBatchTiming::default()));
|
||||
timing.invalidate_critical_path();
|
||||
timing.complete_profile_and_duration_ms().await;
|
||||
let (_, _, profile) = timing.complete_profile_and_duration_ms().await;
|
||||
assert_eq!(
|
||||
profile.critical_path,
|
||||
codex_analytics::TurnCriticalPath::Incomplete
|
||||
);
|
||||
timing.enable_critical_path();
|
||||
drop(guard);
|
||||
assert!(timing.snapshot_path(Instant::now()).is_none());
|
||||
@@ -412,3 +457,30 @@ async fn completed_batch_guard_retains_its_path_across_turns() {
|
||||
assert_eq!(totals.unknown, Duration::ZERO);
|
||||
assert!(totals.tools >= Duration::from_secs(1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn critical_path_report_retains_fractional_millisecond_rounding_and_compaction() {
|
||||
let start = Instant::now();
|
||||
let mut state = TurnProfileState {
|
||||
capture_enabled: true,
|
||||
path: Some(CriticalPath::default()),
|
||||
..Default::default()
|
||||
};
|
||||
state.start(start);
|
||||
state.begin_compaction(start);
|
||||
let end = start + Duration::from_nanos(1_999_999);
|
||||
state.end_phase(end, TurnProfilePhase::Compaction);
|
||||
let profile = state.complete(end);
|
||||
assert_eq!(
|
||||
profile.critical_path,
|
||||
codex_analytics::TurnCriticalPath::Complete(codex_analytics::TurnCriticalPathDurations {
|
||||
duration_ms: 1,
|
||||
inference_ms: 1,
|
||||
auto_review_ms: 0,
|
||||
user_wait_ms: 0,
|
||||
tools_ms: 0,
|
||||
overhead_ms: 0,
|
||||
unknown_ms: 0,
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user