mirror of
https://github.com/openai/codex.git
synced 2026-10-26 18:38:47 +00:00
jif
e244a9d94e
## Why A streamable HTTP MCP handshake could time out while its executor-backed HTTP request continued running, leaving the serial executor blocked for later requests. ## What changed - Track the remaining initialization deadline for streamable HTTP transports. - Apply it to `initialize`, `notifications/initialized`, and discovery HTTP requests, then clear it when the handshake finishes. ## Testing - Added a regression test that stalls the remote MCP endpoint, waits for the handshake timeout, and verifies that a subsequent executor request completes. GitOrigin-RevId: d4980a332e898309c885cbaa4fdd598924b440e1
112 lines
4.1 KiB
Rust
112 lines
4.1 KiB
Rust
//! Integration coverage for the remote Streamable HTTP RMCP path.
|
|
//!
|
|
//! These tests exercise the orchestrator-side RMCP adapter against a real
|
|
//! `exec-server` process so HTTP requests go through the remote runtime path
|
|
//! instead of direct local `reqwest` calls.
|
|
|
|
mod streamable_http_test_support;
|
|
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
use codex_config::types::AuthKeyringBackendKind;
|
|
use codex_config::types::OAuthCredentialsStoreMode;
|
|
use codex_rmcp_client::ElicitationAction;
|
|
use codex_rmcp_client::ElicitationResponse;
|
|
use codex_rmcp_client::RmcpClient;
|
|
use futures::FutureExt as _;
|
|
use pretty_assertions::assert_eq;
|
|
use rmcp::model::ClientCapabilities;
|
|
use rmcp::model::Implementation;
|
|
use rmcp::model::InitializeRequestParams;
|
|
use rmcp::model::ProtocolVersion;
|
|
use wiremock::Mock;
|
|
use wiremock::MockServer;
|
|
use wiremock::ResponseTemplate;
|
|
use wiremock::matchers::method;
|
|
|
|
use streamable_http_test_support::call_echo_tool;
|
|
use streamable_http_test_support::create_remote_client;
|
|
use streamable_http_test_support::expected_echo_result;
|
|
use streamable_http_test_support::spawn_exec_server;
|
|
use streamable_http_test_support::spawn_streamable_http_server;
|
|
|
|
/// What this tests: the RMCP remote Streamable HTTP adapter can initialize
|
|
/// a server and call a tool while every MCP HTTP request goes through a real
|
|
/// exec-server process instead of a direct reqwest transport.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn streamable_http_remote_client_round_trips_through_exec_server() -> anyhow::Result<()> {
|
|
// Phase 1: start the MCP Streamable HTTP test server and a local
|
|
// exec-server process that will own the HTTP network calls.
|
|
let (_server, base_url) = spawn_streamable_http_server().await?;
|
|
let exec_server = spawn_exec_server().await?;
|
|
|
|
// Phase 2: create and initialize the RMCP client using the executor-backed
|
|
// Streamable HTTP transport.
|
|
let client = create_remote_client(&base_url, exec_server.client.clone()).await?;
|
|
|
|
// Phase 3: prove the initialized client can complete a tool call and
|
|
// preserve the normal RMCP response shape.
|
|
let result = call_echo_tool(&client, "remote").await?;
|
|
assert_eq!(result, expected_echo_result("remote"));
|
|
|
|
Ok(())
|
|
}
|
|
|
|
/// A timed-out MCP handshake must release the remote executor for later requests.
|
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
|
async fn streamable_http_handshake_timeout_unblocks_remote_executor() -> anyhow::Result<()> {
|
|
let http_server = MockServer::start().await;
|
|
Mock::given(method("POST"))
|
|
.respond_with(ResponseTemplate::new(200).set_delay(Duration::from_secs(10)))
|
|
.mount(&http_server)
|
|
.await;
|
|
let exec_server = spawn_exec_server().await?;
|
|
let client = RmcpClient::new_streamable_http_client(
|
|
"stalled-remote-http",
|
|
&format!("{}/mcp", http_server.uri()),
|
|
Some("test-bearer".to_string()),
|
|
/*http_headers*/ None,
|
|
/*env_http_headers*/ None,
|
|
OAuthCredentialsStoreMode::File,
|
|
AuthKeyringBackendKind::default(),
|
|
Arc::new(exec_server.client.clone()),
|
|
/*auth_provider*/ None,
|
|
)
|
|
.await?;
|
|
let params = InitializeRequestParams::new(
|
|
ClientCapabilities::default(),
|
|
Implementation::new("codex-test", "0.0.0-test"),
|
|
)
|
|
.with_protocol_version(ProtocolVersion::V_2025_06_18);
|
|
let error = client
|
|
.initialize(
|
|
params,
|
|
Some(Duration::from_millis(500)),
|
|
Box::new(|_, _| {
|
|
async {
|
|
Ok(ElicitationResponse {
|
|
action: ElicitationAction::Accept,
|
|
content: None,
|
|
meta: None,
|
|
})
|
|
}
|
|
.boxed()
|
|
}),
|
|
)
|
|
.await
|
|
.expect_err("the stalled MCP handshake must time out");
|
|
assert!(
|
|
error.to_string().contains("timed out"),
|
|
"unexpected MCP handshake error: {error}"
|
|
);
|
|
|
|
tokio::time::timeout(
|
|
Duration::from_secs(2),
|
|
exec_server.client.environment_info(),
|
|
)
|
|
.await
|
|
.expect("a timed-out handshake must not leave the serial executor blocked")?;
|
|
Ok(())
|
|
}
|