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 {}