Add streamable HTTP MCP failure metric

codex-rs/Cargo.lock

@@ -3559,6 +3559,7 @@ dependencies = [
  "codex-config",
  "codex-exec-server",
  "codex-keyring-store",
+ "codex-otel",
  "codex-protocol",
  "codex-utils-cargo-bin",
  "codex-utils-home-dir",

codex-rs/rmcp-client/Cargo.toml

@@ -20,6 +20,7 @@ codex-client = { workspace = true }
 codex-config = { workspace = true }
 codex-exec-server = { workspace = true }
 codex-keyring-store = { workspace = true }
+codex-otel = { workspace = true }
 codex-protocol = { workspace = true }
 codex-utils-pty = { workspace = true }
 codex-utils-home-dir = { workspace = true }

codex-rs/rmcp-client/src/http_client_adapter.rs

@@ -46,9 +46,15 @@ const EVENT_STREAM_MIME_TYPE: &str = "text/event-stream";
 const JSON_MIME_TYPE: &str = "application/json";
 const HEADER_SESSION_ID: &str = "Mcp-Session-Id";
 const NON_JSON_RESPONSE_BODY_PREVIEW_BYTES: usize = 8_192;
+const STREAMABLE_HTTP_POST_MESSAGE_FAILURE_METRIC: &str =
+    "codex.mcp.streamable_http.post_message.failure";
+const CODEX_APPS_MCP_SERVER_NAME: &str = "codex_apps";
+const CHATGPT_WHAM_APPS_ENDPOINT_LABEL: &str = "chatgpt_wham_apps";
+const OTHER_METRIC_LABEL: &str = "other";
 
 #[derive(Clone)]
 pub(crate) struct StreamableHttpClientAdapter {
+    server_name: String,
     http_client: Arc<dyn HttpClient>,
     default_headers: HeaderMap,
     auth_provider: Option<SharedAuthProvider>,
@@ -66,11 +72,13 @@ pub(crate) enum StreamableHttpClientAdapterError {
 
 impl StreamableHttpClientAdapter {
     pub(crate) fn new(
+        server_name: &str,
         http_client: Arc<dyn HttpClient>,
         default_headers: HeaderMap,
         auth_provider: Option<SharedAuthProvider>,
     ) -> Self {
         Self {
+            server_name: server_name.to_string(),
             http_client,
             default_headers,
             auth_provider,
@@ -140,12 +148,14 @@ impl StreamableHttpClient for StreamableHttpClientAdapter {
         let (response, mut body_stream) = match response {
             Ok(response) => response,
             Err(error) => {
-                log_post_message_http_error(
+                record_post_message_http_error(
+                    &self.server_name,
                     &uri,
                     mcp_method.as_deref(),
                     mcp_request_id.as_deref(),
                     has_session_id,
                     has_authorization_header,
+                    &error,
                 );
                 return Err(StreamableHttpError::Client(
                     StreamableHttpClientAdapterError::from(error),
@@ -385,14 +395,24 @@ fn client_jsonrpc_message_fields(
     }
 }
 
-fn log_post_message_http_error(
+fn record_post_message_http_error(
+    server_name: &str,
     uri: &str,
     mcp_method: Option<&str>,
     mcp_request_id: Option<&str>,
     has_session_id: bool,
     has_authorization_header: bool,
+    error: &ExecServerError,
 ) {
     let parsed_url = reqwest::Url::parse(uri).ok();
+    emit_post_message_failure_metric(PostMessageFailureMetricFields {
+        server_name,
+        parsed_url: parsed_url.as_ref(),
+        mcp_method,
+        has_session_id,
+        has_authorization_header,
+        error,
+    });
     tracing::warn!(
         endpoint_scheme = parsed_url
             .as_ref()
@@ -415,6 +435,98 @@ fn log_post_message_http_error(
     );
 }
 
+struct PostMessageFailureMetricFields<'a> {
+    server_name: &'a str,
+    parsed_url: Option<&'a reqwest::Url>,
+    mcp_method: Option<&'a str>,
+    has_session_id: bool,
+    has_authorization_header: bool,
+    error: &'a ExecServerError,
+}
+
+fn emit_post_message_failure_metric(fields: PostMessageFailureMetricFields<'_>) {
+    let Some(metrics) = codex_otel::global() else {
+        return;
+    };
+    let tags = post_message_failure_metric_tags(fields);
+    let tag_refs: Vec<(&str, &str)> = tags
+        .iter()
+        .map(|(key, value)| (*key, value.as_str()))
+        .collect();
+    let _ = metrics.counter(
+        STREAMABLE_HTTP_POST_MESSAGE_FAILURE_METRIC,
+        /*inc*/ 1,
+        &tag_refs,
+    );
+}
+
+fn post_message_failure_metric_tags(
+    fields: PostMessageFailureMetricFields<'_>,
+) -> Vec<(&'static str, String)> {
+    vec![
+        (
+            "server_name",
+            mcp_server_metric_label(fields.server_name).to_string(),
+        ),
+        (
+            "endpoint",
+            endpoint_metric_label(fields.parsed_url).to_string(),
+        ),
+        (
+            "mcp_method",
+            codex_otel::sanitize_metric_tag_value(fields.mcp_method.unwrap_or("none")),
+        ),
+        ("has_session_id", fields.has_session_id.to_string()),
+        (
+            "has_authorization_header",
+            fields.has_authorization_header.to_string(),
+        ),
+        (
+            "error_type",
+            exec_server_error_type(fields.error).to_string(),
+        ),
+    ]
+}
+
+fn mcp_server_metric_label(server_name: &str) -> &'static str {
+    if server_name == CODEX_APPS_MCP_SERVER_NAME {
+        CODEX_APPS_MCP_SERVER_NAME
+    } else {
+        OTHER_METRIC_LABEL
+    }
+}
+
+fn endpoint_metric_label(parsed_url: Option<&reqwest::Url>) -> &'static str {
+    if parsed_url.is_some_and(|url| {
+        url.scheme() == "https"
+            && url.host_str() == Some("chatgpt.com")
+            && url.path() == "/backend-api/wham/apps"
+    }) {
+        CHATGPT_WHAM_APPS_ENDPOINT_LABEL
+    } else {
+        OTHER_METRIC_LABEL
+    }
+}
+
+fn exec_server_error_type(error: &ExecServerError) -> &'static str {
+    match error {
+        ExecServerError::Spawn(_) => "spawn",
+        ExecServerError::WebSocketConnectTimeout { .. } => "websocket_connect_timeout",
+        ExecServerError::WebSocketConnect { .. } => "websocket_connect",
+        ExecServerError::InitializeTimedOut { .. } => "initialize_timed_out",
+        ExecServerError::Closed => "closed",
+        ExecServerError::Disconnected(_) => "disconnected",
+        ExecServerError::Json(_) => "json",
+        ExecServerError::HttpRequest(_) => "http_request",
+        ExecServerError::Protocol(_) => "protocol",
+        ExecServerError::Server { .. } => "server",
+        ExecServerError::EnvironmentRegistryHttp { .. } => "environment_registry_http",
+        ExecServerError::EnvironmentRegistryConfig(_) => "environment_registry_config",
+        ExecServerError::EnvironmentRegistryAuth(_) => "environment_registry_auth",
+        ExecServerError::EnvironmentRegistryRequest(_) => "environment_registry_request",
+    }
+}
+
 fn insert_header<Error>(
     headers: &mut HeaderMap,
     name: HeaderName,
@@ -490,3 +602,75 @@ fn sse_stream_from_body(
     }))
     .boxed()
 }
+
+#[cfg(test)]
+mod tests {
+    use std::time::Duration;
+
+    use pretty_assertions::assert_eq;
+
+    use super::*;
+
+    #[test]
+    fn post_message_failure_metric_tags_include_initialize_endpoint() {
+        let url = reqwest::Url::parse("https://chatgpt.com/backend-api/wham/apps").unwrap();
+        let error = ExecServerError::HttpRequest("error sending request".to_string());
+
+        let tags = post_message_failure_metric_tags(PostMessageFailureMetricFields {
+            server_name: "codex_apps",
+            parsed_url: Some(&url),
+            mcp_method: Some("initialize"),
+            has_session_id: false,
+            has_authorization_header: true,
+            error: &error,
+        });
+
+        assert_eq!(
+            tags,
+            vec![
+                ("server_name", "codex_apps".to_string()),
+                ("endpoint", "chatgpt_wham_apps".to_string()),
+                ("mcp_method", "initialize".to_string()),
+                ("has_session_id", "false".to_string()),
+                ("has_authorization_header", "true".to_string()),
+                ("error_type", "http_request".to_string()),
+            ]
+        );
+    }
+
+    #[test]
+    fn exec_server_error_type_covers_timeout_variant() {
+        let error = ExecServerError::InitializeTimedOut {
+            timeout: Duration::from_secs(5),
+        };
+
+        assert_eq!(exec_server_error_type(&error), "initialize_timed_out");
+    }
+
+    #[test]
+    fn post_message_failure_metric_tags_bucket_unknown_server_and_endpoint() {
+        let url = reqwest::Url::parse("https://example.com/custom/mcp").unwrap();
+        let error = ExecServerError::HttpRequest("error sending request".to_string());
+
+        let tags = post_message_failure_metric_tags(PostMessageFailureMetricFields {
+            server_name: "user_configured_server",
+            parsed_url: Some(&url),
+            mcp_method: Some("tools/list"),
+            has_session_id: true,
+            has_authorization_header: false,
+            error: &error,
+        });
+
+        assert_eq!(
+            tags,
+            vec![
+                ("server_name", "other".to_string()),
+                ("endpoint", "other".to_string()),
+                ("mcp_method", "tools/list".to_string()),
+                ("has_session_id", "true".to_string()),
+                ("has_authorization_header", "false".to_string()),
+                ("error_type", "http_request".to_string()),
+            ]
+        );
+    }
+}

codex-rs/rmcp-client/src/rmcp_client.rs

@@ -784,6 +784,7 @@ impl RmcpClient {
                                     .auth_header(access_token);
                             let transport = StreamableHttpClientTransport::with_client(
                                 StreamableHttpClientAdapter::new(
+                                    server_name,
                                     Arc::clone(http_client),
                                     default_headers,
                                     /*auth_provider*/ None,
@@ -803,6 +804,7 @@ impl RmcpClient {
 
                     let transport = StreamableHttpClientTransport::with_client(
                         StreamableHttpClientAdapter::new(
+                            server_name,
                             Arc::clone(http_client),
                             default_headers,
                             auth_provider.clone(),
@@ -1040,7 +1042,12 @@ async fn create_oauth_transport_and_runtime(
     };
 
     let auth_client = AuthClient::new(
-        StreamableHttpClientAdapter::new(http_client, default_headers, /*auth_provider*/ None),
+        StreamableHttpClientAdapter::new(
+            server_name,
+            http_client,
+            default_headers,
+            /*auth_provider*/ None,
+        ),
         manager,
     );
     let auth_manager = auth_client.auth_manager.clone();