agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[2/4] Implement executor HTTP request runner (#18582)

Browse Source 
### Why
Remote streamable HTTP MCP needs the executor to perform ordinary HTTP
requests on the executor side. This keeps network placement aligned with
`experimental_environment = "remote"` without adding MCP-specific
executor APIs.

### What
- Add an executor-side `http/request` runner backed by `reqwest`.
- Validate request method and URL scheme, preserving the transport
boundary at plain HTTP.
- Return buffered responses for ordinary calls and emit ordered
`http/request/bodyDelta` notifications for streaming responses.
- Register the request handler in the exec-server router.
- Document the runner entrypoint, conversion helpers, body-stream
bridge, notification sender, timeout behavior, and new integration-test
helpers.
- Add exec-server integration tests with the existing websocket harness
and a local TCP HTTP peer for buffered and streamed responses, with
comments spelling out what each test proves and its
setup/exercise/assert phases.

### Stack
1. #18581 protocol
2. #18582 runner
3. #18583 RMCP client
4. #18584 manager wiring and local/remote coverage

### Verification
- `just fmt`
- `cargo check -p codex-exec-server -p codex-rmcp-client --tests`
- `cargo check -p codex-core --test all` compile-only
- `git diff --check`
- Online full CI is running from the `full-ci` branch, including the
remote Rust test job.

Co-authored-by: Codex <noreply@openai.com>

---------

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Ahmed Ibrahim
2026-04-22 13:36:34 -07:00
committed by GitHub
Unverified
parent 18a26d7bbc
commit 9360f267f3
10 changed files with 1169 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -2589,6 +2589,7 @@ dependencies = [
"ctor 0.6.3",
"futures",
"pretty_assertions",
"reqwest",
"serde",
"serde_json",
"serial_test",
@@ -2597,6 +2598,7 @@ dependencies = [
"thiserror 2.0.18",
"tokio",
"tokio-tungstenite",
"tokio-util",
"tracing",
"uuid",
]
codex-rs/exec-server/Cargo.toml
+2
View File
@@ -21,6 +21,7 @@ codex-sandboxing = { workspace = true }
codex-utils-absolute-path = { workspace = true }
codex-utils-pty = { workspace = true }
futures = { workspace = true }
reqwest = { workspace = true, features = ["rustls-tls", "stream"] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
thiserror = { workspace = true }
@@ -35,6 +36,7 @@ tokio = { workspace = true, features = [
"sync",
"time",
] }
tokio-util = { workspace = true, features = ["rt"] }
tokio-tungstenite = { workspace = true }
tracing = { workspace = true }
uuid = { workspace = true, features = ["v4"] }
codex-rs/exec-server/src/client/http_client.rs
+271 -71
View File
@@ -1,7 +1,15 @@
use std::collections::HashMap;
use std::sync::Arc;
use std::sync::atomic::Ordering;
use std::time::Duration;
use codex_app_server_protocol::JSONRPCErrorError;
use futures::StreamExt;
use reqwest::Method;
use reqwest::Url;
use reqwest::header::HeaderMap;
use reqwest::header::HeaderName;
use reqwest::header::HeaderValue;
use serde_json::Value;
use serde_json::from_value;
use tokio::runtime::Handle;
@@ -12,14 +20,28 @@ use tracing::debug;
use super::ExecServerClient;
use super::ExecServerError;
use super::Inner;
use crate::protocol::HTTP_REQUEST_BODY_DELTA_METHOD;
use crate::protocol::HTTP_REQUEST_METHOD;
use crate::protocol::HttpHeader;
use crate::protocol::HttpRequestBodyDeltaNotification;
use crate::protocol::HttpRequestParams;
use crate::protocol::HttpRequestResponse;
use crate::rpc::RpcNotificationSender;
use crate::rpc::internal_error;
use crate::rpc::invalid_params;
/// Maximum queued body frames per streamed executor HTTP response.
const HTTP_BODY_DELTA_CHANNEL_CAPACITY: usize = 256;
pub(crate) struct ExecutorPendingHttpBodyStream {
pub(crate) request_id: String,
response: reqwest::Response,
}
pub(crate) struct ExecutorHttpRequestRunner {
client: reqwest::Client,
}
/// Request-scoped stream of body chunks for an executor HTTP response.
///
/// The initial `http/request` call returns status and headers. This stream then
@@ -35,6 +57,60 @@ pub struct HttpResponseBodyStream {
closed: bool,
}
impl ExecServerClient {
/// Performs an executor-side HTTP request and buffers the response body.
pub async fn http_request(
&self,
mut params: HttpRequestParams,
) -> Result<HttpRequestResponse, ExecServerError> {
params.stream_response = false;
self.call(HTTP_REQUEST_METHOD, &params).await
}
/// Performs an executor-side HTTP request and returns a body stream.
///
/// The method sets `stream_response` and replaces any caller-supplied
/// `request_id` with a connection-local id, so late deltas from abandoned
/// streams cannot be confused with later requests.
pub async fn http_request_stream(
&self,
mut params: HttpRequestParams,
) -> Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError> {
params.stream_response = true;
let request_id = self.inner.next_http_body_stream_request_id();
params.request_id = request_id.clone();
let (tx, rx) = mpsc::channel(HTTP_BODY_DELTA_CHANNEL_CAPACITY);
self.inner
.insert_http_body_stream(request_id.clone(), tx)
.await?;
let mut registration = HttpBodyStreamRegistration {
inner: Arc::clone(&self.inner),
request_id: request_id.clone(),
active: true,
};
let response = match self.call(HTTP_REQUEST_METHOD, &params).await {
Ok(response) => response,
Err(error) => {
self.inner.remove_http_body_stream(&request_id).await;
registration.active = false;
return Err(error);
}
};
registration.active = false;
Ok((
response,
HttpResponseBodyStream {
inner: Arc::clone(&self.inner),
request_id,
next_seq: 1,
rx,
pending_eof: false,
closed: false,
},
))
}
}
impl HttpResponseBodyStream {
/// Receives the next response-body chunk.
///
@@ -109,75 +185,165 @@ impl Drop for HttpResponseBodyStream {
}
}
/// Active route registration owned while `http_request_stream` awaits headers.
struct HttpBodyStreamRegistration {
inner: Arc<Inner>,
request_id: String,
active: bool,
}
impl Drop for HttpBodyStreamRegistration {
/// Removes the route if the stream request future is cancelled before headers return.
fn drop(&mut self) {
if self.active {
spawn_remove_http_body_stream(Arc::clone(&self.inner), self.request_id.clone());
}
}
}
impl ExecServerClient {
/// Performs an executor-side HTTP request and buffers the response body.
pub async fn http_request(
&self,
mut params: HttpRequestParams,
) -> Result<HttpRequestResponse, ExecServerError> {
params.stream_response = false;
params.request_id = None;
self.call(HTTP_REQUEST_METHOD, &params).await
}
/// Performs an executor-side HTTP request and returns a body stream.
///
/// The method sets `stream_response` and replaces any caller-supplied
/// `request_id` with a connection-local id, so late deltas from abandoned
/// streams cannot be confused with later requests.
pub async fn http_request_stream(
&self,
mut params: HttpRequestParams,
) -> Result<(HttpRequestResponse, HttpResponseBodyStream), ExecServerError> {
params.stream_response = true;
let request_id = self.inner.next_http_body_stream_request_id();
params.request_id = Some(request_id.clone());
let (tx, rx) = mpsc::channel(HTTP_BODY_DELTA_CHANNEL_CAPACITY);
self.inner
.insert_http_body_stream(request_id.clone(), tx)
.await?;
let mut registration = HttpBodyStreamRegistration {
inner: Arc::clone(&self.inner),
request_id: request_id.clone(),
active: true,
};
let response = match self.call(HTTP_REQUEST_METHOD, &params).await {
Ok(response) => response,
Err(error) => {
self.inner.remove_http_body_stream(&request_id).await;
registration.active = false;
return Err(error);
impl ExecutorHttpRequestRunner {
pub(crate) fn new(timeout_ms: Option<u64>) -> Result<Self, JSONRPCErrorError> {
let client = match timeout_ms {
None => reqwest::Client::builder(),
Some(timeout_ms) => {
reqwest::Client::builder().timeout(Duration::from_millis(timeout_ms))
}
};
registration.active = false;
}
.build()
.map_err(|err| internal_error(format!("failed to build http/request client: {err}")))?;
Ok(Self { client })
}
pub(crate) async fn run(
&self,
params: HttpRequestParams,
) -> Result<(HttpRequestResponse, Option<ExecutorPendingHttpBodyStream>), JSONRPCErrorError>
{
let method = Method::from_bytes(params.method.as_bytes())
.map_err(|err| invalid_params(format!("http/request method is invalid: {err}")))?;
let url = Url::parse(&params.url)
.map_err(|err| invalid_params(format!("http/request url is invalid: {err}")))?;
match url.scheme() {
"http" | "https" => {}
scheme => {
return Err(invalid_params(format!(
"http/request only supports http and https URLs, got {scheme}"
)));
}
}
let headers = Self::build_headers(params.headers)?;
let mut request = self.client.request(method, url).headers(headers);
if let Some(body) = params.body {
request = request.body(body.into_inner());
}
let response = request
.send()
.await
.map_err(|err| internal_error(format!("http/request failed: {err}")))?;
let status = response.status().as_u16();
let headers = Self::response_headers(response.headers());
if params.stream_response {
return Ok((
HttpRequestResponse {
status,
headers,
body: Vec::new().into(),
},
Some(ExecutorPendingHttpBodyStream {
request_id: params.request_id,
response,
}),
));
}
let body = response.bytes().await.map_err(|err| {
internal_error(format!("failed to read http/request response body: {err}"))
})?;
Ok((
response,
HttpResponseBodyStream {
inner: Arc::clone(&self.inner),
request_id,
next_seq: 1,
rx,
pending_eof: false,
closed: false,
HttpRequestResponse {
status,
headers,
body: body.to_vec().into(),
},
None,
))
}
fn build_headers(headers: Vec<HttpHeader>) -> Result<HeaderMap, JSONRPCErrorError> {
let mut header_map = HeaderMap::new();
for header in headers {
let name = HeaderName::from_bytes(header.name.as_bytes()).map_err(|err| {
invalid_params(format!("http/request header name is invalid: {err}"))
})?;
let value = HeaderValue::from_str(&header.value).map_err(|err| {
invalid_params(format!(
"http/request header value is invalid for {}: {err}",
header.name
))
})?;
header_map.append(name, value);
}
Ok(header_map)
}
fn response_headers(headers: &HeaderMap) -> Vec<HttpHeader> {
headers
.iter()
.filter_map(|(name, value)| {
Some(HttpHeader {
name: name.as_str().to_string(),
value: value.to_str().ok()?.to_string(),
})
})
.collect()
}
pub(crate) async fn stream_body(
pending_stream: ExecutorPendingHttpBodyStream,
notifications: RpcNotificationSender,
) {
let ExecutorPendingHttpBodyStream {
request_id,
response,
} = pending_stream;
let mut seq = 1;
let mut body = response.bytes_stream();
while let Some(chunk) = body.next().await {
match chunk {
Ok(bytes) => {
if !send_executor_body_delta(
&notifications,
HttpRequestBodyDeltaNotification {
request_id: request_id.clone(),
seq,
delta: bytes.to_vec().into(),
done: false,
error: None,
},
)
.await
{
return;
}
seq += 1;
}
Err(err) => {
let _ = send_executor_body_delta(
&notifications,
HttpRequestBodyDeltaNotification {
request_id,
seq,
delta: Vec::new().into(),
done: true,
error: Some(err.to_string()),
},
)
.await;
return;
}
}
}
let _ = send_executor_body_delta(
&notifications,
HttpRequestBodyDeltaNotification {
request_id,
seq,
delta: Vec::new().into(),
done: true,
error: None,
},
)
.await;
}
}
impl Inner {
@@ -231,13 +397,21 @@ impl Inner {
let streams = streams.as_ref().clone();
self.http_body_streams.store(Arc::new(HashMap::new()));
for (request_id, tx) in streams {
let _ = tx.try_send(HttpRequestBodyDeltaNotification {
request_id,
seq: 1,
delta: Vec::new().into(),
done: true,
error: Some(message.clone()),
});
if tx
.try_send(HttpRequestBodyDeltaNotification {
request_id: request_id.clone(),
seq: 1,
delta: Vec::new().into(),
done: true,
error: Some(message.clone()),
})
.is_err()
{
let mut next_failures = self.http_body_stream_failures.load().as_ref().clone();
next_failures.insert(request_id, message.clone());
self.http_body_stream_failures
.store(Arc::new(next_failures));
}
}
}
@@ -312,6 +486,22 @@ impl Inner {
}
}
/// Active route registration owned while `http_request_stream` awaits headers.
struct HttpBodyStreamRegistration {
inner: Arc<Inner>,
request_id: String,
active: bool,
}
impl Drop for HttpBodyStreamRegistration {
/// Removes the route if the stream request future is cancelled before headers return.
fn drop(&mut self) {
if self.active {
spawn_remove_http_body_stream(Arc::clone(&self.inner), self.request_id.clone());
}
}
}
/// Schedules HTTP body route removal from synchronous drop paths.
fn spawn_remove_http_body_stream(inner: Arc<Inner>, request_id: String) {
if let Ok(handle) = Handle::try_current() {
@@ -320,3 +510,13 @@ fn spawn_remove_http_body_stream(inner: Arc<Inner>, request_id: String) {
});
}
}
async fn send_executor_body_delta(
notifications: &RpcNotificationSender,
delta: HttpRequestBodyDeltaNotification,
) -> bool {
notifications
.notify(HTTP_REQUEST_BODY_DELTA_METHOD, &delta)
.await
.is_ok()
}
codex-rs/exec-server/src/protocol.rs
+55 -5
View File
@@ -286,15 +286,19 @@ pub struct HttpRequestParams {
/// Optional request body bytes.
#[serde(default, rename = "bodyBase64")]
pub body: Option<ByteChunk>,
/// Optional request timeout in milliseconds.
#[serde(default)]
/// Request timeout in milliseconds.
///
/// Omitted or `null` disables the timeout. A number applies that exact
/// millisecond deadline.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub timeout_ms: Option<u64>,
/// Caller-chosen stream id for `http/request/bodyDelta` notifications.
///
/// The id must remain unique on a connection until the terminal body delta
/// arrives, even if the caller stops reading the stream earlier.
#[serde(default)]
pub request_id: Option<String>,
/// arrives, even if the caller stops reading the stream earlier. Buffered
/// requests still send an id so callers can keep one consistent request
/// envelope shape.
pub request_id: String,
/// Return after response headers and stream the response body as deltas.
#[serde(default)]
pub stream_response: bool,
@@ -391,3 +395,49 @@ mod base64_bytes {
.map_err(serde::de::Error::custom)
}
}
#[cfg(test)]
mod tests {
use super::HttpRequestParams;
use pretty_assertions::assert_eq;
#[test]
fn http_request_timeout_treats_omitted_and_null_as_no_timeout() {
let omitted: HttpRequestParams = serde_json::from_value(serde_json::json!({
"method": "GET",
"url": "https://example.test",
"requestId": "req-omitted-timeout",
}))
.expect("omitted timeout should deserialize");
let null_timeout: HttpRequestParams = serde_json::from_value(serde_json::json!({
"method": "GET",
"url": "https://example.test",
"requestId": "req-null-timeout",
"timeoutMs": null,
}))
.expect("null timeout should deserialize");
let explicit_timeout: HttpRequestParams = serde_json::from_value(serde_json::json!({
"method": "GET",
"url": "https://example.test",
"requestId": "req-explicit-timeout",
"timeoutMs": 1234,
}))
.expect("numeric timeout should deserialize");
assert_eq!(
(omitted.request_id.as_str(), omitted.timeout_ms),
("req-omitted-timeout", None)
);
assert_eq!(
(null_timeout.request_id.as_str(), null_timeout.timeout_ms),
("req-null-timeout", None)
);
assert_eq!(
(
explicit_timeout.request_id.as_str(),
explicit_timeout.timeout_ms
),
("req-explicit-timeout", Some(1234))
);
}
}
codex-rs/exec-server/src/rpc.rs
+44 -4
View File
@@ -36,8 +36,9 @@ pub(crate) enum RpcCallError {
type PendingRequest = oneshot::Sender<Result<Value, RpcCallError>>;
type BoxFuture<T> = Pin<Box<dyn Future<Output = T> + Send + 'static>>;
type RequestRoute<S> =
Box<dyn Fn(Arc<S>, JSONRPCRequest) -> BoxFuture<RpcServerOutboundMessage> + Send + Sync>;
type RequestRoute<S> = Box<
dyn Fn(Arc<S>, JSONRPCRequest) -> BoxFuture<Option<RpcServerOutboundMessage>> + Send + Sync,
>;
type NotificationRoute<S> =
Box<dyn Fn(Arc<S>, JSONRPCNotification) -> BoxFuture<Result<(), String>> + Send + Sync>;
@@ -72,6 +73,17 @@ impl RpcNotificationSender {
Self { outgoing_tx }
}
pub(crate) async fn response(
&self,
request_id: RequestId,
result: Value,
) -> Result<(), JSONRPCErrorError> {
self.outgoing_tx
.send(RpcServerOutboundMessage::Response { request_id, result })
.await
.map_err(|_| internal_error("RPC connection closed while sending response".into()))
}
#[allow(dead_code)]
pub(crate) async fn notify<P: Serialize>(
&self,
@@ -131,10 +143,10 @@ where
let response = match response {
Ok(response) => response.await,
Err(error) => {
return RpcServerOutboundMessage::Error { request_id, error };
return Some(RpcServerOutboundMessage::Error { request_id, error });
}
};
match response {
Some(match response {
Ok(result) => match serde_json::to_value(result) {
Ok(result) => RpcServerOutboundMessage::Response { request_id, result },
Err(err) => RpcServerOutboundMessage::Error {
@@ -143,6 +155,34 @@ where
},
},
Err(error) => RpcServerOutboundMessage::Error { request_id, error },
})
})
}),
);
}
pub(crate) fn request_with_id<P, F, Fut>(&mut self, method: &'static str, handler: F)
where
P: DeserializeOwned + Send + 'static,
F: Fn(Arc<S>, RequestId, P) -> Fut + Send + Sync + 'static,
Fut: Future<Output = Result<(), JSONRPCErrorError>> + Send + 'static,
{
self.request_routes.insert(
method,
Box::new(move |state, request| {
let request_id = request.id;
let params = decode_request_params::<P>(request.params)
.map(|params| handler(state, request_id.clone(), params));
Box::pin(async move {
let response = match params {
Ok(response) => response.await,
Err(error) => {
return Some(RpcServerOutboundMessage::Error { request_id, error });
}
};
match response {
Ok(()) => None,
Err(error) => Some(RpcServerOutboundMessage::Error { request_id, error }),
}
})
}),
codex-rs/exec-server/src/server/handler.rs
+99
View File
@@ -4,8 +4,16 @@ use std::sync::atomic::AtomicBool;
use std::sync::atomic::Ordering;
use codex_app_server_protocol::JSONRPCErrorError;
use codex_app_server_protocol::RequestId;
use serde_json::to_value;
use std::collections::HashSet;
use tokio::sync::Mutex;
use tokio_util::sync::CancellationToken;
use tokio_util::task::TaskTracker;
use crate::ExecServerRuntimePaths;
use crate::client::http_client::ExecutorHttpRequestRunner;
use crate::client::http_client::ExecutorPendingHttpBodyStream;
use crate::protocol::ExecParams;
use crate::protocol::ExecResponse;
use crate::protocol::FsCopyParams;
@@ -22,6 +30,7 @@ use crate::protocol::FsRemoveParams;
use crate::protocol::FsRemoveResponse;
use crate::protocol::FsWriteFileParams;
use crate::protocol::FsWriteFileResponse;
use crate::protocol::HttpRequestParams;
use crate::protocol::InitializeParams;
use crate::protocol::InitializeResponse;
use crate::protocol::ReadParams;
@@ -31,6 +40,8 @@ use crate::protocol::TerminateResponse;
use crate::protocol::WriteParams;
use crate::protocol::WriteResponse;
use crate::rpc::RpcNotificationSender;
use crate::rpc::internal_error;
use crate::rpc::invalid_params;
use crate::rpc::invalid_request;
use crate::server::file_system_handler::FileSystemHandler;
use crate::server::session_registry::SessionHandle;
@@ -40,6 +51,9 @@ pub(crate) struct ExecServerHandler {
session_registry: Arc<SessionRegistry>,
notifications: RpcNotificationSender,
session: StdMutex<Option<SessionHandle>>,
active_body_stream_ids: Mutex<HashSet<String>>,
background_task_shutdown: CancellationToken,
background_tasks: TaskTracker,
file_system: FileSystemHandler,
initialize_requested: AtomicBool,
initialized: AtomicBool,
@@ -55,6 +69,9 @@ impl ExecServerHandler {
session_registry,
notifications,
session: StdMutex::new(None),
active_body_stream_ids: Mutex::new(HashSet::new()),
background_task_shutdown: CancellationToken::new(),
background_tasks: TaskTracker::new(),
file_system: FileSystemHandler::new(runtime_paths),
initialize_requested: AtomicBool::new(false),
initialized: AtomicBool::new(false),
@@ -62,6 +79,9 @@ impl ExecServerHandler {
}
pub(crate) async fn shutdown(&self) {
self.background_task_shutdown.cancel();
self.background_tasks.close();
self.background_tasks.wait().await;
if let Some(session) = self.session() {
session.detach().await;
}
@@ -147,6 +167,47 @@ impl ExecServerHandler {
session.process().terminate(params).await
}
pub(crate) async fn http_request(
self: &Arc<Self>,
request_id: RequestId,
params: HttpRequestParams,
) -> Result<(), JSONRPCErrorError> {
self.require_initialized_for("http")?;
let stream_response = params.stream_response;
let http_request_id = params.request_id.clone();
if stream_response {
self.reserve_http_body_stream(&http_request_id).await?;
}
let response = ExecutorHttpRequestRunner::new(params.timeout_ms)?
.run(params)
.await;
if response.is_err() && stream_response {
self.release_http_body_stream(&http_request_id).await;
}
let (response, mut pending_stream) = response?;
let result = match to_value(response) {
Ok(result) => result,
Err(err) => {
if let Some(pending_stream) = pending_stream.take() {
self.release_http_body_stream(&pending_stream.request_id)
.await;
}
return Err(internal_error(err.to_string()));
}
};
if let Err(error) = self.notifications.response(request_id, result).await {
if let Some(pending_stream) = pending_stream.take() {
self.release_http_body_stream(&pending_stream.request_id)
.await;
}
return Err(error);
}
if let Some(pending_stream) = pending_stream {
self.start_http_body_stream(pending_stream).await;
}
Ok(())
}
pub(crate) async fn fs_read_file(
&self,
params: FsReadFileParams,
@@ -242,6 +303,44 @@ impl ExecServerHandler {
.unwrap_or_else(std::sync::PoisonError::into_inner)
.clone()
}
async fn start_http_body_stream(
self: &Arc<Self>,
pending_stream: ExecutorPendingHttpBodyStream,
) {
let request_id = pending_stream.request_id.clone();
if self.background_task_shutdown.is_cancelled() {
self.release_http_body_stream(&request_id).await;
return;
}
let finished_request_id = request_id.clone();
let handler = Arc::clone(self);
let notifications = self.notifications.clone();
let shutdown = self.background_task_shutdown.clone();
self.background_tasks.spawn(async move {
tokio::select! {
_ = shutdown.cancelled() => {}
_ = ExecutorHttpRequestRunner::stream_body(pending_stream, notifications) => {}
}
handler.release_http_body_stream(&finished_request_id).await;
});
}
async fn release_http_body_stream(&self, request_id: &str) {
let mut active_body_stream_ids = self.active_body_stream_ids.lock().await;
active_body_stream_ids.remove(request_id);
}
async fn reserve_http_body_stream(&self, request_id: &str) -> Result<(), JSONRPCErrorError> {
let mut active_body_stream_ids = self.active_body_stream_ids.lock().await;
if active_body_stream_ids.contains(request_id) {
return Err(invalid_params(format!(
"http/request streamResponse requestId `{request_id}` is already active"
)));
}
active_body_stream_ids.insert(request_id.to_string());
Ok(())
}
}
#[cfg(test)]
codex-rs/exec-server/src/server/processor.rs
+3 -1
View File
@@ -103,7 +103,9 @@ async fn run_connection(
break;
}
};
if outgoing_tx.send(message).await.is_err() {
if let Some(message) = message
&& outgoing_tx.send(message).await.is_err()
{
break;
}
} else if outgoing_tx
codex-rs/exec-server/src/server/registry.rs
+8
View File
@@ -19,6 +19,8 @@ use crate::protocol::FsReadDirectoryParams;
use crate::protocol::FsReadFileParams;
use crate::protocol::FsRemoveParams;
use crate::protocol::FsWriteFileParams;
use crate::protocol::HTTP_REQUEST_METHOD;
use crate::protocol::HttpRequestParams;
use crate::protocol::INITIALIZE_METHOD;
use crate::protocol::INITIALIZED_METHOD;
use crate::protocol::InitializeParams;
@@ -42,6 +44,12 @@ pub(crate) fn build_router() -> RpcRouter<ExecServerHandler> {
handler.initialize(params).await
},
);
router.request_with_id(
HTTP_REQUEST_METHOD,
|handler: Arc<ExecServerHandler>, request_id, params: HttpRequestParams| async move {
handler.http_request(request_id, params).await
},
);
router.request(
EXEC_METHOD,
|handler: Arc<ExecServerHandler>, params: ExecParams| async move { handler.exec(params).await },
codex-rs/exec-server/tests/http_client.rs
+117 -24
View File
@@ -42,6 +42,7 @@ const HTTP_REQUEST_BODY_DELTA_METHOD: &str = "http/request/bodyDelta";
const INITIALIZE_METHOD: &str = "initialize";
const INITIALIZED_METHOD: &str = "initialized";
const TEST_TIMEOUT: Duration = Duration::from_secs(5);
const HTTP_BODY_DELTA_CHANNEL_CAPACITY: u64 = 256;
const OVERFLOWING_BODY_DELTA_FRAMES: u64 = 1_024;
/// What this tests: the buffered HTTP helper always sends a buffered
@@ -51,8 +52,8 @@ async fn http_request_forces_buffered_request_params() -> Result<()> {
// Phase 1: start a fake WebSocket exec-server so the test covers the
// public client connection path without depending on the HTTP runner.
let server = spawn_scripted_exec_server(|mut peer| async move {
// Phase 2: verify the buffered helper strips streaming-only fields
// before it sends the JSON-RPC call.
// Phase 2: verify the buffered helper forces buffered mode before it
// sends the JSON-RPC call.
let (request_id, params) = peer.read_http_request().await?;
assert_eq!(
params,
@@ -62,7 +63,7 @@ async fn http_request_forces_buffered_request_params() -> Result<()> {
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: None,
request_id: "ignored-stream-id".to_string(),
stream_response: false,
}
);
@@ -90,7 +91,7 @@ async fn http_request_forces_buffered_request_params() -> Result<()> {
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("ignored-stream-id".to_string()),
request_id: "ignored-stream-id".to_string(),
stream_response: true,
}),
)
@@ -130,7 +131,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta
}],
body: None,
timeout_ms: None,
request_id: Some("http-1".to_string()),
request_id: "http-1".to_string(),
stream_response: true,
}
);
@@ -185,7 +186,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("http-2".to_string()),
request_id: "http-2".to_string(),
stream_response: true,
}
);
@@ -214,7 +215,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta
}],
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
}),
)
@@ -252,7 +253,7 @@ async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_delta
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
}),
)
@@ -288,7 +289,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("http-1".to_string()),
request_id: "http-1".to_string(),
stream_response: true,
}
);
@@ -321,7 +322,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("http-2".to_string()),
request_id: "http-2".to_string(),
stream_response: true,
}
);
@@ -347,7 +348,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
}),
)
@@ -371,7 +372,7 @@ async fn http_response_body_stream_drops_queued_terminal_before_next_generated_i
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
};
let (reuse_response, _reuse_body_stream) =
@@ -410,7 +411,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request()
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("http-1".to_string()),
request_id: "http-1".to_string(),
stream_response: true,
}
);
@@ -429,7 +430,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request()
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("http-2".to_string()),
request_id: "http-2".to_string(),
stream_response: true,
}
);
@@ -473,7 +474,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request()
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
})
.await;
@@ -494,7 +495,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request()
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
}),
)
@@ -540,7 +541,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<()
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("http-1".to_string()),
request_id: "http-1".to_string(),
stream_response: true,
}
);
@@ -579,7 +580,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<()
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("http-2".to_string()),
request_id: "http-2".to_string(),
stream_response: true,
}
);
@@ -614,7 +615,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<()
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
}),
)
@@ -646,7 +647,7 @@ async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<()
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
}),
)
@@ -690,7 +691,7 @@ async fn http_response_body_stream_fails_when_transport_disconnects() -> Result<
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("http-1".to_string()),
request_id: "http-1".to_string(),
stream_response: true,
}
);
@@ -716,7 +717,7 @@ async fn http_response_body_stream_fails_when_transport_disconnects() -> Result<
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
}),
)
@@ -742,6 +743,98 @@ async fn http_response_body_stream_fails_when_transport_disconnects() -> Result<
Ok(())
}
/// What this tests: transport disconnect still records a terminal stream
/// failure even when the client-side body-delta queue is already full.
#[tokio::test]
async fn http_response_body_stream_reports_disconnect_when_queue_is_full() -> Result<()> {
// Phase 1: fill the queued body-delta route exactly to capacity before the
// response headers arrive, then drop the transport without sending EOF.
let server = spawn_scripted_exec_server(|mut peer| async move {
let (request_id, params) = peer.read_http_request().await?;
assert_eq!(
params,
HttpRequestParams {
method: "GET".to_string(),
url: "https://example.test/mcp/disconnect-full-queue".to_string(),
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: "http-1".to_string(),
stream_response: true,
}
);
for seq in 1..=HTTP_BODY_DELTA_CHANNEL_CAPACITY {
peer.write_body_delta(HttpRequestBodyDeltaNotification {
request_id: "http-1".to_string(),
seq,
delta: b"x".to_vec().into(),
done: false,
error: None,
})
.await?;
}
peer.write_response(
request_id,
HttpRequestResponse {
status: 200,
headers: Vec::new(),
body: Vec::new().into(),
},
)
.await
})
.await?;
let client = server.connect_client().await?;
// Phase 2: start the streaming request and receive headers while the
// queue is already full.
let (_response, mut body_stream) = timeout(
TEST_TIMEOUT,
client.http_request_stream(HttpRequestParams {
method: "GET".to_string(),
url: "https://example.test/mcp/disconnect-full-queue".to_string(),
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: "caller-stream-id".to_string(),
stream_response: false,
}),
)
.await
.context("streamed http/request should return headers")??;
// Phase 3: drain the queued chunks and assert the transport disconnect is
// still reported as an error rather than a clean EOF.
let mut chunks = 0;
let error = loop {
match timeout(TEST_TIMEOUT, body_stream.recv())
.await
.context("disconnect should wake the full queued body stream")?
{
Ok(Some(_chunk)) => {
chunks += 1;
}
Ok(None) => bail!("disconnect with a full queue should not look like clean EOF"),
Err(error) => break error,
}
};
assert_eq!(
(
chunks,
error
.to_string()
.starts_with(
"exec-server protocol error: http response stream `http-1` failed: exec-server transport disconnected",
),
),
(HTTP_BODY_DELTA_CHANNEL_CAPACITY as usize, true)
);
drop(client);
server.finish().await?;
Ok(())
}
/// What this tests: body-delta backpressure closes the public body stream as
/// an error rather than letting callers accept a truncated body as clean EOF.
#[tokio::test]
@@ -759,7 +852,7 @@ async fn http_response_body_stream_reports_backpressure_truncation() -> Result<(
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("http-1".to_string()),
request_id: "http-1".to_string(),
stream_response: true,
}
);
@@ -801,7 +894,7 @@ async fn http_response_body_stream_reports_backpressure_truncation() -> Result<(
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: Some("caller-stream-id".to_string()),
request_id: "caller-stream-id".to_string(),
stream_response: false,
}),
)
codex-rs/exec-server/tests/http_request.rs
+568
View File
@@ -0,0 +1,568 @@
#![cfg(unix)]
mod common;
use std::collections::BTreeMap;
use std::io::ErrorKind;
use std::time::Duration;
use codex_app_server_protocol::JSONRPCError;
use codex_app_server_protocol::JSONRPCMessage;
use codex_app_server_protocol::JSONRPCNotification;
use codex_app_server_protocol::JSONRPCResponse;
use codex_app_server_protocol::RequestId;
use codex_exec_server::HttpHeader;
use codex_exec_server::HttpRequestBodyDeltaNotification;
use codex_exec_server::HttpRequestParams;
use codex_exec_server::HttpRequestResponse;
use codex_exec_server::InitializeParams;
use common::exec_server::ExecServerHarness;
use common::exec_server::exec_server;
use pretty_assertions::assert_eq;
use serde::de::DeserializeOwned;
use serde_json::Value;
use tokio::io::AsyncBufReadExt;
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::BufReader;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
use tokio::sync::oneshot;
use tokio::time::timeout;
/// HTTP request captured by the ad-hoc TCP server in these integration tests.
#[derive(Debug)]
struct CapturedHttpRequest {
stream: TcpStream,
request_line: String,
headers: BTreeMap<String, String>,
body: Vec<u8>,
}
/// What this tests: a real exec-server websocket `http/request` performs one
/// HTTP request through the runner and returns the complete response body in
/// the JSON-RPC response.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_http_request_buffers_response_body() -> anyhow::Result<()> {
// Phase 1: start exec-server and complete the JSON-RPC handshake.
let mut server = exec_server().await?;
initialize_exec_server(&mut server).await?;
// Phase 2: start a local HTTP peer and ask exec-server to POST to it.
let listener = TcpListener::bind("127.0.0.1:0").await?;
let url = format!("http://{}/mcp?case=buffered", listener.local_addr()?);
let http_request_id = server
.send_request(
"http/request",
serde_json::to_value(HttpRequestParams {
method: "POST".to_string(),
url,
headers: vec![HttpHeader {
name: "x-codex-test".to_string(),
value: "buffered".to_string(),
}],
body: Some(b"request-body".to_vec().into()),
timeout_ms: Some(5_000),
request_id: "buffered-request".to_string(),
stream_response: false,
})?,
)
.await?;
// Phase 3: assert the HTTP peer observes the expected method, path,
// headers, and body before returning a fixed-length response.
let captured = accept_http_request(&listener).await?;
assert_eq!(
(
captured.request_line.as_str(),
captured.headers.get("x-codex-test").map(String::as_str),
captured.body.as_slice(),
),
(
"POST /mcp?case=buffered HTTP/1.1",
Some("buffered"),
b"request-body".as_slice(),
)
);
respond_with_status_and_headers(
captured.stream,
"201 Created",
&[("x-mcp-test", "buffered")],
b"response-body",
)
.await?;
// Phase 4: assert exec-server returns status, response headers, and the
// full response body in the JSON-RPC result.
let response: HttpRequestResponse = wait_for_response(&mut server, http_request_id).await?;
assert_eq!(
(
response.status,
response_header(&response.headers, "x-mcp-test"),
response.body.into_inner(),
),
(201, Some("buffered".to_string()), b"response-body".to_vec(),)
);
server.shutdown().await?;
Ok(())
}
/// What this tests: a real exec-server websocket `http/request` can return
/// response headers immediately and stream the response body as ordered
/// `http/request/bodyDelta` notifications.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_http_request_streams_response_body_notifications() -> anyhow::Result<()> {
// Phase 1: start exec-server and complete the JSON-RPC handshake.
let mut server = exec_server().await?;
initialize_exec_server(&mut server).await?;
// Phase 2: start a local HTTP peer and ask exec-server for a streamed GET.
let listener = TcpListener::bind("127.0.0.1:0").await?;
let url = format!("http://{}/mcp?case=streaming", listener.local_addr()?);
let http_request_id = server
.send_request(
"http/request",
serde_json::to_value(HttpRequestParams {
method: "GET".to_string(),
url,
headers: vec![HttpHeader {
name: "accept".to_string(),
value: "text/event-stream".to_string(),
}],
body: None,
timeout_ms: Some(5_000),
request_id: "stream-1".to_string(),
stream_response: true,
})?,
)
.await?;
// Phase 3: assert the HTTP peer observes the expected request and then
// respond with chunked transfer encoding to exercise streaming.
let captured = accept_http_request(&listener).await?;
assert_eq!(
(
captured.request_line.as_str(),
captured.headers.get("accept").map(String::as_str),
captured.body,
),
(
"GET /mcp?case=streaming HTTP/1.1",
Some("text/event-stream"),
Vec::new(),
)
);
respond_with_chunked_body(
captured.stream,
&[("x-mcp-test", "streaming")],
&[b"hello ".as_slice(), b"world".as_slice()],
)
.await?;
// Phase 4: assert the JSON-RPC response reaches the wire before any body
// delta notifications, and that it contains status and headers but no
// buffered body when streaming is requested.
let first_event = server.next_event().await?;
let JSONRPCMessage::Response(JSONRPCResponse { id, result }) = first_event else {
anyhow::bail!("expected http/request response before body deltas, got {first_event:?}");
};
assert_eq!(id, http_request_id);
let response: HttpRequestResponse = serde_json::from_value(result)?;
assert_eq!(
(
response.status,
response_header(&response.headers, "x-mcp-test"),
response.body.into_inner(),
),
(200, Some("streaming".to_string()), Vec::new())
);
// Phase 5: assert the body notifications are contiguous, ordered, and end
// with a clean terminal frame.
let deltas = collect_response_body_deltas(&mut server, "stream-1").await?;
let seqs = deltas.iter().map(|delta| delta.seq).collect::<Vec<_>>();
let body = deltas
.iter()
.flat_map(|delta| delta.delta.clone().into_inner())
.collect::<Vec<_>>();
let terminal = deltas.last().map(|delta| (delta.done, delta.error.clone()));
let expected_seqs = (1..=deltas.len() as u64).collect::<Vec<_>>();
assert_eq!(
(seqs, body, terminal),
(expected_seqs, b"hello world".to_vec(), Some((true, None)))
);
server.shutdown().await?;
Ok(())
}
/// What this tests: streamed `requestId`s stay reserved until the body stream
/// finishes, so a second in-flight request cannot reuse the same id.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_http_request_rejects_duplicate_stream_request_ids() -> 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=duplicate-stream-id",
listener.local_addr()?
);
let first_request_id = server
.send_request(
"http/request",
serde_json::to_value(HttpRequestParams {
method: "GET".to_string(),
url: url.clone(),
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: "stream-dup".to_string(),
stream_response: true,
})?,
)
.await?;
let captured = accept_http_request(&listener).await?;
let (finish_tx, finish_rx) = oneshot::channel();
let response_task = tokio::spawn(async move {
respond_with_chunked_body_until_finish(captured.stream, &[], &[b"hello"], finish_rx).await
});
let _: HttpRequestResponse = wait_for_response(&mut server, first_request_id).await?;
let duplicate_request_id = server
.send_request(
"http/request",
serde_json::to_value(HttpRequestParams {
method: "GET".to_string(),
url,
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: "stream-dup".to_string(),
stream_response: true,
})?,
)
.await?;
let duplicate_response = server
.wait_for_event(|event| {
matches!(
event,
JSONRPCMessage::Error(JSONRPCError { id, .. }) if id == &duplicate_request_id
)
})
.await?;
let JSONRPCMessage::Error(JSONRPCError { error, .. }) = duplicate_response else {
anyhow::bail!("expected duplicate requestId error response");
};
assert_eq!(error.code, -32602);
assert_eq!(
error.message,
"http/request streamResponse requestId `stream-dup` is already active"
);
finish_tx
.send(())
.expect("response task should still be waiting");
response_task.await??;
let _ = collect_response_body_deltas(&mut server, "stream-dup").await?;
server.shutdown().await?;
Ok(())
}
/// What this tests: omitting `timeoutMs` leaves the request unbounded, while
/// an explicit short timeout still fails the same delayed response.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_server_http_request_honors_optional_timeout() -> 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 delayed_url = format!(
"http://{}/mcp?case=optional-timeout",
listener.local_addr()?
);
let no_timeout_request_id = server
.send_request(
"http/request",
serde_json::to_value(HttpRequestParams {
method: "GET".to_string(),
url: delayed_url.clone(),
headers: Vec::new(),
body: None,
timeout_ms: None,
request_id: "buffered-request".to_string(),
stream_response: false,
})?,
)
.await?;
let captured = accept_http_request(&listener).await?;
let delayed_response = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(100)).await;
respond_with_status_and_headers(captured.stream, "200 OK", &[], b"slow-success").await
});
let response: HttpRequestResponse =
wait_for_response(&mut server, no_timeout_request_id).await?;
assert_eq!(response.body.into_inner(), b"slow-success".to_vec());
delayed_response.await??;
let timeout_request_id = server
.send_request(
"http/request",
serde_json::to_value(HttpRequestParams {
method: "GET".to_string(),
url: delayed_url,
headers: Vec::new(),
body: None,
timeout_ms: Some(10),
request_id: "buffered-request".to_string(),
stream_response: false,
})?,
)
.await?;
let captured = accept_http_request(&listener).await?;
let delayed_timeout_response = tokio::spawn(async move {
tokio::time::sleep(Duration::from_millis(100)).await;
respond_with_status_and_headers(captured.stream, "200 OK", &[], b"too-late").await
});
let error = wait_for_error_response(&mut server, timeout_request_id).await?;
assert_eq!(error.code, -32603);
assert!(
error.message.starts_with("http/request failed: "),
"unexpected timeout error: {}",
error.message
);
match delayed_timeout_response.await? {
Ok(()) => {}
Err(err) if is_expected_peer_disconnect(&err) => {}
Err(err) => return Err(err),
}
server.shutdown().await?;
Ok(())
}
/// Performs the JSON-RPC initialize handshake required before executor methods.
async fn initialize_exec_server(server: &mut ExecServerHarness) -> anyhow::Result<()> {
let initialize_id = server
.send_request(
"initialize",
serde_json::to_value(InitializeParams {
client_name: "exec-server-http-test".to_string(),
resume_session_id: None,
})?,
)
.await?;
let _: Value = wait_for_response(server, initialize_id).await?;
server
.send_notification("initialized", serde_json::json!({}))
.await?;
Ok(())
}
/// Waits for a typed JSON-RPC response with the requested id.
async fn wait_for_response<T>(
server: &mut ExecServerHarness,
request_id: RequestId,
) -> anyhow::Result<T>
where
T: DeserializeOwned,
{
let response = server
.wait_for_event(|event| {
matches!(
event,
JSONRPCMessage::Response(JSONRPCResponse { id, .. }) if id == &request_id
)
})
.await?;
let JSONRPCMessage::Response(JSONRPCResponse { result, .. }) = response else {
anyhow::bail!("expected JSON-RPC response for {request_id:?}");
};
Ok(serde_json::from_value(result)?)
}
/// Waits for a JSON-RPC error with the requested id.
async fn wait_for_error_response(
server: &mut ExecServerHarness,
request_id: RequestId,
) -> anyhow::Result<codex_app_server_protocol::JSONRPCErrorError> {
let response = server
.wait_for_event(|event| {
matches!(
event,
JSONRPCMessage::Error(JSONRPCError { id, .. }) if id == &request_id
)
})
.await?;
let JSONRPCMessage::Error(JSONRPCError { error, .. }) = response else {
anyhow::bail!("expected JSON-RPC error for {request_id:?}");
};
Ok(error)
}
/// Accepts one HTTP/1.1 request and captures its wire-visible fields.
async fn accept_http_request(listener: &TcpListener) -> anyhow::Result<CapturedHttpRequest> {
let (stream, _) = timeout(Duration::from_secs(5), listener.accept()).await??;
let mut reader = BufReader::new(stream);
let mut request_line = String::new();
reader.read_line(&mut request_line).await?;
let request_line = request_line.trim_end_matches("\r\n").to_string();
let mut headers = BTreeMap::new();
loop {
let mut line = String::new();
reader.read_line(&mut line).await?;
if line == "\r\n" {
break;
}
let line = line.trim_end_matches("\r\n");
let (name, value) = line
.split_once(':')
.ok_or_else(|| anyhow::anyhow!("HTTP header should contain colon: {line}"))?;
headers.insert(name.to_ascii_lowercase(), value.trim().to_string());
}
let content_length = headers
.get("content-length")
.and_then(|value| value.parse::<usize>().ok())
.unwrap_or(0);
let mut body = vec![0; content_length];
reader.read_exact(&mut body).await?;
Ok(CapturedHttpRequest {
stream: reader.into_inner(),
request_line,
headers,
body,
})
}
/// Writes a fixed-length HTTP response to the captured request stream.
async fn respond_with_status_and_headers(
mut stream: TcpStream,
status: &str,
headers: &[(&str, &str)],
body: &[u8],
) -> anyhow::Result<()> {
let extra_headers = headers
.iter()
.map(|(name, value)| format!("{name}: {value}\r\n"))
.collect::<String>();
let response = format!(
"HTTP/1.1 {status}\r\ncontent-type: text/plain\r\ncontent-length: {}\r\nconnection: close\r\n{extra_headers}\r\n",
body.len(),
);
stream.write_all(response.as_bytes()).await?;
stream.write_all(body).await?;
stream.flush().await?;
Ok(())
}
fn is_expected_peer_disconnect(err: &anyhow::Error) -> bool {
err.chain().any(|cause| {
cause
.downcast_ref::<std::io::Error>()
.is_some_and(|io_err| {
matches!(
io_err.kind(),
ErrorKind::BrokenPipe | ErrorKind::ConnectionReset | ErrorKind::UnexpectedEof
)
})
})
}
/// Writes a chunked HTTP response so reqwest must drive the streaming path.
async fn respond_with_chunked_body(
mut stream: TcpStream,
headers: &[(&str, &str)],
chunks: &[&[u8]],
) -> anyhow::Result<()> {
let extra_headers = headers
.iter()
.map(|(name, value)| format!("{name}: {value}\r\n"))
.collect::<String>();
let response = format!(
"HTTP/1.1 200 OK\r\ncontent-type: text/plain\r\ntransfer-encoding: chunked\r\nconnection: close\r\n{extra_headers}\r\n",
);
stream.write_all(response.as_bytes()).await?;
for chunk in chunks {
stream
.write_all(format!("{:x}\r\n", chunk.len()).as_bytes())
.await?;
stream.write_all(chunk).await?;
stream.write_all(b"\r\n").await?;
stream.flush().await?;
}
stream.write_all(b"0\r\n\r\n").await?;
stream.flush().await?;
Ok(())
}
/// Writes a chunked response and keeps the stream open until the test allows EOF.
async fn respond_with_chunked_body_until_finish(
mut stream: TcpStream,
headers: &[(&str, &str)],
chunks: &[&[u8]],
finish_rx: oneshot::Receiver<()>,
) -> anyhow::Result<()> {
let extra_headers = headers
.iter()
.map(|(name, value)| format!("{name}: {value}\r\n"))
.collect::<String>();
let response = format!(
"HTTP/1.1 200 OK\r\ncontent-type: text/plain\r\ntransfer-encoding: chunked\r\nconnection: close\r\n{extra_headers}\r\n",
);
stream.write_all(response.as_bytes()).await?;
for chunk in chunks {
stream
.write_all(format!("{:x}\r\n", chunk.len()).as_bytes())
.await?;
stream.write_all(chunk).await?;
stream.write_all(b"\r\n").await?;
stream.flush().await?;
}
finish_rx.await?;
stream.write_all(b"0\r\n\r\n").await?;
stream.flush().await?;
Ok(())
}
/// Collects streamed response-body notifications until the terminal frame.
async fn collect_response_body_deltas(
server: &mut ExecServerHarness,
request_id: &str,
) -> anyhow::Result<Vec<HttpRequestBodyDeltaNotification>> {
let mut deltas = Vec::new();
loop {
let event = server.next_event().await?;
let JSONRPCMessage::Notification(JSONRPCNotification { method, params }) = event else {
anyhow::bail!("expected http/request body delta notification, got {event:?}");
};
assert_eq!(method, "http/request/bodyDelta");
let delta: HttpRequestBodyDeltaNotification =
serde_json::from_value(params.unwrap_or(Value::Null))?;
assert_eq!(delta.request_id, request_id);
let done = delta.done;
deltas.push(delta);
if done {
return Ok(deltas);
}
}
}
/// Returns a response header value without depending on header-name casing.
fn response_header(headers: &[HttpHeader], name: &str) -> Option<String> {
headers
.iter()
.find(|header| header.name.eq_ignore_ascii_case(name))
.map(|header| header.value.clone())
}