From f58e27a1c244d9ca38229228419c425e15dce718 Mon Sep 17 00:00:00 2001 From: Alex Hancock Date: Thu, 23 Apr 2026 15:36:33 -0400 Subject: [PATCH] feat(acp): Align to new request patterns of ACP Streamable HTTP/WS transport (#8605) Signed-off-by: Alex Hancock --- crates/goose/src/acp/transport/connection.rs | 265 +++++++++++ crates/goose/src/acp/transport/http.rs | 449 ++++++++----------- crates/goose/src/acp/transport/mod.rs | 71 ++- crates/goose/src/acp/transport/websocket.rs | 198 ++++---- ui/sdk/src/http-stream.ts | 354 +++++++++++---- 5 files changed, 859 insertions(+), 478 deletions(-) create mode 100644 crates/goose/src/acp/transport/connection.rs diff --git a/crates/goose/src/acp/transport/connection.rs b/crates/goose/src/acp/transport/connection.rs new file mode 100644 index 00000000..3ab9179f --- /dev/null +++ b/crates/goose/src/acp/transport/connection.rs @@ -0,0 +1,265 @@ +//! Connection-level state shared between HTTP and WebSocket transports. +//! +//! Each connection hosts one ACP agent task. All server→client messages for +//! the connection are multicast through a single broadcast channel; HTTP GET +//! SSE streams and WebSocket sinks subscribe to that channel. POSTs (and WS +//! text frames) forward client→server messages into the agent over an mpsc. + +use std::{ + collections::{HashMap, VecDeque}, + sync::Arc, +}; + +use anyhow::Result; +use tokio::sync::{broadcast, mpsc, Mutex, RwLock}; +use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; +use tracing::{error, info, warn}; + +use crate::acp::adapters::{ReceiverToAsyncRead, SenderToAsyncWrite}; +use crate::acp::server_factory::AcpServer; + +/// Broadcast capacity for agent→client messages. Large enough to buffer a +/// typical prompt's streaming notifications even if the subscriber is briefly +/// slow (e.g. during reconnect). +const OUTBOUND_BROADCAST_CAPACITY: usize = 1024; + +/// Maximum number of server→client messages to retain while no subscriber is +/// attached. In the HTTP flow the client opens `GET /acp` only after receiving +/// the initialize response, so any notifications or server-initiated requests +/// emitted by the agent in that window would otherwise be broadcast to zero +/// subscribers and permanently lost. We buffer them here and replay on the +/// first subscribe. On overflow the oldest message is dropped with a warning. +const PRE_SUBSCRIBE_BUFFER_CAPACITY: usize = 1024; + +pub(crate) struct Connection { + /// Send client→server messages into the agent. + pub to_agent_tx: mpsc::Sender, + /// Subscribe here to receive all server→client messages for this connection. + pub outbound_tx: broadcast::Sender, + /// Pulled exactly once during `initialize` to read the synchronous response + /// that must be returned as the HTTP 200 body before any broadcast + /// subscribers exist. `None` once consumed. + pub init_receiver: Mutex>>, + /// Set once the initialize handler has captured the initialize response and + /// handed ownership of the agent output pump over to the broadcast fan-out. + pub init_complete: Mutex, + /// Handle to the agent task; aborted on connection termination. + pub agent_handle: tokio::task::JoinHandle<()>, + /// Handle to the fan-out pump task; aborted on connection termination. + pub pump_handle: Mutex>>, + pre_subscribe_buffer: Arc>>>, +} + +pub(crate) struct ConnectionRegistry { + pub server: Arc, + connections: RwLock>>, +} + +impl ConnectionRegistry { + pub fn new(server: Arc) -> Self { + Self { + server, + connections: RwLock::new(HashMap::new()), + } + } + + /// Create a new connection, spawn the ACP agent task, and return + /// (connection_id, connection). The initialize request body should be sent + /// via `connection.to_agent_tx` and the synchronous initialize response + /// read via `consume_initialize_response`. + pub async fn create_connection(&self) -> Result<(String, Arc)> { + let (to_agent_tx, to_agent_rx) = mpsc::channel::(256); + let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::(); + let (outbound_tx, _) = broadcast::channel::(OUTBOUND_BROADCAST_CAPACITY); + + let agent = self.server.create_agent().await?; + let connection_id = uuid::Uuid::new_v4().to_string(); + + let read_stream = ReceiverToAsyncRead::new(to_agent_rx); + let write_stream = SenderToAsyncWrite::new(from_agent_tx); + let fut = + crate::acp::server::serve(agent, read_stream.compat(), write_stream.compat_write()); + + let conn_id_for_task = connection_id.clone(); + let agent_handle = tokio::spawn(async move { + if let Err(e) = fut.await { + error!(connection_id = %conn_id_for_task, "ACP agent task error: {}", e); + } + }); + + let connection = Arc::new(Connection { + to_agent_tx, + outbound_tx, + init_receiver: Mutex::new(Some(from_agent_rx)), + init_complete: Mutex::new(false), + agent_handle, + pump_handle: Mutex::new(None), + pre_subscribe_buffer: Arc::new(Mutex::new(Some(VecDeque::new()))), + }); + + self.connections + .write() + .await + .insert(connection_id.clone(), connection.clone()); + + info!(connection_id = %connection_id, "Connection created"); + Ok((connection_id, connection)) + } + + pub async fn get(&self, connection_id: &str) -> Option> { + self.connections.read().await.get(connection_id).cloned() + } + + pub async fn remove(&self, connection_id: &str) -> Option> { + self.connections.write().await.remove(connection_id) + } +} + +impl Connection { + /// After the synchronous initialize response has been consumed, spawn a + /// task that forwards all remaining agent output to the broadcast channel. + /// Idempotent. + pub async fn start_fanout(self: &Arc) { + let mut complete = self.init_complete.lock().await; + if *complete { + return; + } + let Some(mut rx) = self.init_receiver.lock().await.take() else { + return; + }; + let outbound_tx = self.outbound_tx.clone(); + let buffer = self.pre_subscribe_buffer.clone(); + let handle = tokio::spawn(async move { + while let Some(msg) = rx.recv().await { + let mut buf_guard = buffer.lock().await; + match buf_guard.as_mut() { + Some(buf) => { + if buf.len() >= PRE_SUBSCRIBE_BUFFER_CAPACITY { + warn!( + "Pre-subscribe buffer full ({} messages); dropping oldest", + PRE_SUBSCRIBE_BUFFER_CAPACITY + ); + buf.pop_front(); + } + buf.push_back(msg); + } + None => { + drop(buf_guard); + let _ = outbound_tx.send(msg); + } + } + } + }); + *self.pump_handle.lock().await = Some(handle); + *complete = true; + } + + pub async fn subscribe_with_replay(&self) -> (Vec, broadcast::Receiver) { + let mut guard = self.pre_subscribe_buffer.lock().await; + let receiver = self.outbound_tx.subscribe(); + let replay = guard.take().map(Vec::from).unwrap_or_default(); + (replay, receiver) + } + + /// Terminate the connection: abort the agent task and the fan-out pump. + pub async fn shutdown(&self) { + self.agent_handle.abort(); + if let Some(h) = self.pump_handle.lock().await.take() { + h.abort(); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::time::Duration; + use tokio::time::timeout; + + fn fake_connection() -> (Arc, mpsc::UnboundedSender) { + let (to_agent_tx, _to_agent_rx) = mpsc::channel::(256); + let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::(); + let (outbound_tx, _) = broadcast::channel::(OUTBOUND_BROADCAST_CAPACITY); + + let agent_handle = tokio::spawn(async { + std::future::pending::<()>().await; + }); + + let connection = Arc::new(Connection { + to_agent_tx, + outbound_tx, + init_receiver: Mutex::new(Some(from_agent_rx)), + init_complete: Mutex::new(false), + agent_handle, + pump_handle: Mutex::new(None), + pre_subscribe_buffer: Arc::new(Mutex::new(Some(VecDeque::new()))), + }); + + (connection, from_agent_tx) + } + + #[tokio::test] + async fn buffers_messages_emitted_before_first_subscribe() { + let (conn, agent_tx) = fake_connection(); + conn.start_fanout().await; + + agent_tx.send("one".to_string()).unwrap(); + agent_tx.send("two".to_string()).unwrap(); + agent_tx.send("three".to_string()).unwrap(); + + tokio::task::yield_now().await; + tokio::time::sleep(Duration::from_millis(20)).await; + + let (replay, _rx) = conn.subscribe_with_replay().await; + assert_eq!(replay, vec!["one", "two", "three"]); + + conn.shutdown().await; + } + + #[tokio::test] + async fn switches_to_live_broadcast_after_subscribe() { + let (conn, agent_tx) = fake_connection(); + conn.start_fanout().await; + + let (replay, mut rx) = conn.subscribe_with_replay().await; + assert!(replay.is_empty()); + + agent_tx.send("live-one".to_string()).unwrap(); + agent_tx.send("live-two".to_string()).unwrap(); + + let got1 = timeout(Duration::from_secs(1), rx.recv()) + .await + .unwrap() + .unwrap(); + let got2 = timeout(Duration::from_secs(1), rx.recv()) + .await + .unwrap() + .unwrap(); + assert_eq!(got1, "live-one"); + assert_eq!(got2, "live-two"); + + conn.shutdown().await; + } + + #[tokio::test] + async fn pre_subscribe_buffer_is_bounded() { + let (conn, agent_tx) = fake_connection(); + conn.start_fanout().await; + + for i in 0..(PRE_SUBSCRIBE_BUFFER_CAPACITY + 50) { + agent_tx.send(format!("m{}", i)).unwrap(); + } + + tokio::time::sleep(Duration::from_millis(50)).await; + + let (replay, _rx) = conn.subscribe_with_replay().await; + assert_eq!(replay.len(), PRE_SUBSCRIBE_BUFFER_CAPACITY); + assert_eq!( + replay.last().unwrap(), + &format!("m{}", PRE_SUBSCRIBE_BUFFER_CAPACITY + 49) + ); + assert_eq!(replay.first().unwrap(), &format!("m{}", 50)); + + conn.shutdown().await; + } +} diff --git a/crates/goose/src/acp/transport/http.rs b/crates/goose/src/acp/transport/http.rs index 4359a21c..3694cfbe 100644 --- a/crates/goose/src/acp/transport/http.rs +++ b/crates/goose/src/acp/transport/http.rs @@ -1,200 +1,30 @@ -use anyhow::Result; +use std::{convert::Infallible, sync::Arc, time::Duration}; + use axum::{ body::Body, extract::State, - http::{Request, StatusCode}, + http::{HeaderValue, Request, StatusCode}, response::{IntoResponse, Response, Sse}, }; use http_body_util::BodyExt; use serde_json::Value; -use std::{collections::HashMap, convert::Infallible, sync::Arc, time::Duration}; -use tokio::sync::{mpsc, Mutex, RwLock}; -use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; -use tracing::{error, info}; +use tokio::sync::broadcast; +use tracing::{debug, error, info, trace}; +use super::connection::{Connection, ConnectionRegistry}; use super::*; -use crate::acp::adapters::{ReceiverToAsyncRead, SenderToAsyncWrite}; -use crate::acp::server_factory::AcpServer; - -pub(crate) struct HttpState { - server: Arc, - // Keyed by acp_session_id: a connection-scoped UUID serving many Goose sessions. - sessions: RwLock>, -} - -impl HttpState { - pub fn new(server: Arc) -> Self { - Self { - server, - sessions: RwLock::new(HashMap::new()), - } - } - - async fn create_session(&self) -> Result { - let (to_agent_tx, to_agent_rx) = mpsc::channel::(256); - let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::(); - - let agent = self.server.create_agent().await.map_err(|e| { - error!("Failed to create agent: {}", e); - StatusCode::INTERNAL_SERVER_ERROR - })?; - - let acp_session_id = uuid::Uuid::new_v4().to_string(); - - let read_stream = ReceiverToAsyncRead::new(to_agent_rx); - let write_stream = SenderToAsyncWrite::new(from_agent_tx); - let fut = - crate::acp::server::serve(agent, read_stream.compat(), write_stream.compat_write()); - let handle = tokio::spawn(async move { - if let Err(e) = fut.await { - error!("ACP session error: {}", e); - } - }); - - self.sessions.write().await.insert( - acp_session_id.clone(), - TransportSession { - to_agent_tx, - from_agent_rx: Arc::new(Mutex::new(from_agent_rx)), - handle, - }, - ); - - info!(acp_session_id = %acp_session_id, "Session created"); - Ok(acp_session_id) - } - - async fn has_session(&self, acp_session_id: &str) -> bool { - self.sessions.read().await.contains_key(acp_session_id) - } - - async fn remove_session(&self, acp_session_id: &str) { - if let Some(session) = self.sessions.write().await.remove(acp_session_id) { - session.handle.abort(); - info!(acp_session_id = %acp_session_id, "Session removed"); - } - } - - async fn send_message(&self, acp_session_id: &str, message: String) -> Result<(), StatusCode> { - let sessions = self.sessions.read().await; - let session = sessions.get(acp_session_id).ok_or(StatusCode::NOT_FOUND)?; - session - .to_agent_tx - .send(message) - .await - .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR) - } - - async fn get_receiver( - &self, - acp_session_id: &str, - ) -> Result>>, StatusCode> { - let sessions = self.sessions.read().await; - let session = sessions.get(acp_session_id).ok_or(StatusCode::NOT_FOUND)?; - Ok(session.from_agent_rx.clone()) - } -} - -fn create_sse_stream( - receiver: Arc>>, - cleanup: Option<(Arc, String)>, -) -> Sse>> { - let stream = async_stream::stream! { - let mut rx = receiver.lock().await; - while let Some(msg) = rx.recv().await { - yield Ok::<_, Infallible>(axum::response::sse::Event::default().data(msg)); - } - if let Some((state, acp_session_id)) = cleanup { - state.remove_session(&acp_session_id).await; - } - }; - - Sse::new(stream).keep_alive( - axum::response::sse::KeepAlive::new() - .interval(Duration::from_secs(15)) - .text(""), - ) -} - -async fn handle_initialize(state: Arc, json_message: &Value) -> Response { - let acp_session_id = match state.create_session().await { - Ok(id) => id, - Err(status) => return status.into_response(), - }; - - let message_str = serde_json::to_string(json_message).unwrap(); - if let Err(status) = state.send_message(&acp_session_id, message_str).await { - state.remove_session(&acp_session_id).await; - return status.into_response(); - } - - let receiver = match state.get_receiver(&acp_session_id).await { - Ok(r) => r, - Err(status) => { - state.remove_session(&acp_session_id).await; - return status.into_response(); - } - }; - - let sse = create_sse_stream(receiver, Some((state.clone(), acp_session_id.clone()))); - let mut response = sse.into_response(); - response - .headers_mut() - .insert(HEADER_SESSION_ID, acp_session_id.parse().unwrap()); - response -} - -async fn handle_request( - state: Arc, - acp_session_id: String, - json_message: &Value, -) -> Response { - if !state.has_session(&acp_session_id).await { - return (StatusCode::NOT_FOUND, "Session not found").into_response(); - } - - let message_str = serde_json::to_string(json_message).unwrap(); - if let Err(status) = state.send_message(&acp_session_id, message_str).await { - return status.into_response(); - } - - let receiver = match state.get_receiver(&acp_session_id).await { - Ok(r) => r, - Err(status) => return status.into_response(), - }; - - create_sse_stream(receiver, None).into_response() -} - -async fn handle_notification_or_response( - state: Arc, - acp_session_id: String, - json_message: &Value, -) -> Response { - if !state.has_session(&acp_session_id).await { - return (StatusCode::NOT_FOUND, "Session not found").into_response(); - } - - let message_str = serde_json::to_string(json_message).unwrap(); - if let Err(status) = state.send_message(&acp_session_id, message_str).await { - return status.into_response(); - } - - StatusCode::ACCEPTED.into_response() -} +/// POST /acp +/// +/// - `initialize`: creates a new connection, forwards the request, waits for +/// the synchronous initialize response from the agent, and returns it as a +/// 200 OK JSON body with the `Acp-Connection-Id` header set. +/// - All other messages: require `Acp-Connection-Id` (and `Acp-Session-Id` +/// for session-scoped methods), forward to the agent, return 202 Accepted. pub(crate) async fn handle_post( - State(state): State>, + State(registry): State>, request: Request, ) -> Response { - if !accepts_json_and_sse(&request) { - return ( - StatusCode::NOT_ACCEPTABLE, - "Not Acceptable: Client must accept both application/json and text/event-stream", - ) - .into_response(); - } - if !content_type_is_json(&request) { return ( StatusCode::UNSUPPORTED_MEDIA_TYPE, @@ -203,7 +33,8 @@ pub(crate) async fn handle_post( .into_response(); } - let acp_session_id = get_session_id(&request); + let connection_id = header_value(&request, HEADER_CONNECTION_ID); + let session_id = header_value(&request, HEADER_SESSION_ID); let body_bytes = match request.into_body().collect().await { Ok(collected) => collected.to_bytes(), @@ -216,7 +47,6 @@ pub(crate) async fn handle_post( let json_message: Value = match serde_json::from_slice(&body_bytes) { Ok(v) => v, Err(e) => { - error!("Failed to parse JSON: {}", e); return (StatusCode::BAD_REQUEST, format!("Invalid JSON: {}", e)).into_response(); } }; @@ -230,31 +60,128 @@ pub(crate) async fn handle_post( } if is_initialize_request(&json_message) { - handle_initialize(state.clone(), &json_message).await - } else if is_jsonrpc_request(&json_message) { - let Some(id) = acp_session_id else { - return ( - StatusCode::BAD_REQUEST, - "Bad Request: Acp-Session-Id header required", - ) - .into_response(); - }; - handle_request(state.clone(), id, &json_message).await - } else if is_jsonrpc_notification(&json_message) || is_jsonrpc_response(&json_message) { - let Some(id) = acp_session_id else { - return ( - StatusCode::BAD_REQUEST, - "Bad Request: Acp-Session-Id header required", - ) - .into_response(); - }; - handle_notification_or_response(state.clone(), id, &json_message).await - } else { - (StatusCode::BAD_REQUEST, "Invalid JSON-RPC message").into_response() + return handle_initialize(registry, json_message).await; } + + let Some(connection_id) = connection_id else { + return ( + StatusCode::BAD_REQUEST, + "Bad Request: Acp-Connection-Id header required", + ) + .into_response(); + }; + + let Some(connection) = registry.get(&connection_id).await else { + return (StatusCode::NOT_FOUND, "Unknown Acp-Connection-Id").into_response(); + }; + + if let Some(method) = json_message.get("method").and_then(|m| m.as_str()) { + if method_requires_session_header(method) && session_id.is_none() { + return ( + StatusCode::BAD_REQUEST, + "Bad Request: Acp-Session-Id header required for session-scoped methods", + ) + .into_response(); + } + } + + if !is_jsonrpc_request_with_id(&json_message) + && !is_jsonrpc_notification(&json_message) + && !is_jsonrpc_response(&json_message) + { + return (StatusCode::BAD_REQUEST, "Invalid JSON-RPC message").into_response(); + } + + let message_str = serde_json::to_string(&json_message).unwrap(); + trace!(connection_id = %connection_id, payload = %message_str, "POST → agent"); + if connection.to_agent_tx.send(message_str).await.is_err() { + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "Failed to forward message to agent", + ) + .into_response(); + } + + StatusCode::ACCEPTED.into_response() } -pub(crate) async fn handle_get(state: Arc, request: Request) -> Response { +async fn handle_initialize(registry: Arc, json_message: Value) -> Response { + let (connection_id, connection) = match registry.create_connection().await { + Ok(pair) => pair, + Err(e) => { + error!("Failed to create connection: {}", e); + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "Failed to create connection", + ) + .into_response(); + } + }; + + let message_str = serde_json::to_string(&json_message).unwrap(); + trace!(connection_id = %connection_id, payload = %message_str, "initialize → agent"); + if connection.to_agent_tx.send(message_str).await.is_err() { + registry.remove(&connection_id).await; + connection.shutdown().await; + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "Failed to forward initialize to agent", + ) + .into_response(); + } + + // Read exactly one message from the agent: the initialize response. + let init_response = { + let mut guard = connection.init_receiver.lock().await; + let Some(rx) = guard.as_mut() else { + registry.remove(&connection_id).await; + connection.shutdown().await; + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "Initialize receiver already consumed", + ) + .into_response(); + }; + rx.recv().await + }; + + let init_response = match init_response { + Some(msg) => msg, + None => { + registry.remove(&connection_id).await; + connection.shutdown().await; + return ( + StatusCode::INTERNAL_SERVER_ERROR, + "Agent closed before initialize response", + ) + .into_response(); + } + }; + + connection.start_fanout().await; + + let mut response = ( + StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, JSON_MIME_TYPE)], + init_response, + ) + .into_response(); + if let Ok(v) = HeaderValue::from_str(&connection_id) { + response.headers_mut().insert(HEADER_CONNECTION_ID, v); + } + info!(connection_id = %connection_id, "Initialize complete"); + response +} + +/// GET /acp (no Upgrade) +/// +/// Opens the single long-lived SSE stream for a connection. All server→client +/// messages (responses + notifications + server-initiated requests) are +/// delivered here, correlated by their JSON-RPC body fields. +pub(crate) async fn handle_get( + registry: Arc, + request: Request, +) -> Response { if !accepts_mime_type(&request, EVENT_STREAM_MIME_TYPE) { return ( StatusCode::NOT_ACCEPTABLE, @@ -263,61 +190,77 @@ pub(crate) async fn handle_get(state: Arc, request: Request) -> .into_response(); } - let acp_session_id = match get_session_id(&request) { - Some(id) => id, - None => { - return ( - StatusCode::BAD_REQUEST, - "Bad Request: Acp-Session-Id header required", - ) - .into_response(); - } + let Some(connection_id) = header_value(&request, HEADER_CONNECTION_ID) else { + return ( + StatusCode::BAD_REQUEST, + "Bad Request: Acp-Connection-Id header required", + ) + .into_response(); }; - if !state.has_session(&acp_session_id).await { - return (StatusCode::NOT_FOUND, "Session not found").into_response(); + let Some(connection) = registry.get(&connection_id).await else { + return (StatusCode::NOT_FOUND, "Unknown Acp-Connection-Id").into_response(); + }; + + let (replay, receiver) = connection.subscribe_with_replay().await; + let sse = build_sse_stream(connection.clone(), replay, receiver); + + let mut response = sse.into_response(); + if let Ok(v) = HeaderValue::from_str(&connection_id) { + response.headers_mut().insert(HEADER_CONNECTION_ID, v); } + response +} - let receiver = match state.get_receiver(&acp_session_id).await { - Ok(r) => r, - Err(status) => return status.into_response(), - }; - +fn build_sse_stream( + _connection: Arc, + replay: Vec, + mut receiver: broadcast::Receiver, +) -> Sse>> { let stream = async_stream::stream! { - let mut rx = receiver.lock().await; - while let Some(msg) = rx.recv().await { + for msg in replay { + trace!(payload = %msg, "SSE → client (replay)"); yield Ok::<_, Infallible>(axum::response::sse::Event::default().data(msg)); } - }; - - Sse::new(stream) - .keep_alive( - axum::response::sse::KeepAlive::new() - .interval(Duration::from_secs(15)) - .text(""), - ) - .into_response() -} - -pub(crate) async fn handle_delete( - State(state): State>, - request: Request, -) -> Response { - let acp_session_id = match get_session_id(&request) { - Some(id) => id, - None => { - return ( - StatusCode::BAD_REQUEST, - "Bad Request: Acp-Session-Id header required", - ) - .into_response(); + loop { + match receiver.recv().await { + Ok(msg) => { + trace!(payload = %msg, "SSE → client"); + yield Ok::<_, Infallible>(axum::response::sse::Event::default().data(msg)); + } + Err(broadcast::error::RecvError::Lagged(n)) => { + debug!("SSE subscriber lagged {} messages", n); + continue; + } + Err(broadcast::error::RecvError::Closed) => break, + } } }; - if !state.has_session(&acp_session_id).await { - return (StatusCode::NOT_FOUND, "Session not found").into_response(); - } + Sse::new(stream).keep_alive( + axum::response::sse::KeepAlive::new() + .interval(Duration::from_secs(15)) + .text(""), + ) +} - state.remove_session(&acp_session_id).await; +/// DELETE /acp +pub(crate) async fn handle_delete( + State(registry): State>, + request: Request, +) -> Response { + let Some(connection_id) = header_value(&request, HEADER_CONNECTION_ID) else { + return ( + StatusCode::BAD_REQUEST, + "Bad Request: Acp-Connection-Id header required", + ) + .into_response(); + }; + + let Some(connection) = registry.remove(&connection_id).await else { + return (StatusCode::NOT_FOUND, "Unknown Acp-Connection-Id").into_response(); + }; + connection.shutdown().await; + info!(connection_id = %connection_id, "Connection terminated via DELETE"); StatusCode::ACCEPTED.into_response() } diff --git a/crates/goose/src/acp/transport/mod.rs b/crates/goose/src/acp/transport/mod.rs index 3cb1c67a..a9038431 100644 --- a/crates/goose/src/acp/transport/mod.rs +++ b/crates/goose/src/acp/transport/mod.rs @@ -1,3 +1,4 @@ +pub mod connection; pub mod http; pub mod websocket; @@ -9,27 +10,21 @@ use axum::{ ws::{rejection::WebSocketUpgradeRejection, WebSocketUpgrade}, State, }, - http::{header, Method, Request}, + http::{header, HeaderName, Method, Request}, response::Response, routing::{delete, get, post}, Router, }; use serde_json::Value; -use tokio::sync::{mpsc, Mutex}; use tower_http::cors::{Any, CorsLayer}; use crate::acp::server_factory::AcpServer; +pub(crate) const HEADER_CONNECTION_ID: &str = "Acp-Connection-Id"; pub(crate) const HEADER_SESSION_ID: &str = "Acp-Session-Id"; pub(crate) const EVENT_STREAM_MIME_TYPE: &str = "text/event-stream"; pub(crate) const JSON_MIME_TYPE: &str = "application/json"; -pub(crate) struct TransportSession { - pub to_agent_tx: mpsc::Sender, - pub from_agent_rx: Arc>>, - pub handle: tokio::task::JoinHandle<()>, -} - pub(crate) fn accepts_mime_type(request: &Request, mime_type: &str) -> bool { request .headers() @@ -38,16 +33,6 @@ pub(crate) fn accepts_mime_type(request: &Request, mime_type: &str) -> boo .is_some_and(|accept| accept.contains(mime_type)) } -pub(crate) fn accepts_json_and_sse(request: &Request) -> bool { - request - .headers() - .get(axum::http::header::ACCEPT) - .and_then(|v| v.to_str().ok()) - .is_some_and(|accept| { - accept.contains(JSON_MIME_TYPE) && accept.contains(EVENT_STREAM_MIME_TYPE) - }) -} - pub(crate) fn content_type_is_json(request: &Request) -> bool { request .headers() @@ -56,15 +41,15 @@ pub(crate) fn content_type_is_json(request: &Request) -> bool { .is_some_and(|ct| ct.starts_with(JSON_MIME_TYPE)) } -pub(crate) fn get_session_id(request: &Request) -> Option { +pub(crate) fn header_value(request: &Request, name: &str) -> Option { request .headers() - .get(HEADER_SESSION_ID) + .get(name) .and_then(|v| v.to_str().ok()) .map(|s| s.to_string()) } -pub(crate) fn is_jsonrpc_request(value: &Value) -> bool { +pub(crate) fn is_jsonrpc_request_with_id(value: &Value) -> bool { value.get("method").is_some() && value.get("id").is_some() } @@ -73,21 +58,35 @@ pub(crate) fn is_jsonrpc_notification(value: &Value) -> bool { } pub(crate) fn is_jsonrpc_response(value: &Value) -> bool { - value.get("id").is_some() && (value.get("result").is_some() || value.get("error").is_some()) + value.get("id").is_some() + && value.get("method").is_none() + && (value.get("result").is_some() || value.get("error").is_some()) } pub(crate) fn is_initialize_request(value: &Value) -> bool { value.get("method").is_some_and(|m| m == "initialize") && value.get("id").is_some() } +/// Methods that are scoped to a session and require an Acp-Session-Id header. +pub(crate) fn method_requires_session_header(method: &str) -> bool { + matches!( + method, + "session/prompt" + | "session/cancel" + | "session/load" + | "session/set_mode" + | "session/set_model" + ) +} + async fn handle_get( ws_upgrade: Result, - State(state): State<(Arc, Arc)>, + State(state): State>, request: Request, ) -> Response { match ws_upgrade { - Ok(ws) => websocket::handle_get(state.1, ws).await, - Err(_) => http::handle_get(state.0, request).await, + Ok(ws) => websocket::handle_ws_upgrade(state, ws).await, + Err(_) => http::handle_get(state, request).await, } } @@ -96,8 +95,7 @@ async fn health() -> &'static str { } pub fn create_router(server: Arc) -> Router { - let http_state = Arc::new(http::HttpState::new(server.clone())); - let ws_state = Arc::new(websocket::WsState::new(server)); + let registry = Arc::new(connection::ConnectionRegistry::new(server)); let cors = CorsLayer::new() .allow_origin(Any) @@ -105,24 +103,23 @@ pub fn create_router(server: Arc) -> Router { .allow_headers([ header::CONTENT_TYPE, header::ACCEPT, - HEADER_SESSION_ID.parse().unwrap(), + HeaderName::from_static("acp-connection-id"), + HeaderName::from_static("acp-session-id"), header::SEC_WEBSOCKET_VERSION, header::SEC_WEBSOCKET_KEY, header::CONNECTION, header::UPGRADE, + ]) + .expose_headers([ + HeaderName::from_static("acp-connection-id"), + HeaderName::from_static("acp-session-id"), ]); Router::new() .route("/health", get(health)) .route("/status", get(health)) - .route( - "/acp", - post(http::handle_post).with_state(http_state.clone()), - ) - .route( - "/acp", - get(handle_get).with_state((http_state.clone(), ws_state)), - ) - .route("/acp", delete(http::handle_delete).with_state(http_state)) + .route("/acp", post(http::handle_post).with_state(registry.clone())) + .route("/acp", get(handle_get).with_state(registry.clone())) + .route("/acp", delete(http::handle_delete).with_state(registry)) .layer(cors) } diff --git a/crates/goose/src/acp/transport/websocket.rs b/crates/goose/src/acp/transport/websocket.rs index 2a548630..752cb320 100644 --- a/crates/goose/src/acp/transport/websocket.rs +++ b/crates/goose/src/acp/transport/websocket.rs @@ -1,75 +1,29 @@ -use anyhow::Result; +use std::sync::Arc; + use axum::{ extract::ws::{Message, WebSocket, WebSocketUpgrade}, - http::StatusCode, + http::{HeaderValue, StatusCode}, response::{IntoResponse, Response}, }; use futures::{SinkExt, StreamExt}; -use std::{collections::HashMap, sync::Arc}; -use tokio::sync::{mpsc, Mutex, RwLock}; -use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt}; -use tracing::{debug, error, info, warn}; +use tracing::{debug, error, info, trace, warn}; -use super::{TransportSession, HEADER_SESSION_ID}; -use crate::acp::adapters::{ReceiverToAsyncRead, SenderToAsyncWrite}; -use crate::acp::server_factory::AcpServer; +use super::connection::ConnectionRegistry; +use super::HEADER_CONNECTION_ID; -pub(crate) struct WsState { - server: Arc, - // Keyed by acp_session_id: a connection-scoped UUID serving many Goose sessions. - sessions: RwLock>, -} - -impl WsState { - pub fn new(server: Arc) -> Self { - Self { - server, - sessions: RwLock::new(HashMap::new()), - } - } - - async fn create_connection(&self) -> Result { - let (to_agent_tx, to_agent_rx) = mpsc::channel::(256); - let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::(); - - let agent = self.server.create_agent().await?; - - let acp_session_id = uuid::Uuid::new_v4().to_string(); - - let read_stream = ReceiverToAsyncRead::new(to_agent_rx); - let write_stream = SenderToAsyncWrite::new(from_agent_tx); - let fut = - crate::acp::server::serve(agent, read_stream.compat(), write_stream.compat_write()); - let handle = tokio::spawn(async move { - if let Err(e) = fut.await { - error!("ACP WebSocket session error: {}", e); - } - }); - - self.sessions.write().await.insert( - acp_session_id.clone(), - TransportSession { - to_agent_tx, - from_agent_rx: Arc::new(Mutex::new(from_agent_rx)), - handle, - }, - ); - - info!(acp_session_id = %acp_session_id, "WebSocket connection created"); - Ok(acp_session_id) - } - - async fn remove_connection(&self, acp_session_id: &str) { - if let Some(session) = self.sessions.write().await.remove(acp_session_id) { - session.handle.abort(); - info!(acp_session_id = %acp_session_id, "WebSocket connection removed"); - } - } -} - -pub(crate) async fn handle_get(state: Arc, ws: WebSocketUpgrade) -> Response { - let acp_session_id = match state.create_connection().await { - Ok(id) => id, +/// GET /acp with `Upgrade: websocket` +/// +/// Creates a new connection (same lifecycle as Streamable HTTP), upgrades to a +/// WebSocket, and runs a bidirectional message loop. The client still sends +/// `initialize` as the first WS text frame — unlike the HTTP path, the +/// initialize response is streamed back over the same WebSocket rather than +/// returned synchronously. +pub(crate) async fn handle_ws_upgrade( + registry: Arc, + ws: WebSocketUpgrade, +) -> Response { + let (connection_id, connection) = match registry.create_connection().await { + Ok(pair) => pair, Err(e) => { error!("Failed to create WebSocket connection: {}", e); return ( @@ -80,80 +34,102 @@ pub(crate) async fn handle_get(state: Arc, ws: WebSocketUpgrade) -> Res } }; - let mut response = ws.on_upgrade({ - let acp_session_id = acp_session_id.clone(); - move |socket| handle_ws(socket, state, acp_session_id) + // WebSocket does not need the synchronous initialize split — start the + // broadcast fan-out immediately so the WS sink reads from the same stream + // of server→client messages as any HTTP SSE subscribers would. + connection.start_fanout().await; + + let conn_id_for_handler = connection_id.clone(); + let registry_for_handler = registry.clone(); + let mut response = ws.on_upgrade(move |socket| async move { + run_ws( + socket, + registry_for_handler, + conn_id_for_handler, + connection, + ) + .await }); - response - .headers_mut() - .insert(HEADER_SESSION_ID, acp_session_id.parse().unwrap()); + + if let Ok(v) = HeaderValue::from_str(&connection_id) { + response.headers_mut().insert(HEADER_CONNECTION_ID, v); + } + info!(connection_id = %connection_id, "WebSocket connection created"); response } -pub(crate) async fn handle_ws(socket: WebSocket, state: Arc, acp_session_id: String) { +async fn run_ws( + socket: WebSocket, + registry: Arc, + connection_id: String, + connection: Arc, +) { let (mut ws_tx, mut ws_rx) = socket.split(); + let (replay, mut outbound_rx) = connection.subscribe_with_replay().await; - let (to_agent, from_agent) = { - let sessions = state.sessions.read().await; - match sessions.get(&acp_session_id) { - Some(session) => (session.to_agent_tx.clone(), session.from_agent_rx.clone()), - None => { - error!(acp_session_id = %acp_session_id, "Session not found after creation"); - return; + debug!(connection_id = %connection_id, "Starting WebSocket message loop"); + + for text in replay { + trace!(connection_id = %connection_id, payload = %text, "Agent → Client (replay): {} bytes", text.len()); + if ws_tx.send(Message::Text(text.into())).await.is_err() { + error!(connection_id = %connection_id, "WebSocket send failed during replay"); + if let Some(conn) = registry.remove(&connection_id).await { + conn.shutdown().await; } + return; } - }; - - debug!(acp_session_id = %acp_session_id, "Starting bidirectional message loop"); - - let mut from_agent_rx = from_agent.lock().await; + } loop { tokio::select! { - Some(msg_result) = ws_rx.next() => { + msg_result = ws_rx.next() => { match msg_result { - Ok(Message::Text(text)) => { + Some(Ok(Message::Text(text))) => { let text_str = text.to_string(); - debug!(acp_session_id = %acp_session_id, "Client → Agent: {} bytes", text_str.len()); - if let Err(e) = to_agent.send(text_str).await { - error!(acp_session_id = %acp_session_id, "Failed to send to agent: {}", e); + trace!(connection_id = %connection_id, payload = %text_str, "Client → Agent: {} bytes", text_str.len()); + if connection.to_agent_tx.send(text_str).await.is_err() { + error!(connection_id = %connection_id, "Agent channel closed"); break; } } - Ok(Message::Close(frame)) => { - debug!(acp_session_id = %acp_session_id, "Client closed connection: {:?}", frame); + Some(Ok(Message::Close(frame))) => { + debug!(connection_id = %connection_id, "Client closed connection: {:?}", frame); break; } - Ok(Message::Ping(_)) | Ok(Message::Pong(_)) => { - // Axum handles ping/pong automatically + Some(Ok(Message::Ping(_))) | Some(Ok(Message::Pong(_))) => continue, + Some(Ok(Message::Binary(_))) => { + warn!(connection_id = %connection_id, "Ignoring binary message (ACP uses text)"); continue; } - Ok(Message::Binary(_)) => { - warn!(acp_session_id = %acp_session_id, "Ignoring binary message (ACP uses text)"); - continue; - } - Err(e) => { - error!(acp_session_id = %acp_session_id, "WebSocket error: {}", e); + Some(Err(e)) => { + error!(connection_id = %connection_id, "WebSocket error: {}", e); break; } + None => break, } } - Some(text) = from_agent_rx.recv() => { - debug!(acp_session_id = %acp_session_id, "Agent → Client: {} bytes", text.len()); - if let Err(e) = ws_tx.send(Message::Text(text.into())).await { - error!(acp_session_id = %acp_session_id, "Failed to send to client: {}", e); - break; + recv = outbound_rx.recv() => { + match recv { + Ok(text) => { + trace!(connection_id = %connection_id, payload = %text, "Agent → Client: {} bytes", text.len()); + if ws_tx.send(Message::Text(text.into())).await.is_err() { + error!(connection_id = %connection_id, "WebSocket send failed"); + break; + } + } + Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => { + warn!(connection_id = %connection_id, "WebSocket lagged {} messages", n); + continue; + } + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, } } - - else => { - debug!(acp_session_id = %acp_session_id, "Both channels closed"); - break; - } } } - debug!(acp_session_id = %acp_session_id, "Cleaning up connection"); - state.remove_connection(&acp_session_id).await; + debug!(connection_id = %connection_id, "Cleaning up WebSocket connection"); + if let Some(conn) = registry.remove(&connection_id).await { + conn.shutdown().await; + } } diff --git a/ui/sdk/src/http-stream.ts b/ui/sdk/src/http-stream.ts index c9ebfb6e..e934807d 100644 --- a/ui/sdk/src/http-stream.ts +++ b/ui/sdk/src/http-stream.ts @@ -1,27 +1,132 @@ import type { AnyMessage, Stream } from "@agentclientprotocol/sdk"; +const ACP_CONNECTION_HEADER = "Acp-Connection-Id"; const ACP_SESSION_HEADER = "Acp-Session-Id"; +// Enable via `globalThis.ACP_DEBUG = true`, `localStorage.ACP_DEBUG = "1"`, +// or `ACP_DEBUG=1` in the environment. +function acpDebug(label: string, payload: unknown): void { + const g = globalThis as { + ACP_DEBUG?: unknown; + localStorage?: { getItem?: (k: string) => string | null }; + process?: { env?: Record }; + }; + const on = + g.ACP_DEBUG === true || + g.ACP_DEBUG === "1" || + !!g.localStorage?.getItem?.("ACP_DEBUG") || + !!g.process?.env?.ACP_DEBUG; + if (!on) return; + // eslint-disable-next-line no-console + console.debug(`[acp] ${label}`, payload); +} + +// Methods that are scoped to a session and require an Acp-Session-Id header. +const SESSION_SCOPED_METHODS = new Set([ + "session/prompt", + "session/cancel", + "session/load", + "session/set_mode", + "session/set_model", +]); + +function messageMethod(msg: AnyMessage): string | null { + const m = msg as { method?: unknown }; + return typeof m.method === "string" ? m.method : null; +} + +function messageParams(msg: AnyMessage): unknown { + return (msg as { params?: unknown }).params; +} + +function isRequest(msg: AnyMessage): boolean { + const m = msg as { method?: unknown; id?: unknown }; + return typeof m.method === "string" && m.id !== undefined && m.id !== null; +} + +function isNotification(msg: AnyMessage): boolean { + const m = msg as { method?: unknown; id?: unknown }; + return typeof m.method === "string" && (m.id === undefined || m.id === null); +} + +function extractSessionId(value: unknown): string | null { + if (value && typeof value === "object" && "sessionId" in value) { + const sid = (value as { sessionId?: unknown }).sessionId; + if (typeof sid === "string") return sid; + } + return null; +} + +/** + * Create a Stream that speaks the Streamable HTTP ACP transport. + * + * Protocol summary: + * - The first outbound message must be an `initialize` request, sent as a + * regular POST. The server responds synchronously with 200 OK, a JSON body + * containing the initialize response, and an `Acp-Connection-Id` header. + * - After initialize, we open a single long-lived GET SSE stream carrying + * all server → client messages (responses, notifications, server-initiated + * requests) for every session on the connection. + * - All subsequent POSTs carry `Acp-Connection-Id` and return 202 Accepted. + * Session-scoped methods must also carry `Acp-Session-Id`. + * - On close we send DELETE /acp with the connection header. + */ export function createHttpStream(serverUrl: string): Stream { - let sessionId: string | null = null; - const incoming: AnyMessage[] = []; - const waiters: Array<() => void> = []; - const sseAbort = new AbortController(); + const base = serverUrl.replace(/\/+$/, ""); + const endpoint = `${base}/acp`; - function pushMessage(msg: AnyMessage) { - incoming.push(msg); - const w = waiters.shift(); - if (w) w(); + let connectionId: string | null = null; + let getStreamAbort: AbortController | null = null; + let closed = false; + + // Readable-stream plumbing: enqueue-with-buffer until the consumer pulls. + const inbox: AnyMessage[] = []; + let pullResolve: (() => void) | null = null; + + function deliver(msg: AnyMessage) { + inbox.push(msg); + if (pullResolve) { + const r = pullResolve; + pullResolve = null; + r(); + } } - function waitForMessage(): Promise { - if (incoming.length > 0) return Promise.resolve(); - return new Promise((r) => waiters.push(r)); + function waitForInbox(): Promise { + if (inbox.length > 0) return Promise.resolve(); + return new Promise((r) => { + pullResolve = r; + }); } - async function consumeSSE(response: Response) { - if (!response.body) return; - const reader = response.body.getReader(); + async function openGetStream() { + if (!connectionId) return; + getStreamAbort = new AbortController(); + + const response = await fetch(endpoint, { + method: "GET", + headers: { + Accept: "text/event-stream", + [ACP_CONNECTION_HEADER]: connectionId, + }, + signal: getStreamAbort.signal, + }); + + if (!response.ok || !response.body) { + throw new Error( + `Failed to open ACP GET stream: ${response.status} ${response.statusText}`, + ); + } + + void consumeSSE(response.body).catch((err) => { + if (closed) return; + // eslint-disable-next-line no-console + console.error("ACP GET stream error:", err); + }); + } + + async function consumeSSE(body: ReadableStream) { + const reader = body.getReader(); const decoder = new TextDecoder(); let buffer = ""; @@ -31,88 +136,183 @@ export function createHttpStream(serverUrl: string): Stream { if (done) break; buffer += decoder.decode(value, { stream: true }); - const parts = buffer.split("\n\n"); - buffer = parts.pop() || ""; - - for (const part of parts) { - for (const line of part.split("\n")) { - if (line.startsWith("data: ")) { - try { - const msg = JSON.parse(line.slice(6)) as AnyMessage; - pushMessage(msg); - } catch { - // ignore malformed JSON - } - } - } + let idx: number; + while ((idx = buffer.indexOf("\n\n")) >= 0) { + const event = buffer.slice(0, idx); + buffer = buffer.slice(idx + 2); + handleSseEvent(event); } } + if (buffer.length > 0) handleSseEvent(buffer); } catch (e: unknown) { if (e instanceof DOMException && e.name === "AbortError") return; + throw e; } } - let isFirstRequest = true; + function handleSseEvent(event: string) { + const dataLines: string[] = []; + for (const line of event.split("\n")) { + if (line.startsWith("data:")) { + dataLines.push(line.slice(5).replace(/^ /, "")); + } + } + if (dataLines.length === 0) return; + const data = dataLines.join("\n"); + let msg: AnyMessage; + try { + msg = JSON.parse(data) as AnyMessage; + } catch { + return; + } + + acpDebug("SSE → client", msg); + deliver(msg); + } + + async function sendInitialize(msg: AnyMessage) { + acpDebug("initialize → agent", msg); + const response = await fetch(endpoint, { + method: "POST", + headers: { + "Content-Type": "application/json", + Accept: "application/json", + }, + body: JSON.stringify(msg), + }); + + if (!response.ok) { + throw new Error( + `ACP initialize failed: ${response.status} ${response.statusText}`, + ); + } + + const connId = response.headers.get(ACP_CONNECTION_HEADER); + if (!connId) { + throw new Error( + `ACP initialize response missing ${ACP_CONNECTION_HEADER} header`, + ); + } + connectionId = connId; + + const body = (await response.json()) as AnyMessage; + acpDebug("initialize response", body); + await openGetStream(); + // Deliver the initialize response to the SDK *after* the GET stream is + // up, so any immediate server-initiated messages won't be missed. + deliver(body); + } + + async function sendPost(msg: AnyMessage) { + if (!connectionId) { + throw new Error("ACP POST attempted before initialize"); + } + + const headers: Record = { + "Content-Type": "application/json", + Accept: "application/json", + [ACP_CONNECTION_HEADER]: connectionId, + }; + + if (isRequest(msg) || isNotification(msg)) { + const sid = extractSessionId(messageParams(msg)); + if (sid) { + headers[ACP_SESSION_HEADER] = sid; + } else if (isRequest(msg)) { + const method = messageMethod(msg); + if (method && SESSION_SCOPED_METHODS.has(method)) { + throw new Error(`ACP method ${method} requires sessionId in params`); + } + } + } + + acpDebug("POST → agent", msg); + const response = await fetch(endpoint, { + method: "POST", + headers, + body: JSON.stringify(msg), + }); + + if (response.status !== 202 && !response.ok) { + throw new Error( + `ACP POST failed: ${response.status} ${response.statusText}`, + ); + } + // Drain the body so the connection can be reused. + await response.arrayBuffer().catch(() => undefined); + } + + async function sendDelete() { + if (!connectionId) return; + try { + await fetch(endpoint, { + method: "DELETE", + headers: { [ACP_CONNECTION_HEADER]: connectionId }, + }); + } catch { + // best-effort + } + } const readable = new ReadableStream({ async pull(controller) { - await waitForMessage(); - while (incoming.length > 0) { - controller.enqueue(incoming.shift()!); + await waitForInbox(); + while (inbox.length > 0) { + controller.enqueue(inbox.shift()!); + } + if (closed && inbox.length === 0) { + controller.close(); + } + }, + async cancel() { + closed = true; + await sendDelete(); + getStreamAbort?.abort(); + if (pullResolve) { + const r = pullResolve; + pullResolve = null; + r(); } }, }); const writable = new WritableStream({ async write(msg) { - const isRequest = - "method" in msg && - "id" in msg && - msg.id !== undefined && - msg.id !== null; - - const headers: Record = { - "Content-Type": "application/json", - Accept: "application/json, text/event-stream", - }; - if (sessionId) { - headers[ACP_SESSION_HEADER] = sessionId; + if ( + !connectionId && + isRequest(msg) && + messageMethod(msg) === "initialize" + ) { + await sendInitialize(msg); + return; } - - if (isFirstRequest && isRequest) { - isFirstRequest = false; - - const response = await fetch(`${serverUrl}/acp`, { - method: "POST", - headers, - body: JSON.stringify(msg), - signal: sseAbort.signal, - }); - - const sid = response.headers.get(ACP_SESSION_HEADER); - if (sid) sessionId = sid; - - consumeSSE(response); - } else if (isRequest) { - const abort = new AbortController(); - fetch(`${serverUrl}/acp`, { - method: "POST", - headers, - body: JSON.stringify(msg), - signal: abort.signal, - }).catch(() => {}); - setTimeout(() => abort.abort(), 200); - } else { - await fetch(`${serverUrl}/acp`, { - method: "POST", - headers, - body: JSON.stringify(msg), - }); + if (!connectionId) { + throw new Error( + "ACP transport: first outgoing message must be `initialize`", + ); + } + await sendPost(msg); + }, + async close() { + closed = true; + await sendDelete(); + getStreamAbort?.abort(); + // Unblock any pending pull so the readable can close. + if (pullResolve) { + const r = pullResolve; + pullResolve = null; + r(); } }, - - close() { - sseAbort.abort(); + async abort() { + closed = true; + await sendDelete(); + getStreamAbort?.abort(); + if (pullResolve) { + const r = pullResolve; + pullResolve = null; + r(); + } }, });