openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 17:48:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add a critical-path utility for exclusive duration attribution (#55804)

Browse Source 
## What changed

Add `codex-utils-critical-path` to track shared dependency paths and attribute time to inference, automatic review, user waits, tools, overhead, and unknown work. Parallel joins retain the later-finishing branch, preserving its full history without double-counting overlapping work.

Clip attribution to the requested interval and count unrecorded gaps as overhead. Bound retained history by collapsing it into unknown work at the depth limit, with nonrecursive destruction and bounded debug output.

## Testing

Add unit tests for parallel joins, nested dependencies, ties, user waits, interval clipping, overlapping segments, history exhaustion, and safe formatting and concurrent release on small stacks.

GitOrigin-RevId: 2df7453162491d3ae97bf4bee942730c8a4d5e5c
This commit is contained in:
taohong-oai
2026-10-21 07:33:13 +00:00
committed by copyberry
parent c7234c1174
commit af91f6ef2c
6 changed files with 578 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+7
View File
@@ -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"
codex-rs/Cargo.toml
+1
View File
@@ -125,6 +125,7 @@ members = [
"utils/cargo-bin",
"git-utils",
"utils/cache",
"utils/critical-path",
"utils/git-discovery",
"utils/image",
"utils/json-to-toml",
codex-rs/utils/critical-path/BUILD.bazel
+6
View File
@@ -0,0 +1,6 @@
load("//:defs.bzl", "codex_rust_crate")
codex_rust_crate(
name = "critical-path",
crate_name = "codex_utils_critical_path",
)
codex-rs/utils/critical-path/Cargo.toml
+14
View File
@@ -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
codex-rs/utils/critical-path/src/critical_path_tests.rs
+369
View File
@@ -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()
}
);
}
codex-rs/utils/critical-path/src/lib.rs
+181
View File
@@ -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<Arc<Node>>);
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<Instant> {
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;