From 79dcb11c9016a71340064dc12cec210363948516 Mon Sep 17 00:00:00 2001 From: taohong-oai Date: Tue, 3 Nov 2026 22:45:22 +0000 Subject: [PATCH] Track automatic review overhead in critical-path durations (#59251) ## What changed Add `ReviewExecution` observers and an optional `auto_review_overhead` subcomponent to `PathDurations`. Compute overhead by subtracting the union of observed child execution intervals from automatic review time, clipping intervals to the selected path segments and requested window. Track concurrent and retried submissions, preserve unknown attribution for incomplete or unobserved execution, and allow observations to be sealed against late callbacks. ## Testing Add unit coverage for overlapping execution, window clipping, human waits, cached reviews, losing parallel paths, canceled submissions, and sealed observations. GitOrigin-RevId: 73ebe5ed9100a6598f28099ba46cc6d76578d8d9 --- codex-rs/core/src/turn_timing_tests.rs | 1 + .../src/direct_tool_timing_tests.rs | 2 + .../src/recorder/runtime_tests.rs | 1 + .../critical-path/src/critical_path_tests.rs | 245 ++++++++++++++++++ codex-rs/utils/critical-path/src/lib.rs | 54 +++- .../critical-path/src/review_execution.rs | 168 ++++++++++++ 6 files changed, 469 insertions(+), 2 deletions(-) create mode 100644 codex-rs/utils/critical-path/src/review_execution.rs diff --git a/codex-rs/core/src/turn_timing_tests.rs b/codex-rs/core/src/turn_timing_tests.rs index 7c8b3d633d..d3cca5cb81 100644 --- a/codex-rs/core/src/turn_timing_tests.rs +++ b/codex-rs/core/src/turn_timing_tests.rs @@ -380,6 +380,7 @@ fn capture_preserves_phases_and_freezes_missing_dependencies() { PathDurations { sampling: Duration::from_millis(10), unknown: Duration::from_millis(30), + auto_review_overhead: None, overhead: Duration::from_millis(20), ..PathDurations::default() } 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 7a6398774b..4fe8308f2a 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 @@ -40,6 +40,7 @@ fn parallel_winner_retains_its_review_and_shared_initial_path() { PathDurations { sampling: Duration::from_secs(1), auto_review: Duration::from_secs(3), + auto_review_overhead: None, tools: completed.ended_at.duration_since(start) - Duration::from_secs(4), ..PathDurations::default() } @@ -181,6 +182,7 @@ fn human_wait_takes_precedence_over_reviews_and_overlapping_waits() { PathDurations { tools: Duration::from_secs(2), auto_review: Duration::from_secs(2), + auto_review_overhead: None, user_wait: Duration::from_secs(3), ..PathDurations::default() } 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 b620fc9b88..c8b66c88e7 100644 --- a/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs +++ b/codex-rs/direct-tool-timing/src/recorder/runtime_tests.rs @@ -91,6 +91,7 @@ async fn lock_handoff_publishes_review_history_before_admitting_waiter(admission PathDurations { sampling: Duration::from_secs(1), auto_review: Duration::from_secs(3), + auto_review_overhead: None, tools: completed.ended_at.duration_since(start) - Duration::from_secs(4), ..PathDurations::default() } diff --git a/codex-rs/utils/critical-path/src/critical_path_tests.rs b/codex-rs/utils/critical-path/src/critical_path_tests.rs index 6cb8020a09..a4bd109d59 100644 --- a/codex-rs/utils/critical-path/src/critical_path_tests.rs +++ b/codex-rs/utils/critical-path/src/critical_path_tests.rs @@ -93,6 +93,7 @@ fn nested_serial_dependencies_preserve_exclusive_categories() { sampling: Duration::from_millis(100), compaction: Duration::from_millis(100), auto_review: Duration::from_millis(200), + auto_review_overhead: None, tools: Duration::from_millis(200), ..PathDurations::default() } @@ -132,6 +133,7 @@ fn path_retention_exhaustion_marks_unknown_instead_of_claiming_zero() { critical_path.durations(turn_start, turn_end), PathDurations { unknown: turn_end.duration_since(turn_start), + auto_review_overhead: None, ..Default::default() } ); @@ -179,6 +181,7 @@ fn reviewed_branch_wins_even_when_some_review_overlapped_parent_inference() { PathDurations { sampling: Duration::from_millis(100), auto_review: Duration::from_millis(200), + auto_review_overhead: None, tools: Duration::from_millis(100), ..Default::default() } @@ -206,6 +209,7 @@ fn durations_clip_work_to_the_window_and_attribute_gaps_to_overhead() { PathDurations { sampling: Duration::from_millis(20), auto_review: Duration::from_millis(40), + auto_review_overhead: None, overhead: Duration::from_millis(40), ..PathDurations::default() } @@ -215,6 +219,7 @@ fn durations_clip_work_to_the_window_and_attribute_gaps_to_overhead() { PathDurations { sampling: Duration::from_millis(10), auto_review: Duration::from_millis(30), + auto_review_overhead: None, overhead: Duration::from_millis(20), ..PathDurations::default() } @@ -362,9 +367,249 @@ fn human_wait_is_retained_only_on_the_selected_parallel_path() { PathDurations { sampling: Duration::from_millis(100), auto_review: Duration::from_millis(100), + auto_review_overhead: None, user_wait: Duration::from_millis(300), tools: Duration::from_millis(100), ..PathDurations::default() } ); } + +fn observed_review(intervals: &[(Instant, Instant)]) -> ReviewExecution { + let review = ReviewExecution::default(); + review.observe(); + for &(start, end) in intervals { + review.started(start); + review.completed(start, end); + } + review.seal(); + review +} + +#[test] +fn review_overhead_is_the_execution_complement_and_clips_to_the_requested_window() { + let start = Instant::now(); + let at = |ms| start + Duration::from_millis(ms); + let review = observed_review(&[(at(1000), at(4633))]); + let path = CriticalPath::default().auto_review(start, at(5000), vec![review]); + assert_eq!( + path.durations(start, at(5000)), + PathDurations { + auto_review: Duration::from_millis(5000), + auto_review_overhead: Some(Duration::from_millis(1367)), + ..Default::default() + } + ); + assert_eq!( + path.durations(at(500), at(4800)), + PathDurations { + auto_review: Duration::from_millis(4300), + auto_review_overhead: Some(Duration::from_millis(667)), + ..Default::default() + } + ); +} + +#[test] +fn overlapping_reviews_and_retries_subtract_a_clipped_union_once() { + let start = Instant::now(); + let at = |ms| start + Duration::from_millis(ms); + let first = observed_review(&[(at(0), at(30)), (at(25), at(50))]); + let second = observed_review(&[(at(40), at(60)), (at(80), at(120))]); + let path = CriticalPath::default().auto_review(at(10), at(100), vec![first, second]); + assert_eq!( + path.durations(at(10), at(100)), + PathDurations { + auto_review: Duration::from_millis(90), + auto_review_overhead: Some(Duration::from_millis(20)), + ..Default::default() + } + ); +} + +#[test] +fn human_wait_overlap_is_excluded_from_both_review_components() { + let start = Instant::now(); + let at = |ms| start + Duration::from_millis(ms); + let review = observed_review(&[(at(20), at(80))]); + let path = CriticalPath::default() + .auto_review(start, at(40), vec![review.clone()]) + .work(at(40), at(70), WorkKind::UserWait) + .auto_review(at(70), at(100), vec![review]); + assert_eq!( + path.durations(start, at(100)), + PathDurations { + auto_review: Duration::from_millis(70), + auto_review_overhead: Some(Duration::from_millis(40)), + user_wait: Duration::from_millis(30), + ..Default::default() + } + ); +} + +#[test] +fn cached_review_has_only_overhead_and_losing_review_has_no_contribution() { + let start = Instant::now(); + let at = |ms| start + Duration::from_millis(ms); + let reviewed = CriticalPath::default().auto_review(start, at(10), vec![observed_review(&[])]); + assert_eq!( + reviewed.durations(start, at(10)), + PathDurations { + auto_review: Duration::from_millis(10), + auto_review_overhead: Some(Duration::from_millis(10)), + ..Default::default() + } + ); + let winner = CriticalPath::default().work(start, at(20), WorkKind::Tools); + assert_eq!( + reviewed.latest(winner).durations(start, at(20)), + PathDurations { + tools: Duration::from_millis(20), + ..Default::default() + } + ); +} + +#[test] +fn incomplete_or_unobserved_reviews_leave_the_subcomponent_unknown() { + let start = Instant::now(); + let end = start + Duration::from_millis(10); + let unobserved = ReviewExecution::default(); + let canceled = ReviewExecution::default(); + canceled.observe(); + canceled.started(start); + let missing = ReviewExecution::default(); + missing.observe(); + missing.mark_unknown(); + for review in [unobserved, canceled, missing] { + review.seal(); + let path = CriticalPath::default().auto_review(start, end, vec![review.clone()]); + let expected = PathDurations { + auto_review: Duration::from_millis(10), + auto_review_overhead: None, + ..Default::default() + }; + assert_eq!(path.durations(start, end), expected); + review.observe(); + review.completed(start, end); + assert_eq!(path.durations(start, end), expected); + } +} + +#[test] +fn one_completed_retry_does_not_hide_another_canceled_attempt() { + let start = Instant::now(); + let at = |ms| start + Duration::from_millis(ms); + let review = ReviewExecution::default(); + review.observe(); + review.started(at(1)); + review.started(at(2)); + review.completed(at(2), at(5)); + review.seal(); + assert_eq!( + CriticalPath::default() + .auto_review(start, at(10), vec![review]) + .durations(start, at(10)), + PathDurations { + auto_review: Duration::from_millis(10), + auto_review_overhead: None, + ..Default::default() + } + ); +} + +#[test] +fn indistinguishable_concurrent_attempts_remain_unknown_after_one_completes() { + let start = Instant::now(); + let end = start + Duration::from_millis(10); + let review = ReviewExecution::default(); + review.observe(); + review.started(start); + review.started(start); + review.completed(start, end); + review.seal(); + assert_eq!( + CriticalPath::default() + .auto_review(start, end, vec![review]) + .durations(start, end), + PathDurations { + auto_review: Duration::from_millis(10), + auto_review_overhead: None, + ..Default::default() + } + ); +} + +#[test] +fn sealing_before_submission_starts_freezes_unknown_attribution() { + let start = Instant::now(); + let end = start + Duration::from_millis(10); + let review = ReviewExecution::default(); + review.observe(); + let attempt = review.start_attempt(); + review.seal(); + let path = CriticalPath::default().auto_review(start, end, vec![review.clone()]); + let expected = PathDurations { + auto_review: Duration::from_millis(10), + auto_review_overhead: None, + ..Default::default() + }; + assert_eq!(path.durations(start, end), expected); + review.started(start); + review.completed(start, end); + attempt.finish(/*known*/ true); + assert_eq!(path.durations(start, end), expected); +} + +#[test] +fn completed_attempts_resolve_only_when_all_submissions_have_known_outcomes() { + let start = Instant::now(); + let at = |ms| start + Duration::from_millis(ms); + for known in [true, false] { + let review = ReviewExecution::default(); + review.observe(); + let first = review.start_attempt(); + let second = review.start_attempt(); + review.started(at(2)); + review.completed(at(2), at(7)); + first.finish(/*known*/ true); + let path = CriticalPath::default().auto_review(start, at(10), vec![review.clone()]); + assert_eq!(path.durations(start, at(10)).auto_review_overhead, None); + second.finish(known); + review.seal(); + assert_eq!( + path.durations(start, at(10)), + PathDurations { + auto_review: Duration::from_millis(10), + auto_review_overhead: known.then_some(Duration::from_millis(5)), + ..Default::default() + } + ); + } +} + +#[test] +fn incomplete_attempt_does_not_label_unobserved_execution_as_overhead() { + for completed in [false, true] { + let execution = ReviewExecution::default(); + execution.observe(); + let attempt = execution.start_attempt(); + let start = Instant::now(); + let end = start + Duration::from_millis(100); + if completed { + execution.started(start); + execution.completed(start, end); + attempt.finish(/*known*/ true); + } else { + drop(attempt); + } + execution.seal(); + let totals = CriticalPath::default() + .auto_review(start, end, vec![execution]) + .durations(start, end); + assert_eq!( + totals.auto_review_overhead, + completed.then_some(Duration::ZERO) + ); + } +} diff --git a/codex-rs/utils/critical-path/src/lib.rs b/codex-rs/utils/critical-path/src/lib.rs index b8b18d2d7b..1e0920ead5 100644 --- a/codex-rs/utils/critical-path/src/lib.rs +++ b/codex-rs/utils/critical-path/src/lib.rs @@ -5,6 +5,10 @@ use std::sync::Arc; use std::time::Duration; use std::time::Instant; +mod review_execution; +pub use review_execution::ReviewExecution; +pub use review_execution::ReviewExecutionAttempt; + const MAX_PATH_DEPTH: usize = 16_384; /// Category attributed to an observed segment of dependency work. @@ -28,7 +32,8 @@ pub enum WorkKind { Unknown, } -/// An immutable observed dependency path. Forks share their prefix; a join +/// A persistent observed dependency path. Review observers must be sealed before +/// publishing completed paths. Forks share their prefix; a join /// retains only the dependency that actually released the waiting operation. /// Equality compares path identity rather than the contents of each segment. #[derive(Clone, Default)] @@ -59,6 +64,7 @@ impl Eq for CriticalPath {} struct Node { previous: CriticalPath, kind: WorkKind, + reviews: Vec, start: Instant, end: Instant, origin: Instant, @@ -66,19 +72,40 @@ struct Node { } /// Exclusive time attributed to each category within the requested interval. -#[derive(Debug, Default, PartialEq, Eq)] +#[derive(Debug, PartialEq, Eq)] pub struct PathDurations { pub sampling: Duration, pub compaction: Duration, pub auto_review: Duration, + /// Subcomponent of auto_review excluding the union of observed child execution. + /// None means review attribution or dependency history is incomplete. + pub auto_review_overhead: Option, pub user_wait: Duration, pub tools: Duration, pub overhead: Duration, pub unknown: Duration, } +impl Default for PathDurations { + fn default() -> Self { + Self { + sampling: Duration::ZERO, + compaction: Duration::ZERO, + auto_review: Duration::ZERO, + auto_review_overhead: Some(Duration::ZERO), + user_wait: Duration::ZERO, + tools: Duration::ZERO, + overhead: Duration::ZERO, + unknown: Duration::ZERO, + } + } +} + impl PathDurations { fn record(&mut self, kind: WorkKind, duration: Duration) { + if matches!(kind, WorkKind::Unknown) { + self.auto_review_overhead = None; + } let total = match kind { WorkKind::Sampling => &mut self.sampling, WorkKind::Compaction => &mut self.compaction, @@ -110,6 +137,22 @@ impl CriticalPath { /// Append work after the existing path ends, clipping overlap and ignoring empty segments. /// Exceeding the history limit collapses the retained interval into unknown work. pub fn work(self, start: Instant, end: Instant, kind: WorkKind) -> Self { + self.append(start, end, kind, Vec::new()) + } + + /// Adds review time with the exact reviews active during this segment. Their + /// child execution union is clipped to the segment and requested window. + pub fn auto_review(self, start: Instant, end: Instant, reviews: Vec) -> Self { + self.append(start, end, WorkKind::AutoReview, reviews) + } + + fn append( + self, + start: Instant, + end: Instant, + kind: WorkKind, + reviews: Vec, + ) -> Self { let start = self.end().map_or(start, |previous| previous.max(start)); if start >= end { return self; @@ -123,6 +166,7 @@ impl CriticalPath { Self(Some(Arc::new(Node { previous: self, kind, + reviews, start, end, origin, @@ -154,6 +198,12 @@ impl CriticalPath { } if segment_start < segment_end { totals.record(node.kind, segment_end.duration_since(segment_start)); + if matches!(node.kind, WorkKind::AutoReview) { + totals.auto_review_overhead = totals.auto_review_overhead.and_then(|total| { + review_execution::overhead(&node.reviews, segment_start, segment_end) + .map(|overhead| total.saturating_add(overhead)) + }); + } } end = end.min(node.start); path = &node.previous; diff --git a/codex-rs/utils/critical-path/src/review_execution.rs b/codex-rs/utils/critical-path/src/review_execution.rs new file mode 100644 index 0000000000..e88de39272 --- /dev/null +++ b/codex-rs/utils/critical-path/src/review_execution.rs @@ -0,0 +1,168 @@ +//! Collects exact child execution intervals for one automatic review, then seals them. + +use std::collections::BTreeSet; +use std::sync::Arc; +use std::sync::Mutex; +use std::time::Duration; +use std::time::Instant; + +/// Observes child turn execution for one review, including concurrent or retried turns. +/// The owner must seal observations before publishing the completed dependency path. +/// Equality compares review identity, not its collected intervals. +#[derive(Clone, Debug, Default)] +pub struct ReviewExecution(Arc>); + +/// Resolves one submission exactly once, including cancellation before acknowledgement. +#[must_use = "keep the attempt alive until submission completes"] +pub struct ReviewExecutionAttempt { + execution: ReviewExecution, + finished: bool, +} + +impl ReviewExecutionAttempt { + /// Records whether the submitted child's execution boundaries are known. + pub fn finish(mut self, known: bool) { + self.execution.attempt_finished(known); + self.finished = true; + } +} + +impl Drop for ReviewExecutionAttempt { + fn drop(&mut self) { + if !self.finished { + self.execution.attempt_finished(/*known*/ false); + } + } +} + +#[derive(Debug, Default)] +struct State { + observed: bool, + sealed: bool, + unknown: bool, + pending: BTreeSet, + pending_attempts: usize, + intervals: Vec<(Instant, Instant)>, +} + +impl ReviewExecution { + /// Declares that this review's child executions are instrumented, even if none run. + pub fn observe(&self) { + let mut state = self.state(); + if !state.sealed { + state.observed = true; + } + } + + /// Registers an unresolved submission before asynchronous work can be detached. + pub fn start_attempt(&self) -> ReviewExecutionAttempt { + let mut state = self.state(); + if !state.sealed { + state.pending_attempts += 1; + } + ReviewExecutionAttempt { + execution: self.clone(), + finished: false, + } + } + + /// Resolves one submission. Unknown execution remains unknown after it finishes. + fn attempt_finished(&self, known: bool) { + let mut state = self.state(); + if state.sealed { + return; + } + if let Some(pending) = state.pending_attempts.checked_sub(1) { + state.pending_attempts = pending; + } else { + state.unknown = true; + } + state.unknown |= !known; + } + + /// Records the actual child turn start, before it can be canceled. + pub fn started(&self, started_at: Instant) { + let mut state = self.state(); + if !state.sealed && !state.pending.insert(started_at) { + // Equal clock readings cannot distinguish two concurrent attempts. + state.unknown = true; + } + } + + /// Records exact turn bounds. Repeated completion notifications are harmless. + pub fn completed(&self, started_at: Instant, completed_at: Instant) { + let mut state = self.state(); + if state.sealed { + return; + } + state.pending.remove(&started_at); + if completed_at < started_at { + state.unknown = true; + } else { + state.intervals.push((started_at, completed_at)); + } + } + + /// Marks unavailable child attribution, rather than reporting it as review overhead. + pub fn mark_unknown(&self) { + let mut state = self.state(); + if !state.sealed { + state.unknown = true; + } + } + + /// Freezes this review's evidence. Later callbacks cannot change a completed path. + pub fn seal(&self) { + self.state().sealed = true; + } + + fn state(&self) -> std::sync::MutexGuard<'_, State> { + self.0 + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } +} + +pub(crate) fn overhead( + reviews: &[ReviewExecution], + start: Instant, + end: Instant, +) -> Option { + if reviews.is_empty() { + return None; + } + let mut intervals = Vec::new(); + for review in reviews { + let state = review.state(); + if !state.observed + || state.unknown + || !state.pending.is_empty() + || state.pending_attempts > 0 + { + return None; + } + intervals.extend(state.intervals.iter().filter_map(|&(left, right)| { + let left = left.max(start); + let right = right.min(end); + (left < right).then_some((left, right)) + })); + } + intervals.sort_unstable(); + let mut execution = Duration::ZERO; + let mut covered_until = start; + for (left, right) in intervals { + if right > covered_until { + execution += right.duration_since(left.max(covered_until)); + covered_until = right; + } + } + Some(end.duration_since(start).saturating_sub(execution)) +} + +impl PartialEq for ReviewExecution { + fn eq(&self, other: &Self) -> bool { + Arc::ptr_eq(&self.0, &other.0) + } +} + +impl Eq for ReviewExecution {}