Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
35 changes: 31 additions & 4 deletions codex-rs/app-server/src/request_processors/apps_processor.rs
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,8 @@ impl AppsRequestProcessor {
request_id: &ConnectionRequestId,
params: AppsListParams,
) -> Result<Option<AppsListResponse>, 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)
Expand All @@ -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();
Expand All @@ -103,6 +109,7 @@ impl AppsRequestProcessor {
environment_manager,
mcp_manager,
plugins_manager,
installed_start,
) => {}
}
});
Expand All @@ -113,6 +120,7 @@ impl AppsRequestProcessor {
self.shutdown_token.cancel();
}

#[allow(clippy::too_many_arguments)]
async fn apps_list_task(
outgoing: Arc<OutgoingMessageSender>,
request_id: ConnectionRequestId,
Expand All @@ -121,7 +129,9 @@ impl AppsRequestProcessor {
environment_manager: Arc<EnvironmentManager>,
mcp_manager: Arc<McpManager>,
plugins_manager: Arc<PluginsManager>,
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);
Expand All @@ -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);
Expand Down Expand Up @@ -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<AccessibleConnectorsStatus, String>),
Expand Down
21 changes: 10 additions & 11 deletions codex-rs/codex-mcp/src/connection_manager.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<Vec<ToolInfo>> {
let refresh_start = Instant::now();
let managed_client = self
.clients
.get(CODEX_APPS_MCP_SERVER_NAME)
Expand All @@ -539,14 +540,14 @@ 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()
.map(|cache_context| cache_context.begin_fetch(CodexAppsToolsFetchSource::HardRefresh));
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(),
Expand All @@ -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 (
Expand All @@ -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
Expand Down
33 changes: 27 additions & 6 deletions codex-rs/codex-mcp/src/rmcp_client.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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);

Expand Down Expand Up @@ -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());
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -564,10 +575,12 @@ impl From<anyhow::Error> 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<RmcpClient>,
timeout: Option<Duration>,
server_instructions: Option<&str>,
) -> Result<Vec<ToolInfo>> {
let fetch_start = Instant::now();
let resp = client
.list_tools_with_connector_ids(/*params*/ None, timeout)
.await?;
Expand All @@ -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)
}

Expand Down Expand Up @@ -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)) => {
Expand Down
Loading