From 0f1380eede62b6738753a6479edf10abff453c4e Mon Sep 17 00:00:00 2001 From: Alexey Ldokov Date: Thu, 9 Jul 2026 15:08:08 +0100 Subject: [PATCH] [codex] strengthen nonblocking MCP startup coverage --- .../app-server/tests/suite/v2/thread_start.rs | 24 +++++--- codex-rs/core/src/session/tests.rs | 55 +++++++++++++++++++ 2 files changed, 71 insertions(+), 8 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/thread_start.rs b/codex-rs/app-server/tests/suite/v2/thread_start.rs index 18b70cd383..9566a22a82 100644 --- a/codex-rs/app-server/tests/suite/v2/thread_start.rs +++ b/codex-rs/app-server/tests/suite/v2/thread_start.rs @@ -45,6 +45,7 @@ use std::path::Path; use std::path::PathBuf; use tempfile::TempDir; use tokio::net::TcpListener; +use tokio::sync::oneshot; use tokio::time::timeout; use wiremock::Mock; use wiremock::MockServer; @@ -1256,11 +1257,14 @@ async fn thread_start_does_not_wait_for_optional_http_mcp_auth_discovery() -> Re let server = create_mock_responses_server_repeating_assistant("Done").await; let listener = TcpListener::bind("127.0.0.1:0").await?; let mcp_addr = listener.local_addr()?; + let (connection_started_tx, connection_started_rx) = oneshot::channel(); let blackhole_server = tokio::spawn(async move { - let mut connections = Vec::new(); - while let Ok((connection, _)) = listener.accept().await { - connections.push(connection); - } + let Ok((connection, _)) = listener.accept().await else { + return; + }; + let _ = connection_started_tx.send(()); + let _connection = connection; + std::future::pending::<()>().await; }); let codex_home = TempDir::new()?; @@ -1280,16 +1284,20 @@ async fn thread_start_does_not_wait_for_optional_http_mcp_auth_discovery() -> Re .send_thread_start_request_with_auto_env(ThreadStartParams::default()) .await?; - let response: JSONRPCResponse = timeout( - std::time::Duration::from_secs(3), + timeout(DEFAULT_READ_TIMEOUT, connection_started_rx) + .await + .context("optional HTTP MCP never attempted a connection")??; + let response = timeout( + DEFAULT_READ_TIMEOUT, mcp.read_stream_until_response_message(RequestId::Integer(req_id)), ) .await - .context("thread/start waited for optional HTTP MCP auth discovery")??; + .context("thread/start waited for optional HTTP MCP auth discovery"); + blackhole_server.abort(); + let response: JSONRPCResponse = response??; let response: ThreadStartResponse = to_response(response)?; assert!(!response.thread.id.is_empty()); - blackhole_server.abort(); Ok(()) } diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 5c6d3d7bb2..63e06bf0de 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -16,6 +16,7 @@ use crate::skills::SkillRenderSideEffects; use crate::skills::render::SkillMetadataBudget; use crate::test_support::models_manager_with_provider; use crate::tools::format_exec_output_str; +use anyhow::Context; use codex_config::ConfigLayerStack; use codex_config::ConfigLayerStackOrdering; use codex_config::DEFAULT_MCP_SERVER_ENVIRONMENT_ID; @@ -177,7 +178,9 @@ use opentelemetry_sdk::metrics::data::MetricData; use opentelemetry_sdk::metrics::data::ResourceMetrics; use std::path::Path; use std::time::Duration; +use tokio::net::TcpListener; use tokio::sync::Semaphore; +use tokio::sync::oneshot; use tokio::time::sleep; use tokio::time::timeout; use tracing_opentelemetry::OpenTelemetrySpanExt; @@ -7823,6 +7826,58 @@ async fn refresh_mcp_servers_keeps_the_previous_runtime_alive() { ); } +#[tokio::test] +async fn refresh_mcp_servers_does_not_wait_for_optional_http_auth_discovery() -> anyhow::Result<()> +{ + let listener = TcpListener::bind("127.0.0.1:0").await?; + let mcp_addr = listener.local_addr()?; + let (connection_started_tx, connection_started_rx) = oneshot::channel(); + let blackhole_server = tokio::spawn(async move { + let Ok((connection, _)) = listener.accept().await else { + return; + }; + let _ = connection_started_tx.send(()); + let _connection = connection; + std::future::pending::<()>().await; + }); + + let (session, turn_context) = make_session_and_context().await; + let session = Arc::new(session); + let mut refresh_config = turn_context.config.as_ref().clone(); + let refreshed_mcp_servers = + serde_json::from_value::>(json!({ + "optional_http": { + "url": format!("http://{mcp_addr}/mcp"), + "startup_timeout_sec": 60 + } + }))?; + refresh_config + .mcp_servers + .set(refreshed_mcp_servers.clone())?; + + let refresh_result = timeout( + Duration::from_secs(10), + session.refresh_mcp_servers_now( + &turn_context, + &refresh_config, + /*elicitation_reviewer*/ None, + ), + ) + .await; + let connection_result = timeout(Duration::from_secs(10), connection_started_rx).await; + blackhole_server.abort(); + + refresh_result.context("MCP refresh waited for optional HTTP auth discovery")?; + connection_result.context("optional HTTP MCP never attempted a connection")??; + let configured_mcp_servers = + codex_mcp::configured_mcp_servers(session.services.latest_mcp_runtime().config()); + assert_eq!( + configured_mcp_servers.get("optional_http"), + refreshed_mcp_servers.get("optional_http") + ); + Ok(()) +} + #[tokio::test] async fn plugin_availability_change_reuses_the_mcp_manager() { struct ReadyPluginContributor;