fix(http): enforce bounded WebSocket request handling

Cargo.toml

@@ -45,6 +45,7 @@ tokio = { version = "1.52", default-features = false }
 rustix = { version = "1", default-features = false, features = ["std", "process"] }
 tokio-util = { version = "0.7", features = ["compat"] }
 async-tungstenite = { version = "0.35.0", default-features = false, features = ["tokio-rustls-webpki-roots"] }
+tungstenite = "=0.29.0"
 
 # Serialization
 serde = { version = "1.0", features = ["derive", "rc"] }
@@ -67,7 +68,9 @@ rmcp = { version = "2.1.0", features = ["server", "transport-io", "schemars"] }
 clap = { version = "4.5", features = ["derive"] }
 
 # HTTP
-axum = "0.8"
+# The typed capacity-error downcast requires Axum and this crate to share the
+# same Tungstenite 0.29 identity. Update both constraints together.
+axum = "=0.8.9"
 reqwest = { version = "0.13", default-features = false, features = ["rustls", "json"] }
 eventsource-stream = "0.2"
 url = "2.5"

src/agent-client-protocol-http/Cargo.toml

@@ -35,19 +35,22 @@ server = [
     "dep:async-stream",
     "dep:axum",
     "dep:futures",
+    "dep:serde",
     "dep:serde_json",
     "dep:tokio",
     "tokio/macros",
     "tokio/rt",
     "tokio/sync",
     "dep:tower-http",
     "dep:tracing",
+    "dep:tungstenite",
     "dep:uuid",
 ]
 [dependencies]
 agent-client-protocol = { workspace = true, optional = true }
 
 futures = { workspace = true, optional = true }
+serde = { workspace = true, optional = true }
 serde_json = { workspace = true, optional = true }
 thiserror = { workspace = true, optional = true }
 tokio = { workspace = true, optional = true }
@@ -57,6 +60,7 @@ tracing = { workspace = true, optional = true }
 async-stream = { workspace = true, optional = true }
 axum = { workspace = true, features = ["ws", "macros"], optional = true }
 tower-http = { workspace = true, features = ["cors"], optional = true }
+tungstenite = { workspace = true, optional = true }
 uuid = { workspace = true, optional = true }
 
 # Client

src/agent-client-protocol-http/src/connection.rs

@@ -143,6 +143,10 @@ impl Connection {
         self.outbound_transport.subscribe_all_outbound()
     }
 
+    pub(crate) fn enqueue_websocket_text(&self, text: String) -> Result<(), &'static str> {
+        self.outbound_transport.enqueue_websocket_text(text)
+    }
+
     pub(crate) fn subscribe_closed(&self) -> watch::Receiver<bool> {
         self.closed_tx.subscribe()
     }
@@ -154,10 +158,7 @@ impl Connection {
 
     #[cfg(test)]
     pub(crate) fn push_all_outbound_for_test(&self, msg: String) -> Result<(), &'static str> {
-        let OutboundTransport::WebSocket(websocket) = &self.outbound_transport else {
-            return Err("not a WebSocket connection");
-        };
-        websocket.all_outbound.push(msg)
+        self.enqueue_websocket_text(msg)
     }
 
     pub(crate) async fn start_router(self: &Arc<Self>) {
@@ -250,6 +251,13 @@ impl OutboundTransport {
         }
     }
 
