From 91b735018779daed7c40f86aab9bec9abc9922e8 Mon Sep 17 00:00:00 2001 From: Eric Traut Date: Tue, 5 May 2026 09:21:54 -0700 Subject: [PATCH] Add goal lifecycle metrics (#20799) ## Why Adding goal metrics makes it possible to track how often goals are created, completed, and stopped by budget limits, plus the final token and wall-clock usage for terminal outcomes. ## What Changed - Added OpenTelemetry metric constants for goal lifecycle tracking: - `codex.goal.created`: increments each time a new persisted goal is created or an existing goal is replaced with a new objective. - `codex.goal.completed`: increments when a goal transitions to `complete`. - `codex.goal.budget_limited`: increments when a goal transitions to `budget_limited` because its token budget has been reached. - `codex.goal.token_count`: records the final persisted token count when a goal transitions to `complete` or `budget_limited`. - `codex.goal.duration_s`: records the final persisted elapsed wall-clock time, in seconds, when a goal transitions to `complete` or `budget_limited`. - Emitted creation metrics when a goal is created or replaced. - Emitted terminal outcome counters and final usage histograms when a goal transitions to `complete` or `budget_limited`, avoiding double-counting later in-flight accounting for already budget-limited goals. - Added focused `codex-core` tests for create/complete metrics and one-time budget-limit metrics. --- codex-rs/app-server/src/request_processors.rs | 2 + .../thread_goal_processor.rs | 24 ++- codex-rs/core/src/codex_thread.rs | 5 +- codex-rs/core/src/goals.rs | 198 ++++++++++++++---- codex-rs/core/src/lib.rs | 2 + codex-rs/core/src/session/tests.rs | 20 +- codex-rs/otel/src/metrics/names.rs | 5 + 7 files changed, 201 insertions(+), 55 deletions(-) diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index cb713ac6e..a2b76bf0b 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -246,6 +246,8 @@ use codex_config::loader::project_trust_key; use codex_config::types::McpServerTransportConfig; use codex_core::CodexThread; use codex_core::CodexThreadTurnContextOverrides; +use codex_core::ExternalGoalPreviousStatus; +use codex_core::ExternalGoalSet; use codex_core::ForkSnapshot; use codex_core::NewThread; #[cfg(test)] diff --git a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs index 2e7c9909f..0e12e44ce 100644 --- a/codex-rs/app-server/src/request_processors/thread_goal_processor.rs +++ b/codex-rs/app-server/src/request_processors/thread_goal_processor.rs @@ -148,7 +148,7 @@ impl ThreadGoalRequestProcessor { thread.prepare_external_goal_mutation().await; } - let goal = (if let Some(objective) = objective { + let (goal, previous_status) = (if let Some(objective) = objective { let existing_goal = state_db .get_thread_goal(thread_id) .await @@ -157,6 +157,7 @@ impl ThreadGoalRequestProcessor { goal.objective == objective && goal.status != codex_state::ThreadGoalStatus::Complete }) { + let previous_status = ExternalGoalPreviousStatus::Existing(goal.status); state_db .update_thread_goal( thread_id, @@ -174,7 +175,9 @@ impl ThreadGoalRequestProcessor { ) }) }) + .map(|goal| (goal, previous_status)) } else { + let previous_status = ExternalGoalPreviousStatus::NewGoal; state_db .replace_thread_goal( thread_id, @@ -183,8 +186,19 @@ impl ThreadGoalRequestProcessor { params.token_budget.flatten(), ) .await + .map(|goal| (goal, previous_status)) } } else { + let existing_goal = state_db + .get_thread_goal(thread_id) + .await + .map_err(|err| invalid_request(err.to_string()))?; + let Some(existing_goal) = existing_goal else { + return Err(invalid_request(format!( + "cannot update goal for thread {thread_id}: no goal exists" + ))); + }; + let previous_status = ExternalGoalPreviousStatus::Existing(existing_goal.status); state_db .update_thread_goal( thread_id, @@ -200,9 +214,13 @@ impl ThreadGoalRequestProcessor { anyhow::anyhow!("cannot update goal for thread {thread_id}: no goal exists") }) }) + .map(|goal| (goal, previous_status)) }) .map_err(|err| invalid_request(err.to_string()))?; - let goal_status = goal.status; + let external_goal_set = ExternalGoalSet { + goal: goal.clone(), + previous_status, + }; let goal = api_thread_goal_from_state(goal); self.outgoing .send_response( @@ -213,7 +231,7 @@ impl ThreadGoalRequestProcessor { self.emit_thread_goal_updated_ordered(thread_id, goal, listener_command_tx) .await; if let Some(thread) = running_thread.as_ref() { - thread.apply_external_goal_set(goal_status).await; + thread.apply_external_goal_set(external_goal_set).await; } Ok(()) } diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index b4167a0ec..cd96ac927 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -1,6 +1,7 @@ use crate::agent::AgentStatus; use crate::config::ConstraintResult; use crate::file_watcher::WatchRegistration; +use crate::goals::ExternalGoalSet; use crate::goals::GoalRuntimeEvent; use crate::session::Codex; use crate::session::SessionSettingsUpdate; @@ -160,11 +161,11 @@ impl CodexThread { } } - pub async fn apply_external_goal_set(&self, status: codex_state::ThreadGoalStatus) { + pub async fn apply_external_goal_set(&self, external_set: ExternalGoalSet) { if let Err(err) = self .codex .session - .goal_runtime_apply(GoalRuntimeEvent::ExternalSet { status }) + .goal_runtime_apply(GoalRuntimeEvent::ExternalSet { external_set }) .await { tracing::warn!("failed to apply external goal status runtime effects: {err}"); diff --git a/codex-rs/core/src/goals.rs b/codex-rs/core/src/goals.rs index 4165c0514..3cea25f9b 100644 --- a/codex-rs/core/src/goals.rs +++ b/codex-rs/core/src/goals.rs @@ -12,6 +12,11 @@ use crate::state::TurnState; use crate::tasks::RegularTask; use anyhow::Context; use codex_features::Feature; +use codex_otel::GOAL_BUDGET_LIMITED_METRIC; +use codex_otel::GOAL_COMPLETED_METRIC; +use codex_otel::GOAL_CREATED_METRIC; +use codex_otel::GOAL_DURATION_SECONDS_METRIC; +use codex_otel::GOAL_TOKEN_COUNT_METRIC; use codex_protocol::config_types::ModeKind; use codex_protocol::models::ContentItem; use codex_protocol::models::ResponseInputItem; @@ -68,6 +73,27 @@ enum BudgetLimitSteering { Suppressed, } +#[derive(Clone, Copy)] +enum TerminalMetricEmission { + Emit, + Suppress, +} + +/// Describes whether an external goal mutation created a new logical goal or +/// updated an existing one. +#[derive(Clone, Copy)] +pub enum ExternalGoalPreviousStatus { + NewGoal, + Existing(codex_state::ThreadGoalStatus), +} + +/// Runtime effects for an externally persisted goal mutation. +#[derive(Clone)] +pub struct ExternalGoalSet { + pub goal: codex_state::ThreadGoal, + pub previous_status: ExternalGoalPreviousStatus, +} + /// Runtime lifecycle events that can affect goal accounting, scheduling, or /// model-visible steering. /// @@ -96,7 +122,7 @@ pub(crate) enum GoalRuntimeEvent<'a> { }, ExternalMutationStarting, ExternalSet { - status: codex_state::ThreadGoalStatus, + external_set: ExternalGoalSet, }, ExternalClear, ThreadResumed, @@ -293,14 +319,22 @@ impl Session { tool_name, } => Box::pin(async move { if tool_name != codex_tools::UPDATE_GOAL_TOOL_NAME { - self.account_thread_goal_progress(turn_context, BudgetLimitSteering::Allowed) - .await?; + self.account_thread_goal_progress( + turn_context, + BudgetLimitSteering::Allowed, + TerminalMetricEmission::Emit, + ) + .await?; } Ok(()) }), GoalRuntimeEvent::ToolCompletedGoal { turn_context } => Box::pin(async move { - self.account_thread_goal_progress(turn_context, BudgetLimitSteering::Suppressed) - .await?; + self.account_thread_goal_progress( + turn_context, + BudgetLimitSteering::Suppressed, + TerminalMetricEmission::Suppress, + ) + .await?; Ok(()) }), GoalRuntimeEvent::TurnFinished { @@ -331,8 +365,8 @@ impl Session { } Ok(()) }), - GoalRuntimeEvent::ExternalSet { status } => Box::pin(async move { - self.apply_external_thread_goal_status(status).await; + GoalRuntimeEvent::ExternalSet { external_set } => Box::pin(async move { + self.apply_external_thread_goal_status(external_set).await; Ok(()) }), GoalRuntimeEvent::ExternalClear => Box::pin(async move { @@ -384,6 +418,7 @@ impl Session { self.account_thread_goal_wall_clock_usage( &state_db, codex_state::ThreadGoalAccountingMode::ActiveOnly, + TerminalMetricEmission::Emit, ) .await?; let mut replacing_goal = objective.is_some(); @@ -454,6 +489,15 @@ impl Session { let goal_status = goal.status; let goal_id = goal.goal_id.clone(); + let previous_status_for_goal = if replacing_goal { + None + } else { + previous_status + }; + if replacing_goal { + self.emit_goal_created_metric(); + } + self.emit_goal_terminal_metrics_if_status_changed(previous_status_for_goal, &goal); let goal = protocol_goal_from_state(goal); *self.goal_runtime.budget_limit_reported_goal_id.lock().await = None; let newly_active_goal = goal_status == codex_state::ThreadGoalStatus::Active @@ -504,6 +548,7 @@ impl Session { self.account_thread_goal_wall_clock_usage( &state_db, codex_state::ThreadGoalAccountingMode::ActiveOnly, + TerminalMetricEmission::Emit, ) .await?; let goal = state_db @@ -522,6 +567,7 @@ impl Session { })?; let goal_id = goal.goal_id.clone(); + self.emit_goal_created_metric(); let goal = protocol_goal_from_state(goal); *self.goal_runtime.budget_limit_reported_goal_id.lock().await = None; @@ -545,44 +591,30 @@ impl Session { Ok(goal) } - async fn apply_external_thread_goal_status( - self: &Arc, - status: codex_state::ThreadGoalStatus, - ) { + async fn apply_external_thread_goal_status(self: &Arc, external_set: ExternalGoalSet) { + let ExternalGoalSet { + goal, + previous_status, + } = external_set; + let previous_status = match previous_status { + ExternalGoalPreviousStatus::NewGoal => { + self.emit_goal_created_metric(); + None + } + ExternalGoalPreviousStatus::Existing(status) => Some(status), + }; + self.emit_goal_terminal_metrics_if_status_changed(previous_status, &goal); + let goal_id = goal.goal_id; + let status = goal.status; match status { codex_state::ThreadGoalStatus::Active => { - match self.state_db_for_thread_goals().await { - Ok(Some(state_db)) => { - match state_db.get_thread_goal(self.conversation_id).await { - Ok(Some(goal)) - if goal.status == codex_state::ThreadGoalStatus::Active => - { - let turn_id = self - .active_turn_context() - .await - .map(|turn_context| turn_context.sub_id.clone()); - let current_token_usage = - self.total_token_usage().await.unwrap_or_default(); - self.mark_active_goal_accounting( - goal.goal_id, - turn_id, - current_token_usage, - ) - .await; - } - Ok(Some(_)) | Ok(None) => {} - Err(err) => { - tracing::warn!( - "failed to read active goal after external set: {err}" - ); - } - } - } - Err(err) => { - tracing::warn!("failed to open state db after external goal set: {err}"); - } - Ok(None) => {} - } + let turn_id = self + .active_turn_context() + .await + .map(|turn_context| turn_context.sub_id.clone()); + let current_token_usage = self.total_token_usage().await.unwrap_or_default(); + self.mark_active_goal_accounting(goal_id, turn_id, current_token_usage) + .await; self.maybe_continue_goal_if_idle_runtime().await; } codex_state::ThreadGoalStatus::BudgetLimited => { @@ -638,6 +670,57 @@ impl Session { accounting.wall_clock.mark_active_goal(goal_id); } + fn emit_goal_created_metric(&self) { + self.services + .session_telemetry + .counter(GOAL_CREATED_METRIC, /*inc*/ 1, &[]); + } + + fn emit_goal_terminal_metrics_if_status_changed( + &self, + previous_status: Option, + goal: &codex_state::ThreadGoal, + ) { + if previous_status == Some(goal.status) { + return; + } + + let counter = match goal.status { + codex_state::ThreadGoalStatus::BudgetLimited => GOAL_BUDGET_LIMITED_METRIC, + codex_state::ThreadGoalStatus::Complete => GOAL_COMPLETED_METRIC, + codex_state::ThreadGoalStatus::Active | codex_state::ThreadGoalStatus::Paused => { + return; + } + }; + let status_tag = [("status", goal.status.as_str())]; + self.services + .session_telemetry + .counter(counter, /*inc*/ 1, &[]); + self.services.session_telemetry.histogram( + GOAL_TOKEN_COUNT_METRIC, + goal.tokens_used, + &status_tag, + ); + self.services.session_telemetry.histogram( + GOAL_DURATION_SECONDS_METRIC, + goal.time_used_seconds, + &status_tag, + ); + } + + async fn current_goal_status_for_metrics( + &self, + state_db: &StateDbHandle, + expected_goal_id: Option<&str>, + ) -> anyhow::Result> { + let goal = state_db.get_thread_goal(self.conversation_id).await?; + Ok(goal.and_then(|goal| { + expected_goal_id + .is_none_or(|expected_goal_id| goal.goal_id == expected_goal_id) + .then_some(goal.status) + })) + } + async fn active_turn_context(&self) -> Option> { let active = self.active_turn.lock().await; active @@ -732,7 +815,11 @@ impl Session { ) { if turn_completed && let Err(err) = self - .account_thread_goal_progress(turn_context, BudgetLimitSteering::Suppressed) + .account_thread_goal_progress( + turn_context, + BudgetLimitSteering::Suppressed, + TerminalMetricEmission::Emit, + ) .await { tracing::warn!("failed to account thread goal progress at turn end: {err}"); @@ -761,7 +848,11 @@ impl Session { self.take_thread_goal_continuation_turn(&turn_context.sub_id) .await; if let Err(err) = self - .account_thread_goal_progress(turn_context, BudgetLimitSteering::Suppressed) + .account_thread_goal_progress( + turn_context, + BudgetLimitSteering::Suppressed, + TerminalMetricEmission::Emit, + ) .await { tracing::warn!("failed to account thread goal progress after abort: {err}"); @@ -787,6 +878,7 @@ impl Session { &self, turn_context: &TurnContext, budget_limit_steering: BudgetLimitSteering, + terminal_metric_emission: TerminalMetricEmission, ) -> anyhow::Result<()> { if !self.enabled(Feature::Goals) { return Ok(()); @@ -820,6 +912,9 @@ impl Session { if time_delta_seconds == 0 && token_delta <= 0 { return Ok(()); } + let previous_status = self + .current_goal_status_for_metrics(&state_db, expected_goal_id.as_deref()) + .await?; let outcome = state_db .account_thread_goal_usage( self.conversation_id, @@ -862,6 +957,9 @@ impl Session { accounting.wall_clock.clear_active_goal(); } } + if matches!(terminal_metric_emission, TerminalMetricEmission::Emit) { + self.emit_goal_terminal_metrics_if_status_changed(previous_status, &goal); + } goal } codex_state::ThreadGoalAccountingOutcome::Unchanged(_) => return Ok(()), @@ -901,6 +999,7 @@ impl Session { .account_thread_goal_progress( turn_context.as_ref(), BudgetLimitSteering::Suppressed, + TerminalMetricEmission::Emit, ) .await; } @@ -911,6 +1010,7 @@ impl Session { self.account_thread_goal_wall_clock_usage( &state_db, codex_state::ThreadGoalAccountingMode::ActiveOnly, + TerminalMetricEmission::Suppress, ) .await?; Ok(()) @@ -920,6 +1020,7 @@ impl Session { &self, state_db: &StateDbHandle, mode: codex_state::ThreadGoalAccountingMode, + terminal_metric_emission: TerminalMetricEmission, ) -> anyhow::Result> { let _accounting_permit = self.goal_runtime.accounting_permit().await?; let (time_delta_seconds, expected_goal_id) = { @@ -932,6 +1033,9 @@ impl Session { if time_delta_seconds == 0 { return Ok(None); } + let previous_status = self + .current_goal_status_for_metrics(state_db, expected_goal_id.as_deref()) + .await?; match state_db .account_thread_goal_usage( @@ -944,6 +1048,9 @@ impl Session { .await? { codex_state::ThreadGoalAccountingOutcome::Updated(goal) => { + if matches!(terminal_metric_emission, TerminalMetricEmission::Emit) { + self.emit_goal_terminal_metrics_if_status_changed(previous_status, &goal); + } self.goal_runtime .accounting .lock() @@ -989,6 +1096,7 @@ impl Session { self.account_thread_goal_wall_clock_usage( &state_db, codex_state::ThreadGoalAccountingMode::ActiveStatusOnly, + TerminalMetricEmission::Emit, ) .await?; let Some(goal) = state_db diff --git a/codex-rs/core/src/lib.rs b/codex-rs/core/src/lib.rs index cf909c5e8..d8a3fbcf0 100644 --- a/codex-rs/core/src/lib.rs +++ b/codex-rs/core/src/lib.rs @@ -39,6 +39,8 @@ mod flags; #[cfg(test)] mod git_info_tests; mod goals; +pub use goals::ExternalGoalPreviousStatus; +pub use goals::ExternalGoalSet; mod guardian; mod hook_runtime; mod installation_id; diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 0a8ea46f8..dbb0fab95 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -52,6 +52,8 @@ use codex_protocol::request_permissions::PermissionGrantScope; use codex_protocol::request_permissions::RequestPermissionProfile; use tracing::Span; +use crate::goals::ExternalGoalPreviousStatus; +use crate::goals::ExternalGoalSet; use crate::goals::GoalRuntimeEvent; use crate::goals::SetGoalRequest; use crate::rollout::recorder::RolloutRecorder; @@ -7505,19 +7507,24 @@ async fn external_goal_mutation_accounts_active_turn_before_status_change() -> a .expect("goal should remain persisted"); assert_eq!(70, goal.tokens_used); - state_db + let previous_status = goal.status; + let goal_id = goal.goal_id.clone(); + let updated_goal = state_db .update_thread_goal( sess.conversation_id, codex_state::ThreadGoalUpdate { status: Some(codex_state::ThreadGoalStatus::Complete), token_budget: None, - expected_goal_id: Some(goal.goal_id), + expected_goal_id: Some(goal_id), }, ) .await? .expect("goal status update should succeed"); sess.goal_runtime_apply(GoalRuntimeEvent::ExternalSet { - status: codex_state::ThreadGoalStatus::Complete, + external_set: ExternalGoalSet { + goal: updated_goal, + previous_status: ExternalGoalPreviousStatus::Existing(previous_status), + }, }) .await?; @@ -7549,7 +7556,7 @@ async fn external_active_goal_set_marks_current_turn_for_accounting() -> anyhow: set_total_token_usage(&sess, post_goal_token_usage()).await; let state_db = goal_test_state_db(sess.as_ref()).await?; - state_db + let goal = state_db .replace_thread_goal( sess.conversation_id, "Keep improving the benchmark", @@ -7558,7 +7565,10 @@ async fn external_active_goal_set_marks_current_turn_for_accounting() -> anyhow: ) .await?; sess.goal_runtime_apply(GoalRuntimeEvent::ExternalSet { - status: codex_state::ThreadGoalStatus::Active, + external_set: ExternalGoalSet { + goal, + previous_status: ExternalGoalPreviousStatus::NewGoal, + }, }) .await?; diff --git a/codex-rs/otel/src/metrics/names.rs b/codex-rs/otel/src/metrics/names.rs index aca120f1e..dc4937216 100644 --- a/codex-rs/otel/src/metrics/names.rs +++ b/codex-rs/otel/src/metrics/names.rs @@ -27,6 +27,11 @@ pub const TURN_NETWORK_PROXY_METRIC: &str = "codex.turn.network_proxy"; pub const TURN_MEMORY_METRIC: &str = "codex.turn.memory"; pub const TURN_TOOL_CALL_METRIC: &str = "codex.turn.tool.call"; pub const TURN_TOKEN_USAGE_METRIC: &str = "codex.turn.token_usage"; +pub const GOAL_CREATED_METRIC: &str = "codex.goal.created"; +pub const GOAL_COMPLETED_METRIC: &str = "codex.goal.completed"; +pub const GOAL_BUDGET_LIMITED_METRIC: &str = "codex.goal.budget_limited"; +pub const GOAL_TOKEN_COUNT_METRIC: &str = "codex.goal.token_count"; +pub const GOAL_DURATION_SECONDS_METRIC: &str = "codex.goal.duration_s"; pub const PROFILE_USAGE_METRIC: &str = "codex.profile.usage"; pub const CURATED_PLUGINS_STARTUP_SYNC_METRIC: &str = "codex.plugins.startup_sync"; pub const CURATED_PLUGINS_STARTUP_SYNC_FINAL_METRIC: &str = "codex.plugins.startup_sync.final";