From 64b8d019c8881acf1a8bc337db6a1f91eaef3076 Mon Sep 17 00:00:00 2001 From: Alex Kira Date: Thu, 12 Nov 2026 04:05:29 +0000 Subject: [PATCH] Record reasoning replay removal metrics (#61017) ## What changed Parse reasoning removals from response `input_transformations` and carry the counts through completion events to sampling telemetry. - Add `codex.responses.reasoning_replay` to distinguish unavailable reports, no reported removals, prefix mismatches, and other removals. - Add `codex.responses.reasoning_items_removed` histograms by bounded removal reason, grouping unknown or missing reasons under `other`. - Include zero counts for usable reports; exclude missing, null, or malformed reports from histograms without failing the response. - Use the sampling step's session telemetry when recording response events. ## Testing Add SSE and WebSocket integration coverage for report availability, removal counts, unknown reasons, zero observations, and `product_sku` tags in exported metrics. GitOrigin-RevId: 83c70ca6067fa4b34bf5c463958b4632a7a0a188 --- codex-rs/codex-api/src/common.rs | 38 ++++ codex-rs/codex-api/src/lib.rs | 1 + codex-rs/codex-api/src/sse/responses.rs | 33 ++++ codex-rs/codex-api/tests/sse_end_to_end.rs | 1 + codex-rs/core/src/client.rs | 2 + codex-rs/core/src/client_tests.rs | 2 + codex-rs/core/src/compact_remote_v2.rs | 2 + codex-rs/core/src/session/turn.rs | 3 +- codex-rs/core/tests/suite/mod.rs | 1 + .../tests/suite/reasoning_replay_metrics.rs | 185 ++++++++++++++++++ codex-rs/otel/src/events/session_telemetry.rs | 34 ++++ codex-rs/otel/src/metrics/names.rs | 3 + 12 files changed, 304 insertions(+), 1 deletion(-) create mode 100644 codex-rs/core/tests/suite/reasoning_replay_metrics.rs diff --git a/codex-rs/codex-api/src/common.rs b/codex-rs/codex-api/src/common.rs index 455836c2b8..3942f3aa69 100644 --- a/codex-rs/codex-api/src/common.rs +++ b/codex-rs/codex-api/src/common.rs @@ -16,6 +16,7 @@ use serde::Deserialize; use serde::Serialize; use serde_json::Value; use serde_json::value::RawValue; +use std::collections::BTreeMap; use std::collections::HashMap; use std::pin::Pin; use std::sync::Arc; @@ -78,6 +79,40 @@ pub struct MemorySummarizeOutput { #[derive(Clone, Debug)] pub struct ResponseId(pub String); +/// A bounded reason for removing a replayed reasoning item from model input. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub enum ReasoningRemovalReason { + OrganizationMismatch, + EndUserMismatch, + ModelMismatch, + PrefixMismatch, + TurnBoundary, + Other, +} + +impl ReasoningRemovalReason { + /// Every metric category, including unrecognized API reasons. + pub const ALL: [Self; 6] = [ + Self::OrganizationMismatch, + Self::EndUserMismatch, + Self::ModelMismatch, + Self::PrefixMismatch, + Self::TurnBoundary, + Self::Other, + ]; + + pub fn as_str(self) -> &'static str { + match self { + Self::OrganizationMismatch => "organization_mismatch", + Self::EndUserMismatch => "end_user_mismatch", + Self::ModelMismatch => "model_mismatch", + Self::PrefixMismatch => "prefix_mismatch", + Self::TurnBoundary => "turn_boundary", + Self::Other => "other", + } + } +} + #[derive(Debug)] pub enum ResponseEvent { Created { @@ -104,6 +139,9 @@ pub enum ResponseEvent { response_id: String, token_usage: Option, usage_metadata: Option, + /// Reported reasoning removals by bounded reason. Missing or non-array + /// reports are unavailable; an empty map means no removals were reported. + reasoning_removals: Option>, /// Did the model affirmatively end its turn? Some providers do not set this, /// so we rely on fallback logic when this is `None`. end_turn: Option, diff --git a/codex-rs/codex-api/src/lib.rs b/codex-rs/codex-api/src/lib.rs index 68b0c1c4ae..c2e302f9f1 100644 --- a/codex-rs/codex-api/src/lib.rs +++ b/codex-rs/codex-api/src/lib.rs @@ -42,6 +42,7 @@ pub use crate::common::RawMemory; pub use crate::common::RawMemoryMetadata; pub use crate::common::Reasoning; pub use crate::common::ReasoningContext; +pub use crate::common::ReasoningRemovalReason; pub use crate::common::ReasoningSummaryDelivery; pub use crate::common::ResponseCreateWsRequest; pub use crate::common::ResponseEvent; diff --git a/codex-rs/codex-api/src/sse/responses.rs b/codex-rs/codex-api/src/sse/responses.rs index 9c12aaed21..eafc92dd69 100644 --- a/codex-rs/codex-api/src/sse/responses.rs +++ b/codex-rs/codex-api/src/sse/responses.rs @@ -1,4 +1,5 @@ use super::responses_error::parse_failed_response; +use crate::common::ReasoningRemovalReason; use crate::common::ResponseEvent; use crate::common::ResponseStream; use crate::common::SafetyBuffering; @@ -21,6 +22,7 @@ use eventsource_stream::Eventsource; use futures::StreamExt; use serde::Deserialize; use serde_json::Value; +use std::collections::BTreeMap; use std::sync::Arc; use std::sync::OnceLock; use std::time::Duration; @@ -372,6 +374,29 @@ impl ResponsesEventError { } } +// Diagnostic fields are optional and must never make an otherwise valid response fail. +fn reasoning_removals(response: &Value) -> Option> { + let transformations = response.get("input_transformations")?.as_array()?; + let mut counts = BTreeMap::new(); + for transformation in transformations { + if transformation.get("type").and_then(Value::as_str) != Some("reasoning") + || transformation.get("action").and_then(Value::as_str) != Some("removed") + { + continue; + } + let reason = match transformation.get("reason").and_then(Value::as_str) { + Some("organization_mismatch") => ReasoningRemovalReason::OrganizationMismatch, + Some("end_user_mismatch") => ReasoningRemovalReason::EndUserMismatch, + Some("model_mismatch") => ReasoningRemovalReason::ModelMismatch, + Some("prefix_mismatch") => ReasoningRemovalReason::PrefixMismatch, + Some("turn_boundary") => ReasoningRemovalReason::TurnBoundary, + _ => ReasoningRemovalReason::Other, + }; + *counts.entry(reason).or_default() += 1; + } + Some(counts) +} + pub fn process_responses_event( event: ResponsesStreamEvent, ) -> std::result::Result, ResponsesEventError> { @@ -492,6 +517,7 @@ pub fn process_responses_event( } } if let Some(resp_val) = event.response { + let reasoning_removals = reasoning_removals(&resp_val); let metadata = resp_val .get("usage") .filter(|usage| !usage.is_null()) @@ -505,6 +531,7 @@ pub fn process_responses_event( response_id: resp.id, token_usage: resp.usage.map(Into::into), usage_metadata: resp.usage_metadata, + reasoning_removals, end_turn: if interrupted { Some(false) } else { @@ -881,6 +908,7 @@ mod tests { response_id, token_usage, usage_metadata, + reasoning_removals: _, end_turn, }) => { assert_eq!(response_id, "resp1"); @@ -1079,6 +1107,7 @@ mod tests { response_id, token_usage, usage_metadata, + reasoning_removals: _, end_turn, }) => { assert_eq!(response_id, "resp1"); @@ -1675,6 +1704,7 @@ mod tests { response_id, token_usage: None, usage_metadata: None, + reasoning_removals: None, end_turn: None, } if response_id == "resp-1" ); @@ -1716,6 +1746,7 @@ mod tests { response_id, token_usage: None, usage_metadata: None, + reasoning_removals: None, end_turn: None, } if response_id == "resp-1" ); @@ -1752,6 +1783,7 @@ mod tests { response_id, token_usage: None, usage_metadata: None, + reasoning_removals: None, end_turn: None, } if response_id == "resp-1" ); @@ -1788,6 +1820,7 @@ mod tests { response_id, token_usage: None, usage_metadata: None, + reasoning_removals: None, end_turn: None, } if response_id == "resp-1" ); diff --git a/codex-rs/codex-api/tests/sse_end_to_end.rs b/codex-rs/codex-api/tests/sse_end_to_end.rs index 0a7f14801d..cc54fa3e62 100644 --- a/codex-rs/codex-api/tests/sse_end_to_end.rs +++ b/codex-rs/codex-api/tests/sse_end_to_end.rs @@ -174,6 +174,7 @@ async fn responses_stream_parses_items_and_completed_end_to_end() -> Result<()> response_id, token_usage, usage_metadata, + reasoning_removals: _, end_turn, } => { assert_eq!(response_id, "resp1"); diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index f78a98dde3..afe9865f05 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -2842,6 +2842,7 @@ where response_id, token_usage, usage_metadata, + reasoning_removals, end_turn, }) => { feedback_tags!(last_model_response_id = &response_id); @@ -2866,6 +2867,7 @@ where response_id, token_usage, usage_metadata, + reasoning_removals, end_turn, })) .await diff --git a/codex-rs/core/src/client_tests.rs b/codex-rs/core/src/client_tests.rs index 4bf90c4227..5eabca60fa 100644 --- a/codex-rs/core/src/client_tests.rs +++ b/codex-rs/core/src/client_tests.rs @@ -1820,6 +1820,7 @@ async fn response_stream_records_last_model_feedback_ids() { response_id: "resp-123".to_string(), token_usage: None, usage_metadata: None, + reasoning_removals: None, end_turn: Some(true), }), ]); @@ -2864,6 +2865,7 @@ async fn intercepted_output_reaches_trace_and_websocket_bookkeeping() -> anyhow: response_id: "response".into(), token_usage: None, usage_metadata: None, + reasoning_removals: None, end_turn: None, })) .await?; diff --git a/codex-rs/core/src/compact_remote_v2.rs b/codex-rs/core/src/compact_remote_v2.rs index 5656fb486b..ad59b2676d 100644 --- a/codex-rs/core/src/compact_remote_v2.rs +++ b/codex-rs/core/src/compact_remote_v2.rs @@ -1433,6 +1433,7 @@ mod tests { response_id: "resp-compact".into(), token_usage: Some(usage.clone()), usage_metadata: None, + reasoning_removals: None, end_turn: Some(true), })]); let result = collect_compaction_output( @@ -1485,6 +1486,7 @@ mod tests { amount: Some("0.125".to_string()), metadata: Some(serde_json::json!({ "extra": { "label": "example" } })), }), + reasoning_removals: None, end_turn: Some(true), }), ]); diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index f682cc2584..0d3da231c9 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -3178,7 +3178,7 @@ async fn try_run_sampling_request( let Some(event) = resampling.filter(event, &step_context.tool_router) else { continue; }; - sess.services + step_context .session_telemetry .record_responses(&handle_responses, &event); @@ -3532,6 +3532,7 @@ async fn try_run_sampling_request( response_id, token_usage, usage_metadata, + reasoning_removals: _, end_turn, } => { if let Some(accepted) = pending_audio.take() { diff --git a/codex-rs/core/tests/suite/mod.rs b/codex-rs/core/tests/suite/mod.rs index b1f7a42313..756524d6c5 100644 --- a/codex-rs/core/tests/suite/mod.rs +++ b/codex-rs/core/tests/suite/mod.rs @@ -101,6 +101,7 @@ mod guardian_authorization_refresh; mod guardian_cached_score; #[path = "guardian_checkpoint_migration_tests.rs"] mod guardian_checkpoint_migration; +mod reasoning_replay_metrics; // Uses the same command-approval harness as guardian_review below. mod builtin_model_catalog; mod canonical_plugin_connectors; diff --git a/codex-rs/core/tests/suite/reasoning_replay_metrics.rs b/codex-rs/core/tests/suite/reasoning_replay_metrics.rs new file mode 100644 index 0000000000..37dac98522 --- /dev/null +++ b/codex-rs/core/tests/suite/reasoning_replay_metrics.rs @@ -0,0 +1,185 @@ +//! Verifies reported CoT removals through both sampling transports and the metrics exporter. + +use anyhow::Result; +use codex_otel::MetricsClient; +use codex_otel::MetricsConfig; +use codex_otel::REASONING_ITEMS_REMOVED_METRIC; +use codex_otel::REASONING_REPLAY_COUNT_METRIC; +use core_test_support::responses; +use core_test_support::skip_if_no_network; +use core_test_support::test_codex::test_codex; +use opentelemetry_sdk::metrics::InMemoryMetricExporter; +use opentelemetry_sdk::metrics::data::AggregatedMetrics; +use opentelemetry_sdk::metrics::data::MetricData; +use opentelemetry_sdk::metrics::data::ScopeMetrics; +use pretty_assertions::assert_eq; +use serde_json::json; +use std::collections::BTreeMap; + +#[test_case::test_case(false; "sse")] +#[test_case::test_case(true; "websocket")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn reasoning_replay_metrics_preserve_report_availability(use_websocket: bool) -> Result<()> { + skip_if_no_network!(Ok(())); + let reports = [ + None, + Some(json!([])), + Some(json!([ + {"type": "reasoning", "id": "rs_same", "action": "removed", "reason": "prefix_mismatch"}, + {"type": "reasoning", "id": "rs_same", "action": "removed", "reason": "prefix_mismatch"}, + {"type": "reasoning", "action": "removed", "reason": "organization_mismatch"}, + {"type": "reasoning", "action": "removed", "reason": "end_user_mismatch"}, + {"type": "reasoning", "action": "removed", "reason": "model_mismatch"}, + {"type": "reasoning", "action": "removed", "reason": "turn_boundary"}, + {"type": "reasoning", "action": "removed", "reason": "future-private-reason"}, + {"type": "reasoning", "action": "removed"}, + {"type": "message", "action": "removed", "reason": "prefix_mismatch"}, + {"type": "reasoning", "action": "changed", "reason": "prefix_mismatch"}, + null + ])), + Some(json!([{"type": "reasoning", "action": "removed", "reason": "turn_boundary"}])), + Some(json!(null)), + Some(json!({"malformed": true})), + ]; + let events = reports + .iter() + .enumerate() + .map(|(index, report)| { + let id = format!("resp-{index}"); + let mut completed = responses::ev_completed(&id); + if let Some(report) = report { + completed["response"]["input_transformations"] = report.clone(); + } + let mut in_progress = completed.clone(); + in_progress["type"] = json!("response.in_progress"); + vec![responses::ev_response_created(&id), in_progress, completed] + }) + .collect::>(); + let server = responses::start_mock_server().await; + let http_requests = if use_websocket { + None + } else { + Some( + responses::mount_sse_sequence( + &server, + events.iter().cloned().map(responses::sse).collect(), + ) + .await, + ) + }; + let websocket_server = if use_websocket { + let mut frames = vec![vec![ + responses::ev_response_created("prewarm"), + responses::ev_completed("prewarm"), + ]]; + frames.extend(events); + Some(responses::start_websocket_server(vec![frames]).await) + } else { + None + }; + let metrics = MetricsClient::new( + MetricsConfig::in_memory( + "test", + "reasoning-replay", + env!("CARGO_PKG_VERSION"), + InMemoryMetricExporter::default(), + ) + .with_runtime_reader(), + )?; + let mut builder = test_codex().with_config(|config| { + config.apps_mcp_product_sku = Some("codex".to_string()); + }); + let mut test = builder.build_with_auto_env(&server).await?; + let mut options = test.start_thread_options().await; + test.codex.shutdown_and_wait().await?; + // Only the instrumented thread may prewarm the scripted WebSocket connection. + if let Some(server) = &websocket_server { + options.config.model_provider.base_url = Some(format!("{}/v1", server.uri())); + options.config.model_provider.supports_websockets = true; + } + options.thread_extension_init.insert(metrics.clone()); + test.codex = test.thread_manager.start_thread(options).await?.thread; + for _ in &reports { + test.submit_text_turn("continue").await?; + } + if let Some(requests) = http_requests { + assert_eq!(requests.requests().len(), reports.len()); + } + let snapshot = metrics.snapshot()?; + let mut reports = BTreeMap::new(); + let mut removals = BTreeMap::new(); + for metric in snapshot.scope_metrics().flat_map(ScopeMetrics::metrics) { + match metric.name() { + REASONING_REPLAY_COUNT_METRIC => { + let AggregatedMetrics::U64(MetricData::Sum(sum)) = metric.data() else { + panic!("report availability must be a counter"); + }; + for point in sum.data_points() { + let tags = point + .attributes() + .map(|attribute| { + (attribute.key.as_str(), attribute.value.as_str().to_string()) + }) + .collect::>(); + assert_eq!(tags["product_sku"], "codex"); + reports.insert(tags["report"].clone(), point.value()); + } + } + REASONING_ITEMS_REMOVED_METRIC => { + let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = metric.data() else { + panic!("reasoning removal counts must be histograms"); + }; + for point in histogram.data_points() { + let tags = point + .attributes() + .map(|attribute| { + (attribute.key.as_str(), attribute.value.as_str().to_string()) + }) + .collect::>(); + assert_eq!(tags["product_sku"], "codex"); + removals.insert( + tags["reason"].clone(), + (point.count(), point.sum(), point.min(), point.max()), + ); + } + } + _ => {} + } + } + assert_eq!( + reports, + BTreeMap::from([ + ("unavailable".to_string(), 3), + ("no_removals_reported".to_string(), 1), + ("prefix_mismatch_reported".to_string(), 1), + ("removals_reported".to_string(), 1), + ]) + ); + // Three usable reports contribute to every reason, including their zero counts. + // Missing, null, and malformed reports contribute no histogram observations. + assert_eq!( + removals, + BTreeMap::from([ + ( + "prefix_mismatch".to_string(), + (3, 2.0, Some(0.0), Some(2.0)) + ), + ( + "organization_mismatch".to_string(), + (3, 1.0, Some(0.0), Some(1.0)) + ), + ( + "end_user_mismatch".to_string(), + (3, 1.0, Some(0.0), Some(1.0)) + ), + ("model_mismatch".to_string(), (3, 1.0, Some(0.0), Some(1.0))), + ("turn_boundary".to_string(), (3, 2.0, Some(0.0), Some(1.0))), + ("other".to_string(), (3, 2.0, Some(0.0), Some(2.0))), + ]) + ); + test.codex.shutdown_and_wait().await?; + if let Some(server) = websocket_server { + server.shutdown().await; + } + Ok(()) +} diff --git a/codex-rs/otel/src/events/session_telemetry.rs b/codex-rs/otel/src/events/session_telemetry.rs index f6099b84df..61d6248a6e 100644 --- a/codex-rs/otel/src/events/session_telemetry.rs +++ b/codex-rs/otel/src/events/session_telemetry.rs @@ -12,6 +12,8 @@ use crate::metrics::MetricsConfig; use crate::metrics::MetricsError; use crate::metrics::PLUGIN_INSTALL_ELICITATION_SENT_METRIC; use crate::metrics::PLUGIN_INSTALL_SUGGESTION_METRIC; +use crate::metrics::REASONING_ITEMS_REMOVED_METRIC; +use crate::metrics::REASONING_REPLAY_COUNT_METRIC; use crate::metrics::RESPONSES_API_ENGINE_IAPI_TBT_DURATION_METRIC; use crate::metrics::RESPONSES_API_ENGINE_IAPI_TTFT_DURATION_METRIC; use crate::metrics::RESPONSES_API_ENGINE_SERVICE_TBT_DURATION_METRIC; @@ -40,6 +42,7 @@ use crate::sanitize_metric_tag_value; use crate::tool_result::ToolResultEvent; use crate::tool_result::emit_tool_result; use codex_api::AgentIdentityTelemetry; +use codex_api::ReasoningRemovalReason; use codex_api::ResponseEvent; use codex_protocol::AgentPath; use codex_protocol::ThreadId; @@ -647,6 +650,37 @@ impl SessionTelemetry { pub fn record_responses(&self, handle_responses_span: &Span, event: &ResponseEvent) { handle_responses_span.record("otel.name", SessionTelemetry::responses_type(event)); + if let ResponseEvent::Completed { + reasoning_removals, .. + } = event + { + let report = match reasoning_removals { + None => "unavailable", + Some(counts) if counts.is_empty() => "no_removals_reported", + Some(counts) if counts.contains_key(&ReasoningRemovalReason::PrefixMismatch) => { + "prefix_mismatch_reported" + } + Some(_) => "removals_reported", + }; + let mut tags = vec![("report", report)]; + if let Some(product_sku) = self.metadata.product_sku { + tags.push(("product_sku", product_sku)); + } + self.counter(REASONING_REPLAY_COUNT_METRIC, /*inc*/ 1, &tags); + if let Some(counts) = reasoning_removals { + for reason in ReasoningRemovalReason::ALL { + let count = counts.get(&reason).copied().unwrap_or_default(); + tags[0] = ("reason", reason.as_str()); + self.histogram_with_boundaries( + REASONING_ITEMS_REMOVED_METRIC, + count, + &[0.0, 1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0], + &tags, + ); + } + } + } + match event { ResponseEvent::OutputItemDone(item) => { handle_responses_span.record("from", "output_item_done"); diff --git a/codex-rs/otel/src/metrics/names.rs b/codex-rs/otel/src/metrics/names.rs index aa6dbf66ac..f8d9374f69 100644 --- a/codex-rs/otel/src/metrics/names.rs +++ b/codex-rs/otel/src/metrics/names.rs @@ -34,6 +34,9 @@ pub const RESPONSES_API_ENGINE_IAPI_TBT_DURATION_METRIC: &str = "codex.responses_api_engine_iapi_tbt.duration_ms"; pub const RESPONSES_API_ENGINE_SERVICE_TBT_DURATION_METRIC: &str = "codex.responses_api_engine_service_tbt.duration_ms"; +pub const REASONING_REPLAY_COUNT_METRIC: &str = "codex.responses.reasoning_replay"; +/// Histogram of reported removals per sampling response and reason, including zeros. +pub const REASONING_ITEMS_REMOVED_METRIC: &str = "codex.responses.reasoning_items_removed"; pub const TURN_E2E_DURATION_METRIC: &str = "codex.turn.e2e_duration_ms"; pub const TURN_TTFT_DURATION_METRIC: &str = "codex.turn.ttft.duration_ms"; pub const TURN_TTFM_DURATION_METRIC: &str = "codex.turn.ttfm.duration_ms";