+    fn enqueue_websocket_text(&self, text: String) -> Result<(), &'static str> {
+        let Self::WebSocket(websocket) = self else {
+            return Err("not a WebSocket connection");
+        };
+        websocket.all_outbound.push(text)
+    }
+
     #[cfg(test)]
     fn push_connection_stream_for_test(&self, msg: String) -> Result<(), &'static str> {
         let Self::Http(http) = self else {

src/agent-client-protocol-http/src/lib.rs

@@ -16,4 +16,4 @@ mod websocket_server;
 #[cfg(feature = "client")]
 pub use client::{HttpClient, HttpClientError};
 #[cfg(feature = "server")]
-pub use server::{AcpHttpServer, CorsOptions, ServerOptions};
+pub use server::{AcpHttpServer, CorsOptions, ServerOptions, WebSocketLimits};

src/agent-client-protocol-http/src/server.rs

@@ -13,6 +13,73 @@ use tower_http::cors::{AllowOrigin, CorsLayer};
 
 use crate::connection::ConnectionRegistry;
 
+/// Finite resource limits for an ACP WebSocket connection.
+///
+/// Frame and message limits are enforced by the WebSocket transport. The lower
+/// JSON-RPC request limit applies to one complete WebSocket text value, including
+/// an entire JSON-RPC batch. An oversized call, or every response-bearing entry
+/// in an oversized mixed batch, is rejected without forwarding any part of that
+/// text value. Correlated errors leave the physical connection available for
+/// other sessions when the bounded error response fits `max_message_size`;
+/// otherwise the connection closes with WebSocket code 1009.
+/// Limits apply per connection; concurrent connections and later protocol
+/// parsing can multiply total process memory use.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct WebSocketLimits {
+    max_frame_size: usize,
+    max_message_size: usize,
+    max_json_rpc_request_size: usize,
+}
+
+impl WebSocketLimits {
+    /// Creates finite hard frame/message limits and a lower soft request limit.
+    ///
+    /// # Panics
+    ///
+    /// Panics if any limit is zero or if `max_json_rpc_request_size` exceeds
+    /// `max_message_size`.
+    #[must_use]
+    pub const fn new(
+        max_frame_size: usize,
+        max_message_size: usize,
+        max_json_rpc_request_size: usize,
+    ) -> Self {
+        assert!(max_frame_size > 0, "max_frame_size must be positive");
+        assert!(max_message_size > 0, "max_message_size must be positive");
+        assert!(
+            max_json_rpc_request_size > 0,
+            "max_json_rpc_request_size must be positive"
+        );
+        assert!(
+            max_json_rpc_request_size <= max_message_size,
+            "soft request limit must not exceed hard message limit"
+        );
+        Self {
+            max_frame_size,
+            max_message_size,
+            max_json_rpc_request_size,
+        }
+    }
+
+    /// Returns the maximum accepted WebSocket frame size in bytes.
+    #[must_use]
+    pub const fn max_frame_size(self) -> usize {
+        self.max_frame_size
+    }
+
+    /// Returns the maximum accepted reassembled WebSocket message size in bytes.
+    #[must_use]
+    pub const fn max_message_size(self) -> usize {
+        self.max_message_size
+    }
+
+    /// Returns the maximum text value or aggregate batch size before correlated rejection.
+    #[must_use]
+    pub const fn max_json_rpc_request_size(self) -> usize {
+        self.max_json_rpc_request_size
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct ServerOptions {
     pub path: String,
@@ -85,17 +152,20 @@ impl CorsOptions {
 struct ServerState {
     registry: Arc<ConnectionRegistry>,
     cors: CorsOptions,
+    websocket_limits: Option<WebSocketLimits>,
 }
 
 pub struct AcpHttpServer {
     registry: Arc<ConnectionRegistry>,
     options: ServerOptions,
+    websocket_limits: Option<WebSocketLimits>,
 }
 
 impl std::fmt::Debug for AcpHttpServer {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("AcpHttpServer")
             .field("options", &self.options)
+            .field("websocket_limits", &self.websocket_limits)
             .finish_non_exhaustive()
     }
 }
@@ -109,6 +179,7 @@ impl AcpHttpServer {
         Self {
             registry: Arc::new(ConnectionRegistry::new(Arc::new(factory))),
             options: ServerOptions::default(),
+            websocket_limits: None,
         }
     }
 
@@ -118,13 +189,26 @@ impl AcpHttpServer {
         self
     }
 
+    /// Opts into finite WebSocket and JSON-RPC request limits.
+    ///
+    /// Without this builder, the server retains its existing WebSocket defaults.
+    /// For example, call this with
+    /// `WebSocketLimits::new(64 * 1024 * 1024, 64 * 1024 * 1024, 32 * 1024 * 1024)`
+    /// to set 64 MiB hard transport limits and a 32 MiB soft request limit.
+    #[must_use]
+    pub fn with_websocket_limits(mut self, limits: WebSocketLimits) -> Self {
+        self.websocket_limits = Some(limits);
+        self
+    }
+
     pub fn into_router(self) -> Router {
         let registry = self.registry.clone();
         let path = self.options.path.clone();
         let cors = self.options.cors.clone();
         let state = ServerState {
             registry: registry.clone(),
             cors: cors.clone(),
+            websocket_limits: self.websocket_limits,
         };
 
         let mut router = Router::new()
@@ -187,7 +271,13 @@ async fn handle_get(
             {
                 return (StatusCode::FORBIDDEN, "WebSocket origin not allowed").into_response();
             }
-            crate::websocket_server::handle_ws_upgrade(state.registry, ws)
+            let ws = if let Some(limits) = state.websocket_limits {
+                ws.max_frame_size(limits.max_frame_size())
+                    .max_message_size(limits.max_message_size())
+            } else {
+                ws
+            };
+            crate::websocket_server::handle_ws_upgrade(state.registry, ws, state.websocket_limits)
         }
         Err(_) => crate::http_server::handle_get(state.registry, request).await,
     }