From 27e4a05cd34f069ce043770b8e79969c992e87d4 Mon Sep 17 00:00:00 2001 From: jif Date: Fri, 7 Aug 2026 18:28:56 +0000 Subject: [PATCH] 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 | 1 + codex-rs/app-server/README.md | 7 ++++++- codex-rs/app-server/src/outgoing_message.rs | 9 ++++++++ .../app-server/src/request_serialization.rs | 11 +++++++++- .../tests/suite/v2/server_diagnostics.rs | 21 ++++++++++++++----- codex-rs/codex-mcp/Cargo.toml | 1 + codex-rs/codex-mcp/src/connection_manager.rs | 6 ++++++ .../codex-mcp/src/connection_manager_tests.rs | 4 ++++ codex-rs/core/src/session/input_queue.rs | 6 ++++++ codex-rs/core/src/state/turn.rs | 2 ++ codex-rs/core/src/tasks/mod.rs | 3 +++ 11 files changed, 64 insertions(+), 7 deletions(-) diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 251447eaad..e2f5c6bd58 100644 --- a/codex-rs/Cargo.lock +++ b/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", diff --git a/codex-rs/app-server/README.md b/codex-rs/app-server/README.md index d9cf12a8b7..e7d4abfaaa 100644 --- a/codex-rs/app-server/README.md +++ b/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: diff --git a/codex-rs/app-server/src/outgoing_message.rs b/codex-rs/app-server/src/outgoing_message.rs index 0b4933509b..d0dc22a12a 100644 --- a/codex-rs/app-server/src/outgoing_message.rs +++ b/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; +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, + _diagnostics_guard: Arc, } 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, thread_id: Option, 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(), }, ); } diff --git a/codex-rs/app-server/src/request_serialization.rs b/codex-rs/app-server/src/request_serialization.rs index e3038bbc30..5b1d5af2dc 100644 --- a/codex-rs/app-server/src/request_serialization.rs +++ b/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 + 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) { diff --git a/codex-rs/app-server/tests/suite/v2/server_diagnostics.rs b/codex-rs/app-server/tests/suite/v2/server_diagnostics.rs index fa26135994..3fba0104f7 100644 --- a/codex-rs/app-server/tests/suite/v2/server_diagnostics.rs +++ b/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(()) } diff --git a/codex-rs/codex-mcp/Cargo.toml b/codex-rs/codex-mcp/Cargo.toml index a61e71c51c..2fd886d232 100644 --- a/codex-rs/codex-mcp/Cargo.toml +++ b/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 } diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index d342d33a65..6880484b95 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/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, client: AsyncManagedClient, startup_trigger: Option>, + _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, diff --git a/codex-rs/codex-mcp/src/connection_manager_tests.rs b/codex-rs/codex-mcp/src/connection_manager_tests.rs index 571bbb808a..a2e2a7c4a1 100644 --- a/codex-rs/codex-mcp/src/connection_manager_tests.rs +++ b/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); diff --git a/codex-rs/core/src/session/input_queue.rs b/codex-rs/core/src/session/input_queue.rs index a133ec106f..9f35a335e2 100644 --- a/codex-rs/core/src/session/input_queue.rs +++ b/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, + _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); } diff --git a/codex-rs/core/src/state/turn.rs b/codex-rs/core/src/state/turn.rs index b284199d96..490dc042d1 100644 --- a/codex-rs/core/src/state/turn.rs +++ b/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, pub(crate) _agent_execution_guard: Option, + pub(crate) _diagnostics_guard: GaugeGuard, // Timer recorded when the task drops to capture the full turn duration. pub(crate) _timer: Option, } diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 0a2b39c49c..3cf2907b42 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/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>; @@ -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);