Decouple exec-server HTTP from reqwest types (#35059)

## What changed

- Rename `ReqwestHttpClient` to `RouteAwareHttpClient` to reflect that delegated HTTP uses Codex's shared route-aware transport.
- Use `codex_http_client` response and error types plus transport-neutral `http` and `url` types, removing the exec server's direct `reqwest` dependency.

## Testing

- Cover fragment stripping and Unicode hostname normalization for delegated HTTP requests.

GitOrigin-RevId: 8b0fc60a76004feb57198bfb4afb1371c9ceb1bd
This commit is contained in:
Celia Chen
2026-07-24 00:53:05 +00:00
committed by copyberry
parent 94ebae725e
commit 09241ae4db
18 changed files with 175 additions and 64 deletions
+1 -1
View File
@@ -31,7 +31,6 @@ codex-utils-rustls-provider = { workspace = true }
codex-websocket-client = { workspace = true }
futures = { workspace = true }
http = { workspace = true }
reqwest = { workspace = true, features = ["json", "rustls-tls", "stream"] }
prost = "0.14.3"
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
@@ -51,6 +50,7 @@ tokio = { workspace = true, features = [
tokio-util = { workspace = true, features = ["io", "rt"] }
tokio-tungstenite = { workspace = true }
tracing = { workspace = true }
url = { workspace = true }
uuid = { workspace = true, features = ["v4"] }
[target.'cfg(unix)'.dependencies]
@@ -2,7 +2,7 @@
//!
//! This module is the facade for the environment-owned [`crate::HttpClient`]
//! capability:
//! - [`ReqwestHttpClient`] executes requests directly with `reqwest`
//! - [`RouteAwareHttpClient`] executes requests through the shared transport
//! - [`ExecServerClient`] forwards requests over the JSON-RPC transport
//! - [`HttpResponseBodyStream`] presents buffered local bodies and streamed
//! remote `http/request/bodyDelta` notifications through one byte-stream API
@@ -13,14 +13,14 @@
//! - remote runtime: serves the `http/request` RPC and runs the concrete local
//! HTTP request there when the orchestrator uses [`ExecServerClient`]
#[path = "reqwest_http_client.rs"]
mod reqwest_http_client;
#[path = "http_response_body_stream.rs"]
pub(crate) mod response_body_stream;
#[path = "route_aware_http_client.rs"]
mod route_aware_http_client;
#[path = "rpc_http_client.rs"]
mod rpc_http_client;
pub(crate) use reqwest_http_client::PendingReqwestHttpBodyStream;
pub use reqwest_http_client::ReqwestHttpClient;
pub(crate) use reqwest_http_client::ReqwestHttpRequestRunner;
pub use response_body_stream::HttpResponseBodyStream;
pub(crate) use route_aware_http_client::PendingRouteAwareHttpBodyStream;
pub use route_aware_http_client::RouteAwareHttpClient;
pub(crate) use route_aware_http_client::RouteAwareHttpRequestRunner;
@@ -10,8 +10,9 @@ use std::sync::Arc;
use std::sync::atomic::Ordering;
use bytes::Bytes;
use codex_http_client::HttpError;
use codex_http_client::HttpResponse;
use futures::StreamExt;
use reqwest::Response;
use serde_json::Value;
use serde_json::from_value;
use tokio::runtime::Handle;
@@ -55,7 +56,7 @@ pub(super) struct HttpBodyStreamRegistration {
enum HttpResponseBodyStreamInner {
Local {
body: Pin<Box<dyn futures::Stream<Item = Result<Bytes, reqwest::Error>> + Send>>,
body: Pin<Box<dyn futures::Stream<Item = Result<Bytes, HttpError>> + Send>>,
},
Remote {
inner: Arc<Inner>,
@@ -77,7 +78,7 @@ pub struct HttpResponseBodyStream {
}
impl HttpResponseBodyStream {
pub(super) fn local(response: Response) -> Self {
pub(super) fn local(response: HttpResponse) -> Self {
Self {
inner: HttpResponseBodyStreamInner::Local {
body: Box::pin(response.bytes_stream()),
@@ -1,4 +1,4 @@
//! `reqwest`-backed `HttpClient` implementation.
//! Route-aware local HTTP capability implementation.
//!
//! This code runs wherever the real network request should originate:
//! - in a local environment, that means the orchestrator process
@@ -15,12 +15,12 @@ use codex_http_client::RouteAwareRequestError;
use futures::FutureExt;
use futures::StreamExt;
use futures::future::BoxFuture;
use reqwest::Method;
use reqwest::Url;
use reqwest::header::HeaderMap;
use reqwest::header::HeaderName;
use reqwest::header::HeaderValue;
use http::HeaderMap;
use http::HeaderName;
use http::HeaderValue;
use http::Method;
use tracing::Instrument;
use url::Url;
use super::HttpResponseBodyStream;
use super::response_body_stream::send_body_delta;
@@ -38,25 +38,25 @@ use crate::rpc::invalid_params;
/// HTTP capability implementation backed by the shared route-aware transport.
#[derive(Clone)]
pub struct ReqwestHttpClient {
pub struct RouteAwareHttpClient {
follow_redirects: RouteAwareClientPool,
stop_redirects: RouteAwareClientPool,
}
/// Streaming response state held between the initial HTTP response and
/// downstream body-delta forwarding.
pub(crate) struct PendingReqwestHttpBodyStream {
pub(crate) struct PendingRouteAwareHttpBodyStream {
pub(crate) request_id: String,
pub(crate) response: reqwest::Response,
pub(crate) response: codex_http_client::HttpResponse,
}
/// Validates `http/request` parameters and runs the actual `reqwest` call used
/// Validates `http/request` parameters and runs the actual HTTP call used
/// by the exec-server route and the local [`HttpClient`] backend.
pub(crate) struct ReqwestHttpRequestRunner {
pub(crate) struct RouteAwareHttpRequestRunner {
client: RouteAwareClientPool,
}
impl ReqwestHttpClient {
impl RouteAwareHttpClient {
pub fn new(http_client_factory: HttpClientFactory) -> Self {
Self {
follow_redirects: RouteAwareClientPool::with_chatgpt_cloudflare_cookies_without_request_logging(
@@ -73,16 +73,19 @@ impl ReqwestHttpClient {
}
}
pub(crate) fn runner(&self, redirect_policy: HttpRedirectPolicy) -> ReqwestHttpRequestRunner {
pub(crate) fn runner(
&self,
redirect_policy: HttpRedirectPolicy,
) -> RouteAwareHttpRequestRunner {
let client = match redirect_policy {
HttpRedirectPolicy::Follow => self.follow_redirects.clone(),
HttpRedirectPolicy::Stop => self.stop_redirects.clone(),
};
ReqwestHttpRequestRunner { client }
RouteAwareHttpRequestRunner { client }
}
}
impl HttpClient for ReqwestHttpClient {
impl HttpClient for RouteAwareHttpClient {
fn http_request(
&self,
params: HttpRequestParams,
@@ -128,11 +131,11 @@ impl HttpClient for ReqwestHttpClient {
}
}
impl ReqwestHttpRequestRunner {
impl RouteAwareHttpRequestRunner {
pub(crate) async fn run(
&self,
params: HttpRequestParams,
) -> Result<(HttpRequestResponse, Option<PendingReqwestHttpBodyStream>), JSONRPCErrorError>
) -> Result<(HttpRequestResponse, Option<PendingRouteAwareHttpBodyStream>), JSONRPCErrorError>
{
let method = Method::from_bytes(params.method.as_bytes())
.map_err(|error| invalid_params(format!("http/request method is invalid: {error}")))?;
@@ -188,7 +191,7 @@ impl ReqwestHttpRequestRunner {
headers,
body: Vec::new().into(),
},
Some(PendingReqwestHttpBodyStream {
Some(PendingRouteAwareHttpBodyStream {
request_id: params.request_id,
response,
}),
@@ -212,10 +215,10 @@ impl ReqwestHttpRequestRunner {
}
pub(crate) async fn stream_body(
pending_stream: PendingReqwestHttpBodyStream,
pending_stream: PendingRouteAwareHttpBodyStream,
notifications: RpcNotificationSender,
) {
let PendingReqwestHttpBodyStream {
let PendingRouteAwareHttpBodyStream {
request_id,
response,
} = pending_stream;
@@ -6,7 +6,7 @@ use super::*;
use crate::protocol::ExecOutputStream;
use crate::protocol::ProcessOutputChunk;
fn registry_error(status: reqwest::StatusCode, code: Option<&str>) -> ExecServerError {
fn registry_error(status: http::StatusCode, code: Option<&str>) -> ExecServerError {
ExecServerError::EnvironmentRegistryHttp {
status,
code: code.map(str::to_string),
@@ -37,7 +37,7 @@ fn registry_recovery_retry_delay_exponentially_backs_off_and_caps() {
#[test]
fn recovery_retries_transient_registry_errors() {
let error = registry_error(reqwest::StatusCode::TOO_MANY_REQUESTS, /*code*/ None);
let error = registry_error(http::StatusCode::TOO_MANY_REQUESTS, /*code*/ None);
assert!(is_retryable_registry_error(&error));
assert!(is_retryable_recovery_error(&error));
@@ -45,7 +45,7 @@ fn recovery_retries_transient_registry_errors() {
#[test]
fn recovery_retries_environment_offline_conflicts() {
let error = registry_error(reqwest::StatusCode::CONFLICT, Some("environment_offline"));
let error = registry_error(http::StatusCode::CONFLICT, Some("environment_offline"));
assert!(is_retryable_registry_error(&error));
assert!(is_retryable_recovery_error(&error));
@@ -53,7 +53,7 @@ fn recovery_retries_environment_offline_conflicts() {
#[test]
fn recovery_does_not_retry_other_registry_conflicts() {
let error = registry_error(reqwest::StatusCode::CONFLICT, Some("registration_conflict"));
let error = registry_error(http::StatusCode::CONFLICT, Some("registration_conflict"));
assert!(!is_retryable_registry_error(&error));
assert!(!is_retryable_recovery_error(&error));
+3 -3
View File
@@ -20,7 +20,7 @@ use crate::HttpClient;
use crate::NoiseChannelIdentity;
use crate::NoiseRendezvousConnectProvider;
use crate::client::LazyRemoteExecServerClient;
use crate::client::http_client::ReqwestHttpClient;
use crate::client::http_client::RouteAwareHttpClient;
use crate::client_api::DEFAULT_REMOTE_EXEC_SERVER_CONNECT_TIMEOUT;
use crate::client_api::ExecServerTransportParams;
use crate::environment_bootstrap::PreparedEnvironmentManager;
@@ -632,7 +632,7 @@ impl Environment {
exec_backend: Arc::new(LocalProcess::default()),
filesystem: Arc::new(LocalFileSystem::unsandboxed()),
// Test-only construction has no application config from which to resolve proxy policy.
http_client: Arc::new(ReqwestHttpClient::new(HttpClientFactory::new(
http_client: Arc::new(RouteAwareHttpClient::new(HttpClientFactory::new(
OutboundProxyPolicy::ReqwestDefault,
))),
local_runtime_paths: None,
@@ -713,7 +713,7 @@ impl Environment {
filesystem: Arc::new(LocalFileSystem::with_runtime_paths(
local_runtime_paths.clone(),
)),
http_client: Arc::new(ReqwestHttpClient::new(http_client_factory)),
http_client: Arc::new(RouteAwareHttpClient::new(http_client_factory)),
local_runtime_paths: Some(local_runtime_paths),
}
}
+1 -1
View File
@@ -47,7 +47,7 @@ pub use capability_discovery_cache::ExecutorCapabilityDiscoveryCache;
pub use client::ExecServerClient;
pub use client::ExecServerError;
pub use client::http_client::HttpResponseBodyStream;
pub use client::http_client::ReqwestHttpClient;
pub use client::http_client::RouteAwareHttpClient;
pub use client_api::ExecServerClientConnectOptions;
pub use client_api::HttpClient;
pub use client_api::NoiseRendezvousConnectArgs;
+1 -1
View File
@@ -12,7 +12,7 @@ use futures::FutureExt;
use http::HeaderMap;
use http::HeaderName;
use http::HeaderValue;
use reqwest::StatusCode;
use http::StatusCode;
use serde::Deserialize;
use tokio::time::sleep;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
+7 -7
View File
@@ -13,9 +13,9 @@ use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;
use crate::ExecServerRuntimePaths;
use crate::client::http_client::PendingReqwestHttpBodyStream;
use crate::client::http_client::ReqwestHttpClient;
use crate::client::http_client::ReqwestHttpRequestRunner;
use crate::client::http_client::PendingRouteAwareHttpBodyStream;
use crate::client::http_client::RouteAwareHttpClient;
use crate::client::http_client::RouteAwareHttpRequestRunner;
use crate::protocol::CapabilityRootsDiscoverParams;
use crate::protocol::CapabilityRootsDiscoverResponse;
use crate::protocol::EnvironmentInfo;
@@ -75,7 +75,7 @@ pub(crate) struct ExecServerHandler {
background_tasks: TaskTracker,
file_system: FileSystemHandler,
runtime_paths: ExecServerRuntimePaths,
http_client: ReqwestHttpClient,
http_client: RouteAwareHttpClient,
initialize_requested: AtomicBool,
initialized: AtomicBool,
}
@@ -96,7 +96,7 @@ impl ExecServerHandler {
background_tasks: TaskTracker::new(),
file_system: FileSystemHandler::new(runtime_paths.clone()),
runtime_paths,
http_client: ReqwestHttpClient::new(http_client_factory),
http_client: RouteAwareHttpClient::new(http_client_factory),
initialize_requested: AtomicBool::new(false),
initialized: AtomicBool::new(false),
}
@@ -405,7 +405,7 @@ impl ExecServerHandler {
async fn start_http_body_stream(
self: &Arc<Self>,
pending_stream: PendingReqwestHttpBodyStream,
pending_stream: PendingRouteAwareHttpBodyStream,
) {
let request_id = pending_stream.request_id.clone();
if self.background_task_shutdown.is_cancelled() {
@@ -419,7 +419,7 @@ impl ExecServerHandler {
self.background_tasks.spawn(async move {
tokio::select! {
_ = shutdown.cancelled() => {}
_ = ReqwestHttpRequestRunner::stream_body(pending_stream, notifications) => {}
_ = RouteAwareHttpRequestRunner::stream_body(pending_stream, notifications) => {}
}
handler.release_http_body_stream(&finished_request_id).await;
});
+2 -2
View File
@@ -1,5 +1,5 @@
use reqwest::header::HeaderMap;
use reqwest::header::HeaderValue;
use http::HeaderMap;
use http::HeaderValue;
pub(crate) fn current_trace_context_headers() -> HeaderMap {
let mut headers = HeaderMap::new();
@@ -65,7 +65,7 @@ struct TlsInterceptingProxy {
}
/// Exercises the same `http/request` route used by remotely executed Streamable HTTP MCP calls.
/// Each RPC builds a fresh reqwest client. The first response sets `__cflb`, and the second response
/// Each RPC uses the shared route-aware client. The first response sets `__cflb`, and the second response
/// replaces it, proving cross-client persistence through the shared cookie store.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_replays_only_chatgpt_cloudflare_cookies() -> anyhow::Result<()> {
+6 -2
View File
@@ -20,8 +20,12 @@ async fn exec_server_serves_readyz_alongside_websocket_endpoint() -> anyhow::Res
.strip_prefix("ws://")
.expect("websocket URL should use ws://");
let response = reqwest::get(format!("http://{http_base_url}/readyz")).await?;
assert_eq!(response.status(), reqwest::StatusCode::OK);
let client = codex_http_client::HttpClientBuilder::new().build_direct()?;
let response = client
.get(format!("http://{http_base_url}/readyz"))
.send()
.await?;
assert_eq!(response.status(), http::StatusCode::OK);
server.shutdown().await?;
Ok(())
+105 -1
View File
@@ -113,6 +113,48 @@ async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()>
Ok(())
}
/// What this tests: delegated HTTP accepts URLs with client-side fragments and
/// does not include those fragments in the request sent over the network.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_http_request_omits_url_fragment() -> anyhow::Result<()> {
let mut server = exec_server().await?;
initialize_exec_server(&mut server).await?;
let listener = TcpListener::bind("127.0.0.1:0").await?;
let url = format!(
"http://{}/mcp?case=fragment#client-section",
listener.local_addr()?
);
let http_request_id = server
.send_request(
"http/request",
serde_json::to_value(HttpRequestParams {
method: "GET".to_string(),
url,
headers: Vec::new(),
body: None,
timeout_ms: Some(5_000),
redirect_policy: HttpRedirectPolicy::Follow,
request_id: "fragment-request".to_string(),
stream_response: false,
})?,
)
.await?;
let captured = accept_http_request(&listener).await?;
assert_eq!(captured.request_line, "GET /mcp?case=fragment HTTP/1.1");
respond_with_status_and_headers(captured.stream, "200 OK", &[], b"fragment-response").await?;
let response: HttpRequestResponse = wait_for_response(&mut server, http_request_id).await?;
assert_eq!(
(response.status, response.body.into_inner()),
(200, b"fragment-response".to_vec())
);
server.shutdown().await?;
Ok(())
}
/// What this tests: a configured system-proxy factory survives the complete
/// executor transport, processor, and handler chain for delegated HTTP.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -168,6 +210,68 @@ async fn exec_server_http_request_uses_configured_system_proxy() -> anyhow::Resu
Ok(())
}
/// What this tests: delegated HTTP preserves WHATWG IDNA normalization before
/// selecting a configured system-proxy route.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_http_request_normalizes_unicode_hostname() -> anyhow::Result<()> {
let proxy_listener = TcpListener::bind("127.0.0.1:0").await?;
let proxy_url = format!("http://{}", proxy_listener.local_addr()?);
let request_url = "http://münich.invalid/mcp?route=unicode";
let normalized_url = "http://xn--mnich-kva.invalid/mcp?route=unicode";
let mut server = exec_server_with_env([
(SYSTEM_PROXY_REQUEST_URL_ENV, normalized_url),
(SYSTEM_PROXY_URL_ENV, proxy_url.as_str()),
("HTTP_PROXY", ""),
("http_proxy", ""),
("HTTPS_PROXY", ""),
("https_proxy", ""),
("ALL_PROXY", ""),
("all_proxy", ""),
("NO_PROXY", ""),
("no_proxy", ""),
])
.await?;
initialize_exec_server(&mut server).await?;
let http_request_id = server
.send_request(
"http/request",
serde_json::to_value(HttpRequestParams {
method: "GET".to_string(),
url: request_url.to_string(),
headers: Vec::new(),
body: None,
timeout_ms: Some(5_000),
redirect_policy: HttpRedirectPolicy::Follow,
request_id: "unicode-hostname-request".to_string(),
stream_response: false,
})?,
)
.await?;
let captured = accept_http_request(&proxy_listener).await?;
assert_eq!(
(
captured.request_line.as_str(),
captured.headers.get("host").map(String::as_str),
),
(
"GET http://xn--mnich-kva.invalid/mcp?route=unicode HTTP/1.1",
Some("xn--mnich-kva.invalid"),
)
);
respond_with_status_and_headers(captured.stream, "200 OK", &[], b"unicode-response").await?;
let response: HttpRequestResponse = wait_for_response(&mut server, http_request_id).await?;
assert_eq!(
(response.status, response.body.into_inner()),
(200, b"unicode-response".to_vec())
);
server.shutdown().await?;
Ok(())
}
/// What this tests: OAuth callers can inspect redirect responses without the
/// executor following the Location header.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -662,7 +766,7 @@ fn is_expected_peer_disconnect(err: &anyhow::Error) -> bool {
})
}
/// Writes a chunked HTTP response so reqwest must drive the streaming path.
/// Writes a chunked HTTP response so the shared client must drive the streaming path.
async fn respond_with_chunked_body(
mut stream: TcpStream,
headers: &[(&str, &str)],
@@ -5,7 +5,7 @@ use std::sync::Mutex;
use codex_exec_server::HttpClient;
use codex_exec_server::HttpRedirectPolicy;
use codex_exec_server::HttpRequestParams;
use codex_exec_server::ReqwestHttpClient;
use codex_exec_server::RouteAwareHttpClient;
use codex_http_client::HttpClientFactory;
use codex_http_client::OutboundProxyPolicy;
use pretty_assertions::assert_eq;
@@ -34,7 +34,7 @@ async fn delegated_http_success_logs_do_not_expose_sensitive_request_or_response
let _guard = tracing::subscriber::set_default(subscriber);
tracing::debug!(target: "codex_exec_server", "log capture sentinel");
let client =
ReqwestHttpClient::new(HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault));
RouteAwareHttpClient::new(HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault));
for (redirect_policy, status, query_secret, cookie_secret, location_secret) in [
(
@@ -128,7 +128,7 @@ async fn delegated_http_failure_warning_redacts_request_url() -> anyhow::Result<
let unavailable_address = unavailable_server.local_addr()?;
drop(unavailable_server);
let client =
ReqwestHttpClient::new(HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault));
RouteAwareHttpClient::new(HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault));
let error = client
.http_request(HttpRequestParams {