Record truncation metrics for accepted Guardian review inputs (#53628)

## What changed

Emit `codex.guardian.review.truncation` counts and `codex.guardian.review.truncation.bytes` histograms after successful input finalization. Aggregate observations across reduction stages by component and disposition (`truncated` or `omitted`), recording original, retained, and omitted byte totals. Exclude feasibility checks and failed retries from loss telemetry.

Keep blocking command-hook cancellation fixtures in the local environment and use double quotes for the Python command.

## Testing

Add metric-export tests covering aggregation across multiple reductions, separate truncation and omission totals, no loss metrics for rejected inputs, and no duplicate reporting on repeated finalization.

GitOrigin-RevId: 69182fc7afe230be69dd2469caedb5ae4494947b

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,

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::<BTreeMap<_, _>>();
+                    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::<BTreeMap<_, _>>();
+                    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 {

codex-rs/core/tests/suite/dynamic_tool_cancellation.rs

@@ -61,29 +61,33 @@ async fn interrupt_completes_cancelled_dynamic_call(hook: Option<HookEventName>)
     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 {

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";