Report auto-review overhead in turn critical-path analytics (#59260)

## What changed

Track Guardian compaction and review turn execution within each tool's auto-review interval, and expose `critical_path_auto_review_overhead_ms` in turn analytics. This measures the portion of `auto_review_ms` outside observed Guardian turn execution; it is a subset, not an additional duration.

Freeze review observations when the review or tool invocation ends. Report unknown overhead as `null`, including when denial reconsideration runs an unobserved follow-up turn.

## Testing

Add coverage for shared compaction and review observations, cached reviews, unfinished reviews, reused tool call IDs, and execution timing with child critical-path capture enabled or disabled. Extend analytics and app-server tests to check the new field.

GitOrigin-RevId: a63dec2a75571a3167c8c3f0c1d21e11e4a8d050
This commit is contained in:
taohong-oai
2026-11-03 23:49:01 +00:00
committed by copyberry
parent 444359fa2d
commit db14d7508d
17 changed files with 437 additions and 24 deletions
@@ -194,6 +194,7 @@ fn turn_event_serializes_expected_shape() {
"critical_path_sampling_ms": null,
"critical_path_compaction_ms": null,
"critical_path_auto_review_ms": null,
"critical_path_auto_review_overhead_ms": null,
"critical_path_user_wait_ms": null,
"critical_path_tools_ms": null,
"critical_path_overhead_ms": null,
@@ -1010,6 +1011,7 @@ async fn critical_path_payload_preserves_capture_status_and_emits_once() {
sampling_ms: 500,
compaction_ms: 200,
auto_review_ms: 200,
auto_review_overhead_ms: Some(40),
user_wait_ms: 100,
tools_ms: 150,
overhead_ms: 80,
@@ -1024,6 +1026,7 @@ async fn critical_path_payload_preserves_capture_status_and_emits_once() {
sampling_ms: 1234,
compaction_ms: 0,
auto_review_ms: 0,
auto_review_overhead_ms: Some(0),
user_wait_ms: 0,
tools_ms: 0,
overhead_ms: 0,
@@ -1044,6 +1047,10 @@ async fn critical_path_payload_preserves_capture_status_and_emits_once() {
)
.await;
let mut profile = sample_turn_profile();
let expected_review_overhead = match &critical_path {
TurnCriticalPath::Complete(value) => value.auto_review_overhead_ms,
_ => None,
};
profile.critical_path = critical_path;
reducer
.ingest(
@@ -1087,6 +1094,10 @@ async fn critical_path_payload_preserves_capture_status_and_emits_once() {
]),
expected
);
assert_eq!(
params.get("critical_path_auto_review_overhead_ms"),
Some(&json!(expected_review_overhead))
);
assert_eq!(params["tool_blocking_ms"], json!(250));
assert_eq!(params["duration_ms"], json!(1234));
}
@@ -40,6 +40,8 @@ pub struct TurnCriticalPathDurations {
pub sampling_ms: u64,
pub compaction_ms: u64,
pub auto_review_ms: u64,
/// Portion of auto_review_ms outside observed Guardian turn execution; not additive.
pub auto_review_overhead_ms: Option<u64>,
pub user_wait_ms: u64,
pub tools_ms: u64,
pub overhead_ms: u64,
@@ -54,6 +56,7 @@ pub(crate) struct CriticalPathEventParams {
critical_path_sampling_ms: Option<u64>,
critical_path_compaction_ms: Option<u64>,
critical_path_auto_review_ms: Option<u64>,
critical_path_auto_review_overhead_ms: Option<u64>,
critical_path_user_wait_ms: Option<u64>,
critical_path_tools_ms: Option<u64>,
critical_path_overhead_ms: Option<u64>,
@@ -70,6 +73,8 @@ impl From<TurnCriticalPath> for CriticalPathEventParams {
critical_path_sampling_ms: durations.map(|value| value.sampling_ms),
critical_path_compaction_ms: durations.map(|value| value.compaction_ms),
critical_path_auto_review_ms: durations.map(|value| value.auto_review_ms),
critical_path_auto_review_overhead_ms: durations
.and_then(|value| value.auto_review_overhead_ms),
critical_path_user_wait_ms: durations.map(|value| value.user_wait_ms),
critical_path_tools_ms: durations.map(|value| value.tools_ms),
critical_path_overhead_ms: durations.map(|value| value.overhead_ms),
@@ -147,6 +147,10 @@ async fn critical_path_reaches_turn_event(finish: Finish, capture: Capture) -> R
];
if matches!(capture, Capture::Enabled) {
assert_eq!(p["critical_path_status"], "complete");
assert_eq!(
p.get("critical_path_auto_review_overhead_ms"),
Some(&json!(0))
);
let inference = p["critical_path_inference_ms"]
.as_u64()
.expect("inference duration");
@@ -181,6 +185,10 @@ async fn critical_path_reaches_turn_event(finish: Finish, capture: Capture) -> R
);
} else {
assert_eq!(p["critical_path_status"], "disabled");
assert_eq!(
p.get("critical_path_auto_review_overhead_ms"),
Some(&serde_json::Value::Null)
);
assert_eq!(
p.get("critical_path_inference_ms"),
Some(&serde_json::Value::Null)
+4 -1
View File
@@ -50,7 +50,7 @@ pub(crate) fn decide_approval(
reasons: ApprovalRequestReasons,
mut options: GuardianReviewOptions,
) -> futures::future::BoxFuture<'static, Option<ReviewDecision>> {
let context: GuardianReviewContext = context.into();
let mut context: GuardianReviewContext = context.into();
let request: ReviewAction = request.into();
// Bind the invocation before scheduling. Async review cleanup can outlive
// the decision, but must not extend the time blocking this tool.
@@ -90,6 +90,9 @@ pub(crate) fn decide_approval(
} else {
None
};
context.review_execution = review_timing
.as_ref()
.and_then(codex_direct_tool_timing::ToolBlockingTimingGuard::review_execution);
Box::pin(async move {
let _review_timing = review_timing;
let (_, history_reset) = session.history_reset().await;
+4
View File
@@ -95,6 +95,7 @@ pub(crate) struct GuardianReviewContext {
/// The latest response ID and its connection affinity when review was requested.
pub(crate) parent_response: Option<Arc<ParentResponse>>,
turn: Arc<TurnContext>,
review_execution: Option<codex_direct_tool_timing::ReviewExecution>,
environments: TurnEnvironmentSnapshot,
project_instructions: Option<Arc<LoadedAgentsMd>>,
// Model and reasoning inputs are carried for the follow-up Guardian and V2 migrations.
@@ -136,6 +137,7 @@ impl GuardianReviewContext {
) -> Self {
Self {
serving_assessment: None,
review_execution: None,
parent_response: turn.extension_data.get::<ParentResponse>(),
environments: environments.clone(),
project_instructions: None,
@@ -162,6 +164,7 @@ impl From<&Arc<StepContext>> for GuardianReviewContext {
fn from(step: &Arc<StepContext>) -> Self {
Self {
serving_assessment: step.serving_assessment.clone(),
review_execution: None,
parent_response: step.turn.extension_data.get::<ParentResponse>(),
turn: Arc::clone(&step.turn),
environments: step.environments.clone(),
@@ -180,6 +183,7 @@ impl From<Arc<TurnContext>> for GuardianReviewContext {
fn from(turn: Arc<TurnContext>) -> Self {
Self {
serving_assessment: None,
review_execution: None,
parent_response: turn.extension_data.get::<ParentResponse>(),
environments: turn.initial_environments.clone(),
project_instructions: None,
@@ -334,6 +334,11 @@ async fn run_review_on_session(
false
};
if should_compact {
let mut execution_attempt =
crate::turn_timing::review_execution::ReviewExecutionObservation::install(
Arc::clone(&review_session.session),
params.parent_context.review_execution.clone(),
);
let compact_submission = run_before_review_deadline(
deadline,
params.external_cancel.as_ref(),
@@ -369,6 +374,7 @@ async fn run_review_on_session(
&mut analytics_result,
)
.await;
execution_attempt.finish(result.turn_completed);
if !matches!(
result.outcome,
GuardianReviewSessionOutcome::Completed(Ok(_))
@@ -572,6 +578,11 @@ async fn run_review_on_session(
.insert(super::PendingReviewRouting(
params.parent_context.parent_response.clone(),
));
let mut execution_attempt =
crate::turn_timing::review_execution::ReviewExecutionObservation::install(
Arc::clone(&review_session.session),
params.parent_context.review_execution.clone(),
);
let request = codex_guardian_reviewer::ReviewerTurn {
items,
environments: codex_protocol::protocol::TurnEnvironmentSelections::new(
@@ -625,6 +636,7 @@ async fn run_review_on_session(
.services
.thread_extension_data
.remove::<super::PendingReviewRouting>();
review_session
.session
.services
@@ -653,8 +665,19 @@ async fn run_review_on_session(
&mut analytics_result,
)
.await;
execution_attempt.finish(turn_result.turn_completed);
let turn_result = match reconsideration {
Some(request) => {
if let GuardianReviewSessionOutcome::Completed(Ok(Some(response))) =
&turn_result.outcome
&& response.assessment().is_ok_and(|assessment| {
assessment.outcome == codex_protocol::protocol::GuardianAssessmentOutcome::Deny
})
&& let Some(execution) = &params.parent_context.review_execution
{
// Reconsideration's second turn does not yet have an execution observer.
execution.mark_unknown();
}
// The new turn does not own the first turn's retained evidence backup.
// Keep that evidence intact; an over-budget follow-up keeps the denial.
let store = &review_session.session.services.thread_extension_data;
@@ -684,6 +707,7 @@ async fn run_review_on_session(
.services
.thread_extension_data
.remove::<super::PendingReviewRouting>();
// Finalizing input consumes this context; failures before admission leave it pending.
let transcript_admitted = review_session
.session
@@ -1126,3 +1126,122 @@ async fn parent_checkpoint_recovery_stops_after_one_restart() {
assert_eq!(observed_spawns.load(Ordering::SeqCst), 2);
assert!(pool.trunk().await.is_none());
}
#[test_case::test_case(None; "single_review")]
#[test_case::test_case(Some(r#"{"outcome":"allow"}"#); "reconsideration_allows")]
#[test_case::test_case(Some("invalid assessment"); "reconsideration_keeps_denial")]
#[tokio::test]
async fn compaction_and_review_submissions_share_execution_observer(
reconsidered_assessment: Option<&str>,
) {
let (review_session, events, submissions) = test_review_session().await;
let child_session = Arc::clone(&review_session.session);
let mut params = test_review_params().await;
params
.spawn_config
.features
.enable(Feature::TokenBudget)
.unwrap();
params.spawn_config.model_auto_compact_token_limit = Some(0);
params.spawn_config.token_budget = None;
if reconsidered_assessment.is_some() {
params
.spawn_config
.features
.enable(Feature::GuardianDenialReconsideration)
.unwrap();
}
let turn = Arc::get_mut(&mut params.parent_context.turn).expect("unshared turn fixture");
turn.configured_token_budget = None;
turn.use_model_token_budget_defaults = false;
let execution = codex_direct_tool_timing::ReviewExecution::default();
execution.observe();
params.parent_context.review_execution = Some(execution.clone());
let review = tokio::spawn(async move {
run_review_on_session(
&review_session,
&params,
GuardianReviewSessionKind::TrunkReused,
codex_guardian_reviewer::ReviewDeadline::At(
tokio::time::Instant::now() + Duration::from_secs(30),
),
)
.await
});
let start = std::time::Instant::now();
let mut child_intervals = vec![(10, 30), (40, 70)];
if reconsidered_assessment.is_some() {
child_intervals.push((75, 90));
}
for (index, (left, right)) in child_intervals.into_iter().enumerate() {
let submission = tokio::time::timeout(Duration::from_secs(10), submissions.recv())
.await
.expect("child submission deadline")
.expect("child submission");
if index == 0 {
assert!(matches!(submission.op, Op::Compact));
} else {
let Op::TurnInput { reply, .. } = submission.op else {
panic!("expected review input after compaction");
};
reply
.send(Ok(TurnInputSubmission::Started {
root_turn_id: submission.id.clone(),
turn_id: submission.id.clone(),
}))
.unwrap();
}
let pending = child_session
.services
.thread_extension_data
.remove::<crate::turn_timing::review_execution::PendingReviewExecution>();
if index < 2 {
let pending = pending.expect("instrumented child");
assert_eq!(pending.0, execution);
let child_start = start + Duration::from_millis(left);
pending.0.started(child_start);
pending
.0
.completed(child_start, start + Duration::from_millis(right));
} else {
assert!(pending.is_none());
}
let assessment = match (index, reconsidered_assessment) {
(1, Some(_)) => r#"{"outcome":"deny"}"#,
(2, Some(assessment)) => assessment,
_ => "complete",
};
events
.send(turn_complete_event(
&submission.id,
Some(assessment),
/*time_to_first_token_ms*/ None,
))
.await
.unwrap();
}
let result = review.await.unwrap();
assert!(matches!(
result.outcome,
GuardianReviewSessionOutcome::Completed(Ok(_))
));
assert!(
child_session
.services
.thread_extension_data
.get::<crate::turn_timing::review_execution::PendingReviewExecution>()
.is_none()
);
execution.seal();
let end = start + Duration::from_millis(100);
let totals = codex_utils_critical_path::CriticalPath::default()
.auto_review(start, end, vec![execution])
.durations(start, end);
assert_eq!(totals.auto_review, Duration::from_millis(100));
assert_eq!(
totals.auto_review_overhead,
reconsidered_assessment
.is_none()
.then_some(Duration::from_millis(50))
);
}
+5
View File
@@ -315,6 +315,11 @@ impl Session {
let task: Arc<dyn AnySessionTask> = Arc::new(task);
let task_kind = task.kind();
let span_name = task.span_name();
crate::turn_timing::review_execution::observe_pending_review_execution(
self,
&turn_context.turn_timing_state,
)
.await;
let started_at = Instant::now();
let turn_started_at_unix_ms = turn_context
.turn_timing_state
+1
View File
@@ -678,6 +678,7 @@ async fn turn_critical_path_metrics_count_each_capture_once() {
sampling_ms: 31,
compaction_ms: 0,
auto_review_ms: 2,
auto_review_overhead_ms: None,
tools_ms: 10,
user_wait_ms: 0,
overhead_ms: 14,
+15
View File
@@ -95,6 +95,7 @@ pub(crate) struct TurnTimingState {
#[derive(Debug, Default)]
struct TurnTimingStateInner {
review_execution: Option<codex_direct_tool_timing::ReviewExecution>,
started_at: Option<Instant>,
started_at_unix_secs: Option<i64>,
completed_at: Option<(Instant, i64)>,
@@ -164,9 +165,19 @@ impl TurnTimingState {
}
}
pub(crate) async fn observe_review_execution(
&self,
execution: codex_direct_tool_timing::ReviewExecution,
) {
self.state.lock().await.review_execution = Some(execution);
}
pub(crate) async fn mark_turn_started(&self, started_at: Instant) -> i64 {
let started_at_unix_ms = now_unix_timestamp_ms();
let mut state = self.state.lock().await;
if let Some(execution) = &state.review_execution {
execution.started(started_at);
}
state.started_at = Some(started_at);
state.completed_at = None;
state.started_at_unix_secs = Some(started_at_unix_ms / 1000);
@@ -211,6 +222,9 @@ impl TurnTimingState {
)
.unwrap_or(i64::MAX)
});
if let (Some(execution), Some(started_at)) = (&state.review_execution, state.started_at) {
execution.completed(started_at, completed_at_instant);
}
let profile = self.profile_state().complete(completed_at_instant);
(Some(completed_at), duration_ms, profile)
}
@@ -567,6 +581,7 @@ fn response_item_records_turn_ttft(item: &ResponseItem) -> bool {
}
mod report;
pub(crate) mod review_execution;
#[cfg(test)]
#[path = "turn_timing_tests.rs"]
+1
View File
@@ -23,6 +23,7 @@ pub(super) fn completed_critical_path(state: &TurnProfileState, end: Instant) ->
sampling_ms: duration_to_u64_ms(totals.sampling),
compaction_ms: duration_to_u64_ms(totals.compaction),
auto_review_ms: duration_to_u64_ms(totals.auto_review),
auto_review_overhead_ms: totals.auto_review_overhead.map(duration_to_u64_ms),
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),
@@ -0,0 +1,67 @@
//! Attaches submitted review execution to its child turn before measurement starts.
use super::TurnTimingState;
use crate::session::session::Session;
use codex_utils_critical_path::ReviewExecution;
use codex_utils_critical_path::ReviewExecutionAttempt;
use std::sync::Arc;
pub(crate) struct PendingReviewExecution(pub(crate) ReviewExecution);
/// Keeps a submission unresolved until completion and removes only its pending observer.
pub(crate) struct ReviewExecutionObservation {
session: Arc<Session>,
execution: Option<ReviewExecution>,
attempt: Option<ReviewExecutionAttempt>,
}
impl ReviewExecutionObservation {
pub(crate) fn install(session: Arc<Session>, execution: Option<ReviewExecution>) -> Self {
let attempt = execution.as_ref().map(ReviewExecution::start_attempt);
if let Some(execution) = &execution {
session
.services
.thread_extension_data
.insert(PendingReviewExecution(execution.clone()));
}
Self {
session,
execution,
attempt,
}
}
pub(crate) fn finish(&mut self, completed: bool) {
if let Some(attempt) = self.attempt.take() {
attempt.finish(completed);
}
}
}
impl Drop for ReviewExecutionObservation {
fn drop(&mut self) {
if let Some(execution) = &self.execution
&& self
.session
.services
.thread_extension_data
.get::<PendingReviewExecution>()
.is_some_and(|pending| pending.0 == *execution)
{
self.session
.services
.thread_extension_data
.remove::<PendingReviewExecution>();
}
}
}
pub(crate) async fn observe_pending_review_execution(session: &Session, timing: &TurnTimingState) {
if let Some(execution) = session
.services
.thread_extension_data
.remove::<PendingReviewExecution>()
{
timing.observe_review_execution(execution.0.clone()).await;
}
}
+33
View File
@@ -291,6 +291,7 @@ fn turn_profile_counts_compaction_as_an_exclusive_phase() {
sampling_ms: 300,
compaction_ms: 500,
auto_review_ms: 0,
auto_review_overhead_ms: Some(0),
user_wait_ms: 0,
tools_ms: 0,
overhead_ms: 300,
@@ -338,6 +339,7 @@ async fn turn_profile_and_duration_share_a_completion_instant() {
sampling_ms: 0,
compaction_ms: 0,
auto_review_ms: 0,
auto_review_overhead_ms: Some(0),
user_wait_ms: 0,
tools_ms: 0,
overhead_ms: duration_ms.unwrap() as u64,
@@ -393,6 +395,7 @@ fn capture_preserves_phases_and_freezes_missing_dependencies() {
sampling_ms: 10,
compaction_ms: 0,
auto_review_ms: 0,
auto_review_overhead_ms: None,
user_wait_ms: 0,
tools_ms: 0,
overhead_ms: 20,
@@ -490,6 +493,7 @@ fn critical_path_report_retains_fractional_millisecond_rounding_and_compaction()
sampling_ms: 0,
compaction_ms: 1,
auto_review_ms: 0,
auto_review_overhead_ms: Some(0),
user_wait_ms: 0,
tools_ms: 0,
overhead_ms: 0,
@@ -497,3 +501,32 @@ fn critical_path_report_retains_fractional_millisecond_rounding_and_compaction()
})
);
}
#[test_case::test_case(false; "child_capture_disabled")]
#[test_case::test_case(true; "child_capture_enabled")]
#[tokio::test]
async fn review_execution_uses_frozen_profile_window(capture_enabled: bool) {
let timing = super::TurnTimingState::default();
let execution = codex_direct_tool_timing::ReviewExecution::default();
execution.observe();
timing.observe_review_execution(execution.clone()).await;
if capture_enabled {
timing.enable_critical_path();
}
let start = Instant::now() - Duration::from_millis(25);
timing.mark_turn_started(start).await;
let first = timing.complete_profile_and_duration_ms().await;
let end = timing.state.lock().await.completed_at.unwrap().0;
assert_eq!(timing.complete_profile_and_duration_ms().await, first);
execution.seal();
let review_start = start - Duration::from_millis(100);
let review_end = end + Duration::from_millis(200);
let path = CriticalPath::default().auto_review(review_start, review_end, vec![execution]);
let totals = path.durations(review_start, review_end);
assert_eq!(
totals.auto_review_overhead,
Some(Duration::from_millis(300))
);
assert_eq!(totals.auto_review, review_end.duration_since(review_start));
assert_eq!(first.1, Some(end.duration_since(start).as_millis() as i64));
}
@@ -27,9 +27,9 @@ fn parallel_winner_retains_its_review_and_shared_initial_path() {
let mut state = timings.state();
let work = state.unique_work("winner").unwrap();
work.record_until(start + Duration::from_secs(2));
work.active_auto_reviews = 1;
work.active_auto_reviews = vec![ReviewExecution::default()];
work.record_until(start + Duration::from_secs(5));
work.active_auto_reviews = 0;
work.active_auto_reviews.clear();
}
drop(losing);
assert!(batch.completed().is_none());
@@ -65,7 +65,8 @@ fn duplicate_ids_and_late_review_drop_do_not_complete_another_invocation() {
.state()
.unique_work("same")
.unwrap()
.active_auto_reviews,
.active_auto_reviews
.len(),
0
);
assert!(batch.completed().is_none());
@@ -165,7 +166,7 @@ fn human_wait_takes_precedence_over_reviews_and_overlapping_waits() {
let mut state = timings.state();
let work = state.unique_work("call").unwrap();
work.record_until(at(1));
work.active_auto_reviews = 1;
work.active_auto_reviews = vec![ReviewExecution::default()];
work.record_until(at(2));
work.active_user_waits = 1;
work.record_until(at(3));
@@ -175,7 +176,7 @@ fn human_wait_takes_precedence_over_reviews_and_overlapping_waits() {
work.record_until(at(5));
work.active_user_waits = 0;
work.record_until(at(6));
work.active_auto_reviews = 0;
work.active_auto_reviews.clear();
work.record_until(at(7));
assert_eq!(
work.path.as_ref().unwrap().durations(start, at(7)),
@@ -238,3 +239,79 @@ fn user_wait_guards_balance_and_do_not_follow_reused_call_ids() {
drop(next);
assert!(batch.completed().is_some());
}
#[test]
fn observed_cached_review_is_all_overhead_and_human_wait_has_no_observer() {
let timings = DirectToolTimings::default();
let batch = ToolBatchTiming::default();
let start = Instant::now();
let invocation =
timings.start_direct_call("cached", batch, start, Some(CriticalPath::default()));
let first = invocation.start_review().unwrap();
let second = invocation.start_review().unwrap();
let first_observer = first.review_execution().unwrap();
let second_observer = second.review_execution().unwrap();
assert_ne!(first_observer, second_observer);
let human = invocation.start_user_wait().unwrap();
assert!(human.review_execution().is_none());
drop(first);
assert_eq!(
timings
.state()
.unique_work("cached")
.unwrap()
.active_auto_reviews,
vec![second_observer]
);
drop(human);
drop(second);
let completed = invocation.finish().unwrap();
let durations = completed.path.unwrap().durations(start, completed.ended_at);
assert_eq!(durations.auto_review_overhead, Some(durations.auto_review));
assert!(durations.auto_review > Duration::ZERO);
assert!(durations.user_wait > Duration::ZERO);
}
#[test]
fn invocation_finish_freezes_unfinished_review_and_id_reuse_gets_new_observation() {
let timings = DirectToolTimings::default();
let batch = ToolBatchTiming::default();
let start = Instant::now();
let first =
timings.start_direct_call("same", batch.clone(), start, Some(CriticalPath::default()));
let old_guard = first.start_review().unwrap();
let old_observer = old_guard.review_execution().unwrap();
let child_start = Instant::now();
old_observer.started(child_start);
let first_result = first.finish().unwrap();
let old_path = first_result.path.unwrap();
let old_durations = old_path.durations(start, first_result.ended_at);
assert_eq!(old_durations.auto_review_overhead, None);
let next_start = Instant::now();
let next = timings.start_direct_call("same", batch, next_start, Some(CriticalPath::default()));
let next_guard = next.start_review().unwrap();
let next_observer = next_guard.review_execution().unwrap();
assert_ne!(old_observer, next_observer);
old_observer.completed(child_start, Instant::now());
drop(old_guard);
assert_eq!(
old_path.durations(start, first_result.ended_at),
old_durations
);
assert_eq!(
timings
.state()
.unique_work("same")
.unwrap()
.active_auto_reviews,
vec![next_observer]
);
drop(next_guard);
let next_result = next.finish().unwrap();
let durations = next_result
.path
.unwrap()
.durations(next_start, next_result.ended_at);
assert_eq!(durations.auto_review_overhead, Some(durations.auto_review));
}
+2
View File
@@ -9,3 +9,5 @@ pub use recorder::ToolTimingGuard;
pub use recorder::AdmittedToolTiming;
pub use recorder::ToolAdmission;
pub use recorder::ToolGateTiming;
pub use codex_utils_critical_path::ReviewExecution;
+53 -16
View File
@@ -17,6 +17,7 @@ pub use runtime::ToolAdmission;
pub use runtime::ToolGateTiming;
use codex_utils_critical_path::CriticalPath;
use codex_utils_critical_path::ReviewExecution;
use codex_utils_critical_path::WorkKind;
/// Completion time and dependency history for one direct invocation.
@@ -89,7 +90,7 @@ struct DirectToolInvocation {
identity: Arc<()>,
path: Option<CriticalPath>,
segment_started_at: Instant,
active_auto_reviews: usize,
active_auto_reviews: Vec<ReviewExecution>,
active_user_waits: usize,
batch: ToolBatchTiming,
}
@@ -110,6 +111,7 @@ enum BlockingKind {
/// Records a blocking interval until dropped; user waits take precedence over reviews.
pub struct ToolBlockingTimingGuard {
kind: BlockingKind,
review: Option<ReviewExecution>,
identity: Arc<()>,
timings: DirectToolTimings,
call_id: String,
@@ -121,15 +123,22 @@ impl DirectToolInvocation {
// to the human wait; all remaining invocation time is elapsed tool work.
let kind = if self.active_user_waits > 0 {
WorkKind::UserWait
} else if self.active_auto_reviews > 0 {
} else if !self.active_auto_reviews.is_empty() {
WorkKind::AutoReview
} else {
WorkKind::Tools
};
self.path = self
.path
.take()
.map(|path| path.work(self.segment_started_at, now, kind));
self.path = self.path.take().map(|path| {
if matches!(kind, WorkKind::AutoReview) {
path.auto_review(
self.segment_started_at,
now,
self.active_auto_reviews.clone(),
)
} else {
path.work(self.segment_started_at, now, kind)
}
});
self.segment_started_at = now;
}
@@ -181,7 +190,7 @@ impl DirectToolTimings {
identity: Arc::clone(&identity),
path,
segment_started_at: started_at,
active_auto_reviews: 0,
active_auto_reviews: Vec::new(),
active_user_waits: 0,
batch,
});
@@ -255,6 +264,9 @@ impl DirectToolTimings {
}
let ended_at = Instant::now();
work.record_until(ended_at);
for review in &work.active_auto_reviews {
review.seal();
}
let completed = CompletedWork {
ended_at,
path: work.path,
@@ -292,12 +304,20 @@ impl ToolTimingGuard {
.iter_mut()
.find(|work| Arc::ptr_eq(&work.identity, &self.identity))?;
work.record_until(Instant::now());
match kind {
BlockingKind::AutoReview => work.active_auto_reviews += 1,
BlockingKind::UserWait => work.active_user_waits += 1,
}
let review = match kind {
BlockingKind::AutoReview => {
let review = ReviewExecution::default();
work.active_auto_reviews.push(review.clone());
Some(review)
}
BlockingKind::UserWait => {
work.active_user_waits += 1;
None
}
};
Some(ToolBlockingTimingGuard {
kind,
review,
identity: Arc::clone(&self.identity),
timings: self.timings.clone(),
call_id: self.call_id.clone(),
@@ -316,19 +336,36 @@ impl Drop for ToolTimingGuard {
}
}
impl ToolBlockingTimingGuard {
/// Returns the observer for this exact review; human waits have no observer.
/// Call before submitting child work, including when a cached review runs no child.
pub fn review_execution(&self) -> Option<ReviewExecution> {
let review = self.review.as_ref()?;
review.observe();
Some(review.clone())
}
}
impl Drop for ToolBlockingTimingGuard {
fn drop(&mut self) {
if let Some(review) = &self.review {
review.seal();
}
if let Some(calls) = self.timings.state().calls.get_mut(&self.call_id)
&& let Some(work) = calls
.iter_mut()
.find(|work| Arc::ptr_eq(&work.identity, &self.identity))
{
work.record_until(Instant::now());
let counter = match self.kind {
BlockingKind::AutoReview => &mut work.active_auto_reviews,
BlockingKind::UserWait => &mut work.active_user_waits,
};
*counter = counter.saturating_sub(1);
match self.kind {
BlockingKind::AutoReview => {
work.active_auto_reviews
.retain(|review| Some(review) != self.review.as_ref());
}
BlockingKind::UserWait => {
work.active_user_waits = work.active_user_waits.saturating_sub(1);
}
}
}
}
}
@@ -6,6 +6,7 @@ use super::super::ToolBatchTiming;
use super::*;
use codex_utils_critical_path::CriticalPath;
use codex_utils_critical_path::PathDurations;
use codex_utils_critical_path::ReviewExecution;
use codex_utils_critical_path::WorkKind;
use pretty_assertions::assert_eq;
use std::future::Future;
@@ -43,9 +44,9 @@ async fn lock_handoff_publishes_review_history_before_admitting_waiter(admission
let mut state = timings.state();
let work = state.unique_work("releasing").unwrap();
work.record_until(start + Duration::from_secs(2));
work.active_auto_reviews = 1;
work.active_auto_reviews = vec![ReviewExecution::default()];
work.record_until(start + Duration::from_secs(5));
work.active_auto_reviews = 0;
work.active_auto_reviews.clear();
}
let waiting_work = timings.start_turn_direct_call(
"waiting",