diff --git a/codex-rs/core/src/guardian/input_budget.rs b/codex-rs/core/src/guardian/input_budget.rs index 529c0d65a5..9c5fbfdb97 100644 --- a/codex-rs/core/src/guardian/input_budget.rs +++ b/codex-rs/core/src/guardian/input_budget.rs @@ -3,7 +3,9 @@ //! recording. Budget failures preserve it for a compaction retry; successful //! finalization consumes it. Checkpoint recovery skips startup compaction and //! budgets the fresh full input against the preserved checkpoint. +//! Loss telemetry describes accepted inputs, excluding feasibility checks and failed retries. +use std::collections::BTreeMap; use std::sync::Arc; use std::sync::atomic::AtomicBool; use std::sync::atomic::Ordering; @@ -224,7 +226,7 @@ pub(crate) async fn finalize( } else { history_truncation }; - let context = context + let mut context = context .enforce_budget(budget, GuardianBudgetOmission.render(), history_truncation) .map_err(|error| { session @@ -255,6 +257,7 @@ pub(crate) async fn finalize( ); } } + let observations = std::mem::take(&mut context.truncations); match context.clone().into_annotated_user_inputs() { Ok((user_input, metadata)) => { if metadata.is_some() { @@ -304,6 +307,47 @@ pub(crate) async fn finalize( } Err(error) => return Err(CodexErr::InvalidRequest(error.to_string())), } + // Match async classification accounting: one count per component/disposition + // for this accepted input, with byte totals across all reduction stages. + let mut truncations: BTreeMap<(&str, &str), (usize, usize)> = BTreeMap::new(); + for observation in observations { + let disposition = if observation.retained_bytes == 0 { + "omitted" + } else { + "truncated" + }; + let totals = truncations + .entry((observation.component, disposition)) + .or_default(); + totals.0 = totals.0.saturating_add(observation.original_bytes); + totals.1 = totals.1.saturating_add(observation.retained_bytes); + } + for ((component, disposition), (original_bytes, retained_bytes)) in truncations { + session.services.session_telemetry.counter( + codex_otel::GUARDIAN_REVIEW_TRUNCATION_METRIC, + /*inc*/ 1, + &[("component", component), ("disposition", disposition)], + ); + for (measurement, bytes) in [ + ("original", original_bytes), + ("retained", retained_bytes), + ("omitted", original_bytes.saturating_sub(retained_bytes)), + ] { + session + .services + .session_telemetry + .histogram_with_boundaries( + codex_otel::GUARDIAN_REVIEW_TRUNCATION_BYTES_METRIC, + i64::try_from(bytes).unwrap_or(i64::MAX), + codex_guardian_context::SECTION_COST_BOUNDARIES, + &[ + ("component", component), + ("disposition", disposition), + ("measurement", measurement), + ], + ); + } + } step.turn.extension_data.insert(RetainedReviewContext { context: pending.0.retained_instructions(), history_version, diff --git a/codex-rs/core/src/guardian/input_budget_tests.rs b/codex-rs/core/src/guardian/input_budget_tests.rs index 3f0b66f0c4..d623337dc9 100644 --- a/codex-rs/core/src/guardian/input_budget_tests.rs +++ b/codex-rs/core/src/guardian/input_budget_tests.rs @@ -5,10 +5,141 @@ use codex_guardian_context::ContextPresentation; use codex_guardian_context::ContextProfile; use codex_guardian_context::PlannedAction; use codex_guardian_context::PlannedActionKind; +use codex_guardian_context::TruncationObservation; +use codex_otel::MetricsClient; +use codex_otel::MetricsConfig; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::TurnAbortReason; +use opentelemetry_sdk::metrics::InMemoryMetricExporter; +use opentelemetry_sdk::metrics::data::AggregatedMetrics; +use opentelemetry_sdk::metrics::data::MetricData; use pretty_assertions::assert_eq; use std::sync::Arc; +use test_case::test_case; + +#[tokio::test] +#[test_case(128_000; "accepted_input")] +#[test_case(1; "rejected_input")] +async fn finalization_exports_loss_totals_only_for_accepted_input(window: i64) { + let metrics = MetricsClient::new( + MetricsConfig::in_memory( + "test", + "codex-core", + env!("CARGO_PKG_VERSION"), + InMemoryMetricExporter::default(), + ) + .with_runtime_reader(), + ) + .unwrap(); + let (mut session, mut turn) = crate::session::tests::make_session_and_context().await; + session.services.session_telemetry = session + .services + .session_telemetry + .clone() + .with_metrics(metrics.clone()); + Arc::make_mut(&mut turn.config).model_context_window = Some(window); + let session = Arc::new(session); + let step = session + .capture_step_context(Arc::new(turn), &tokio_util::sync::CancellationToken::new()) + .await + .unwrap(); + let mut context = required_context("required action".to_owned()); + // Composition and final selection can reduce the same component multiple times. + context.truncations = [(256, 100), (64, 40), (16, 0)] + .into_iter() + .map(|(original_bytes, retained_bytes)| TruncationObservation { + component: "conversation_transcript", + original_bytes, + retained_bytes, + }) + .collect(); + let mut input = vec![TurnInput::UserInput { + metadata: Default::default(), + content: context.clone().into_user_inputs().unwrap(), + client_id: None, + }]; + session + .services + .thread_extension_data + .insert(PendingReviewContext(context)); + assert_eq!( + finalize(&session, &step, &mut input, HistoryTruncation::Preserve) + .await + .is_ok(), + window > 1, + ); + // A second finalization has no pending input and must not emit the loss twice. + if window > 1 { + finalize(&session, &step, &mut input, HistoryTruncation::Preserve) + .await + .unwrap(); + } + let snapshot = metrics.snapshot().unwrap(); + let mut counts = BTreeMap::new(); + let mut bytes = BTreeMap::new(); + for metric in snapshot + .scope_metrics() + .flat_map(opentelemetry_sdk::metrics::data::ScopeMetrics::metrics) + { + match metric.name() { + codex_otel::GUARDIAN_REVIEW_TRUNCATION_METRIC => { + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else { + panic!("expected loss counter"); + }; + for point in sum.data_points() { + let tags = point + .attributes() + .filter(|attr| matches!(attr.key.as_str(), "component" | "disposition")) + .map(|attr| (attr.key.as_str().to_owned(), attr.value.to_string())) + .collect::>(); + counts.insert(tags, point.value()); + } + } + codex_otel::GUARDIAN_REVIEW_TRUNCATION_BYTES_METRIC => { + let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() else { + panic!("expected loss histogram"); + }; + for point in histogram.data_points() { + let tags = point + .attributes() + .filter(|attr| { + matches!( + attr.key.as_str(), + "component" | "disposition" | "measurement" + ) + }) + .map(|attr| (attr.key.as_str().to_owned(), attr.value.to_string())) + .collect::>(); + bytes.insert(tags, (point.count(), point.sum())); + } + } + _ => {} + } + } + let mut expected_counts = BTreeMap::new(); + let mut expected_bytes = BTreeMap::new(); + if window > 1 { + for (disposition, original, retained) in + [("omitted", 16.0, 0.0), ("truncated", 320.0, 140.0)] + { + let tags = BTreeMap::from([ + ("component".to_owned(), "conversation_transcript".to_owned()), + ("disposition".to_owned(), disposition.to_owned()), + ]); + expected_counts.insert(tags.clone(), 1); + for (measurement, value) in [ + ("original", original), + ("retained", retained), + ("omitted", original - retained), + ] { + let mut tags = tags.clone(); + tags.insert("measurement".to_owned(), measurement.to_owned()); + expected_bytes.insert(tags, (1, value)); + } + } + } + assert_eq!((counts, bytes), (expected_counts, expected_bytes)); +} fn required_context(text: String) -> ComposedContext { let action = PlannedAction { diff --git a/codex-rs/core/tests/suite/dynamic_tool_cancellation.rs b/codex-rs/core/tests/suite/dynamic_tool_cancellation.rs index 7edda77edb..f187b8a83d 100644 --- a/codex-rs/core/tests/suite/dynamic_tool_cancellation.rs +++ b/codex-rs/core/tests/suite/dynamic_tool_cancellation.rs @@ -61,29 +61,33 @@ async fn interrupt_completes_cancelled_dynamic_call(hook: Option) let outcomes = Arc::new(ToolOutcomes::default()); let mut extensions = ExtensionRegistryBuilder::new(); extensions.tool_lifecycle_contributor(outcomes.clone()); - let test = - test_codex() - .with_extensions(Arc::new(extensions.build())) - .with_pre_build_hook(move |home| { - if let Some(hook) = hook { - let name = match hook { - HookEventName::PreToolUse => "PreToolUse", - HookEventName::PostToolUse => "PostToolUse", - _ => unreachable!(), - }; - std::fs::write(home.join("hooks.json"), json!({"hooks": { + let mut builder = test_codex() + .with_extensions(Arc::new(extensions.build())) + .with_pre_build_hook(move |home| { + if let Some(hook) = hook { + let name = match hook { + HookEventName::PreToolUse => "PreToolUse", + HookEventName::PostToolUse => "PostToolUse", + _ => unreachable!(), + }; + std::fs::write(home.join("hooks.json"), json!({"hooks": { name: [{"matcher": "^gate$", "hooks": [{"type": "command", - "command": "python3 -c 'import threading; threading.Event().wait()'"}]}] + "command": r#"python3 -c "import threading; threading.Event().wait()""#}]}] }}).to_string()).expect("write blocking hook fixture"); - } - }) - .with_config(move |config| { - if hook.is_some() { - trust_discovered_hooks(config); - } - }) - .build_with_auto_env(&server) - .await?; + } + }) + .with_config(move |config| { + if hook.is_some() { + trust_discovered_hooks(config); + } + }); + // Command hooks execute on the host. Keep their blocking fixture in a local + // environment so a remote Windows shell is not launched on a Linux host. + let test = if hook.is_some() { + builder.build(&server).await? + } else { + builder.build_with_auto_env(&server).await? + }; let thread = test .thread_manager .start_thread(StartThreadOptions { diff --git a/codex-rs/otel/src/metrics/names.rs b/codex-rs/otel/src/metrics/names.rs index d4aa1a6e69..f7c6242ee9 100644 --- a/codex-rs/otel/src/metrics/names.rs +++ b/codex-rs/otel/src/metrics/names.rs @@ -46,6 +46,8 @@ pub const GUARDIAN_REVIEW_COUNT_METRIC: &str = "codex.guardian.review"; pub const GUARDIAN_REVIEW_DURATION_METRIC: &str = "codex.guardian.review.duration_ms"; pub const GUARDIAN_REVIEW_TTFT_DURATION_METRIC: &str = "codex.guardian.review.ttft.duration_ms"; pub const GUARDIAN_REVIEW_TOKEN_USAGE_METRIC: &str = "codex.guardian.review.token_usage"; +pub const GUARDIAN_REVIEW_TRUNCATION_METRIC: &str = "codex.guardian.review.truncation"; +pub const GUARDIAN_REVIEW_TRUNCATION_BYTES_METRIC: &str = "codex.guardian.review.truncation.bytes"; pub const GOAL_CREATED_METRIC: &str = "codex.goal.created"; pub const GOAL_RESUMED_METRIC: &str = "codex.goal.resumed"; pub const GOAL_COMPLETED_METRIC: &str = "codex.goal.completed";