diff --git a/codex-rs/analytics/src/tests/suite/turns/events_tests.rs b/codex-rs/analytics/src/tests/suite/turns/events_tests.rs index 5336744bbe..8f2bd1be7a 100644 --- a/codex-rs/analytics/src/tests/suite/turns/events_tests.rs +++ b/codex-rs/analytics/src/tests/suite/turns/events_tests.rs @@ -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)); } diff --git a/codex-rs/analytics/src/turn_critical_path.rs b/codex-rs/analytics/src/turn_critical_path.rs index 4d10fca22c..649a2f4f25 100644 --- a/codex-rs/analytics/src/turn_critical_path.rs +++ b/codex-rs/analytics/src/turn_critical_path.rs @@ -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, 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, critical_path_compaction_ms: Option, critical_path_auto_review_ms: Option, + critical_path_auto_review_overhead_ms: Option, critical_path_user_wait_ms: Option, critical_path_tools_ms: Option, critical_path_overhead_ms: Option, @@ -70,6 +73,8 @@ impl From 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), diff --git a/codex-rs/app-server/tests/suite/v2/critical_path_analytics_tests.rs b/codex-rs/app-server/tests/suite/v2/critical_path_analytics_tests.rs index d681b72a87..e47dcdfba9 100644 --- a/codex-rs/app-server/tests/suite/v2/critical_path_analytics_tests.rs +++ b/codex-rs/app-server/tests/suite/v2/critical_path_analytics_tests.rs @@ -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) diff --git a/codex-rs/core/src/guardian/decision.rs b/codex-rs/core/src/guardian/decision.rs index ad3dd37105..38477da6f2 100644 --- a/codex-rs/core/src/guardian/decision.rs +++ b/codex-rs/core/src/guardian/decision.rs @@ -50,7 +50,7 @@ pub(crate) fn decide_approval( reasons: ApprovalRequestReasons, mut options: GuardianReviewOptions, ) -> futures::future::BoxFuture<'static, Option> { - 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; diff --git a/codex-rs/core/src/guardian/mod.rs b/codex-rs/core/src/guardian/mod.rs index 34f3be4fb4..855a899016 100644 --- a/codex-rs/core/src/guardian/mod.rs +++ b/codex-rs/core/src/guardian/mod.rs @@ -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>, turn: Arc, + review_execution: Option, environments: TurnEnvironmentSnapshot, project_instructions: Option>, // 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::(), environments: environments.clone(), project_instructions: None, @@ -162,6 +164,7 @@ impl From<&Arc> for GuardianReviewContext { fn from(step: &Arc) -> Self { Self { serving_assessment: step.serving_assessment.clone(), + review_execution: None, parent_response: step.turn.extension_data.get::(), turn: Arc::clone(&step.turn), environments: step.environments.clone(), @@ -180,6 +183,7 @@ impl From> for GuardianReviewContext { fn from(turn: Arc) -> Self { Self { serving_assessment: None, + review_execution: None, parent_response: turn.extension_data.get::(), environments: turn.initial_environments.clone(), project_instructions: None, diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 3c07629cfc..a1a92a32e1 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -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::(); + 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) = ¶ms.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::(); + // Finalizing input consumes this context; failures before admission leave it pending. let transcript_admitted = review_session .session diff --git a/codex-rs/core/src/guardian/review_session_tests.rs b/codex-rs/core/src/guardian/review_session_tests.rs index fcaa4b1d5d..7d211f24eb 100644 --- a/codex-rs/core/src/guardian/review_session_tests.rs +++ b/codex-rs/core/src/guardian/review_session_tests.rs @@ -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, + ¶ms, + 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::(); + 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::() + .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)) + ); +} diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 5e1774ff14..ea5b243107 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -315,6 +315,11 @@ impl Session { let task: Arc = 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 diff --git a/codex-rs/core/src/tasks/mod_tests.rs b/codex-rs/core/src/tasks/mod_tests.rs index 64c9fff2e3..701607eb46 100644 --- a/codex-rs/core/src/tasks/mod_tests.rs +++ b/codex-rs/core/src/tasks/mod_tests.rs @@ -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, diff --git a/codex-rs/core/src/turn_timing.rs b/codex-rs/core/src/turn_timing.rs index ddac4878fa..f666d7fcc7 100644 --- a/codex-rs/core/src/turn_timing.rs +++ b/codex-rs/core/src/turn_timing.rs @@ -95,6 +95,7 @@ pub(crate) struct TurnTimingState { #[derive(Debug, Default)] struct TurnTimingStateInner { + review_execution: Option, started_at: Option, started_at_unix_secs: Option, 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"] diff --git a/codex-rs/core/src/turn_timing/report.rs b/codex-rs/core/src/turn_timing/report.rs index 21703e114b..1928c44a53 100644 --- a/codex-rs/core/src/turn_timing/report.rs +++ b/codex-rs/core/src/turn_timing/report.rs @@ -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), diff --git a/codex-rs/core/src/turn_timing/review_execution.rs b/codex-rs/core/src/turn_timing/review_execution.rs new file mode 100644 index 0000000000..08df4ead02 --- /dev/null +++ b/codex-rs/core/src/turn_timing/review_execution.rs @@ -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, + execution: Option, + attempt: Option, +} + +impl ReviewExecutionObservation { + pub(crate) fn install(session: Arc, execution: Option) -> 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::() + .is_some_and(|pending| pending.0 == *execution) + { + self.session + .services + .thread_extension_data + .remove::(); + } + } +} + +pub(crate) async fn observe_pending_review_execution(session: &Session, timing: &TurnTimingState) { + if let Some(execution) = session + .services + .thread_extension_data + .remove::() + { + timing.observe_review_execution(execution.0.clone()).await; + } +} diff --git a/codex-rs/core/src/turn_timing_tests.rs b/codex-rs/core/src/turn_timing_tests.rs index d3cca5cb81..10a7cb927e 100644 --- a/codex-rs/core/src/turn_timing_tests.rs +++ b/codex-rs/core/src/turn_timing_tests.rs @@ -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)); +} diff --git a/codex-rs/direct-tool-timing/src/direct_tool_timing_tests.rs b/codex-rs/direct-tool-timing/src/direct_tool_timing_tests.rs index 4fe8308f2a..e5f00260b2 100644 --- a/codex-rs/direct-tool-timing/src/direct_tool_timing_tests.rs +++ b/codex-rs/direct-tool-timing/src/direct_tool_timing_tests.rs @@ -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)); +} diff --git a/codex-rs/direct-tool-timing/src/lib.rs b/codex-rs/direct-tool-timing/src/lib.rs index 193ff6a66a..1f43976471 100644 --- a/codex-rs/direct-tool-timing/src/lib.rs +++ b/codex-rs/direct-tool-timing/src/lib.rs @@ -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; diff --git a/codex-rs/direct-tool-timing/src/recorder.rs b/codex-rs/direct-tool-timing/src/recorder.rs index 22af5b8d9e..e3d23e511a 100644 --- a/codex-rs/direct-tool-timing/src/recorder.rs +++ b/codex-rs/direct-tool-timing/src/recorder.rs @@ -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, segment_started_at: Instant, - active_auto_reviews: usize, + active_auto_reviews: Vec, 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, 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 { + 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); + } + } } } } diff --git a/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs b/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs index c8b66c88e7..07cb9ee053 100644 --- a/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs +++ b/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs @@ -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",