From d6af7a6c038d8fc88e8293ec64071d3ee744f059 Mon Sep 17 00:00:00 2001 From: Ahmed Ibrahim Date: Mon, 20 Apr 2026 19:21:08 -0700 Subject: [PATCH 1/4] [1/4] Add executor HTTP request protocol (#18581) ### Why Remote streamable HTTP MCP needs a transport-shaped executor primitive before the MCP client can move network I/O to the executor. This layer keeps the executor unaware of MCP and gives later PRs an ordered streaming surface for response bodies. ### What - Add typed `http/request` and `http/request/bodyDelta` protocol payloads. - Add executor client helpers for buffered and streamed HTTP responses. - Route body-delta notifications to request-scoped streams with sequence validation and cleanup when a stream finishes or is dropped. - Document the new protocol constants, transport structs, public client methods, body-stream lifecycle, and request-scoped routing helpers. - Add in-memory JSON-RPC client coverage for streamed HTTP response-body notifications, with comments spelling out what the test proves and each setup/exercise/assert phase. ### 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/exec-server/src/client.rs | 33 +- .../exec-server/src/client/http_client.rs | 322 ++++++ codex-rs/exec-server/src/lib.rs | 4 + codex-rs/exec-server/src/protocol.rs | 81 ++ codex-rs/exec-server/tests/http_client.rs | 1026 +++++++++++++++++ 5 files changed, 1464 insertions(+), 2 deletions(-) create mode 100644 codex-rs/exec-server/src/client/http_client.rs create mode 100644 codex-rs/exec-server/tests/http_client.rs diff --git a/codex-rs/exec-server/src/client.rs b/codex-rs/exec-server/src/client.rs index 8cb094fb24..4e282c8fd3 100644 --- a/codex-rs/exec-server/src/client.rs +++ b/codex-rs/exec-server/src/client.rs @@ -3,12 +3,14 @@ use std::collections::HashMap; use std::sync::Arc; use std::sync::Mutex as StdMutex; use std::sync::OnceLock; +use std::sync::atomic::AtomicU64; use std::time::Duration; use arc_swap::ArcSwap; use codex_app_server_protocol::JSONRPCNotification; use serde_json::Value; use tokio::sync::Mutex; +use tokio::sync::mpsc; use tokio::sync::watch; use tokio::time::timeout; @@ -55,6 +57,8 @@ use crate::protocol::FsRemoveParams; use crate::protocol::FsRemoveResponse; use crate::protocol::FsWriteFileParams; use crate::protocol::FsWriteFileResponse; +use crate::protocol::HTTP_REQUEST_BODY_DELTA_METHOD; +use crate::protocol::HttpRequestBodyDeltaNotification; use crate::protocol::INITIALIZE_METHOD; use crate::protocol::INITIALIZED_METHOD; use crate::protocol::InitializeParams; @@ -70,6 +74,8 @@ use crate::rpc::RpcCallError; use crate::rpc::RpcClient; use crate::rpc::RpcClientEvent; +pub(crate) mod http_client; + const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); const INITIALIZE_TIMEOUT: Duration = Duration::from_secs(10); const PROCESS_EVENT_CHANNEL_CAPACITY: usize = 256; @@ -145,6 +151,14 @@ struct Inner { // with the same canonical message. This client never reconnects, so the // latch only moves from unset to set once. disconnected: OnceLock, + // Streaming HTTP responses are keyed by a client-generated request id + // because they share the same connection-global notification channel as + // process output. Keep the routing table local to the client so higher + // layers can consume body chunks like a normal byte stream. + http_body_streams: ArcSwap>>, + http_body_stream_failures: ArcSwap>, + http_body_streams_write_lock: Mutex<()>, + http_body_stream_next_id: AtomicU64, session_id: std::sync::RwLock>, reader_task: tokio::task::JoinHandle<()>, } @@ -380,7 +394,7 @@ impl ExecServerClient { &inner, format!("exec-server notification handling failed: {err}"), ); - fail_all_sessions(&inner, message).await; + fail_all_in_flight_work(&inner, message).await; return; } } @@ -390,7 +404,7 @@ impl ExecServerClient { &inner, disconnected_message(reason.as_deref()), ); - fail_all_sessions(&inner, message).await; + fail_all_in_flight_work(&inner, message).await; } return; } @@ -403,6 +417,10 @@ impl ExecServerClient { sessions: ArcSwap::from_pointee(HashMap::new()), sessions_write_lock: Mutex::new(()), disconnected: OnceLock::new(), + http_body_streams: ArcSwap::from_pointee(HashMap::new()), + http_body_stream_failures: ArcSwap::from_pointee(HashMap::new()), + http_body_streams_write_lock: Mutex::new(()), + http_body_stream_next_id: AtomicU64::new(1), session_id: std::sync::RwLock::new(None), reader_task, } @@ -733,6 +751,12 @@ async fn fail_all_sessions(inner: &Arc, message: String) { } } +/// Fails all in-flight work that depends on the shared JSON-RPC transport. +async fn fail_all_in_flight_work(inner: &Arc, message: String) { + fail_all_sessions(inner, message.clone()).await; + inner.fail_all_http_body_streams(message).await; +} + async fn handle_server_notification( inner: &Arc, notification: JSONRPCNotification, @@ -783,6 +807,11 @@ async fn handle_server_notification( } } } + HTTP_REQUEST_BODY_DELTA_METHOD => { + inner + .handle_http_body_delta_notification(notification.params) + .await?; + } other => { debug!("ignoring unknown exec-server notification: {other}"); } diff --git a/codex-rs/exec-server/src/client/http_client.rs b/codex-rs/exec-server/src/client/http_client.rs new file mode 100644 index 0000000000..a02a8be9a3 --- /dev/null +++ b/codex-rs/exec-server/src/client/http_client.rs @@ -0,0 +1,322 @@ +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::Ordering; + +use serde_json::Value; +use serde_json::from_value; +use tokio::runtime::Handle; +use tokio::sync::mpsc; +use tokio::sync::mpsc::error::TrySendError; +use tracing::debug; + +use super::ExecServerClient; +use super::ExecServerError; +use super::Inner; +use crate::protocol::HTTP_REQUEST_METHOD; +use crate::protocol::HttpRequestBodyDeltaNotification; +use crate::protocol::HttpRequestParams; +use crate::protocol::HttpRequestResponse; + +/// Maximum queued body frames per streamed executor HTTP response. +const HTTP_BODY_DELTA_CHANNEL_CAPACITY: usize = 256; + +/// Request-scoped stream of body chunks for an executor HTTP response. +/// +/// The initial `http/request` call returns status and headers. This stream then +/// receives the ordered `http/request/bodyDelta` notifications for that request +/// id until EOF or a terminal error. +pub struct HttpResponseBodyStream { + inner: Arc, + request_id: String, + next_seq: u64, + rx: mpsc::Receiver, + // Terminal frames can carry a final chunk; return that once, then EOF. + pending_eof: bool, + closed: bool, +} + +impl HttpResponseBodyStream { + /// Receives the next response-body chunk. + /// + /// Returns `Ok(None)` at EOF and converts sequence gaps or executor-side + /// stream errors into protocol errors. + pub async fn recv(&mut self) -> Result>, ExecServerError> { + if self.pending_eof { + self.pending_eof = false; + self.finish().await; + return Ok(None); + } + + let Some(delta) = self.rx.recv().await else { + self.finish().await; + if let Some(error) = self + .inner + .take_http_body_stream_failure(&self.request_id) + .await + { + return Err(ExecServerError::Protocol(format!( + "http response stream `{}` failed: {error}", + self.request_id + ))); + } + return Ok(None); + }; + if delta.seq != self.next_seq { + self.finish().await; + return Err(ExecServerError::Protocol(format!( + "http response stream `{}` received seq {}, expected {}", + self.request_id, delta.seq, self.next_seq + ))); + } + self.next_seq += 1; + let chunk = delta.delta.into_inner(); + + if let Some(error) = delta.error { + self.finish().await; + return Err(ExecServerError::Protocol(format!( + "http response stream `{}` failed: {error}", + self.request_id + ))); + } + if delta.done { + self.finish().await; + if chunk.is_empty() { + return Ok(None); + } + self.pending_eof = true; + } + Ok(Some(chunk)) + } + + /// Removes this stream from the connection routing table once it reaches EOF. + async fn finish(&mut self) { + if self.closed { + return; + } + self.closed = true; + self.inner.remove_http_body_stream(&self.request_id).await; + } +} + +impl Drop for HttpResponseBodyStream { + /// Schedules stream-route removal if the consumer drops before EOF. + fn drop(&mut self) { + if self.closed { + return; + } + self.closed = true; + spawn_remove_http_body_stream(Arc::clone(&self.inner), self.request_id.clone()); + } +} + +/// 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); + } + }; + registration.active = false; + Ok(( + response, + HttpResponseBodyStream { + inner: Arc::clone(&self.inner), + request_id, + next_seq: 1, + rx, + pending_eof: false, + closed: false, + }, + )) + } +} + +impl Inner { + /// Routes one streamed HTTP body notification into its request-local receiver. + pub(super) async fn handle_http_body_delta_notification( + &self, + params: Option, + ) -> Result<(), ExecServerError> { + let params: HttpRequestBodyDeltaNotification = from_value(params.unwrap_or(Value::Null))?; + // Unknown request ids are ignored intentionally: a stream may have already + // reached EOF and released its route. + if let Some(tx) = self + .http_body_streams + .load() + .get(¶ms.request_id) + .cloned() + { + let request_id = params.request_id.clone(); + let terminal_delta = params.done || params.error.is_some(); + match tx.try_send(params) { + Ok(()) => { + if terminal_delta { + self.remove_http_body_stream(&request_id).await; + } + } + Err(TrySendError::Closed(_)) => { + self.remove_http_body_stream(&request_id).await; + debug!("http response stream receiver dropped before body delta delivery"); + } + Err(TrySendError::Full(_)) => { + self.record_http_body_stream_failure( + &request_id, + "body delta channel filled before delivery".to_string(), + ) + .await; + self.remove_http_body_stream(&request_id).await; + debug!( + "closing http response stream `{request_id}` after body delta backpressure" + ); + } + } + } + Ok(()) + } + + /// Fails active streamed HTTP bodies so callers do not wait forever after a + /// transport disconnect or notification handling failure. + pub(super) async fn fail_all_http_body_streams(&self, message: String) { + let _streams_write_guard = self.http_body_streams_write_lock.lock().await; + let streams = self.http_body_streams.load(); + 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()), + }); + } + } + + /// Allocates a connection-local streamed HTTP response id. + fn next_http_body_stream_request_id(&self) -> String { + let id = self + .http_body_stream_next_id + .fetch_add(1, Ordering::Relaxed); + format!("http-{id}") + } + + /// Registers a request id before issuing an executor streaming HTTP call. + async fn insert_http_body_stream( + &self, + request_id: String, + tx: mpsc::Sender, + ) -> Result<(), ExecServerError> { + let _streams_write_guard = self.http_body_streams_write_lock.lock().await; + let streams = self.http_body_streams.load(); + if streams.contains_key(&request_id) { + return Err(ExecServerError::Protocol(format!( + "http response stream already registered for request {request_id}" + ))); + } + let mut next_streams = streams.as_ref().clone(); + next_streams.insert(request_id.clone(), tx); + self.http_body_streams.store(Arc::new(next_streams)); + let failures = self.http_body_stream_failures.load(); + if failures.contains_key(&request_id) { + let mut next_failures = failures.as_ref().clone(); + next_failures.remove(&request_id); + self.http_body_stream_failures + .store(Arc::new(next_failures)); + } + Ok(()) + } + + /// Removes a request id after EOF, terminal error, or request failure. + async fn remove_http_body_stream( + &self, + request_id: &str, + ) -> Option> { + let _streams_write_guard = self.http_body_streams_write_lock.lock().await; + let streams = self.http_body_streams.load(); + let stream = streams.get(request_id).cloned(); + stream.as_ref()?; + let mut next_streams = streams.as_ref().clone(); + next_streams.remove(request_id); + self.http_body_streams.store(Arc::new(next_streams)); + stream + } + + async fn record_http_body_stream_failure(&self, request_id: &str, message: String) { + let _streams_write_guard = self.http_body_streams_write_lock.lock().await; + let failures = self.http_body_stream_failures.load(); + let mut next_failures = failures.as_ref().clone(); + next_failures.insert(request_id.to_string(), message); + self.http_body_stream_failures + .store(Arc::new(next_failures)); + } + + async fn take_http_body_stream_failure(&self, request_id: &str) -> Option { + let _streams_write_guard = self.http_body_streams_write_lock.lock().await; + let failures = self.http_body_stream_failures.load(); + let error = failures.get(request_id).cloned(); + error.as_ref()?; + let mut next_failures = failures.as_ref().clone(); + next_failures.remove(request_id); + self.http_body_stream_failures + .store(Arc::new(next_failures)); + error + } +} + +/// 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() { + handle.spawn(async move { + inner.remove_http_body_stream(&request_id).await; + }); + } +} diff --git a/codex-rs/exec-server/src/lib.rs b/codex-rs/exec-server/src/lib.rs index 1d024fb5aa..067fa0a7c1 100644 --- a/codex-rs/exec-server/src/lib.rs +++ b/codex-rs/exec-server/src/lib.rs @@ -65,6 +65,10 @@ pub use protocol::FsRemoveParams; pub use protocol::FsRemoveResponse; pub use protocol::FsWriteFileParams; pub use protocol::FsWriteFileResponse; +pub use protocol::HttpHeader; +pub use protocol::HttpRequestBodyDeltaNotification; +pub use protocol::HttpRequestParams; +pub use protocol::HttpRequestResponse; pub use protocol::InitializeParams; pub use protocol::InitializeResponse; pub use protocol::ProcessOutputChunk; diff --git a/codex-rs/exec-server/src/protocol.rs b/codex-rs/exec-server/src/protocol.rs index 9f4234e649..24d1ee1c1e 100644 --- a/codex-rs/exec-server/src/protocol.rs +++ b/codex-rs/exec-server/src/protocol.rs @@ -26,6 +26,10 @@ pub const FS_GET_METADATA_METHOD: &str = "fs/getMetadata"; pub const FS_READ_DIRECTORY_METHOD: &str = "fs/readDirectory"; pub const FS_REMOVE_METHOD: &str = "fs/remove"; pub const FS_COPY_METHOD: &str = "fs/copy"; +/// JSON-RPC request method for executor-side HTTP requests. +pub const HTTP_REQUEST_METHOD: &str = "http/request"; +/// JSON-RPC notification method for streamed executor HTTP response bodies. +pub const HTTP_REQUEST_BODY_DELTA_METHOD: &str = "http/request/bodyDelta"; #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(transparent)] @@ -254,6 +258,83 @@ pub struct FsCopyParams { #[serde(rename_all = "camelCase")] pub struct FsCopyResponse {} +/// HTTP header represented in the executor protocol. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct HttpHeader { + /// Header name as it appears on the HTTP wire. + pub name: String, + /// Header value after UTF-8 conversion. + pub value: String, +} + +/// Executor-side HTTP request envelope. +/// +/// This intentionally stays transport-shaped rather than MCP-shaped so callers +/// can use it for Streamable HTTP, OAuth discovery, and future executor-owned +/// HTTP probes without introducing one protocol method per higher-level use. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct HttpRequestParams { + /// HTTP method, for example `GET`, `POST`, or `DELETE`. + pub method: String, + /// Absolute `http://` or `https://` URL. + pub url: String, + /// Ordered request headers. Repeated header names are preserved. + #[serde(default)] + pub headers: Vec, + /// Optional request body bytes. + #[serde(default, rename = "bodyBase64")] + pub body: Option, + /// Optional request timeout in milliseconds. + #[serde(default)] + 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, + /// Return after response headers and stream the response body as deltas. + #[serde(default)] + pub stream_response: bool, +} + +/// HTTP response envelope returned from an executor `http/request` call. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct HttpRequestResponse { + /// Numeric HTTP response status code. + pub status: u16, + /// Ordered response headers. Repeated header names are preserved. + pub headers: Vec, + /// Buffered response body bytes. Empty when `streamResponse` is true. + #[serde(rename = "bodyBase64")] + pub body: ByteChunk, +} + +/// Ordered response-body frame for `streamResponse` HTTP requests. +/// +/// Headers are returned in the `http/request` response so the caller can choose +/// a parser immediately; body bytes then arrive on this notification stream. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub struct HttpRequestBodyDeltaNotification { + /// Request id from the streamed `http/request` call. + pub request_id: String, + /// Monotonic one-based body frame sequence number. + pub seq: u64, + /// Response-body bytes carried by this frame. + #[serde(rename = "deltaBase64")] + pub delta: ByteChunk, + /// Marks response-body EOF. No later deltas are expected for this request. + #[serde(default)] + pub done: bool, + /// Terminal stream error. Set only on the final notification. + #[serde(default)] + pub error: Option, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub enum ExecOutputStream { diff --git a/codex-rs/exec-server/tests/http_client.rs b/codex-rs/exec-server/tests/http_client.rs new file mode 100644 index 0000000000..1462c80168 --- /dev/null +++ b/codex-rs/exec-server/tests/http_client.rs @@ -0,0 +1,1026 @@ +use std::future::Future; +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use anyhow::bail; +use codex_app_server_protocol::JSONRPCMessage; +use codex_app_server_protocol::JSONRPCNotification; +use codex_app_server_protocol::JSONRPCRequest; +use codex_app_server_protocol::JSONRPCResponse; +use codex_app_server_protocol::RequestId; +use codex_exec_server::ExecServerClient; +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 codex_exec_server::InitializeResponse; +use codex_exec_server::RemoteExecServerConnectArgs; +use futures::SinkExt; +use futures::StreamExt; +use pretty_assertions::assert_eq; +use serde::Serialize; +use serde::de::DeserializeOwned; +use serde_json::from_slice; +use serde_json::from_str; +use serde_json::from_value; +use serde_json::to_string; +use serde_json::to_value; +use tokio::net::TcpListener; +use tokio::net::TcpStream; +use tokio::sync::oneshot; +use tokio::task::JoinHandle; +use tokio::time::timeout; +use tokio_tungstenite::WebSocketStream; +use tokio_tungstenite::accept_async; +use tokio_tungstenite::tungstenite::Message; + +const CLIENT_NAME: &str = "test-exec-server-client"; +const HTTP_REQUEST_METHOD: &str = "http/request"; +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 OVERFLOWING_BODY_DELTA_FRAMES: u64 = 1_024; + +/// What this tests: the buffered HTTP helper always sends a buffered +/// `http/request`, even when a caller accidentally provides streaming flags. +#[tokio::test] +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. + let (request_id, params) = peer.read_http_request().await?; + assert_eq!( + params, + HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/buffered".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: None, + stream_response: false, + } + ); + + peer.write_response( + request_id, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: b"buffered".to_vec().into(), + }, + ) + .await + }) + .await?; + let client = server.connect_client().await?; + + // Phase 3: call the buffered helper with streaming-only fields populated + // and assert callers still receive the buffered response body. + let response = timeout( + TEST_TIMEOUT, + client.http_request(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/buffered".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("ignored-stream-id".to_string()), + stream_response: true, + }), + ) + .await + .context("buffered http/request should complete")??; + assert_eq!( + response, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: b"buffered".to_vec().into(), + } + ); + + drop(client); + server.finish().await?; + Ok(()) +} + +/// What this tests: streamed executor HTTP response frames are routed by the +/// client's generated request id, delivered in sequence, and concatenated by +/// the caller. +#[tokio::test] +async fn http_response_body_stream_uses_generated_ids_and_receives_ordered_deltas() -> Result<()> { + // Phase 1: script two requests. The caller supplies reusable ids, but the + // client replaces them with connection-local ids on the wire. + 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".to_string(), + headers: vec![HttpHeader { + name: "accept".to_string(), + value: "text/event-stream".to_string(), + }], + body: None, + timeout_ms: None, + request_id: Some("http-1".to_string()), + stream_response: true, + } + ); + + // Phase 2: return headers first, then body notifications in the order + // the public body stream should expose them. + peer.write_response( + request_id, + HttpRequestResponse { + status: 200, + headers: vec![HttpHeader { + name: "content-type".to_string(), + value: "text/event-stream".to_string(), + }], + body: Vec::new().into(), + }, + ) + .await?; + for delta in [ + HttpRequestBodyDeltaNotification { + request_id: "http-1".to_string(), + seq: 1, + delta: b"hello ".to_vec().into(), + done: false, + error: None, + }, + HttpRequestBodyDeltaNotification { + request_id: "http-1".to_string(), + seq: 2, + delta: b"world".to_vec().into(), + done: false, + error: None, + }, + HttpRequestBodyDeltaNotification { + request_id: "http-1".to_string(), + seq: 3, + delta: b"!".to_vec().into(), + done: true, + error: None, + }, + ] { + peer.write_body_delta(delta).await?; + } + + // Phase 3: accept the next generated request id after EOF. + let (request_id, params) = peer.read_http_request().await?; + assert_eq!( + params, + HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/reuse".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("http-2".to_string()), + stream_response: true, + } + ); + peer.write_response( + request_id, + HttpRequestResponse { + status: 204, + headers: Vec::new(), + body: Vec::new().into(), + }, + ) + .await + }) + .await?; + let client = server.connect_client().await?; + + // Phase 4: start a streaming HTTP request through the public client API. + let (response, mut body_stream) = timeout( + TEST_TIMEOUT, + client.http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp".to_string(), + headers: vec![HttpHeader { + name: "accept".to_string(), + value: "text/event-stream".to_string(), + }], + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }), + ) + .await + .context("streamed http/request should return headers")??; + assert_eq!( + response, + HttpRequestResponse { + status: 200, + headers: vec![HttpHeader { + name: "content-type".to_string(), + value: "text/event-stream".to_string(), + }], + body: Vec::new().into(), + } + ); + + // Phase 5: drain the body stream and verify the caller-visible byte order. + let mut body = Vec::new(); + while let Some(chunk) = timeout(TEST_TIMEOUT, body_stream.recv()) + .await + .context("http response body delta should arrive")?? + { + body.extend_from_slice(&chunk); + } + assert_eq!(body, b"hello world!".to_vec()); + + // Phase 6: start another stream through the public API to validate cleanup + // after EOF without reaching into the client routing table. + let (reuse_response, _reuse_body_stream) = timeout( + TEST_TIMEOUT, + client.http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/reuse".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }), + ) + .await + .context("second streamed http/request should return headers")??; + assert_eq!( + reuse_response, + HttpRequestResponse { + status: 204, + headers: Vec::new(), + body: Vec::new().into(), + } + ); + + drop(client); + server.finish().await?; + Ok(()) +} + +/// What this tests: dropping a body stream with a queued terminal frame removes +/// the old route while the next stream gets a fresh generated id. +#[tokio::test] +async fn http_response_body_stream_drops_queued_terminal_before_next_generated_id() -> Result<()> { + // Phase 1: send terminal EOF before the header response so the public body + // stream starts with EOF already queued but unread. + 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/queued-terminal".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("http-1".to_string()), + stream_response: true, + } + ); + peer.write_body_delta(HttpRequestBodyDeltaNotification { + request_id: "http-1".to_string(), + seq: 1, + delta: Vec::new().into(), + done: true, + error: None, + }) + .await?; + peer.write_response( + request_id, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: Vec::new().into(), + }, + ) + .await?; + + // Phase 2: accept another stream after the client drops the unread + // body. The second request receives a distinct generated id. + let (request_id, params) = peer.read_http_request().await?; + assert_eq!( + params, + HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/retry-queued-terminal".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("http-2".to_string()), + stream_response: true, + } + ); + peer.write_response( + request_id, + HttpRequestResponse { + status: 204, + headers: Vec::new(), + body: Vec::new().into(), + }, + ) + .await + }) + .await?; + let client = server.connect_client().await?; + + // Phase 3: drop the body stream without reading the queued EOF frame. + let (response, body_stream) = timeout( + TEST_TIMEOUT, + client.http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/queued-terminal".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }), + ) + .await + .context("streamed http/request should return headers")??; + assert_eq!( + response, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: Vec::new().into(), + } + ); + drop(body_stream); + + // Phase 4: start another stream through the public API. The caller-provided + // id is ignored, so the request uses the next generated route id. + let params = HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/retry-queued-terminal".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }; + let (reuse_response, _reuse_body_stream) = + timeout(TEST_TIMEOUT, client.http_request_stream(params)) + .await + .context("second streamed http/request should return headers")??; + assert_eq!( + reuse_response, + HttpRequestResponse { + status: 204, + headers: Vec::new(), + body: Vec::new().into(), + } + ); + + drop(client); + server.finish().await?; + Ok(()) +} + +/// What this tests: cancelling a streaming HTTP request while it is waiting for +/// headers drops its route, and a later stream gets a fresh generated id. +#[tokio::test] +async fn http_response_body_stream_ignores_late_deltas_after_cancelled_request() -> Result<()> { + // Phase 1: coordinate cancellation after the fake server observes the + // first request but before it returns headers. The server later sends a + // stale delta for the cancelled id before serving the fresh stream. + let (request_seen_tx, request_seen_rx) = oneshot::channel(); + 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/cancel".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("http-1".to_string()), + stream_response: true, + } + ); + request_seen_tx + .send(()) + .expect("test should wait for the first request"); + + // Phase 2: the next stream uses a new generated id. A late body delta + // for the cancelled id is ignored by the client-side router. + let (request_id, params) = peer.read_http_request().await?; + assert_eq!( + params, + HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/retry-cancelled".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("http-2".to_string()), + stream_response: true, + } + ); + peer.write_body_delta(HttpRequestBodyDeltaNotification { + request_id: "http-1".to_string(), + seq: 1, + delta: b"stale".to_vec().into(), + done: false, + error: None, + }) + .await?; + peer.write_response( + request_id, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: Vec::new().into(), + }, + ) + .await?; + peer.write_body_delta(HttpRequestBodyDeltaNotification { + request_id: "http-2".to_string(), + seq: 1, + delta: b"fresh".to_vec().into(), + done: true, + error: None, + }) + .await + }) + .await?; + let client = server.connect_client().await?; + + // Phase 3: start a streaming request and abort the caller future while it + // is blocked waiting for response headers. + let client_for_request = client.clone(); + let stream_task = tokio::spawn(async move { + let _ = client_for_request + .http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/cancel".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }) + .await; + }); + request_seen_rx + .await + .expect("server should observe the first http/request"); + stream_task.abort(); + let _ = stream_task.await; + + // Phase 4: start a new stream immediately. It receives only the fresh body + // bytes for its generated id. + let (response, mut body_stream) = timeout( + TEST_TIMEOUT, + client.http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/retry-cancelled".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }), + ) + .await + .context("second streamed http/request should return headers")??; + assert_eq!( + response, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: Vec::new().into(), + } + ); + let mut body = Vec::new(); + while let Some(chunk) = timeout(TEST_TIMEOUT, body_stream.recv()) + .await + .context("fresh http response body delta should arrive")?? + { + body.extend_from_slice(&chunk); + } + assert_eq!(body, b"fresh".to_vec()); + + drop(client); + server.finish().await?; + Ok(()) +} + +/// What this tests: dropping a returned body stream before EOF removes its +/// route and prevents stale body deltas from reaching the next stream. +#[tokio::test] +async fn http_response_body_stream_ignores_late_deltas_after_drop() -> Result<()> { + // Phase 1: script two requests. The first returns only headers; after the + // client drops its body receiver, the server sends a stale body delta. + let (body_dropped_tx, body_dropped_rx) = oneshot::channel(); + let (stale_delta_sent_tx, stale_delta_sent_rx) = oneshot::channel(); + 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/drop".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("http-1".to_string()), + stream_response: true, + } + ); + peer.write_response( + request_id, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: Vec::new().into(), + }, + ) + .await?; + body_dropped_rx + .await + .expect("test should drop the first body stream"); + peer.write_body_delta(HttpRequestBodyDeltaNotification { + request_id: "http-1".to_string(), + seq: 1, + delta: b"stale".to_vec().into(), + done: false, + error: None, + }) + .await?; + stale_delta_sent_tx + .send(()) + .expect("test should wait for the stale delta"); + + // Phase 2: accept the next request with a new generated id. The new + // stream must receive only fresh body bytes. + let (request_id, params) = peer.read_http_request().await?; + assert_eq!( + params, + HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/retry-dropped".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("http-2".to_string()), + stream_response: true, + } + ); + peer.write_response( + request_id, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: Vec::new().into(), + }, + ) + .await?; + peer.write_body_delta(HttpRequestBodyDeltaNotification { + request_id: "http-2".to_string(), + seq: 1, + delta: b"fresh".to_vec().into(), + done: true, + error: None, + }) + .await + }) + .await?; + let client = server.connect_client().await?; + + // Phase 3: receive headers for the first stream, then drop the body stream + // without reading any body frames. + let (response, body_stream) = timeout( + TEST_TIMEOUT, + client.http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/drop".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }), + ) + .await + .context("streamed http/request should return headers")??; + assert_eq!( + response, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: Vec::new().into(), + } + ); + drop(body_stream); + body_dropped_tx + .send(()) + .expect("server should wait for the body stream drop"); + stale_delta_sent_rx + .await + .expect("server should send one stale nonterminal delta"); + + // Phase 4: start the next stream immediately. The caller-provided id is + // ignored, and the fresh generated id isolates it from stale bytes. + let (reuse_response, mut reuse_body_stream) = timeout( + TEST_TIMEOUT, + client.http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/retry-dropped".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }), + ) + .await + .context("second streamed http/request should return headers")??; + assert_eq!( + reuse_response, + HttpRequestResponse { + status: 200, + headers: Vec::new(), + body: Vec::new().into(), + } + ); + let mut body = Vec::new(); + while let Some(chunk) = timeout(TEST_TIMEOUT, reuse_body_stream.recv()) + .await + .context("fresh http response body delta should arrive")?? + { + body.extend_from_slice(&chunk); + } + assert_eq!(body, b"fresh".to_vec()); + + drop(client); + server.finish().await?; + Ok(()) +} + +/// What this tests: an in-flight streamed HTTP body is failed when the shared +/// JSON-RPC transport disconnects before a terminal body frame. +#[tokio::test] +async fn http_response_body_stream_fails_when_transport_disconnects() -> Result<()> { + // Phase 1: return response headers for a streaming request, then drop the + // fake server 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".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("http-1".to_string()), + stream_response: true, + } + ); + 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 a streaming HTTP request and receive headers. + let (_response, mut body_stream) = timeout( + TEST_TIMEOUT, + client.http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/disconnect".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }), + ) + .await + .context("streamed http/request should return headers")??; + + // Phase 3: assert transport disconnect wakes the body stream with a + // terminal error instead of hanging. + let error = timeout(TEST_TIMEOUT, body_stream.recv()) + .await + .context("disconnect should wake http body stream")? + .expect_err("disconnect should fail the http body stream"); + let error_message = error.to_string(); + assert_eq!( + error_message.starts_with( + "exec-server protocol error: http response stream `http-1` failed: exec-server transport disconnected" + ), + 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] +async fn http_response_body_stream_reports_backpressure_truncation() -> Result<()> { + // Phase 1: send enough body frames before headers to overflow the bounded + // client-side route while the public request future is still pending. + let (finish_tx, finish_rx) = oneshot::channel(); + 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/backpressure".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("http-1".to_string()), + stream_response: true, + } + ); + for seq in 1..=OVERFLOWING_BODY_DELTA_FRAMES { + 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?; + + // Phase 2: keep the transport connected so the body stream reports the + // backpressure failure rather than a disconnect. + finish_rx.await.expect("test should finish server task"); + Ok(()) + }) + .await?; + let client = server.connect_client().await?; + + // Phase 3: start the streaming request; the server overfills the route + // before returning the body stream to this consumer. + let (_response, mut body_stream) = timeout( + TEST_TIMEOUT, + client.http_request_stream(HttpRequestParams { + method: "GET".to_string(), + url: "https://example.test/mcp/backpressure".to_string(), + headers: Vec::new(), + body: None, + timeout_ms: None, + request_id: Some("caller-stream-id".to_string()), + stream_response: false, + }), + ) + .await + .context("streamed http/request should return headers")??; + + // Phase 4: drain queued chunks and assert the truncated stream ends in an + // explicit error, not a clean EOF. + let mut chunks = 0; + let error = loop { + match timeout(TEST_TIMEOUT, body_stream.recv()) + .await + .context("backpressure should close http body stream")? + { + Ok(Some(_chunk)) => { + chunks += 1; + } + Ok(None) => bail!("backpressure truncation should not look like clean EOF"), + Err(error) => break error, + } + }; + assert_eq!( + ( + chunks < OVERFLOWING_BODY_DELTA_FRAMES as usize, + error.to_string(), + ), + ( + true, + "exec-server protocol error: http response stream `http-1` failed: body delta channel filled before delivery".to_string(), + ) + ); + + finish_tx + .send(()) + .expect("server task should wait for test completion"); + drop(client); + server.finish().await?; + Ok(()) +} + +/// Fake WebSocket exec-server used by the integration tests. +/// +/// The helper exercises `ExecServerClient::connect_websocket`, including the +/// initialize handshake, while each test controls the exact JSON-RPC traffic +/// that follows. +struct ScriptedExecServer { + websocket_url: String, + task: JoinHandle>, +} + +impl ScriptedExecServer { + /// Connects the public exec-server client to this fake WebSocket endpoint. + async fn connect_client(&self) -> Result { + ExecServerClient::connect_websocket(RemoteExecServerConnectArgs::new( + self.websocket_url.clone(), + CLIENT_NAME.to_string(), + )) + .await + .context("client should connect to fake exec-server") + } + + /// Waits for the scripted fake server to finish. + async fn finish(self) -> Result<()> { + self.task + .await + .context("fake exec-server task should join")??; + Ok(()) + } +} + +/// Starts a fake exec-server that accepts one WebSocket client. +async fn spawn_scripted_exec_server(script: F) -> Result +where + F: FnOnce(JsonRpcPeer) -> Fut + Send + 'static, + Fut: Future> + Send + 'static, +{ + let listener = TcpListener::bind("127.0.0.1:0") + .await + .context("fake exec-server should bind")?; + let websocket_url = format!("ws://{}", listener.local_addr()?); + let task = tokio::spawn(async move { + let (stream, _) = timeout(TEST_TIMEOUT, listener.accept()) + .await + .context("fake exec-server should accept a client")??; + let websocket = accept_async(stream) + .await + .context("fake exec-server websocket handshake should complete")?; + let mut peer = JsonRpcPeer { websocket }; + peer.complete_initialize().await?; + script(peer).await + }); + Ok(ScriptedExecServer { + websocket_url, + task, + }) +} + +/// JSON-RPC peer for the fake exec-server WebSocket. +struct JsonRpcPeer { + websocket: WebSocketStream, +} + +impl JsonRpcPeer { + /// Completes and validates the client initialize handshake. + async fn complete_initialize(&mut self) -> Result<()> { + let request = self.read_request(INITIALIZE_METHOD).await?; + let params: InitializeParams = decode_request_params(&request)?; + assert_eq!( + params, + InitializeParams { + client_name: CLIENT_NAME.to_string(), + resume_session_id: None, + } + ); + self.write_response( + request.id, + InitializeResponse { + session_id: "session-1".to_string(), + }, + ) + .await?; + self.read_notification(INITIALIZED_METHOD).await?; + Ok(()) + } + + /// Reads one typed `http/request` call from the client. + async fn read_http_request(&mut self) -> Result<(RequestId, HttpRequestParams)> { + let request = self.read_request(HTTP_REQUEST_METHOD).await?; + let params = decode_request_params(&request)?; + Ok((request.id, params)) + } + + /// Reads a JSON-RPC request and validates its method. + async fn read_request(&mut self, expected_method: &str) -> Result { + let message = self.read_message().await?; + let JSONRPCMessage::Request(request) = message else { + bail!("expected JSON-RPC request `{expected_method}`, got {message:?}"); + }; + if request.method != expected_method { + bail!( + "expected JSON-RPC request `{expected_method}`, got `{}`", + request.method + ); + } + Ok(request) + } + + /// Reads a JSON-RPC notification and validates its method. + async fn read_notification(&mut self, expected_method: &str) -> Result { + let message = self.read_message().await?; + let JSONRPCMessage::Notification(notification) = message else { + bail!("expected JSON-RPC notification `{expected_method}`, got {message:?}"); + }; + if notification.method != expected_method { + bail!( + "expected JSON-RPC notification `{expected_method}`, got `{}`", + notification.method + ); + } + Ok(notification) + } + + /// Sends a successful JSON-RPC response. + async fn write_response(&mut self, id: RequestId, result: T) -> Result<()> + where + T: Serialize, + { + self.write_message(JSONRPCMessage::Response(JSONRPCResponse { + id, + result: to_value(result)?, + })) + .await + } + + /// Sends one streamed HTTP body notification. + async fn write_body_delta(&mut self, delta: HttpRequestBodyDeltaNotification) -> Result<()> { + self.write_message(JSONRPCMessage::Notification(JSONRPCNotification { + method: HTTP_REQUEST_BODY_DELTA_METHOD.to_string(), + params: Some(to_value(delta)?), + })) + .await + } + + /// Reads one WebSocket JSON-RPC message. + async fn read_message(&mut self) -> Result { + let message = timeout(TEST_TIMEOUT, self.websocket.next()) + .await + .context("timed out waiting for JSON-RPC message")? + .context("client websocket closed before JSON-RPC message arrived")? + .context("failed to read websocket message")?; + match message { + Message::Text(text) => from_str(text.as_ref()).context("text JSON-RPC"), + Message::Binary(bytes) => from_slice(bytes.as_ref()).context("binary JSON-RPC"), + Message::Close(frame) => bail!("client websocket closed: {frame:?}"), + other => bail!("expected text or binary JSON-RPC message, got {other:?}"), + } + } + + /// Writes one WebSocket JSON-RPC message. + async fn write_message(&mut self, message: JSONRPCMessage) -> Result<()> { + let encoded = to_string(&message)?; + timeout( + TEST_TIMEOUT, + self.websocket.send(Message::Text(encoded.into())), + ) + .await + .context("timed out writing JSON-RPC message")? + .context("failed to write JSON-RPC message") + } +} + +/// Decodes a request params object into its typed protocol payload. +fn decode_request_params(request: &JSONRPCRequest) -> Result +where + T: DeserializeOwned, +{ + let params = request + .params + .clone() + .context("JSON-RPC request should include params")?; + from_value(params).context("JSON-RPC request params should decode") +} From ff05532723004e1e8e3c9f44314a3d7a90ac050c Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Mon, 20 Apr 2026 19:23:30 -0700 Subject: [PATCH 2/4] refactor: narrow async lock scopes (#18418) ## Why This is part of the follow-up work from #18178 to make Codex ready for Clippy's [`await_holding_lock`](https://rust-lang.github.io/rust-clippy/master/index.html#await_holding_lock) / [`await_holding_invalid_type`](https://rust-lang.github.io/rust-clippy/master/index.html#await_holding_invalid_type) lints. This bottom PR keeps the scope intentionally small: `NetworkProxyState::record_blocked()` only needs the state write lock while it mutates the blocked-request ring buffer and counters. The debug log payload and `BlockedRequestObserver` callback can be produced after that lock is released. ## What changed - Copies the blocked-request snapshot values needed for logging while updating the state. - Releases the `RwLockWriteGuard` before logging or notifying the observer. ## Verification - `cargo test -p codex-network-proxy` --- [//]: # (BEGIN SAPLING FOOTER) Stack created with [Sapling](https://sapling-scm.com). Best reviewed with [ReviewStack](https://reviewstack.dev/openai/codex/pull/18418). * #18698 * #18423 * __->__ #18418 --- codex-rs/network-proxy/src/runtime.rs | 31 ++++++++++++--------------- 1 file changed, 14 insertions(+), 17 deletions(-) diff --git a/codex-rs/network-proxy/src/runtime.rs b/codex-rs/network-proxy/src/runtime.rs index da090a69d4..daafeec737 100644 --- a/codex-rs/network-proxy/src/runtime.rs +++ b/codex-rs/network-proxy/src/runtime.rs @@ -420,32 +420,29 @@ impl NetworkProxyState { let blocked_for_observer = entry.clone(); let blocked_request_observer = self.blocked_request_observer.read().await.clone(); let violation_line = blocked_request_violation_log_line(&entry); - let mut guard = self.state.write().await; let host = entry.host.clone(); let reason = entry.reason.clone(); let decision = entry.decision.clone(); let source = entry.source.clone(); let protocol = entry.protocol.clone(); let port = entry.port; - guard.blocked.push_back(entry); - guard.blocked_total = guard.blocked_total.saturating_add(1); - let total = guard.blocked_total; - while guard.blocked.len() > MAX_BLOCKED_EVENTS { - guard.blocked.pop_front(); - } + let (total, buffered) = { + let mut guard = self.state.write().await; + guard.blocked.push_back(entry); + guard.blocked_total = guard.blocked_total.saturating_add(1); + let total = guard.blocked_total; + while guard.blocked.len() > MAX_BLOCKED_EVENTS { + guard.blocked.pop_front(); + } + (total, guard.blocked.len()) + }; debug!( - "recorded blocked request telemetry (total={}, host={}, reason={}, decision={:?}, source={:?}, protocol={}, port={:?}, buffered={})", - total, - host, - reason, - decision, - source, - protocol, - port, - guard.blocked.len() + "recorded blocked request telemetry (\ + total={total}, host={host}, reason={reason}, \ + decision={decision:?}, source={source:?}, \ + protocol={protocol}, port={port:?}, buffered={buffered})" ); debug!("{violation_line}"); - drop(guard); if let Some(observer) = blocked_request_observer { observer.on_blocked_request(blocked_for_observer).await; From aca6ccc4eef694a92b869e5940a2a7eeb50fe74d Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Mon, 20 Apr 2026 19:23:36 -0700 Subject: [PATCH 3/4] chore: document intentional await-holding cases --- .../app-server/src/codex_message_processor.rs | 8 +++ .../src/message_processor/tracing_tests.rs | 8 +++ .../src/transport/remote_control/websocket.rs | 12 ++++ codex-rs/clippy.toml | 5 ++ .../src/endpoint/responses_websocket.rs | 4 ++ codex-rs/connectors/src/lib.rs | 4 ++ codex-rs/core/src/guardian/review_session.rs | 4 ++ codex-rs/core/src/mcp_tool_call.rs | 12 ++++ codex-rs/core/src/session/handlers.rs | 4 ++ codex-rs/core/src/session/mcp.rs | 28 +++++++++ codex-rs/core/src/session/mod.rs | 60 +++++++++++++++++++ codex-rs/core/src/session/session.rs | 4 ++ codex-rs/core/src/session/tests.rs | 4 ++ .../core/src/session/tests/guardian_tests.rs | 4 ++ codex-rs/core/src/session/turn.rs | 8 +++ codex-rs/core/src/tools/code_mode/mod.rs | 4 ++ codex-rs/core/src/tools/handlers/dynamic.rs | 4 ++ .../core/src/tools/handlers/mcp_resource.rs | 8 +++ .../core/src/tools/handlers/tool_suggest.rs | 8 +++ codex-rs/core/src/tools/js_repl/mod.rs | 16 +++++ codex-rs/core/src/tools/registry.rs | 4 ++ codex-rs/core/src/tools/router_tests.rs | 12 ++++ codex-rs/login/src/auth/external_bearer.rs | 4 ++ codex-rs/rmcp-client/src/oauth.rs | 8 +++ 24 files changed, 237 insertions(+) diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 52a3d2fcb3..8cdb287702 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -8058,6 +8058,10 @@ impl CodexMessageProcessor { .await } + #[expect( + clippy::await_holding_invalid_type, + reason = "listener subscription must be serialized against pending thread unloads" + )] async fn ensure_conversation_listener_task( listener_task_context: ListenerTaskContext, conversation_id: ThreadId, @@ -8865,6 +8869,10 @@ async fn handle_thread_listener_command( } #[allow(clippy::too_many_arguments)] +#[expect( + clippy::await_holding_invalid_type, + reason = "running-thread resume subscription must be serialized against pending unloads" +)] async fn handle_pending_thread_resume_request( conversation_id: ThreadId, conversation: &Arc, diff --git a/codex-rs/app-server/src/message_processor/tracing_tests.rs b/codex-rs/app-server/src/message_processor/tracing_tests.rs index adf583d8f1..83f8bc98d7 100644 --- a/codex-rs/app-server/src/message_processor/tracing_tests.rs +++ b/codex-rs/app-server/src/message_processor/tracing_tests.rs @@ -505,6 +505,10 @@ where } #[tokio::test(flavor = "current_thread")] +#[expect( + clippy::await_holding_invalid_type, + reason = "test serializes access to global tracing state for its full duration" +)] async fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() -> Result<()> { let _guard = tracing_test_guard().lock().await; let mut harness = TracingHarness::new().await?; @@ -584,6 +588,10 @@ async fn thread_start_jsonrpc_span_exports_server_span_and_parents_children() -> } #[tokio::test(flavor = "current_thread")] +#[expect( + clippy::await_holding_invalid_type, + reason = "test serializes access to global tracing state for its full duration" +)] async fn turn_start_jsonrpc_span_parents_core_turn_spans() -> Result<()> { let _guard = tracing_test_guard().lock().await; let mut harness = TracingHarness::new().await?; diff --git a/codex-rs/app-server/src/transport/remote_control/websocket.rs b/codex-rs/app-server/src/transport/remote_control/websocket.rs index ac91882f52..dee177d30e 100644 --- a/codex-rs/app-server/src/transport/remote_control/websocket.rs +++ b/codex-rs/app-server/src/transport/remote_control/websocket.rs @@ -205,6 +205,10 @@ impl RemoteControlWebsocket { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "remote-control client shutdown must serialize tracker state" + )] pub(crate) async fn run( mut self, app_server_client_name_rx: Option>, @@ -416,6 +420,10 @@ impl RemoteControlWebsocket { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "remote-control server event receiver is shared across reconnects" + )] async fn run_server_writer_inner( state: Arc>, server_event_rx: Arc>>, @@ -557,6 +565,10 @@ impl RemoteControlWebsocket { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "remote-control client tracking must stay serialized while processing inbound events" + )] async fn run_websocket_reader_inner( client_tracker: Arc>, state: Arc>, diff --git a/codex-rs/clippy.toml b/codex-rs/clippy.toml index a2aae6591b..2feed8a487 100644 --- a/codex-rs/clippy.toml +++ b/codex-rs/clippy.toml @@ -1,5 +1,10 @@ allow-expect-in-tests = true allow-unwrap-in-tests = true +await-holding-invalid-types = [ + "tokio::sync::MutexGuard", + "tokio::sync::RwLockReadGuard", + "tokio::sync::RwLockWriteGuard", +] disallowed-methods = [ { path = "ratatui::style::Color::Rgb", reason = "Use ANSI colors, which work better in various terminal themes." }, { path = "ratatui::style::Color::Indexed", reason = "Use ANSI colors, which work better in various terminal themes." }, diff --git a/codex-rs/codex-api/src/endpoint/responses_websocket.rs b/codex-rs/codex-api/src/endpoint/responses_websocket.rs index bc72aee411..aea9b917fb 100644 --- a/codex-rs/codex-api/src/endpoint/responses_websocket.rs +++ b/codex-rs/codex-api/src/endpoint/responses_websocket.rs @@ -229,6 +229,10 @@ impl ResponsesWebsocketConnection { let current_span = Span::current(); tokio::spawn( + #[expect( + clippy::await_holding_invalid_type, + reason = "the guard serializes exclusive use of the websocket stream for the lifetime of the response stream" + )] async move { if let Some(model) = server_model { let _ = tx_event.send(Ok(ResponseEvent::ServerModel(model))).await; diff --git a/codex-rs/connectors/src/lib.rs b/codex-rs/connectors/src/lib.rs index e13270a78c..53956e17cb 100644 --- a/codex-rs/connectors/src/lib.rs +++ b/codex-rs/connectors/src/lib.rs @@ -446,6 +446,10 @@ mod tests { } #[tokio::test] + #[expect( + clippy::await_holding_invalid_type, + reason = "test serializes access to the shared connector cache for its full duration" + )] async fn list_all_connectors_uses_shared_cache() -> anyhow::Result<()> { let _cache_guard = ALL_CONNECTORS_CACHE_TEST_LOCK.lock().await; diff --git a/codex-rs/core/src/guardian/review_session.rs b/codex-rs/core/src/guardian/review_session.rs index 5bb3e9d1b8..a66d79e865 100644 --- a/codex-rs/core/src/guardian/review_session.rs +++ b/codex-rs/core/src/guardian/review_session.rs @@ -265,6 +265,10 @@ impl GuardianReviewSessionManager { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "review session selection and trunk spawning must stay serialized" + )] pub(crate) async fn run_review( &self, params: GuardianReviewSessionParams, diff --git a/codex-rs/core/src/mcp_tool_call.rs b/codex-rs/core/src/mcp_tool_call.rs index 32c7cf879e..c5d4068d27 100644 --- a/codex-rs/core/src/mcp_tool_call.rs +++ b/codex-rs/core/src/mcp_tool_call.rs @@ -494,6 +494,10 @@ async fn execute_mcp_tool_call( ) } +#[expect( + clippy::await_holding_invalid_type, + reason = "MCP sandbox metadata reads through the session-owned manager guard" +)] async fn augment_mcp_tool_request_meta_with_sandbox_state( sess: &Session, turn_context: &TurnContext, @@ -1067,6 +1071,10 @@ fn mcp_tool_approval_callsite_mode( } } +#[expect( + clippy::await_holding_invalid_type, + reason = "MCP approval metadata reads through the session-owned manager guard" +)] pub(crate) async fn lookup_mcp_tool_metadata( sess: &Session, turn_context: &TurnContext, @@ -1149,6 +1157,10 @@ fn get_mcp_app_resource_uri( }) } +#[expect( + clippy::await_holding_invalid_type, + reason = "MCP app metadata reads through the session-owned manager guard" +)] async fn lookup_mcp_app_usage_metadata( sess: &Session, server: &str, diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 4d69167dab..d3cd686258 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -468,6 +468,10 @@ pub async fn reload_user_config(sess: &Arc) { sess.reload_user_config_layer().await; } +#[expect( + clippy::await_holding_invalid_type, + reason = "MCP tool listing reads through the session-owned manager guard" +)] pub async fn list_mcp_tools(sess: &Session, config: &Arc, sub_id: String) { let mcp_connection_manager = sess.services.mcp_connection_manager.read().await; let auth = sess.services.auth_manager.auth().await; diff --git a/codex-rs/core/src/session/mcp.rs b/codex-rs/core/src/session/mcp.rs index d15ca2cae8..be7504d9e7 100644 --- a/codex-rs/core/src/session/mcp.rs +++ b/codex-rs/core/src/session/mcp.rs @@ -1,6 +1,10 @@ use super::*; impl Session { + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn request_mcp_server_elicitation( &self, turn_context: &TurnContext, @@ -80,6 +84,10 @@ impl Session { rx_response.await.ok() } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and manager fallback must stay serialized" + )] pub async fn resolve_elicitation( &self, server_name: String, @@ -111,6 +119,10 @@ impl Session { .await } + #[expect( + clippy::await_holding_invalid_type, + reason = "MCP resource calls are serialized through the session-owned manager guard" + )] pub async fn list_resources( &self, server: &str, @@ -124,6 +136,10 @@ impl Session { .await } + #[expect( + clippy::await_holding_invalid_type, + reason = "MCP resource calls are serialized through the session-owned manager guard" + )] pub async fn list_resource_templates( &self, server: &str, @@ -137,6 +153,10 @@ impl Session { .await } + #[expect( + clippy::await_holding_invalid_type, + reason = "MCP resource calls are serialized through the session-owned manager guard" + )] pub async fn read_resource( &self, server: &str, @@ -150,6 +170,10 @@ impl Session { .await } + #[expect( + clippy::await_holding_invalid_type, + reason = "MCP tool calls are serialized through the session-owned manager guard" + )] pub async fn call_tool( &self, server: &str, @@ -165,6 +189,10 @@ impl Session { .await } + #[expect( + clippy::await_holding_invalid_type, + reason = "MCP tool metadata reads through the session-owned manager guard" + )] pub(crate) async fn resolve_mcp_tool_info(&self, tool_name: &ToolName) -> Option { self.services .mcp_connection_manager diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index f159546575..aaaf8d75bd 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1786,6 +1786,10 @@ impl Session { /// be used to derive the available decisions via /// [ExecApprovalRequestEvent::default_available_decisions]. #[allow(clippy::too_many_arguments)] + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn request_command_approval( &self, turn_context: &TurnContext, @@ -1857,6 +1861,10 @@ impl Session { rx_approve.await.unwrap_or(ReviewDecision::Abort) } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn request_patch_approval( &self, turn_context: &TurnContext, @@ -1893,6 +1901,10 @@ impl Session { rx_approve } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn request_permissions( &self, turn_context: &TurnContext, @@ -1948,6 +1960,10 @@ impl Session { rx_response.await.ok() } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn request_user_input( &self, turn_context: &TurnContext, @@ -1980,6 +1996,10 @@ impl Session { rx_response.await.ok() } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn notify_user_input_response( &self, sub_id: &str, @@ -2005,6 +2025,10 @@ impl Session { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn notify_request_permissions_response( &self, call_id: &str, @@ -2046,6 +2070,10 @@ impl Session { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn reads must stay consistent with the matching turn state" + )] pub(crate) async fn granted_turn_permissions(&self) -> Option { let active = self.active_turn.lock().await; let active = active.as_ref()?; @@ -2058,6 +2086,10 @@ impl Session { state.granted_permissions() } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn notify_dynamic_tool_response(&self, call_id: &str, response: DynamicToolResponse) { let entry = { let mut active = self.active_turn.lock().await; @@ -2079,6 +2111,10 @@ impl Session { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn notify_approval(&self, approval_id: &str, decision: ReviewDecision) { let entry = { let mut active = self.active_turn.lock().await; @@ -2239,6 +2275,10 @@ impl Session { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "MCP app context rendering reads through the session-owned manager guard" + )] pub(crate) async fn build_initial_context( &self, turn_context: &TurnContext, @@ -2701,6 +2741,10 @@ impl Session { /// Inject additional user input into the currently active turn. /// /// Returns the active turn id when accepted. + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn steer_input( &self, input: Vec, @@ -2760,6 +2804,10 @@ impl Session { } /// Returns the input if there was no task running to inject into. + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn inject_response_items( &self, input: Vec, @@ -2833,6 +2881,10 @@ impl Session { self.mailbox_rx.lock().await.has_pending_trigger_turn() } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn prepend_pending_input(&self, input: Vec) -> Result<(), ()> { let mut active = self.active_turn.lock().await; match active.as_mut() { @@ -2845,6 +2897,10 @@ impl Session { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state updates must remain atomic" + )] pub async fn get_pending_input(&self) -> Vec { let (pending_input, accepts_mailbox_delivery) = { let mut active = self.active_turn.lock().await; @@ -2900,6 +2956,10 @@ impl Session { !self.idle_pending_input.lock().await.is_empty() } + #[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and turn state reads must remain atomic" + )] pub async fn has_pending_input(&self) -> bool { let (has_turn_pending_input, accepts_mailbox_delivery) = { let active = self.active_turn.lock().await; diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index a2080bbeab..16e86e3aea 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -209,6 +209,10 @@ pub(crate) struct AppServerClientMetadata { impl Session { #[instrument(name = "session_init", level = "info", skip_all)] #[allow(clippy::too_many_arguments)] + #[expect( + clippy::await_holding_invalid_type, + reason = "session initialization must serialize access through session-owned manager guards" + )] pub(crate) async fn new( mut session_configuration: SessionConfiguration, config: Arc, diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index bd41da9f20..40daa54e25 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -6320,6 +6320,10 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() { } #[tokio::test] +#[expect( + clippy::await_holding_invalid_type, + reason = "test builds a router from session-owned MCP manager state" +)] async fn fatal_tool_error_stops_turn_and_reports_error() { let (session, turn_context, _rx) = make_session_and_context_with_rx().await; let tools = { diff --git a/codex-rs/core/src/session/tests/guardian_tests.rs b/codex-rs/core/src/session/tests/guardian_tests.rs index 9ccd6eb4e0..97543105b6 100644 --- a/codex-rs/core/src/session/tests/guardian_tests.rs +++ b/codex-rs/core/src/session/tests/guardian_tests.rs @@ -298,6 +298,10 @@ async fn process_compacted_history_preserves_separate_guardian_developer_message #[tokio::test] #[cfg(unix)] +#[expect( + clippy::await_holding_invalid_type, + reason = "test mutates active turn state directly to seed granted permissions" +)] async fn shell_handler_allows_sticky_turn_permissions_without_inline_request_permissions_feature() { let (mut session, turn_context_raw) = make_session_and_context().await; session diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index efbcca6716..643007787a 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -127,6 +127,10 @@ use tracing::warn; /// - If the model sends only an assistant message, we record it in the /// conversation history and consider the turn complete. /// +#[expect( + clippy::await_holding_invalid_type, + reason = "turn execution must keep active-turn state transitions atomic" +)] pub(crate) async fn run_turn( sess: Arc, turn_context: Arc, @@ -1143,6 +1147,10 @@ async fn run_sampling_request( } } +#[expect( + clippy::await_holding_invalid_type, + reason = "tool router construction reads through the session-owned manager guard" +)] pub(crate) async fn built_tools( sess: &Session, turn_context: &TurnContext, diff --git a/codex-rs/core/src/tools/code_mode/mod.rs b/codex-rs/core/src/tools/code_mode/mod.rs index 5049836393..575d3f9a89 100644 --- a/codex-rs/core/src/tools/code_mode/mod.rs +++ b/codex-rs/core/src/tools/code_mode/mod.rs @@ -259,6 +259,10 @@ pub(super) async fn build_enabled_tools( collect_code_mode_tool_definitions(&specs) } +#[expect( + clippy::await_holding_invalid_type, + reason = "nested tool router construction reads through the session-owned manager guard" +)] async fn build_nested_router(exec: &ExecContext) -> ToolRouter { let nested_tools_config = exec.turn.tools_config.for_code_mode_nested_tools(); let listed_mcp_tools = exec diff --git a/codex-rs/core/src/tools/handlers/dynamic.rs b/codex-rs/core/src/tools/handlers/dynamic.rs index 140d586be9..c989d70f19 100644 --- a/codex-rs/core/src/tools/handlers/dynamic.rs +++ b/codex-rs/core/src/tools/handlers/dynamic.rs @@ -71,6 +71,10 @@ impl ToolHandler for DynamicToolHandler { } } +#[expect( + clippy::await_holding_invalid_type, + reason = "active turn checks and dynamic tool response registration must remain atomic" +)] async fn request_dynamic_tool( session: &Session, turn_context: &TurnContext, diff --git a/codex-rs/core/src/tools/handlers/mcp_resource.rs b/codex-rs/core/src/tools/handlers/mcp_resource.rs index 67f44cb699..fa4a066741 100644 --- a/codex-rs/core/src/tools/handlers/mcp_resource.rs +++ b/codex-rs/core/src/tools/handlers/mcp_resource.rs @@ -240,6 +240,10 @@ impl ToolHandler for McpResourceHandler { } } +#[expect( + clippy::await_holding_invalid_type, + reason = "MCP resource listing reads through the session-owned manager guard" +)] async fn handle_list_resources( session: Arc, turn: Arc, @@ -344,6 +348,10 @@ async fn handle_list_resources( } } +#[expect( + clippy::await_holding_invalid_type, + reason = "MCP resource template listing reads through the session-owned manager guard" +)] async fn handle_list_resource_templates( session: Arc, turn: Arc, diff --git a/codex-rs/core/src/tools/handlers/tool_suggest.rs b/codex-rs/core/src/tools/handlers/tool_suggest.rs index cc433601bc..5c51c90d66 100644 --- a/codex-rs/core/src/tools/handlers/tool_suggest.rs +++ b/codex-rs/core/src/tools/handlers/tool_suggest.rs @@ -34,6 +34,10 @@ impl ToolHandler for ToolSuggestHandler { ToolKind::Function } + #[expect( + clippy::await_holding_invalid_type, + reason = "tool suggestion discovery reads through the session-owned manager guard" + )] async fn handle(&self, invocation: ToolInvocation) -> Result { let ToolInvocation { payload, @@ -193,6 +197,10 @@ async fn verify_tool_suggestion_completed( } } +#[expect( + clippy::await_holding_invalid_type, + reason = "connector cache refresh reads through the session-owned manager guard" +)] async fn refresh_missing_suggested_connectors( session: &crate::session::session::Session, turn: &crate::session::turn_context::TurnContext, diff --git a/codex-rs/core/src/tools/js_repl/mod.rs b/codex-rs/core/src/tools/js_repl/mod.rs index 23f4906e5f..fffa7028dc 100644 --- a/codex-rs/core/src/tools/js_repl/mod.rs +++ b/codex-rs/core/src/tools/js_repl/mod.rs @@ -836,6 +836,10 @@ impl JsReplManager { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "js_repl kernel initialization must be serialized with kernel state" + )] pub async fn execute( &self, session: Arc, @@ -1162,6 +1166,10 @@ impl JsReplManager { Ok(kernel_path) } + #[expect( + clippy::await_holding_invalid_type, + reason = "js_repl stdin writes must be serialized per kernel" + )] async fn write_message( stdin: &Arc>, msg: &HostToKernel, @@ -1209,6 +1217,10 @@ impl JsReplManager { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "js_repl child shutdown must serialize process inspection and termination" + )] async fn kill_kernel_child(child: &Arc>, reason: &'static str) { let mut guard = child.lock().await; let pid = guard.id(); @@ -1532,6 +1544,10 @@ impl JsReplManager { } } + #[expect( + clippy::await_holding_invalid_type, + reason = "nested js_repl tool routing reads through the session-owned manager guard" + )] async fn run_tool_request(exec: ExecContext, req: RunToolRequest) -> RunToolResult { if is_js_repl_internal_tool(&req.tool_name) { let error = "js_repl cannot invoke itself".to_string(); diff --git a/codex-rs/core/src/tools/registry.rs b/codex-rs/core/src/tools/registry.rs index 4c0e3fbea8..aca1d10144 100644 --- a/codex-rs/core/src/tools/registry.rs +++ b/codex-rs/core/src/tools/registry.rs @@ -239,6 +239,10 @@ impl ToolRegistry { // } // } + #[expect( + clippy::await_holding_invalid_type, + reason = "tool dispatch must keep active-turn accounting atomic" + )] pub(crate) async fn dispatch_any( &self, invocation: ToolInvocation, diff --git a/codex-rs/core/src/tools/router_tests.rs b/codex-rs/core/src/tools/router_tests.rs index 1a59f1e1ba..14536e3985 100644 --- a/codex-rs/core/src/tools/router_tests.rs +++ b/codex-rs/core/src/tools/router_tests.rs @@ -14,6 +14,10 @@ use super::ToolRouter; use super::ToolRouterParams; #[tokio::test] +#[expect( + clippy::await_holding_invalid_type, + reason = "test builds a router from session-owned MCP manager state" +)] async fn js_repl_tools_only_blocks_direct_tool_calls() -> anyhow::Result<()> { let (session, mut turn) = make_session_and_context().await; turn.tools_config.js_repl_tools_only = true; @@ -68,6 +72,10 @@ async fn js_repl_tools_only_blocks_direct_tool_calls() -> anyhow::Result<()> { } #[tokio::test] +#[expect( + clippy::await_holding_invalid_type, + reason = "test builds a router from session-owned MCP manager state" +)] async fn js_repl_tools_only_allows_js_repl_source_calls() -> anyhow::Result<()> { let (session, mut turn) = make_session_and_context().await; turn.tools_config.js_repl_tools_only = true; @@ -171,6 +179,10 @@ async fn js_repl_tools_only_blocks_namespaced_js_repl_tool() -> anyhow::Result<( } #[tokio::test] +#[expect( + clippy::await_holding_invalid_type, + reason = "test builds a router from session-owned MCP manager state" +)] async fn parallel_support_does_not_match_namespaced_local_tool_names() -> anyhow::Result<()> { let (session, turn) = make_session_and_context().await; let mcp_tools = session diff --git a/codex-rs/login/src/auth/external_bearer.rs b/codex-rs/login/src/auth/external_bearer.rs index c179f48619..c528596014 100644 --- a/codex-rs/login/src/auth/external_bearer.rs +++ b/codex-rs/login/src/auth/external_bearer.rs @@ -33,6 +33,10 @@ impl ExternalAuth for BearerTokenRefresher { AuthMode::ApiKey } + #[expect( + clippy::await_holding_invalid_type, + reason = "external bearer cache misses intentionally hold cached_token across the provider command to avoid duplicate refreshes" + )] async fn resolve(&self) -> io::Result> { let access_token = { let mut cached = self.state.cached_token.lock().await; diff --git a/codex-rs/rmcp-client/src/oauth.rs b/codex-rs/rmcp-client/src/oauth.rs index 979a7ad217..5adfae9347 100644 --- a/codex-rs/rmcp-client/src/oauth.rs +++ b/codex-rs/rmcp-client/src/oauth.rs @@ -283,6 +283,10 @@ impl OAuthPersistor { /// Persists the latest stored credentials if they have changed. /// Deletes the credentials if they are no longer present. + #[expect( + clippy::await_holding_invalid_type, + reason = "AuthorizationManager async access must be serialized through its mutex" + )] pub(crate) async fn persist_if_needed(&self) -> Result<()> { let (client_id, maybe_credentials) = { let manager = self.inner.authorization_manager.clone(); @@ -335,6 +339,10 @@ impl OAuthPersistor { Ok(()) } + #[expect( + clippy::await_holding_invalid_type, + reason = "AuthorizationManager async access must be serialized through its mutex" + )] pub(crate) async fn refresh_if_needed(&self) -> Result<()> { let expires_at = { let guard = self.inner.last_credentials.lock().await; From 2a33df043c4de80f34a521bf436f66234cc64495 Mon Sep 17 00:00:00 2001 From: Michael Bolin Date: Mon, 20 Apr 2026 19:24:26 -0700 Subject: [PATCH 4/4] chore: enable await-holding clippy lints Follow-up to https://github.com/openai/codex/pull/18178, where we said the await-holding clippy rule would be enabled separately. Enable `await_holding_lock` and `await_holding_invalid_type` after the preceding commits fixed or explicitly documented the current offenders. --- .bazelrc | 2 ++ codex-rs/Cargo.toml | 2 ++ 2 files changed, 4 insertions(+) diff --git a/.bazelrc b/.bazelrc index 3a49a22c20..fb6e93e58c 100644 --- a/.bazelrc +++ b/.bazelrc @@ -82,6 +82,8 @@ build:clippy --@rules_rust//rust/settings:clippy.toml=//codex-rs:clippy.toml # in their own `Cargo.toml`, but `rules_rust` Bazel clippy does not read Cargo lint levels. # `clippy.toml` can configure lint behavior, but it cannot set allow/warn/deny/forbid levels. build:clippy --@rules_rust//rust/settings:clippy_flag=-Dwarnings +build:clippy --@rules_rust//rust/settings:clippy_flag=--deny=clippy::await_holding_invalid_type +build:clippy --@rules_rust//rust/settings:clippy_flag=--deny=clippy::await_holding_lock build:clippy --@rules_rust//rust/settings:clippy_flag=--deny=clippy::expect_used build:clippy --@rules_rust//rust/settings:clippy_flag=--deny=clippy::identity_op build:clippy --@rules_rust//rust/settings:clippy_flag=--deny=clippy::manual_clamp diff --git a/codex-rs/Cargo.toml b/codex-rs/Cargo.toml index fd203e6259..a911c39fc0 100644 --- a/codex-rs/Cargo.toml +++ b/codex-rs/Cargo.toml @@ -383,6 +383,8 @@ zeroize = "1.8.2" rust = {} [workspace.lints.clippy] +await_holding_invalid_type = "deny" +await_holding_lock = "deny" expect_used = "deny" identity_op = "deny" manual_clamp = "deny"