From 23784749bd78517e7e9ff768e6d8e72c41a4e4c7 Mon Sep 17 00:00:00 2001 From: Matthew Zeng Date: Mon, 6 Jul 2026 16:38:04 -0700 Subject: [PATCH] add connector runtime latency metrics --- .../src/request_processors/apps_processor.rs | 35 ++++++++++++++++--- codex-rs/codex-mcp/src/connection_manager.rs | 21 ++++++----- codex-rs/codex-mcp/src/rmcp_client.rs | 33 +++++++++++++---- 3 files changed, 68 insertions(+), 21 deletions(-) diff --git a/codex-rs/app-server/src/request_processors/apps_processor.rs b/codex-rs/app-server/src/request_processors/apps_processor.rs index 67d768fa57b0..5ff7ec40db5a 100644 --- a/codex-rs/app-server/src/request_processors/apps_processor.rs +++ b/codex-rs/app-server/src/request_processors/apps_processor.rs @@ -47,6 +47,8 @@ impl AppsRequestProcessor { request_id: &ConnectionRequestId, params: AppsListParams, ) -> Result, JSONRPCErrorError> { + let installed_start = Instant::now(); + let reload = params.force_refetch; let thread = if let Some(thread_id) = params.thread_id.as_deref() { let (_, loaded_thread) = self.load_thread(thread_id).await?; Some(loaded_thread) @@ -70,20 +72,24 @@ impl AppsRequestProcessor { .features .apps_enabled_for_auth(auth.as_ref().is_some_and(CodexAuth::uses_codex_backend)) { - return Ok(Some(AppsListResponse { + let response = AppsListResponse { data: Vec::new(), next_cursor: None, - })); + }; + record_legacy_apps_installed_duration(installed_start, reload); + return Ok(Some(response)); } if !self .workspace_codex_plugins_enabled(&config, auth.as_ref()) .await { - return Ok(Some(AppsListResponse { + let response = AppsListResponse { data: Vec::new(), next_cursor: None, - })); + }; + record_legacy_apps_installed_duration(installed_start, reload); + return Ok(Some(response)); } let request = request_id.clone(); @@ -103,6 +109,7 @@ impl AppsRequestProcessor { environment_manager, mcp_manager, plugins_manager, + installed_start, ) => {} } }); @@ -113,6 +120,7 @@ impl AppsRequestProcessor { self.shutdown_token.cancel(); } + #[allow(clippy::too_many_arguments)] async fn apps_list_task( outgoing: Arc, request_id: ConnectionRequestId, @@ -121,7 +129,9 @@ impl AppsRequestProcessor { environment_manager: Arc, mcp_manager: Arc, plugins_manager: Arc, + installed_start: Instant, ) { + let reload = params.force_refetch; let retry_params = params.clone(); let retry_config = config.clone(); let retry_environment_manager = Arc::clone(&environment_manager); @@ -136,6 +146,9 @@ impl AppsRequestProcessor { plugins_manager, ) .await; + if result.is_ok() { + record_legacy_apps_installed_duration(installed_start, reload); + } let should_retry = result .as_ref() .is_ok_and(|(_, codex_apps_ready)| !codex_apps_ready); @@ -363,6 +376,20 @@ impl AppsRequestProcessor { } const APP_LIST_LOAD_TIMEOUT: Duration = Duration::from_secs(90); +// `app/list` is the legacy request-path baseline for the future `app/installed` endpoint; +// `path=legacy` keeps it separate from the new snapshot-backed implementation in dashboards. +const APPS_INSTALLED_DURATION_METRIC: &str = "codex.apps.installed.duration_ms"; + +fn record_legacy_apps_installed_duration(started_at: Instant, reload: bool) { + let reload = if reload { "true" } else { "false" }; + if let Some(metrics) = codex_otel::global() { + let _ = metrics.record_duration( + APPS_INSTALLED_DURATION_METRIC, + started_at.elapsed(), + &[("path", "legacy"), ("reload", reload)], + ); + } +} enum AppListLoadResult { Accessible(Result), diff --git a/codex-rs/codex-mcp/src/connection_manager.rs b/codex-rs/codex-mcp/src/connection_manager.rs index 2143e3b57765..223d4296d51e 100644 --- a/codex-rs/codex-mcp/src/connection_manager.rs +++ b/codex-rs/codex-mcp/src/connection_manager.rs @@ -23,8 +23,8 @@ use crate::elicitation::ElicitationReviewerHandle; use crate::mcp::CODEX_APPS_MCP_SERVER_NAME; use crate::mcp::ToolPluginProvenance; use crate::rmcp_client::AsyncManagedClient; +use crate::rmcp_client::CODEX_APPS_REFRESH_DURATION_METRIC; use crate::rmcp_client::DEFAULT_STARTUP_TIMEOUT; -use crate::rmcp_client::MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC; use crate::rmcp_client::MCP_TOOLS_LIST_DURATION_METRIC; use crate::rmcp_client::ManagedClient; use crate::rmcp_client::StartupOutcomeError; @@ -530,6 +530,7 @@ impl McpConnectionManager { /// cache is enabled and the latest filtered tools are returned directly to /// the caller. On failure, existing shared cache contents remain unchanged. pub async fn hard_refresh_codex_apps_tools_cache(&self) -> Result> { + let refresh_start = Instant::now(); let managed_client = self .clients .get(CODEX_APPS_MCP_SERVER_NAME) @@ -539,7 +540,6 @@ impl McpConnectionManager { .context("failed to get client")?; let list_start = Instant::now(); - let fetch_start = Instant::now(); let fetch_ticket = managed_client .codex_apps_tools_cache_context .as_ref() @@ -547,6 +547,7 @@ impl McpConnectionManager { let tools = list_tools_for_client_uncached( CODEX_APPS_MCP_SERVER_NAME, /*is_codex_apps_mcp_server*/ true, + /*codex_apps_refresh_trigger*/ "explicit", &managed_client.client, managed_client.tool_timeout, managed_client.server_instructions.as_deref(), @@ -555,11 +556,6 @@ impl McpConnectionManager { .with_context(|| { format!("failed to refresh tools for MCP server '{CODEX_APPS_MCP_SERVER_NAME}'") })?; - emit_duration( - MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC, - fetch_start.elapsed(), - &[], - ); let tools = match ( @@ -582,10 +578,13 @@ impl McpConnectionManager { tool.tool = tool_with_model_visible_input_schema(&tool.tool); self.with_server_metadata(tool) }); - Ok(normalize_tools_for_model_with_prefix( - tools, - self.prefix_mcp_tool_names, - )) + let tools = normalize_tools_for_model_with_prefix(tools, self.prefix_mcp_tool_names); + emit_duration( + CODEX_APPS_REFRESH_DURATION_METRIC, + refresh_start.elapsed(), + &[("path", "legacy"), ("trigger", "explicit")], + ); + Ok(tools) } /// Returns resources from servers selected by `include_server`. Each key diff --git a/codex-rs/codex-mcp/src/rmcp_client.rs b/codex-rs/codex-mcp/src/rmcp_client.rs index 37819203f6ac..6fe820b4fe3f 100644 --- a/codex-rs/codex-mcp/src/rmcp_client.rs +++ b/codex-rs/codex-mcp/src/rmcp_client.rs @@ -82,6 +82,7 @@ pub const OPENAI_FORM_CAPABILITY: &str = "openai/form"; pub(crate) const MCP_TOOLS_LIST_DURATION_METRIC: &str = "codex.mcp.tools.list.duration_ms"; pub(crate) const MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC: &str = "codex.mcp.tools.fetch_uncached.duration_ms"; +pub(crate) const CODEX_APPS_REFRESH_DURATION_METRIC: &str = "codex.apps.refresh.duration_ms"; pub(crate) const DEFAULT_STARTUP_TIMEOUT: Duration = Duration::from_secs(30); pub(crate) const DEFAULT_TOOL_TIMEOUT: Duration = Duration::from_secs(300); @@ -304,6 +305,7 @@ impl ManagedClientStartup { .unwrap_or_default(); let cancel_token_for_fut = cancel_token; async move { + let refresh_start = is_codex_apps_mcp_server.then(Instant::now); let outcome = match async { if let Err(error) = validate_mcp_server_name(&server_name) { return Err(error.into()); @@ -349,6 +351,15 @@ impl ManagedClientStartup { Ok(result) => result, Err(CancelErr::Cancelled) => Err(StartupOutcomeError::Cancelled), }; + if outcome.is_ok() + && let Some(refresh_start) = refresh_start + { + emit_duration( + CODEX_APPS_REFRESH_DURATION_METRIC, + refresh_start.elapsed(), + &[("path", "legacy"), ("trigger", "initial")], + ); + } startup_complete.store(true, Ordering::Release); outcome @@ -564,10 +575,12 @@ impl From for StartupOutcomeError { pub(crate) async fn list_tools_for_client_uncached( server_name: &str, is_codex_apps_mcp_server: bool, + codex_apps_refresh_trigger: &'static str, client: &Arc, timeout: Option, server_instructions: Option<&str>, ) -> Result> { + let fetch_start = Instant::now(); let resp = client .list_tools_with_connector_ids(/*params*/ None, timeout) .await?; @@ -583,6 +596,19 @@ pub(crate) async fn list_tools_for_client_uncached( ) }) .collect(); + if is_codex_apps_mcp_server { + emit_duration( + MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC, + fetch_start.elapsed(), + &[("trigger", codex_apps_refresh_trigger)], + ); + } else { + emit_duration( + MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC, + fetch_start.elapsed(), + &[], + ); + } Ok(tools) } @@ -820,24 +846,19 @@ async fn start_server_task( .and_then(|exp| exp.get(MCP_SANDBOX_STATE_META_CAPABILITY)) .is_some(); let list_start = Instant::now(); - let fetch_start = Instant::now(); let fetch_ticket = codex_apps_tools_cache_context .as_ref() .map(|cache_context| cache_context.begin_fetch(CodexAppsToolsFetchSource::Startup)); let tools = list_tools_for_client_uncached( &server_name, is_codex_apps_mcp_server, + /*codex_apps_refresh_trigger*/ "initial", &client, startup_timeout, initialize_result.instructions.as_deref(), ) .await .map_err(StartupOutcomeError::from)?; - emit_duration( - MCP_TOOLS_FETCH_UNCACHED_DURATION_METRIC, - fetch_start.elapsed(), - &[], - ); let server_info = mcp_server_info_from_implementation(initialize_result.server_info); let tools = match (codex_apps_tools_cache_context.as_ref(), fetch_ticket) { (Some(cache_context), Some(fetch_ticket)) => {