Add websocket telemetry metrics and labels (#10316)

Summary
- expose websocket telemetry hooks through the responses client so
request durations and event processing can be reported
- record websocket request/event metrics and emit runtime telemetry
events that the history UI now surfaces
- improve tests to cover websocket telemetry reporting and guard runtime
summary updates


<img width="824" height="79" alt="Screenshot 2026-01-31 at 5 28 12 PM"
src="https://github.com/user-attachments/assets/ea9a7965-d8b4-4e3c-a984-ef4fdc44c81d"
/>

codex-rs/core/Cargo.toml

@@ -90,6 +90,7 @@ tokio = { workspace = true, features = [
     "signal",
 ] }
 tokio-util = { workspace = true, features = ["rt"] }
+tokio-tungstenite = { workspace = true }
 toml = { workspace = true }
 toml_edit = { workspace = true }
 tracing = { workspace = true, features = ["log"] }
@@ -145,6 +146,10 @@ image = { workspace = true, features = ["jpeg", "png"] }
 maplit = { workspace = true }
 predicates = { workspace = true }
 pretty_assertions = { workspace = true }
+opentelemetry_sdk = { workspace = true, features = [
+    "experimental_metrics_custom_reader",
+    "metrics",
+] }
 serial_test = { workspace = true }
 tempfile = { workspace = true }
 tracing-subscriber = { workspace = true }

codex-rs/core/src/client.rs

@@ -21,6 +21,7 @@ use codex_api::ResponsesWebsocketClient as ApiWebSocketResponsesClient;
 use codex_api::ResponsesWebsocketConnection as ApiWebSocketConnection;
 use codex_api::SseTelemetry;
 use codex_api::TransportError;
+use codex_api::WebsocketTelemetry;
 use codex_api::build_conversation_headers;
 use codex_api::common::Reasoning;
 use codex_api::common::ResponsesWsRequest;
@@ -46,6 +47,8 @@ use reqwest::StatusCode;
 use serde_json::Value;
 use std::time::Duration;
 use tokio::sync::mpsc;
+use tokio_tungstenite::tungstenite::Error;
+use tokio_tungstenite::tungstenite::Message;
 use tracing::warn;
 
 use crate::AuthManager;
@@ -451,9 +454,14 @@ impl ModelClientSession {
         if needs_new {
             let mut headers = options.extra_headers.clone();
             headers.extend(build_conversation_headers(options.conversation_id.clone()));
+            let websocket_telemetry = self.build_websocket_telemetry();
             let new_conn: ApiWebSocketConnection =
                 ApiWebSocketResponsesClient::new(api_provider, api_auth)
-                    .connect(headers, options.turn_state.clone())
+                    .connect(
+                        headers,
+                        options.turn_state.clone(),
+                        Some(websocket_telemetry),
+                    )
                     .await?;
             self.connection = Some(new_conn);
         }
@@ -650,6 +658,13 @@ impl ModelClientSession {
         let sse_telemetry: Arc<dyn SseTelemetry> = telemetry;
         (request_telemetry, sse_telemetry)
     }
+
+    /// Builds telemetry for the Responses API WebSocket transport.
+    fn build_websocket_telemetry(&self) -> Arc<dyn WebsocketTelemetry> {
+        let telemetry = Arc::new(ApiTelemetry::new(self.state.otel_manager.clone()));
+        let websocket_telemetry: Arc<dyn WebsocketTelemetry> = telemetry;
+        websocket_telemetry
+    }
 }
 
 impl ModelClient {
@@ -849,3 +864,19 @@ impl SseTelemetry for ApiTelemetry {
         self.otel_manager.log_sse_event(result, duration);
     }
 }
+
+impl WebsocketTelemetry for ApiTelemetry {
+    fn on_ws_request(&self, duration: Duration, error: Option<&ApiError>) {
+        let error_message = error.map(std::string::ToString::to_string);
+        self.otel_manager
+            .record_websocket_request(duration, error_message.as_deref());
+    }
+
+    fn on_ws_event(
+        &self,
+        result: &std::result::Result<Option<std::result::Result<Message, Error>>, ApiError>,
+        duration: Duration,
+    ) {
+        self.otel_manager.record_websocket_event(result, duration);
+    }
+}

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

@@ -14,6 +14,8 @@ use codex_core::features::Feature;
 use codex_core::models_manager::manager::ModelsManager;
 use codex_core::protocol::SessionSource;
 use codex_otel::OtelManager;
+use codex_otel::metrics::MetricsClient;
+use codex_otel::metrics::MetricsConfig;
 use codex_protocol::ThreadId;
 use codex_protocol::config_types::ReasoningSummary;
 use core_test_support::load_default_config_for_test;
@@ -25,15 +27,19 @@ use core_test_support::responses::start_websocket_server;
 use core_test_support::responses::start_websocket_server_with_headers;
 use core_test_support::skip_if_no_network;
 use futures::StreamExt;
+use opentelemetry_sdk::metrics::InMemoryMetricExporter;
 use pretty_assertions::assert_eq;
 use std::sync::Arc;
+use std::time::Duration;
 use tempfile::TempDir;
+use tracing_test::traced_test;
 
 const MODEL: &str = "gpt-5.2-codex";
 
 struct WebsocketTestHarness {
     _codex_home: TempDir,
     client: ModelClient,
+    otel_manager: OtelManager,
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -64,6 +70,38 @@ async fn responses_websocket_streams_request() {
     server.shutdown().await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[traced_test]
+async fn responses_websocket_emits_websocket_telemetry_events() {
+    skip_if_no_network!();
+
+    let server = start_websocket_server(vec![vec![vec![
+        ev_response_created("resp-1"),
+        ev_completed("resp-1"),
+    ]]])
+    .await;
+
+    let harness = websocket_harness(&server).await;
+    harness.otel_manager.reset_runtime_metrics();
+    let mut session = harness.client.new_session();
+    let prompt = prompt_with_input(vec![message_item("hello")]);
+
+    stream_until_complete(&mut session, &prompt).await;
+
+    tokio::time::sleep(Duration::from_millis(10)).await;
+
+    let summary = harness
+        .otel_manager
+        .runtime_metrics_summary()
+        .expect("runtime metrics summary");
+    assert_eq!(summary.api_calls.count, 0);
+    assert_eq!(summary.streaming_events.count, 0);
+    assert_eq!(summary.websocket_calls.count, 1);
+    assert_eq!(summary.websocket_events.count, 2);
+
+    server.shutdown().await;
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn responses_websocket_emits_reasoning_included_event() {
     skip_if_no_network!();
@@ -211,6 +249,12 @@ async fn websocket_harness(server: &WebSocketTestServer) -> WebsocketTestHarness
     let model_info = ModelsManager::construct_model_info_offline(MODEL, &config);
     let conversation_id = ThreadId::new();
     let auth_manager = AuthManager::from_auth_for_testing(CodexAuth::from_api_key("Test API Key"));
+    let exporter = InMemoryMetricExporter::default();
+    let metrics = MetricsClient::new(
+        MetricsConfig::in_memory("test", "codex-core", env!("CARGO_PKG_VERSION"), exporter)
+            .with_runtime_reader(),
+    )
+    .expect("in-memory metrics client");
     let otel_manager = OtelManager::new(
         conversation_id,
         MODEL,
@@ -221,12 +265,13 @@ async fn websocket_harness(server: &WebSocketTestServer) -> WebsocketTestHarness
         false,
         "test".to_string(),
         SessionSource::Exec,
-    );
+    )
+    .with_metrics(metrics);
     let client = ModelClient::new(
         Arc::clone(&config),
         None,
         model_info,
-        otel_manager,
+        otel_manager.clone(),
         provider.clone(),
         None,
         ReasoningSummary::Auto,
@@ -238,6 +283,7 @@ async fn websocket_harness(server: &WebSocketTestServer) -> WebsocketTestHarness
     WebsocketTestHarness {
         _codex_home: codex_home,
         client,
+        otel_manager,
     }
 }