Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions codex-rs/analytics/src/analytics_client_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -441,9 +441,10 @@ fn sample_turn_profile() -> TurnProfile {
TurnProfile {
before_first_sampling_ms: 100,
sampling_ms: 700,
compaction_ms: 40,
between_sampling_overhead_ms: 50,
tool_blocking_ms: 250,
after_last_sampling_ms: 134,
after_last_sampling_ms: 94,
sampling_request_count: 2,
sampling_retry_count: 1,
}
Expand Down Expand Up @@ -4065,9 +4066,10 @@ fn turn_event_serializes_expected_shape() {
total_tokens: None,
before_first_sampling_ms: 100,
sampling_ms: 700,
compaction_ms: 40,
between_sampling_overhead_ms: 50,
tool_blocking_ms: 250,
after_last_sampling_ms: 134,
after_last_sampling_ms: 94,
sampling_request_count: 2,
sampling_retry_count: 1,
duration_ms: Some(1234),
Expand Down Expand Up @@ -4138,9 +4140,10 @@ fn turn_event_serializes_expected_shape() {
"total_tokens": null,
"before_first_sampling_ms": 100,
"sampling_ms": 700,
"compaction_ms": 40,
"between_sampling_overhead_ms": 50,
"tool_blocking_ms": 250,
"after_last_sampling_ms": 134,
"after_last_sampling_ms": 94,
"sampling_request_count": 2,
"sampling_retry_count": 1,
"duration_ms": 1234,
Expand Down
1 change: 1 addition & 0 deletions codex-rs/analytics/src/events.rs
Original file line number Diff line number Diff line change
Expand Up @@ -914,6 +914,7 @@ pub(crate) struct CodexTurnEventParams {
pub(crate) total_tokens: Option<i64>,
pub(crate) before_first_sampling_ms: u64,
pub(crate) sampling_ms: u64,
pub(crate) compaction_ms: u64,
pub(crate) between_sampling_overhead_ms: u64,
pub(crate) tool_blocking_ms: u64,
pub(crate) after_last_sampling_ms: u64,
Expand Down
1 change: 1 addition & 0 deletions codex-rs/analytics/src/facts.rs
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,7 @@ pub struct TurnTokenUsageFact {
pub struct TurnProfile {
pub before_first_sampling_ms: u64,
pub sampling_ms: u64,
pub compaction_ms: u64,
pub between_sampling_overhead_ms: u64,
pub tool_blocking_ms: u64,
pub after_last_sampling_ms: u64,
Expand Down
2 changes: 2 additions & 0 deletions codex-rs/analytics/src/reducer.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2636,6 +2636,7 @@ fn codex_turn_event_params(
let TurnProfile {
before_first_sampling_ms,
sampling_ms,
compaction_ms,
between_sampling_overhead_ms,
tool_blocking_ms,
after_last_sampling_ms,
Expand Down Expand Up @@ -2708,6 +2709,7 @@ fn codex_turn_event_params(
.map(|token_usage| token_usage.total_tokens),
before_first_sampling_ms,
sampling_ms,
compaction_ms,
between_sampling_overhead_ms,
tool_blocking_ms,
after_last_sampling_ms,
Expand Down
1 change: 1 addition & 0 deletions codex-rs/core/src/session/turn.rs
Original file line number Diff line number Diff line change
Expand Up @@ -989,6 +989,7 @@ async fn run_auto_compact(
phase: CompactionPhase,
) -> CodexResult<()> {
let turn_context = &step_context.turn;
let _profile_guard = turn_context.turn_timing_state.begin_compaction();
if turn_context.config.features.enabled(Feature::TokenBudget) {
// Compaction is the reset request, so force a new context window
// instead of consuming a pending `new_context` tool request.
Expand Down
1 change: 1 addition & 0 deletions codex-rs/core/src/tasks/compact.rs
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@ impl SessionTask for CompactTask {
_cancellation_token: CancellationToken,
) -> SessionTaskResult {
let session = session.clone_session();
let _profile_guard = ctx.turn_timing_state.begin_compaction();
if ctx.config.features.enabled(Feature::TokenBudget) {
crate::compact_token_budget::run_manual_compact_task(session, ctx).await?;
return Ok(None);
Expand Down
12 changes: 6 additions & 6 deletions codex-rs/core/src/tasks/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -758,15 +758,15 @@ impl Session {
turn_had_memory_citation,
);
let started_at = turn_context.turn_timing_state.started_at_unix_secs().await;
let (completed_at, duration_ms) = turn_context
let (completed_at, duration_ms, profile) = turn_context
.turn_timing_state
.completed_at_and_duration_ms()
.complete_profile_and_duration_ms()
.await;
self.services
.analytics_events_client
.track_turn_profile(TurnProfileFact {
turn_id: turn_context.sub_id.clone(),
profile: turn_context.turn_timing_state.complete_profile(),
profile,
});
let event = if let Some(reason) = abort_reason {
self.emit_turn_abort_lifecycle(reason.clone(), turn_context.extension_data.as_ref())
Expand Down Expand Up @@ -907,16 +907,16 @@ impl Session {
.turn_timing_state
.started_at_unix_secs()
.await;
let (completed_at, duration_ms) = task
let (completed_at, duration_ms, profile) = task
.turn_context
.turn_timing_state
.completed_at_and_duration_ms()
.complete_profile_and_duration_ms()
.await;
self.services
.analytics_events_client
.track_turn_profile(TurnProfileFact {
turn_id: task.turn_context.sub_id.clone(),
profile: task.turn_context.turn_timing_state.complete_profile(),
profile,
});
let event = EventMsg::TurnAborted(TurnAbortedEvent {
turn_id: Some(task.turn_context.sub_id.clone()),
Expand Down
50 changes: 41 additions & 9 deletions codex-rs/core/src/turn_timing.rs
Original file line number Diff line number Diff line change
Expand Up @@ -61,6 +61,7 @@ struct TurnProfileState {
seen_sampling: bool,
before_first_sampling: Duration,
sampling: Duration,
compaction: Duration,
between_sampling_overhead: Duration,
tool_blocking: Duration,
pending_idle_after_sampling: Duration,
Expand All @@ -72,6 +73,7 @@ struct TurnProfileState {
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum TurnProfilePhase {
Sampling,
Compaction,
ToolBlocking,
}

Expand All @@ -98,13 +100,22 @@ impl TurnTimingState {
self.state.lock().await.started_at_unix_secs
}

pub(crate) async fn completed_at_and_duration_ms(&self) -> (Option<i64>, Option<i64>) {
pub(crate) async fn complete_profile_and_duration_ms(
&self,
) -> (Option<i64>, Option<i64>, TurnProfile) {
let completed_at_instant = Instant::now();
let state = self.state.lock().await;
let completed_at = Some(now_unix_timestamp_secs());
let duration_ms = state
.started_at
.map(|started_at| i64::try_from(started_at.elapsed().as_millis()).unwrap_or(i64::MAX));
(completed_at, duration_ms)
let duration_ms = state.started_at.map(|started_at| {
i64::try_from(
completed_at_instant
.saturating_duration_since(started_at)
.as_millis(),
)
.unwrap_or(i64::MAX)
});
let profile = self.profile_state().complete(completed_at_instant);
(completed_at, duration_ms, profile)
}

pub(crate) async fn time_to_first_token_ms(&self) -> Option<i64> {
Expand All @@ -114,10 +125,6 @@ impl TurnTimingState {
.map(|duration| i64::try_from(duration.as_millis()).unwrap_or(i64::MAX))
}

pub(crate) fn complete_profile(&self) -> TurnProfile {
self.profile_state().complete(Instant::now())
}

pub(crate) fn begin_sampling(self: &Arc<Self>) -> TurnProfileTimingGuard {
let active = self.profile_state().begin_sampling(Instant::now());
TurnProfileTimingGuard {
Expand All @@ -131,6 +138,15 @@ impl TurnTimingState {
self.profile_state().record_sampling_retry();
}

pub(crate) fn begin_compaction(self: &Arc<Self>) -> TurnProfileTimingGuard {
let active = self.profile_state().begin_compaction(Instant::now());
TurnProfileTimingGuard {
timing: Arc::clone(self),
phase: TurnProfilePhase::Compaction,
active,
}
}

pub(crate) fn begin_tool_blocking(self: &Arc<Self>) -> TurnProfileTimingGuard {
let active = self.profile_state().begin_tool_blocking(Instant::now());
TurnProfileTimingGuard {
Expand Down Expand Up @@ -235,6 +251,18 @@ impl TurnProfileState {
true
}

fn begin_compaction(&mut self, now: Instant) -> bool {
if self.completed_profile.is_some()
|| self.started_at.is_none()
|| self.active_phase.is_some()
{
return false;
}
self.advance(now);
self.active_phase = Some(TurnProfilePhase::Compaction);
true
}

fn end_phase(&mut self, now: Instant, phase: TurnProfilePhase) {
if self.completed_profile.is_some() || self.active_phase != Some(phase) {
return;
Expand All @@ -250,6 +278,7 @@ impl TurnProfileState {
let elapsed = now.saturating_duration_since(previous);
match self.active_phase {
Some(TurnProfilePhase::Sampling) => self.sampling += elapsed,
Some(TurnProfilePhase::Compaction) => self.compaction += elapsed,
Some(TurnProfilePhase::ToolBlocking) => self.tool_blocking += elapsed,
None if self.seen_sampling => self.pending_idle_after_sampling += elapsed,
None => self.before_first_sampling += elapsed,
Expand All @@ -272,6 +301,7 @@ impl TurnProfileState {
let mut profile = TurnProfile {
before_first_sampling_ms: duration_to_u64_ms(self.before_first_sampling),
sampling_ms: duration_to_u64_ms(self.sampling),
compaction_ms: duration_to_u64_ms(self.compaction),
between_sampling_overhead_ms: duration_to_u64_ms(self.between_sampling_overhead),
tool_blocking_ms: duration_to_u64_ms(self.tool_blocking),
after_last_sampling_ms: duration_to_u64_ms(after_last_sampling),
Expand All @@ -285,12 +315,14 @@ impl TurnProfileState {
let classified_ms = profile
.before_first_sampling_ms
.saturating_add(profile.sampling_ms)
.saturating_add(profile.compaction_ms)
.saturating_add(profile.between_sampling_overhead_ms)
.saturating_add(profile.tool_blocking_ms)
.saturating_add(profile.after_last_sampling_ms);
let rounding_ms = total_ms.saturating_sub(classified_ms);
match final_phase {
Some(TurnProfilePhase::Sampling) => profile.sampling_ms += rounding_ms,
Some(TurnProfilePhase::Compaction) => profile.compaction_ms += rounding_ms,
Some(TurnProfilePhase::ToolBlocking) => profile.tool_blocking_ms += rounding_ms,
None if self.seen_sampling => profile.after_last_sampling_ms += rounding_ms,
None => profile.before_first_sampling_ms += rounding_ms,
Expand Down
55 changes: 55 additions & 0 deletions codex-rs/core/src/turn_timing_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -186,6 +186,7 @@ fn turn_profile_breaks_down_sampling_blocking_and_retry_overhead() {
TurnProfile {
before_first_sampling_ms: 100,
sampling_ms: 700,
compaction_ms: 0,
between_sampling_overhead_ms: 100,
tool_blocking_ms: 300,
after_last_sampling_ms: 100,
Expand All @@ -194,3 +195,57 @@ fn turn_profile_breaks_down_sampling_blocking_and_retry_overhead() {
}
);
}

#[test]
fn turn_profile_counts_compaction_as_an_exclusive_phase() {
let started_at = Instant::now();
let mut state = TurnProfileState::default();
state.start(started_at);

let _ = state.begin_compaction(started_at + Duration::from_millis(100));
state.end_phase(
started_at + Duration::from_millis(300),
TurnProfilePhase::Compaction,
);
let _ = state.begin_sampling(started_at + Duration::from_millis(400));
state.end_phase(
started_at + Duration::from_millis(700),
TurnProfilePhase::Sampling,
);
let _ = state.begin_compaction(started_at + Duration::from_millis(800));

assert_eq!(
state.complete(started_at + Duration::from_millis(1_100)),
TurnProfile {
before_first_sampling_ms: 200,
sampling_ms: 300,
compaction_ms: 500,
between_sampling_overhead_ms: 0,
tool_blocking_ms: 0,
after_last_sampling_ms: 100,
sampling_request_count: 1,
sampling_retry_count: 0,
}
);
}

#[tokio::test]
async fn turn_profile_and_duration_share_a_completion_instant() {
let state = TurnTimingState::default();
state
.mark_turn_started(Instant::now() - Duration::from_millis(100))
.await;

let (_, duration_ms, profile) = state.complete_profile_and_duration_ms().await;
let classified_ms = profile.before_first_sampling_ms
+ profile.sampling_ms
+ profile.compaction_ms
+ profile.between_sampling_overhead_ms
+ profile.tool_blocking_ms
+ profile.after_last_sampling_ms;

assert_eq!(
duration_ms,
Some(i64::try_from(classified_ms).expect("profile duration should fit i64"))
);
}
Loading