diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index 71b3e60b7..19b3c1762 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -46,7 +46,6 @@ use codex_mcp::auth_elicitation_completed_result; use codex_mcp::build_auth_elicitation_plan; use codex_mcp::declared_openai_file_input_param_names; use codex_mcp::mcp_permission_prompt_is_auto_approved; -use codex_otel::sanitize_metric_tag_value; use codex_protocol::items::McpToolCallError; use codex_protocol::items::McpToolCallItem; use codex_protocol::items::McpToolCallStatus; @@ -95,8 +94,15 @@ use tracing::error; use tracing::field::Empty; use url::Url; -const MCP_CALL_COUNT_METRIC: &str = "codex.mcp.call"; -const MCP_CALL_DURATION_METRIC: &str = "codex.mcp.call.duration_ms"; +mod telemetry; + +use telemetry::MCP_CALL_COUNT_METRIC; +use telemetry::McpCallMetricOutcome; +use telemetry::McpErrorCodeSource; +use telemetry::emit_mcp_call_metrics; +use telemetry::mcp_call_metric_outcome; +use telemetry::record_mcp_call_outcome_span_telemetry; + const MCP_RESULT_TELEMETRY_META_KEY: &str = "codex/telemetry"; const MCP_RESULT_TELEMETRY_SPAN_KEY: &str = "span"; const MCP_RESULT_TELEMETRY_TARGET_ID_KEY: &str = "target_id"; @@ -273,9 +279,10 @@ pub(crate) async fn handle_mcp_tool_call( }; let status = if result.is_ok() { "ok" } else { "error" }; + let outcome = McpCallMetricOutcome::from_status(status); emit_mcp_call_metrics( turn_context.as_ref(), - status, + &outcome, &tool_name, connector_id.as_deref(), connector_name.as_deref(), @@ -345,12 +352,18 @@ async fn handle_approved_mcp_tool_call( let arguments_value = invocation.arguments.clone(); let connector_id = metadata.and_then(|metadata| metadata.connector_id.as_deref()); let connector_name = metadata.and_then(|metadata| metadata.connector_name.as_deref()); - let server_origin = sess - .services - .mcp_connection_manager - .load_full() - .server_origin(&server) - .map(str::to_string); + let (server_origin, error_code_source) = { + let mcp_connection_manager = sess.services.mcp_connection_manager.load_full(); + let server_origin = mcp_connection_manager + .server_origin(&server) + .map(str::to_string); + let error_code_source = if mcp_connection_manager.is_host_owned_codex_apps_server(&server) { + McpErrorCodeSource::HostedPluginService + } else { + McpErrorCodeSource::Untrusted + }; + (server_origin, error_code_source) + }; let start = Instant::now(); let rewrite = rewrite_mcp_tool_arguments_for_openai_files( @@ -367,20 +380,23 @@ async fn handle_approved_mcp_tool_call( .unwrap_or_else(|| JsonValue::Object(serde_json::Map::new())), }; let result = async { - let rewritten_arguments = rewrite?; - let request_meta = - build_mcp_tool_call_request_meta(turn_context, &server, call_id, metadata); - let result = execute_mcp_tool_call( - sess, - turn_context, - call_id, - &invocation, - rewritten_arguments, - metadata, - request_meta, - ) + let result = async { + let rewritten_arguments = rewrite?; + let request_meta = + build_mcp_tool_call_request_meta(turn_context, &server, call_id, metadata); + execute_mcp_tool_call( + sess, + turn_context, + call_id, + &invocation, + rewritten_arguments, + metadata, + request_meta, + ) + .await + } .await; - record_mcp_result_span_telemetry(&Span::current(), result.as_ref().ok()); + record_mcp_result_span_telemetry(&Span::current(), &result, error_code_source); result } .instrument(mcp_tool_call_span( @@ -412,10 +428,10 @@ async fn handle_approved_mcp_tool_call( .await; maybe_track_codex_app_used(sess, turn_context, &server, &tool_name).await; - let status = if result.is_ok() { "ok" } else { "error" }; + let outcome = mcp_call_metric_outcome(&result, error_code_source); emit_mcp_call_metrics( turn_context, - status, + &outcome, &tool_name, connector_id, connector_name, @@ -428,51 +444,6 @@ async fn handle_approved_mcp_tool_call( } } -fn emit_mcp_call_metrics( - turn_context: &TurnContext, - status: &str, - tool_name: &str, - connector_id: Option<&str>, - connector_name: Option<&str>, - duration: Option, -) { - let tags = mcp_call_metric_tags(status, tool_name, connector_id, connector_name); - let tag_refs: Vec<(&str, &str)> = tags - .iter() - .map(|(key, value)| (*key, value.as_str())) - .collect(); - turn_context - .session_telemetry - .counter(MCP_CALL_COUNT_METRIC, /*inc*/ 1, &tag_refs); - if let Some(duration) = duration { - turn_context.session_telemetry.record_duration( - MCP_CALL_DURATION_METRIC, - duration, - &tag_refs, - ); - } -} - -fn mcp_call_metric_tags( - status: &str, - tool_name: &str, - connector_id: Option<&str>, - connector_name: Option<&str>, -) -> Vec<(&'static str, String)> { - let mut tags = vec![ - ("status", sanitize_metric_tag_value(status)), - ("tool", sanitize_metric_tag_value(tool_name)), - ]; - if let Some(connector_id) = connector_id.filter(|connector_id| !connector_id.is_empty()) { - tags.push(("connector_id", sanitize_metric_tag_value(connector_id))); - } - if let Some(connector_name) = connector_name.filter(|connector_name| !connector_name.is_empty()) - { - tags.push(("connector_name", sanitize_metric_tag_value(connector_name))); - } - tags -} - fn mcp_tool_call_span( session: &Session, turn_context: &TurnContext, @@ -503,6 +474,8 @@ fn mcp_tool_call_span( server.port = Empty, codex.mcp.target.id = Empty, codex.mcp.server_user_flow.triggered = Empty, + error.type = Empty, + codex.mcp.error.code = Empty, ); record_server_fields(&span, fields.server_origin); span @@ -532,8 +505,16 @@ fn record_server_fields(span: &Span, url: Option<&str>) { } } -fn record_mcp_result_span_telemetry(span: &Span, result: Option<&CallToolResult>) { +fn record_mcp_result_span_telemetry( + span: &Span, + result: &Result, + error_code_source: McpErrorCodeSource, +) { + record_mcp_call_outcome_span_telemetry(span, result, error_code_source); + let Some(span_telemetry) = result + .as_ref() + .ok() .and_then(|result| result.meta.as_ref()) .and_then(JsonValue::as_object) .and_then(|meta| meta.get(MCP_RESULT_TELEMETRY_META_KEY)) diff --git a/codex-rs/core/src/mcp_tool_call/telemetry.rs b/codex-rs/core/src/mcp_tool_call/telemetry.rs new file mode 100644 index 000000000..9d9f4ec46 --- /dev/null +++ b/codex-rs/core/src/mcp_tool_call/telemetry.rs @@ -0,0 +1,179 @@ +use std::time::Duration; + +use crate::session::turn_context::TurnContext; +use codex_mcp::MCP_TOOL_CODEX_APPS_META_KEY; +use codex_otel::sanitize_metric_tag_value; +use codex_protocol::mcp::CallToolResult; +use serde_json::Value as JsonValue; +use tracing::Span; + +pub(super) const MCP_CALL_COUNT_METRIC: &str = "codex.mcp.call"; +const MCP_CALL_DURATION_METRIC: &str = "codex.mcp.call.duration_ms"; +const MCP_CALL_ERROR_COUNT_METRIC: &str = "codex.mcp.call.error"; +// No CallToolResult was received. This includes request setup, transport, timeout, protocol, and +// JSON-RPC failures; it does not imply that the request never reached the MCP server. +const MCP_CALL_ERROR_TYPE_MCP_REQUEST: &str = "mcp_request"; +// The MCP server returned a CallToolResult with isError=true. +const MCP_CALL_ERROR_TYPE_TOOL_RESULT: &str = "tool_result"; +const MCP_CALL_ERROR_CODE_UNKNOWN: &str = "unknown"; +const MCP_CALL_ERROR_CODE_MAX_CHARS: usize = 256; +const MCP_CALL_ERROR_TYPE_SPAN_ATTR: &str = "error.type"; +const MCP_CALL_ERROR_CODE_SPAN_ATTR: &str = "codex.mcp.error.code"; + +#[derive(Debug, PartialEq, Eq)] +pub(super) struct McpCallMetricOutcome { + status: &'static str, + error_type: Option<&'static str>, + error_code: Option, +} + +impl McpCallMetricOutcome { + pub(super) fn from_status(status: &'static str) -> Self { + Self { + status, + error_type: None, + error_code: None, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum McpErrorCodeSource { + HostedPluginService, + Untrusted, +} + +pub(super) fn emit_mcp_call_metrics( + turn_context: &TurnContext, + outcome: &McpCallMetricOutcome, + tool_name: &str, + connector_id: Option<&str>, + connector_name: Option<&str>, + duration: Option, +) { + let tags = mcp_call_metric_tags(outcome.status, tool_name, connector_id, connector_name); + let tag_refs: Vec<(&str, &str)> = tags + .iter() + .map(|(key, value)| (*key, value.as_str())) + .collect(); + turn_context + .session_telemetry + .counter(MCP_CALL_COUNT_METRIC, /*inc*/ 1, &tag_refs); + if let Some(duration) = duration { + turn_context.session_telemetry.record_duration( + MCP_CALL_DURATION_METRIC, + duration, + &tag_refs, + ); + } + + let (Some(error_type), Some(error_code)) = (outcome.error_type, outcome.error_code.as_deref()) + else { + return; + }; + let mut error_tags = tags; + error_tags.push(("error_type", sanitize_metric_tag_value(error_type))); + error_tags.push(("error_code", error_code.to_string())); + let error_tag_refs: Vec<(&str, &str)> = error_tags + .iter() + .map(|(key, value)| (*key, value.as_str())) + .collect(); + turn_context.session_telemetry.counter( + MCP_CALL_ERROR_COUNT_METRIC, + /*inc*/ 1, + &error_tag_refs, + ); +} + +fn mcp_call_metric_tags( + status: &str, + tool_name: &str, + connector_id: Option<&str>, + connector_name: Option<&str>, +) -> Vec<(&'static str, String)> { + let mut tags = vec![ + ("status", sanitize_metric_tag_value(status)), + ("tool", sanitize_metric_tag_value(tool_name)), + ]; + if let Some(connector_id) = connector_id.filter(|connector_id| !connector_id.is_empty()) { + tags.push(("connector_id", sanitize_metric_tag_value(connector_id))); + } + if let Some(connector_name) = connector_name.filter(|connector_name| !connector_name.is_empty()) + { + tags.push(("connector_name", sanitize_metric_tag_value(connector_name))); + } + tags +} + +pub(super) fn mcp_call_metric_outcome( + result: &Result, + error_code_source: McpErrorCodeSource, +) -> McpCallMetricOutcome { + match result { + Ok(result) if result.is_error.unwrap_or(false) => { + let error_code = if error_code_source == McpErrorCodeSource::HostedPluginService { + result + .structured_content + .as_ref() + .and_then(JsonValue::as_object) + .and_then(|structured_content| structured_content.get("error_code")) + .and_then(JsonValue::as_str) + .filter(|error_code| !error_code.is_empty()) + .or_else(|| { + result + .meta + .as_ref() + .and_then(JsonValue::as_object) + .and_then(|meta| meta.get(MCP_TOOL_CODEX_APPS_META_KEY)) + .and_then(JsonValue::as_object) + .and_then(|codex_apps| codex_apps.get("connector_auth_failure")) + .and_then(JsonValue::as_object) + .filter(|auth_failure| { + auth_failure + .get("is_auth_failure") + .and_then(JsonValue::as_bool) + == Some(true) + }) + .and_then(|auth_failure| auth_failure.get("error_code")) + .and_then(JsonValue::as_str) + .filter(|error_code| !error_code.is_empty()) + }) + } else { + None + }; + let error_code: String = error_code + .unwrap_or(MCP_CALL_ERROR_CODE_UNKNOWN) + .chars() + .take(MCP_CALL_ERROR_CODE_MAX_CHARS) + .collect(); + McpCallMetricOutcome { + status: "error", + error_type: Some(MCP_CALL_ERROR_TYPE_TOOL_RESULT), + error_code: Some(sanitize_metric_tag_value(&error_code)), + } + } + Ok(_) => McpCallMetricOutcome::from_status("ok"), + Err(_) => McpCallMetricOutcome { + status: "error", + error_type: Some(MCP_CALL_ERROR_TYPE_MCP_REQUEST), + error_code: Some(MCP_CALL_ERROR_CODE_UNKNOWN.to_string()), + }, + } +} + +pub(super) fn record_mcp_call_outcome_span_telemetry( + span: &Span, + result: &Result, + error_code_source: McpErrorCodeSource, +) { + let outcome = mcp_call_metric_outcome(result, error_code_source); + let (Some(error_type), Some(error_code)) = (outcome.error_type, outcome.error_code) else { + return; + }; + span.record(MCP_CALL_ERROR_TYPE_SPAN_ATTR, error_type); + span.record(MCP_CALL_ERROR_CODE_SPAN_ATTR, error_code); +} + +#[cfg(test)] +#[path = "telemetry_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/mcp_tool_call/telemetry_tests.rs b/codex-rs/core/src/mcp_tool_call/telemetry_tests.rs new file mode 100644 index 000000000..b11df28d1 --- /dev/null +++ b/codex-rs/core/src/mcp_tool_call/telemetry_tests.rs @@ -0,0 +1,117 @@ +use super::*; +use pretty_assertions::assert_eq; + +fn metric_call_tool_result( + is_error: bool, + structured_content: Option, +) -> CallToolResult { + CallToolResult { + content: Vec::new(), + structured_content, + is_error: Some(is_error), + meta: None, + } +} + +#[test] +fn mcp_call_metric_outcome_distinguishes_request_and_tool_errors() { + assert_eq!( + mcp_call_metric_outcome( + &Ok(metric_call_tool_result( + /*is_error*/ false, /*structured_content*/ None, + )), + McpErrorCodeSource::HostedPluginService, + ), + McpCallMetricOutcome { + status: "ok", + error_type: None, + error_code: None, + } + ); + assert_eq!( + mcp_call_metric_outcome( + &Ok(metric_call_tool_result( + /*is_error*/ true, + Some(serde_json::json!({"error_code": "RATE_LIMITED"})), + )), + McpErrorCodeSource::HostedPluginService, + ), + McpCallMetricOutcome { + status: "error", + error_type: Some(MCP_CALL_ERROR_TYPE_TOOL_RESULT), + error_code: Some("RATE_LIMITED".to_string()), + } + ); + assert_eq!( + mcp_call_metric_outcome( + &Err("connection closed".to_string()), + McpErrorCodeSource::HostedPluginService, + ), + McpCallMetricOutcome { + status: "error", + error_type: Some(MCP_CALL_ERROR_TYPE_MCP_REQUEST), + error_code: Some(MCP_CALL_ERROR_CODE_UNKNOWN.to_string()), + } + ); +} + +#[test] +fn mcp_call_metric_outcome_ignores_untrusted_tool_error_codes() { + let result = Ok(metric_call_tool_result( + /*is_error*/ true, + Some(serde_json::json!({"error_code": "arbitrary-user-value"})), + )); + + assert_eq!( + mcp_call_metric_outcome(&result, McpErrorCodeSource::Untrusted), + McpCallMetricOutcome { + status: "error", + error_type: Some(MCP_CALL_ERROR_TYPE_TOOL_RESULT), + error_code: Some(MCP_CALL_ERROR_CODE_UNKNOWN.to_string()), + } + ); +} + +#[test] +fn mcp_call_metric_outcome_reads_hosted_auth_error_code_from_meta() { + let result = CallToolResult { + content: Vec::new(), + structured_content: None, + is_error: Some(true), + meta: Some(serde_json::json!({ + MCP_TOOL_CODEX_APPS_META_KEY: { + "connector_auth_failure": { + "is_auth_failure": true, + "error_code": "UNAUTHORIZED", + }, + }, + })), + }; + + assert_eq!( + mcp_call_metric_outcome(&Ok(result), McpErrorCodeSource::HostedPluginService), + McpCallMetricOutcome { + status: "error", + error_type: Some(MCP_CALL_ERROR_TYPE_TOOL_RESULT), + error_code: Some("UNAUTHORIZED".to_string()), + } + ); +} + +#[test] +fn mcp_call_metric_outcome_bounds_and_sanitizes_hosted_error_code() { + let raw_error_code = format!("BAD CODE {}", "x".repeat(300)); + let result = Ok(metric_call_tool_result( + /*is_error*/ true, + Some(serde_json::json!({"error_code": raw_error_code})), + )); + + assert_eq!( + mcp_call_metric_outcome(&result, McpErrorCodeSource::HostedPluginService), + McpCallMetricOutcome { + status: "error", + error_type: Some(MCP_CALL_ERROR_TYPE_TOOL_RESULT), + error_code: Some(format!("BAD_CODE_{}", "x".repeat(247))), + } + ); +} diff --git a/codex-rs/core/src/mcp_tool_call_tests.rs b/codex-rs/core/src/mcp_tool_call_tests.rs index 89b4f9de6..9c941d89a 100644 --- a/codex-rs/core/src/mcp_tool_call_tests.rs +++ b/codex-rs/core/src/mcp_tool_call_tests.rs @@ -461,6 +461,57 @@ async fn mcp_tool_call_span_records_expected_fields() { ); } +#[tokio::test] +async fn mcp_tool_call_span_records_error_type_and_hosted_error_code() { + let buffer: &'static std::sync::Mutex> = + Box::leak(Box::new(std::sync::Mutex::new(Vec::new()))); + let subscriber = tracing_subscriber::fmt() + .with_level(true) + .with_ansi(false) + .with_max_level(Level::TRACE) + .with_span_events(FmtSpan::FULL) + .with_writer(MockWriter::new(buffer)) + .finish(); + let _guard = tracing::subscriber::set_default(subscriber); + + let (session, turn_context) = make_session_and_context().await; + let result = Ok(CallToolResult { + content: Vec::new(), + structured_content: Some(serde_json::json!({"error_code": "RATE_LIMITED"})), + is_error: Some(true), + meta: None, + }); + let span = mcp_tool_call_span( + &session, + &turn_context, + McpToolCallSpanFields { + server_name: CODEX_APPS_MCP_SERVER_NAME, + tool_name: "calendar_search", + call_id: "call-123", + server_origin: Some("https://chatgpt.com/api/codex/ps/mcp"), + connector_id: Some("calendar"), + connector_name: Some("Calendar"), + }, + ); + + async { + record_mcp_result_span_telemetry( + &Span::current(), + &result, + McpErrorCodeSource::HostedPluginService, + ); + } + .instrument(span) + .await; + + let logs = String::from_utf8(buffer.lock().expect("buffer lock").clone()).expect("utf8 logs"); + assert!( + logs.contains("error.type=\"tool_result\"") + && logs.contains("codex.mcp.error.code=\"RATE_LIMITED\""), + "missing MCP tool error span fields\nlogs:\n{logs}" + ); +} + async fn mcp_result_telemetry_span_logs(meta: Option) -> String { let buffer: &'static std::sync::Mutex> = Box::leak(Box::new(std::sync::Mutex::new(Vec::new()))); @@ -474,12 +525,12 @@ async fn mcp_result_telemetry_span_logs(meta: Option) -> Stri let _guard = tracing::subscriber::set_default(subscriber); let (session, turn_context) = make_session_and_context().await; - let result = CallToolResult { + let result = Ok(CallToolResult { content: Vec::new(), structured_content: None, is_error: None, meta, - }; + }); { let span = mcp_tool_call_span( @@ -496,7 +547,11 @@ async fn mcp_result_telemetry_span_logs(meta: Option) -> Stri ); async { - record_mcp_result_span_telemetry(&Span::current(), Some(&result)); + record_mcp_result_span_telemetry( + &Span::current(), + &result, + McpErrorCodeSource::Untrusted, + ); } .instrument(span) .await;