Expose runtime activity in server diagnostics (#37486)

## What changed

- Add lifecycle-backed gauges for in-flight and queued app requests, pending server requests and mailbox messages, active turns, and live MCP connections.
- Register each gauge on first use and decrement it when the tracked work or resource is dropped.
- Document the new `server/diagnostics` gauge names and account for the diagnostics request itself in the response example and test.

GitOrigin-RevId: 4bb82a3918af0b502d149b5d761ad669b2083f39

codex-rs/Cargo.lock

@@ -3481,6 +3481,7 @@ dependencies = [
  "codex-async-utils",
  "codex-config",
  "codex-connectors",
+ "codex-diagnostics",
  "codex-exec-server",
  "codex-exec-server-test-support",
  "codex-login",

codex-rs/app-server/README.md

@@ -499,10 +499,15 @@ Enable `capabilities.experimentalApi` during initialization, then use `thread/li
         "residentMemoryBytes": 4194304,
         "physicalFootprintBytes": 5242880
     },
-    "gauges": [{ "name": "core.threads.live", "value": 1 }]
+    "gauges": [
+        { "name": "app.requests.in_flight", "value": 1 },
+        { "name": "core.threads.live", "value": 1 }
+    ]
 } }
 ```
 
+Gauges register when first used. Depending on process activity, the snapshot can also include `app.requests.queued`, `app.server_requests.pending`, `core.mailbox.pending`, `core.turns.active`, and `mcp.connections.live`. The diagnostics request itself is included in `app.requests.in_flight`.
+
 ### Example: Track thread status changes
 
 `thread/status/changed` is emitted whenever a loaded thread's status changes after it has already been introduced to the client:

codex-rs/app-server/src/outgoing_message.rs

@@ -15,6 +15,8 @@ use codex_app_server_protocol::ServerNotificationEnvelope;
 use codex_app_server_protocol::ServerRequest;
 use codex_app_server_protocol::ServerRequestPayload;
 use codex_app_server_protocol::ServerResponse;
+use codex_diagnostics::Gauge;
+use codex_diagnostics::GaugeGuard;
 use codex_otel::span_w3c_trace_context;
 use codex_protocol::ThreadId;
 use codex_protocol::protocol::W3cTraceContext;
@@ -39,6 +41,9 @@ use codex_protocol::account::PlanType;
 
 pub(crate) type ClientRequestResult = std::result::Result<Result, JSONRPCErrorError>;
 
+static IN_FLIGHT_REQUESTS: Gauge = Gauge::new("app.requests.in_flight");
+static PENDING_SERVER_REQUESTS: Gauge = Gauge::new("app.server_requests.pending");
+
 /// Stable identifier for a client request scoped to a transport connection.
 #[derive(Clone, Debug, Eq, Hash, PartialEq)]
 pub(crate) struct ConnectionRequestId {
@@ -53,6 +58,7 @@ pub(crate) struct RequestContext {
     request_id: ConnectionRequestId,
     span: Span,
     parent_trace: Option<W3cTraceContext>,
+    _diagnostics_guard: Arc<GaugeGuard>,
 }
 
 impl RequestContext {
@@ -65,6 +71,7 @@ impl RequestContext {
             request_id,
             span,
             parent_trace,
+            _diagnostics_guard: Arc::new(IN_FLIGHT_REQUESTS.track()),
         }
     }
 
@@ -116,6 +123,7 @@ struct PendingCallbackEntry {
     callback: oneshot::Sender<ClientRequestResult>,
     thread_id: Option<ThreadId>,
     request: ServerRequest,
+    _diagnostics_guard: GaugeGuard,
 }
 
 impl ThreadScopedOutgoingMessageSender {
@@ -303,6 +311,7 @@ impl OutgoingMessageSender {
                     callback: tx_approve,
                     thread_id,
                     request: request.clone(),
+                    _diagnostics_guard: PENDING_SERVER_REQUESTS.track(),
                 },
             );
         }

codex-rs/app-server/src/request_serialization.rs

@@ -6,6 +6,8 @@ use std::pin::Pin;
 use std::sync::Arc;
 
 use codex_app_server_protocol::ClientRequestSerializationScope;
+use codex_diagnostics::Gauge;
+use codex_diagnostics::GaugeGuard;
 use futures::future::join_all;
 use tokio::sync::Mutex;
 use tracing::Instrument;
@@ -15,6 +17,8 @@ use crate::outgoing_message::ConnectionId;
 
 type BoxFutureUnit = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
 
+static QUEUED_REQUESTS: Gauge = Gauge::new("app.requests.queued");
+
 #[derive(Clone, Debug, Eq, Hash, PartialEq)]
 pub(crate) enum RequestSerializationQueueKey {
     Global(&'static str),
@@ -138,6 +142,7 @@ impl QueuedInitializedRequest {
 struct QueuedSerializedRequest {
     access: RequestSerializationAccess,
     request: QueuedInitializedRequest,
+    _diagnostics_guard: GaugeGuard,
 }
 
 #[derive(Clone, Default)]
@@ -167,7 +172,11 @@ impl RequestSerializationQueues {
         access: RequestSerializationAccess,
         request: QueuedInitializedRequest,
     ) {
-        let request = QueuedSerializedRequest { access, request };
+        let request = QueuedSerializedRequest {
+            access,
+            request,
+            _diagnostics_guard: QUEUED_REQUESTS.track(),
+        };
         let should_spawn = {
             let mut queues = self.inner.lock().await;
             match queues.get_mut(&key) {

codex-rs/app-server/tests/suite/v2/server_diagnostics.rs

@@ -47,13 +47,24 @@ async fn server_diagnostics_exposes_process_and_registered_thread_gauge() -> Res
     assert!(diagnostics.process.physical_footprint_bytes.is_some());
     #[cfg(not(target_os = "macos"))]
     assert_eq!(diagnostics.process.physical_footprint_bytes, None);
-    assert_eq!(
-        diagnostics.gauges,
-        vec![ServerDiagnosticsGauge {
+    for expected_gauge in [
+        ServerDiagnosticsGauge {
+            name: "app.requests.in_flight".to_string(),
+            value: 1,
+        },
+        ServerDiagnosticsGauge {
             name: "core.threads.live".to_string(),
             value: 1,
-        }]
-    );
+        },
+    ] {
+        assert_eq!(
+            diagnostics
+                .gauges
+                .iter()
+                .find(|gauge| gauge.name == expected_gauge.name),
+            Some(&expected_gauge)
+        );
+    }
 
     Ok(())
 }

codex-rs/codex-mcp/Cargo.toml

@@ -20,6 +20,7 @@ codex-async-utils = { workspace = true }
 codex-api = { workspace = true }
 codex-config = { workspace = true }
 codex-connectors = { workspace = true }
+codex-diagnostics = { workspace = true }
 codex-exec-server = { workspace = true }
 codex-login = { workspace = true }
 codex-model-provider = { workspace = true }

codex-rs/codex-mcp/src/connection_manager.rs

@@ -54,6 +54,8 @@ use anyhow::Context;
 use anyhow::Result;
 use anyhow::anyhow;
 use codex_config::McpServerTransportConfig;
+use codex_diagnostics::Gauge;
+use codex_diagnostics::GaugeGuard;
 use codex_protocol::mcp::CallToolResult;
 use codex_protocol::mcp::McpServerInfo;
 use codex_protocol::models::PermissionProfile;
@@ -71,10 +73,13 @@ use tokio::sync::watch;
 use tokio::task::JoinSet;
 use tracing::warn;
 
+static LIVE_CONNECTIONS: Gauge = Gauge::new("mcp.connections.live");
+
 pub(crate) struct McpServerConnection {
     identity: Option<McpServerConnectionIdentity>,
     client: AsyncManagedClient,
     startup_trigger: Option<watch::Sender<bool>>,
+    _diagnostics_guard: GaugeGuard,
 }
 
 impl McpServerConnection {
@@ -440,6 +445,7 @@ impl McpConnectionSet {
                         identity: Some(connection_identity),
                         client: async_managed_client.clone(),
                         startup_trigger,
+                        _diagnostics_guard: LIVE_CONNECTIONS.track(),
                     }),
                     metadata,
                     tool_filter: configured_tool_filter,

codex-rs/codex-mcp/src/connection_manager_tests.rs

@@ -116,6 +116,7 @@ impl McpConnectionSet {
                     identity: None,
                     client,
                     startup_trigger: None,
+                    _diagnostics_guard: LIVE_CONNECTIONS.track(),
                 }),
                 metadata: McpServerMetadata {
                     environment_id: String::new(),
@@ -3508,6 +3509,7 @@ async fn manager_with_reusable_ready_server(
                 identity: Some(reusable_server_identity(config, runtime_context)),
                 client: create_ready_async_managed_client(tools).await,
                 startup_trigger: None,
+                _diagnostics_guard: LIVE_CONNECTIONS.track(),
             }),
             metadata: McpServerMetadata::from(&server),
             tool_filter: ToolFilter::from_config(config),
@@ -3641,6 +3643,7 @@ async fn reconciliation_reuses_connection_without_relisting_regular_tools() -> a
                     cancel_token: CancellationToken::new(),
                 },
                 startup_trigger: None,
+                _diagnostics_guard: LIVE_CONNECTIONS.track(),
             }),
             metadata: McpServerMetadata::from(&server),
             tool_filter: ToolFilter::from_config(&config),
@@ -3948,6 +3951,7 @@ async fn reconciliation_replaces_closed_connections() -> anyhow::Result<()> {
             cancel_token: CancellationToken::new(),
         },
         startup_trigger: None,
+        _diagnostics_guard: LIVE_CONNECTIONS.track(),
     });
 
     assert!(!client.is_closed().await);

codex-rs/core/src/session/input_queue.rs

@@ -1,6 +1,8 @@
 use crate::state::ActiveTurn;
 use crate::state::MailboxDeliveryPhase;
 use crate::state::TurnState;
+use codex_diagnostics::Gauge;
+use codex_diagnostics::GaugeGuard;
 use codex_protocol::models::ResponseItem;
 use codex_protocol::protocol::InterAgentCommunication;
 use codex_protocol::user_input::UserInput;
@@ -11,6 +13,8 @@ use std::sync::Arc;
 use tokio::sync::Mutex;
 use tokio::sync::watch;
 
+static PENDING_MAILBOX_MESSAGES: Gauge = Gauge::new("core.mailbox.pending");
+
 /// Input consumed by a regular turn.
 #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
 pub enum TurnInput {
@@ -43,6 +47,7 @@ pub(crate) struct InputQueue {
 struct PendingMailboxCommunication {
     communication: InterAgentCommunication,
     parent_turn_id: Option<String>,
+    _diagnostics_guard: GaugeGuard,
 }
 
 impl InputQueue {
@@ -88,6 +93,7 @@ impl InputQueue {
             .push_back(PendingMailboxCommunication {
                 communication,
                 parent_turn_id,
+                _diagnostics_guard: PENDING_MAILBOX_MESSAGES.track(),
             });
         self.activity_tx.send_replace(InputQueueActivity::Mailbox);
     }

codex-rs/core/src/state/turn.rs

@@ -7,6 +7,7 @@ use tokio::sync::Notify;
 use tokio_util::sync::CancellationToken;
 use tokio_util::task::AbortOnDropHandle;
 
+use codex_diagnostics::GaugeGuard;
 use codex_protocol::dynamic_tools::DynamicToolResponse;
 use codex_protocol::protocol::TurnEnvironmentSelection;
 use codex_protocol::request_permissions::RequestPermissionProfile;
@@ -81,6 +82,7 @@ pub(crate) struct RunningTask {
     pub(crate) handle: AbortOnDropHandle<()>,
     pub(crate) turn_context: Arc<TurnContext>,
     pub(crate) _agent_execution_guard: Option<AgentExecutionGuard>,
+    pub(crate) _diagnostics_guard: GaugeGuard,
     // Timer recorded when the task drops to capture the full turn duration.
     pub(crate) _timer: Option<codex_otel::Timer>,
 }

codex-rs/core/src/tasks/mod.rs

@@ -8,6 +8,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
 
+use codex_diagnostics::Gauge;
 use codex_extension_api::ThreadIdleCause;
 use futures::future::BoxFuture;
 use tokio::select;
@@ -63,6 +64,7 @@ pub(crate) use user_shell::execute_user_shell_command;
 
 const GRACEFULL_INTERRUPTION_TIMEOUT_MS: u64 = 100;
 const TASK_COMPACT_METRIC: &str = "codex.task.compact";
+static ACTIVE_TURNS: Gauge = Gauge::new("core.turns.active");
 
 pub(crate) type SessionTaskResult = CodexResult<Option<String>>;
 
@@ -421,6 +423,7 @@ impl Session {
             cancellation_token,
             turn_context: Arc::clone(&turn_context),
             _agent_execution_guard: agent_execution_guard,
+            _diagnostics_guard: ACTIVE_TURNS.track(),
             _timer: timer,
         };
         turn.task = Some(running_task);