diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index a22a6ec662..923eaefa2e 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -5103,6 +5103,13 @@ dependencies = [ "toml 0.9.11+spec-1.1.0", ] +[[package]] +name = "codex-utils-critical-path" +version = "0.0.0" +dependencies = [ + "pretty_assertions", +] + [[package]] name = "codex-utils-elapsed" version = "0.0.0" diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index 8392dea799..33c12ad70d 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -125,6 +125,7 @@ members = [ "utils/cargo-bin", "git-utils", "utils/cache", + "utils/critical-path", "utils/git-discovery", "utils/image", "utils/json-to-toml", diff --git a/codex-rs/utils/critical-path/BUILD.bazel b/codex-rs/utils/critical-path/BUILD.bazel new file mode 100644 index 0000000000..299f272e84 --- /dev/null +++ b/codex-rs/utils/critical-path/BUILD.bazel @@ -0,0 +1,6 @@ +load("//:defs.bzl", "codex_rust_crate") + +codex_rust_crate( + name = "critical-path", + crate_name = "codex_utils_critical_path", +) diff --git a/codex-rs/utils/critical-path/Cargo.toml b/codex-rs/utils/critical-path/Cargo.toml new file mode 100644 index 0000000000..a670985e65 --- /dev/null +++ b/codex-rs/utils/critical-path/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "codex-utils-critical-path" +version.workspace = true +edition.workspace = true +license.workspace = true + +[lints] +workspace = true + +[lib] +doctest = false + +[dev-dependencies] +pretty_assertions.workspace = true diff --git a/codex-rs/utils/critical-path/src/critical_path_tests.rs b/codex-rs/utils/critical-path/src/critical_path_tests.rs new file mode 100644 index 0000000000..937d5bba8e --- /dev/null +++ b/codex-rs/utils/critical-path/src/critical_path_tests.rs @@ -0,0 +1,369 @@ +//! Reducer coverage for exclusive attribution, parallel joins, ties, and incomplete evidence. + +use super::*; +use pretty_assertions::assert_eq; + +#[test] +fn join_replaces_overlapping_work_with_the_full_winning_branch() { + let turn_start = Instant::now(); + let time_after_turn_start = |milliseconds| turn_start + Duration::from_millis(milliseconds); + let shared_initial_path = CriticalPath::default().work( + time_after_turn_start(0), + time_after_turn_start(100), + WorkKind::Inference, + ); + let reviewed_tool_path = shared_initial_path + .clone() + .work( + time_after_turn_start(100), + time_after_turn_start(300), + WorkKind::AutoReview, + ) + .work( + time_after_turn_start(300), + time_after_turn_start(400), + WorkKind::Tools, + ); + let parallel_tool_path = shared_initial_path.clone().work( + time_after_turn_start(100), + time_after_turn_start(700), + WorkKind::Tools, + ); + let critical_path = shared_initial_path + .work( + time_after_turn_start(100), + time_after_turn_start(200), + WorkKind::Inference, + ) + .join( + time_after_turn_start(200), + time_after_turn_start(700), + reviewed_tool_path.latest(parallel_tool_path), + ); + assert_eq!( + critical_path.durations(time_after_turn_start(0), time_after_turn_start(700)), + PathDurations { + inference: Duration::from_millis(100), + tools: Duration::from_millis(600), + ..PathDurations::default() + } + ); +} + +#[test] +fn nested_serial_dependencies_preserve_exclusive_categories() { + let turn_start = Instant::now(); + let time_after_turn_start = |milliseconds| turn_start + Duration::from_millis(milliseconds); + let shared_initial_path = CriticalPath::default().work( + time_after_turn_start(0), + time_after_turn_start(100), + WorkKind::Inference, + ); + let child_path = shared_initial_path + .clone() + .work( + time_after_turn_start(100), + time_after_turn_start(300), + WorkKind::AutoReview, + ) + .work( + time_after_turn_start(300), + time_after_turn_start(400), + WorkKind::Tools, + ) + .work( + time_after_turn_start(400), + time_after_turn_start(500), + WorkKind::Inference, + ); + let critical_path = shared_initial_path + .work( + time_after_turn_start(100), + time_after_turn_start(200), + WorkKind::Inference, + ) + .join( + time_after_turn_start(200), + time_after_turn_start(600), + child_path, + ); + assert_eq!( + critical_path.durations(time_after_turn_start(0), time_after_turn_start(600)), + PathDurations { + inference: Duration::from_millis(200), + auto_review: Duration::from_millis(200), + tools: Duration::from_millis(200), + ..PathDurations::default() + } + ); +} + +#[test] +fn equal_finish_times_select_one_path_without_double_counting() { + let turn_start = Instant::now(); + let turn_end = turn_start + Duration::from_millis(100); + let reviewed_tool_path = + CriticalPath::default().work(turn_start, turn_end, WorkKind::AutoReview); + let inference_path = CriticalPath::default().work(turn_start, turn_end, WorkKind::Inference); + // A tie has multiple valid paths. Stable left selection does not sum both. + assert_eq!( + reviewed_tool_path + .clone() + .latest(inference_path) + .durations(turn_start, turn_end), + reviewed_tool_path.durations(turn_start, turn_end) + ); +} + +#[test] +fn path_retention_exhaustion_marks_unknown_instead_of_claiming_zero() { + let turn_start = Instant::now(); + let mut critical_path = CriticalPath::default(); + for segment_index in 0..=MAX_PATH_DEPTH { + critical_path = critical_path.work( + turn_start + Duration::from_micros(segment_index as u64), + turn_start + Duration::from_micros(segment_index as u64 + 1), + WorkKind::Inference, + ); + } + let turn_end = turn_start + Duration::from_micros(MAX_PATH_DEPTH as u64 + 1); + assert_eq!( + critical_path.durations(turn_start, turn_end), + PathDurations { + unknown: turn_end.duration_since(turn_start), + ..Default::default() + } + ); +} + +#[test] +fn reviewed_branch_wins_even_when_some_review_overlapped_parent_inference() { + let turn_start = Instant::now(); + let time_after_turn_start = |milliseconds| turn_start + Duration::from_millis(milliseconds); + let shared_initial_path = CriticalPath::default().work( + time_after_turn_start(0), + time_after_turn_start(100), + WorkKind::Inference, + ); + let reviewed_tool_path = shared_initial_path + .clone() + .work( + time_after_turn_start(100), + time_after_turn_start(300), + WorkKind::AutoReview, + ) + .work( + time_after_turn_start(300), + time_after_turn_start(400), + WorkKind::Tools, + ); + let parallel_tool_path = shared_initial_path.clone().work( + time_after_turn_start(100), + time_after_turn_start(350), + WorkKind::Tools, + ); + let critical_path = shared_initial_path + .work( + time_after_turn_start(100), + time_after_turn_start(200), + WorkKind::Inference, + ) + .join( + time_after_turn_start(200), + time_after_turn_start(400), + reviewed_tool_path.latest(parallel_tool_path), + ); + assert_eq!( + critical_path.durations(time_after_turn_start(0), time_after_turn_start(400)), + PathDurations { + inference: Duration::from_millis(100), + auto_review: Duration::from_millis(200), + tools: Duration::from_millis(100), + ..Default::default() + } + ); +} + +#[test] +fn durations_clip_work_to_the_window_and_attribute_gaps_to_overhead() { + let turn_start = Instant::now(); + let time_after_turn_start = |milliseconds| turn_start + Duration::from_millis(milliseconds); + let critical_path = CriticalPath::default() + .work( + time_after_turn_start(10), + time_after_turn_start(30), + WorkKind::Inference, + ) + .work( + time_after_turn_start(50), + time_after_turn_start(90), + WorkKind::AutoReview, + ); + + assert_eq!( + critical_path.durations(time_after_turn_start(0), time_after_turn_start(100)), + PathDurations { + inference: Duration::from_millis(20), + auto_review: Duration::from_millis(40), + overhead: Duration::from_millis(40), + ..PathDurations::default() + } + ); + assert_eq!( + critical_path.durations(time_after_turn_start(20), time_after_turn_start(80)), + PathDurations { + inference: Duration::from_millis(10), + auto_review: Duration::from_millis(30), + overhead: Duration::from_millis(20), + ..PathDurations::default() + } + ); +} + +#[test] +fn overlapping_work_only_attributes_time_after_the_existing_path() { + let turn_start = Instant::now(); + let time_after_turn_start = |milliseconds| turn_start + Duration::from_millis(milliseconds); + let critical_path = CriticalPath::default() + .work( + time_after_turn_start(0), + time_after_turn_start(10), + WorkKind::Inference, + ) + .work( + time_after_turn_start(5), + time_after_turn_start(20), + WorkKind::Tools, + ); + + assert_eq!( + critical_path.durations(time_after_turn_start(0), time_after_turn_start(20)), + PathDurations { + inference: Duration::from_millis(10), + tools: Duration::from_millis(10), + ..PathDurations::default() + } + ); + assert_eq!( + critical_path.clone().work( + time_after_turn_start(5), + time_after_turn_start(15), + WorkKind::AutoReview + ), + critical_path + ); + assert_eq!( + critical_path.clone().work( + time_after_turn_start(20), + time_after_turn_start(20), + WorkKind::Unknown + ), + critical_path + ); +} + +#[test] +fn long_path_debug_output_is_bounded_on_a_small_stack() { + // Match a typical 2 MiB worker stack rather than nextest's larger test stack. + std::thread::Builder::new() + .stack_size(2 * 1024 * 1024) + .spawn(|| { + let start = Instant::now(); + let mut path = CriticalPath::default(); + for index in 0..10_000 { + path = path.work( + start + Duration::from_micros(index), + start + Duration::from_micros(index + 1), + WorkKind::Inference, + ); + } + assert!(format!("{path:?}").len() < 512); + assert!(format!("{path:#?}").len() < 512); + }) + .unwrap() + .join() + .unwrap(); +} + +#[test] +fn retained_prefixes_can_be_released_concurrently_on_small_stacks() { + for _ in 0..8 { + let start = Instant::now(); + let mut path = CriticalPath::default(); + let mut prefixes = [Vec::new(), Vec::new()]; + for index in 0..MAX_PATH_DEPTH { + path = path.work( + start + Duration::from_micros(index as u64), + start + Duration::from_micros(index as u64 + 1), + WorkKind::Inference, + ); + prefixes[index % 2].push(path.clone()); + } + // Only the workers retain nodes when they begin releasing prefixes. + drop(path); + let ready = Arc::new(std::sync::Barrier::new(3)); + let workers = prefixes.map(|mut prefixes| { + let ready = Arc::clone(&ready); + std::thread::Builder::new() + .stack_size(2 * 1024 * 1024) + .spawn(move || { + ready.wait(); + // Release newest first so adjacent retained prefixes race + // to become the last owner of the shared chain. + while let Some(prefix) = prefixes.pop() { + drop(prefix); + } + }) + .unwrap() + }); + ready.wait(); + for worker in workers { + worker.join().unwrap(); + } + } +} + +#[test] +fn human_wait_is_retained_only_on_the_selected_parallel_path() { + let start = Instant::now(); + let time_after_turn_start = |milliseconds| start + Duration::from_millis(milliseconds); + let prefix = + CriticalPath::default().work(start, time_after_turn_start(100), WorkKind::Inference); + let reviewed_tool = prefix + .clone() + .work( + time_after_turn_start(100), + time_after_turn_start(200), + WorkKind::AutoReview, + ) + .work( + time_after_turn_start(200), + time_after_turn_start(500), + WorkKind::UserWait, + ) + .work( + time_after_turn_start(500), + time_after_turn_start(600), + WorkKind::Tools, + ); + let parallel_tool = prefix.clone().work( + time_after_turn_start(100), + time_after_turn_start(400), + WorkKind::Tools, + ); + let path = prefix.join( + time_after_turn_start(100), + time_after_turn_start(600), + parallel_tool.latest(reviewed_tool), + ); + assert_eq!( + path.durations(start, time_after_turn_start(600)), + PathDurations { + inference: Duration::from_millis(100), + auto_review: Duration::from_millis(100), + user_wait: Duration::from_millis(300), + tools: Duration::from_millis(100), + ..PathDurations::default() + } + ); +} diff --git a/codex-rs/utils/critical-path/src/lib.rs b/codex-rs/utils/critical-path/src/lib.rs new file mode 100644 index 0000000000..439d008120 --- /dev/null +++ b/codex-rs/utils/critical-path/src/lib.rs @@ -0,0 +1,181 @@ +//! Persistent dependency paths and exclusive attribution over the observed turn interval. +//! History is depth-limited; formatting and destruction never recurse through the chain. + +use std::sync::Arc; +use std::time::Duration; +use std::time::Instant; + +const MAX_PATH_DEPTH: usize = 16_384; + +/// Category attributed to an observed segment of dependency work. +#[derive(Clone, Copy, Debug)] +pub enum WorkKind { + /// Model inference work on the selected dependency path. + Inference, + /// Automatic review work, such as reviewing a tool call before execution. + AutoReview, + /// Time awaiting a human answer or approval on the selected dependency path. + UserWait, + /// Tool execution work, including work after a dependency join resumes. + /// This is an aggregate category; tool-specific subtypes such as browser or + /// computer use are not represented by this calculator. + Tools, + /// Time within the observed turn interval not covered by a recorded work segment. + Overhead, + /// Work whose category is unavailable, including history collapsed at the depth limit. + Unknown, +} + +/// An immutable observed dependency path. 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)] +pub struct CriticalPath(Option>); + +impl std::fmt::Debug for CriticalPath { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // Summarize the tail without traversing shared dependency history. + f.debug_struct("CriticalPath") + .field("depth", &self.0.as_ref().map_or(0, |node| node.depth)) + .field("end", &self.end()) + .finish() + } +} + +impl PartialEq for CriticalPath { + fn eq(&self, other: &Self) -> bool { + match (&self.0, &other.0) { + (Some(left), Some(right)) => Arc::ptr_eq(left, right), + (None, None) => true, + (Some(_), None) | (None, Some(_)) => false, + } + } +} + +impl Eq for CriticalPath {} + +struct Node { + previous: CriticalPath, + kind: WorkKind, + start: Instant, + end: Instant, + origin: Instant, + depth: usize, +} + +/// Exclusive time attributed to each category within the requested interval. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct PathDurations { + pub inference: Duration, + pub auto_review: Duration, + pub user_wait: Duration, + pub tools: Duration, + pub overhead: Duration, + pub unknown: Duration, +} + +impl PathDurations { + fn record(&mut self, kind: WorkKind, duration: Duration) { + let total = match kind { + WorkKind::Inference => &mut self.inference, + WorkKind::AutoReview => &mut self.auto_review, + WorkKind::UserWait => &mut self.user_wait, + WorkKind::Tools => &mut self.tools, + WorkKind::Overhead => &mut self.overhead, + WorkKind::Unknown => &mut self.unknown, + }; + *total = total.saturating_add(duration); + } +} + +impl CriticalPath { + /// End of the last recorded segment, or `None` for an empty path. + pub fn end(&self) -> Option { + self.0.as_ref().map(|node| node.end) + } + + /// Retain the later-finishing path, choosing `self` on a tie. + pub fn latest(self, other: Self) -> Self { + if other.end() > self.end() { + other + } else { + self + } + } + + /// 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 { + let start = self.end().map_or(start, |previous| previous.max(start)); + if start >= end { + return self; + } + let origin = self.0.as_ref().map_or(start, |node| node.origin); + let previous_depth = self.0.as_ref().map_or(0, |node| node.depth); + let depth = previous_depth + 1; + if depth > MAX_PATH_DEPTH { + return Self::default().work(origin, end, WorkKind::Unknown); + } + Self(Some(Arc::new(Node { + previous: self, + kind, + start, + end, + origin, + depth, + }))) + } + + /// Select the full branch that released this join, not just its exposed + /// waiting tail. Both branches carry their common causal prefix. + pub fn join(self, start: Instant, end: Instant, dependency: Self) -> Self { + let path = self.latest(dependency); + let resumed_at = path.end().unwrap_or(start).max(start); + path.work(resumed_at, end, WorkKind::Tools) + } + + /// Attribute the requested interval exclusively, counting unrecorded gaps as overhead. + pub fn durations(&self, start: Instant, end: Instant) -> PathDurations { + let mut totals = PathDurations::default(); + let mut path = self; + let mut end = end; + while let Some(node) = path.0.as_ref() { + if end <= start { + break; + } + let segment_end = end.min(node.end); + let segment_start = start.max(node.start); + if end > node.end { + totals.record(WorkKind::Overhead, end.duration_since(start.max(node.end))); + } + if segment_start < segment_end { + totals.record(node.kind, segment_end.duration_since(segment_start)); + } + end = end.min(node.start); + path = &node.previous; + } + if start < end { + totals.record(WorkKind::Overhead, end.duration_since(start)); + } + totals + } +} + +// Long sequential turns must not recursively destroy one Arc per operation. +// into_inner consumes each reference atomically: concurrent releases cannot +// leave an Err(Arc) whose subsequent drop recursively destroys the next node. +impl Drop for Node { + fn drop(&mut self) { + let mut previous = self.previous.0.take(); + while let Some(node) = previous { + let Some(mut node) = Arc::into_inner(node) else { + break; + }; + previous = node.previous.0.take(); + } + } +} + +#[cfg(test)] +#[path = "critical_path_tests.rs"] +mod tests;