From 9360f267f329d80fd9e55e75b3110a38e59a8f9b Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Wed, 22 Apr 2026 13:36:34 -0700 Subject: [PATCH] [2/4] Implement executor HTTP request runner (#18582) ### 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 --------- Co-authored-by: Codex --- codex-rs/Cargo.lock | 2 + codex-rs/exec-server/Cargo.toml | 2 + .../exec-server/src/client/http_client.rs | 342 ++++++++--- codex-rs/exec-server/src/protocol.rs | 60 +- codex-rs/exec-server/src/rpc.rs | 48 +- codex-rs/exec-server/src/server/handler.rs | 99 +++ codex-rs/exec-server/src/server/processor.rs | 4 +- codex-rs/exec-server/src/server/registry.rs | 8 + codex-rs/exec-server/tests/http_client.rs | 141 ++++- codex-rs/exec-server/tests/http_request.rs | 568 ++++++++++++++++++ 10 files changed, 1169 insertions(+), 105 deletions(-) create mode 100644 codex-rs/exec-server/tests/http_request.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index c3b607eb6..6bc53a49f 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -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", ] diff --git a/codex-rs/exec-server/Cargo.toml b/codex-rs/exec-server/Cargo.toml index 5ca265c6b..968806327 100644 --- a/codex-rs/exec-server/Cargo.toml +++ b/codex-rs/exec-server/Cargo.toml @@ -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"] } diff --git a/codex-rs/exec-server/src/client/http_client.rs b/codex-rs/exec-server/src/client/http_client.rs index a02a8be9a..1e91fa448 100644 --- a/codex-rs/exec-server/src/client/http_client.rs +++ b/codex-rs/exec-server/src/client/http_client.rs @@ -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 { + params.stream_response = false; + self.call(HTTP_REQUEST_METHOD, ¶ms).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, ¶ms).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, - 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 { - params.stream_response = false; - params.request_id = None; - self.call(HTTP_REQUEST_METHOD, ¶ms).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, ¶ms).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) -> Result { + 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), 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(¶ms.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) -> Result { + 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 { + 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( + ¬ifications, + 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( + ¬ifications, + HttpRequestBodyDeltaNotification { + request_id, + seq, + delta: Vec::new().into(), + done: true, + error: Some(err.to_string()), + }, + ) + .await; + return; + } + } + } + + let _ = send_executor_body_delta( + ¬ifications, + 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, + 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, request_id: String) { if let Ok(handle) = Handle::try_current() { @@ -320,3 +510,13 @@ fn spawn_remove_http_body_stream(inner: Arc, 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() +} diff --git a/codex-rs/exec-server/src/protocol.rs b/codex-rs/exec-server/src/protocol.rs index 24d1ee1c1..435187d05 100644 --- a/codex-rs/exec-server/src/protocol.rs +++ b/codex-rs/exec-server/src/protocol.rs @@ -286,15 +286,19 @@ pub struct HttpRequestParams { /// Optional request body bytes. #[serde(default, rename = "bodyBase64")] pub body: Option, - /// 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, /// 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, + /// 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)) + ); + } +} diff --git a/codex-rs/exec-server/src/rpc.rs b/codex-rs/exec-server/src/rpc.rs index e82b4a0ea..723b99f50 100644 --- a/codex-rs/exec-server/src/rpc.rs +++ b/codex-rs/exec-server/src/rpc.rs @@ -36,8 +36,9 @@ pub(crate) enum RpcCallError { type PendingRequest = oneshot::Sender>; type BoxFuture = Pin + Send + 'static>>; -type RequestRoute = - Box, JSONRPCRequest) -> BoxFuture + Send + Sync>; +type RequestRoute = Box< + dyn Fn(Arc, JSONRPCRequest) -> BoxFuture> + Send + Sync, +>; type NotificationRoute = Box, JSONRPCNotification) -> BoxFuture> + 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( &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(&mut self, method: &'static str, handler: F) + where + P: DeserializeOwned + Send + 'static, + F: Fn(Arc, RequestId, P) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + { + self.request_routes.insert( + method, + Box::new(move |state, request| { + let request_id = request.id; + let params = decode_request_params::

(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 }), } }) }), diff --git a/codex-rs/exec-server/src/server/handler.rs b/codex-rs/exec-server/src/server/handler.rs index 46f7af90a..15035335e 100644 --- a/codex-rs/exec-server/src/server/handler.rs +++ b/codex-rs/exec-server/src/server/handler.rs @@ -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, notifications: RpcNotificationSender, session: StdMutex>, + active_body_stream_ids: Mutex>, + 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, + 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, + 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)] diff --git a/codex-rs/exec-server/src/server/processor.rs b/codex-rs/exec-server/src/server/processor.rs index 1153bc83e..dc1a9b9ff 100644 --- a/codex-rs/exec-server/src/server/processor.rs +++ b/codex-rs/exec-server/src/server/processor.rs @@ -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 diff --git a/codex-rs/exec-server/src/server/registry.rs b/codex-rs/exec-server/src/server/registry.rs index a57704c50..87dee6aa5 100644 --- a/codex-rs/exec-server/src/server/registry.rs +++ b/codex-rs/exec-server/src/server/registry.rs @@ -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 { handler.initialize(params).await }, ); + router.request_with_id( + HTTP_REQUEST_METHOD, + |handler: Arc, request_id, params: HttpRequestParams| async move { + handler.http_request(request_id, params).await + }, + ); router.request( EXEC_METHOD, |handler: Arc, params: ExecParams| async move { handler.exec(params).await }, diff --git a/codex-rs/exec-server/tests/http_client.rs b/codex-rs/exec-server/tests/http_client.rs index 1462c8016..6a087dd11 100644 --- a/codex-rs/exec-server/tests/http_client.rs +++ b/codex-rs/exec-server/tests/http_client.rs @@ -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, }), ) diff --git a/codex-rs/exec-server/tests/http_request.rs b/codex-rs/exec-server/tests/http_request.rs new file mode 100644 index 000000000..f45261dad --- /dev/null +++ b/codex-rs/exec-server/tests/http_request.rs @@ -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, + body: Vec, +} + +/// 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::>(); + let body = deltas + .iter() + .flat_map(|delta| delta.delta.clone().into_inner()) + .collect::>(); + let terminal = deltas.last().map(|delta| (delta.done, delta.error.clone())); + let expected_seqs = (1..=deltas.len() as u64).collect::>(); + 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( + server: &mut ExecServerHarness, + request_id: RequestId, +) -> anyhow::Result +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 { + 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 { + 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::().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::(); + 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::() + .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::(); + 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::(); + 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> { + 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 { + headers + .iter() + .find(|header| header.name.eq_ignore_ascii_case(name)) + .map(|header| header.value.clone()) +}