Attribute tool telemetry to the invoking turn's product SKU (#47741)

## Why

Tool telemetry needs to distinguish product SKUs without turning arbitrary configuration values into metric labels.

## What changed

- Read `apps_mcp_product_sku` from the invoking turn when dispatching tools, replacing any stale SKU in the telemetry metadata.
- Add `product_sku` to tool call count and duration metrics and tool result log and trace events.
- Preserve `codex`, map other nonempty values to `other`, and omit the attribute for empty or missing values.

## Testing

Add coverage for SKU separation on a shared metrics client, normalization, and successful and failing dispatches with stale telemetry metadata. Extend export routing tests to check SKU attributes in logs and matching traces.

GitOrigin-RevId: 3fc58c9f7245fd5b65f7cba122e08d7ea838b2d6
This commit is contained in:
Rennie
2026-09-24 04:10:08 +00:00
committed by copyberry
parent 7dae8c53d9
commit 5f2ce5e02c
6 changed files with 214 additions and 4 deletions
+5 -1
View File
@@ -527,7 +527,11 @@ impl ToolRegistry {
) -> Result<AnyToolResult, FunctionCallError> {
let tool_name = invocation.tool_name.clone();
let call_id_owned = invocation.call_id.clone();
let otel = invocation.step_context.session_telemetry.clone();
let otel = invocation
.step_context
.session_telemetry
.clone()
.with_product_sku(invocation.turn.config.apps_mcp_product_sku.as_deref());
// TODO(anp): Reconcile these tags with TurnEnvironment::sandbox_context
// instead of reporting the thread-wide backend for environment-scoped tools.
let sandbox_tags = invocation.turn.turn_metadata_state.sandbox_tags;
+97
View File
@@ -1,11 +1,20 @@
use super::*;
use crate::session::session::Session;
use crate::session::step_context::StepContext;
use codex_otel::MetricsClient;
use codex_otel::MetricsConfig;
use codex_otel::TOOL_CALL_COUNT_METRIC;
use codex_otel::TOOL_CALL_DURATION_METRIC;
use codex_protocol::DEFAULT_FUNCTION_NAMESPACE;
use codex_protocol::models::ResponseItem;
use codex_utils_output_truncation::TruncationPolicy;
use futures::future::BoxFuture;
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 std::collections::BTreeMap;
use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
@@ -819,6 +828,94 @@ async fn dispatch_uses_canonical_tool_names_for_lifecycle_contributors() -> anyh
Ok(())
}
#[tokio::test]
async fn dispatch_uses_the_invoking_turn_product_sku() -> anyhow::Result<()> {
let metrics = MetricsClient::new(
MetricsConfig::in_memory(
"test",
"codex-core",
env!("CARGO_PKG_VERSION"),
InMemoryMetricExporter::default(),
)
.with_runtime_reader(),
)?;
let registry = ToolRegistry::from_tools([
Arc::new(LifecycleTestHandler {
tool_name: codex_tools::ToolName::plain("ok_tool"),
result: LifecycleTestResult::Ok { success: true },
}) as Arc<dyn CoreToolRuntime>,
Arc::new(LifecycleTestHandler {
tool_name: codex_tools::ToolName::plain("failing_tool"),
result: LifecycleTestResult::Err,
}) as Arc<dyn CoreToolRuntime>,
]);
let mut expected = BTreeMap::new();
let product_skus = [Some("codex"), None];
for sku in product_skus {
let (session, mut turn) = crate::session::tests::make_session_and_context().await;
Arc::make_mut(&mut turn.config).apps_mcp_product_sku = sku.map(str::to_owned);
turn.session_telemetry = turn
.session_telemetry
.clone()
.with_product_sku(Some("stale-product"))
.with_metrics(metrics.clone());
let session = Arc::new(session);
let turn = Arc::new(turn);
for (tool, success) in [("ok_tool", true), ("failing_tool", false)] {
let result = registry
.dispatch_any_with_state(
test_invocation(
Arc::clone(&session),
Arc::clone(&turn),
"sku-call",
codex_tools::ToolName::plain(tool).with_default_namespace(),
),
/*call_state*/ None,
)
.await;
assert_eq!(result.is_ok(), success);
expected.insert((sku.map(str::to_owned), success.to_string()), 1);
}
}
let snapshot = metrics.snapshot()?;
for name in [TOOL_CALL_COUNT_METRIC, TOOL_CALL_DURATION_METRIC] {
let metric = snapshot
.scope_metrics()
.flat_map(ScopeMetrics::metrics)
.find(|metric| metric.name() == name)
.expect("tool metric");
let points = match metric.data() {
AggregatedMetrics::U64(MetricData::Sum(sum)) => sum
.data_points()
.map(|point| (point.attributes().collect::<Vec<_>>(), point.value()))
.collect::<Vec<_>>(),
AggregatedMetrics::F64(MetricData::Histogram(histogram)) => histogram
.data_points()
.map(|point| (point.attributes().collect::<Vec<_>>(), point.count()))
.collect::<Vec<_>>(),
_ => panic!("unexpected tool metric aggregation"),
};
let actual = points
.into_iter()
.map(|(attributes, count)| {
let labels = attributes
.into_iter()
.map(|attribute| (attribute.key.as_str(), attribute.value.as_str().to_string()))
.collect::<BTreeMap<_, _>>();
(
(
labels.get("product_sku").cloned(),
labels["success"].clone(),
),
count,
)
})
.collect::<BTreeMap<_, _>>();
assert_eq!(actual, expected);
}
Ok(())
}
fn test_invocation(
session: Arc<crate::session::session::Session>,
turn: Arc<crate::session::turn_context::TurnContext>,
+16 -1
View File
@@ -100,6 +100,7 @@ pub struct SessionTelemetryMetadata {
pub(crate) account_id: Option<String>,
pub(crate) account_email: Option<String>,
pub(crate) originator: String,
pub(crate) product_sku: Option<&'static str>,
pub(crate) service_name: Option<String>,
pub(crate) session_source: String,
pub(crate) model: String,
@@ -151,6 +152,16 @@ impl SessionTelemetry {
self
}
/// Attributes tool telemetry without turning arbitrary configuration into metric labels.
pub fn with_product_sku(mut self, product_sku: Option<&str>) -> Self {
self.metadata.product_sku = match product_sku {
None | Some("") => None,
Some("codex") => Some("codex"),
Some(_) => Some("other"),
};
self
}
pub fn with_metrics(mut self, metrics: MetricsClient) -> Self {
self.metrics = Some(metrics);
self.metrics_use_metadata_tags = true;
@@ -522,6 +533,7 @@ impl SessionTelemetry {
account_id,
account_email,
originator: sanitize_metric_tag_value(originator.as_str()),
product_sku: None,
service_name: None,
session_source: session_source.to_string(),
model: model.to_owned(),
@@ -1210,10 +1222,13 @@ impl SessionTelemetry {
) {
let flat_tool_name = tool_name.to_string();
let success_str = if success { "true" } else { "false" };
let mut tags = Vec::with_capacity(2 + extra_tags.len());
let mut tags = Vec::with_capacity(3 + extra_tags.len());
tags.push(("tool", flat_tool_name.as_str()));
tags.push(("success", success_str));
tags.extend_from_slice(extra_tags);
if let Some(product_sku) = self.metadata.product_sku {
tags.push(("product_sku", product_sku));
}
self.counter(TOOL_CALL_COUNT_METRIC, /*inc*/ 1, &tags);
self.record_duration(TOOL_CALL_DURATION_METRIC, duration, &tags);
let mcp_server = trace_field_value(extra_trace_fields, "mcp_server").unwrap_or("");
+1
View File
@@ -59,6 +59,7 @@ pub(crate) fn emit_tool_result(
tool_name = %tool_name.name,
tool_namespace = %tool_namespace,
call_id = %call_id,
product_sku = telemetry.metadata.product_sku,
duration_ms = %duration.as_millis(),
success = %success,
output_truncated = preview.truncated,
@@ -7,13 +7,97 @@ use codex_otel::PLUGIN_INSTALL_ELICITATION_SENT_METRIC;
use codex_otel::PLUGIN_INSTALL_SUGGESTION_METRIC;
use codex_otel::Result;
use codex_otel::SessionTelemetry;
use codex_otel::TOOL_CALL_COUNT_METRIC;
use codex_otel::TOOL_CALL_DURATION_METRIC;
use codex_otel::TelemetryAuthMode;
use codex_protocol::ThreadId;
use codex_protocol::ToolName;
use codex_protocol::protocol::SessionSource;
use opentelemetry_sdk::metrics::data::AggregatedMetrics;
use opentelemetry_sdk::metrics::data::MetricData;
use pretty_assertions::assert_eq;
use std::collections::BTreeMap;
use std::time::Duration;
#[test]
fn tool_metrics_keep_product_skus_separate_on_a_shared_client() -> Result<()> {
let (metrics, exporter) = build_metrics_with_defaults(&[])?;
let manager = SessionTelemetry::new(
ThreadId::new(),
"gpt-5.1",
"gpt-5.1",
/*account_id*/ None,
/*account_email*/ None,
/*auth_mode*/ None,
"test_originator".to_string(),
/*log_user_prompts*/ false,
"tty".to_string(),
SessionSource::Cli,
)
.with_metrics_without_metadata_tags(metrics);
let mut expected_counts = BTreeMap::new();
let mut expected_durations = BTreeMap::new();
for (sku, expected_sku, success, duration_ms) in [
(Some("codex"), Some("codex"), true, 11),
(Some("customer-specific-value"), Some("other"), false, 37),
(Some("another-unrecognized-value"), Some("other"), false, 41),
(Some(""), None, true, 43),
(None, None, true, 47),
] {
let telemetry = manager.clone().with_product_sku(sku);
telemetry.tool_result_with_tags(
&ToolName::plain("spawn_agent"),
"call-1",
"{}",
Duration::from_millis(duration_ms),
success,
"result",
&[("sandbox", "workspace")],
&[],
);
let mut labels = BTreeMap::from([
("tool".to_string(), "spawn_agent".to_string()),
("success".to_string(), success.to_string()),
("sandbox".to_string(), "workspace".to_string()),
]);
if let Some(sku) = expected_sku {
labels.insert("product_sku".to_string(), sku.to_string());
}
*expected_counts.entry(labels.clone()).or_insert(0) += 1;
let (count, sum) = expected_durations.entry(labels).or_insert((0, 0.0));
*count += 1;
*sum += duration_ms as f64;
}
manager.shutdown_metrics()?;
let resource_metrics = latest_metrics(&exporter);
let counter = find_metric(&resource_metrics, TOOL_CALL_COUNT_METRIC).expect("tool counter");
let AggregatedMetrics::U64(MetricData::Sum(sum)) = counter.data() else {
panic!("expected tool counter");
};
assert_eq!(
sum.data_points()
.map(|point| (attributes_to_map(point.attributes()), point.value()))
.collect::<BTreeMap<_, _>>(),
expected_counts,
);
let duration =
find_metric(&resource_metrics, TOOL_CALL_DURATION_METRIC).expect("tool duration");
let AggregatedMetrics::F64(MetricData::Histogram(histogram)) = duration.data() else {
panic!("expected tool duration histogram");
};
assert_eq!(
histogram
.data_points()
.map(|point| (
attributes_to_map(point.attributes()),
(point.count(), point.sum()),
))
.collect::<BTreeMap<_, _>>(),
expected_durations,
);
Ok(())
}
// Ensures SessionTelemetry attaches metadata tags when forwarding metrics.
#[test]
@@ -256,7 +256,7 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() {
tracing::subscriber::with_default(subscriber, || {
tracing::callsite::rebuild_interest_cache();
let manager = make_manager(root_id, SessionSource::Cli);
let manager = make_manager(root_id, SessionSource::Cli).with_product_sku(Some("codex"));
let root_span = tracing::info_span!("root");
let _root_guard = root_span.enter();
let manager = manager.with_tool_result_log_config(ToolResultLogConfig {
@@ -284,7 +284,8 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() {
agent_nickname: Some("legacy nickname".to_string()),
agent_role: None,
}),
);
)
.with_product_sku(Some("customer-specific-value"));
child.tool_result_with_tags(
&ToolName::plain("shell"),
"call-2",
@@ -356,6 +357,13 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() {
.map(|log| log_attributes(&log.record))
.filter(|attrs| attrs.get("event.name").map(String::as_str) == Some("codex.tool_result"))
.collect();
assert_eq!(
tool_logs
.iter()
.map(|attrs| attrs.get("product_sku").map(String::as_str))
.collect::<Vec<_>>(),
vec![Some("codex"), Some("other"), Some("codex"), None, None],
);
assert_eq!(
tool_logs
.iter()
@@ -418,6 +426,7 @@ fn otel_export_routing_policy_routes_tool_result_log_and_trace_events() {
assert_eq!(attrs.get("tool_name"), log.get("tool_name"));
assert_eq!(attrs.get("tool_namespace"), log.get("tool_namespace"));
assert_eq!(attrs.get("tool_result_seq"), log.get("tool_result_seq"));
assert_eq!(attrs.get("product_sku"), log.get("product_sku"));
assert_eq!(attrs.get("output_truncated"), log.get("output_truncated"));
assert!(!attrs.contains_key("agent_name"));
}