feat(acp): Align to new request patterns of ACP Streamable HTTP/WS transport (#8605)

Signed-off-by: Alex Hancock <alexhancock@block.xyz>

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<String>,
+    /// Subscribe here to receive all server→client messages for this connection.
+    pub outbound_tx: broadcast::Sender<String>,
+    /// 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<Option<mpsc::UnboundedReceiver<String>>>,
+    /// 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<bool>,
+    /// 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<Option<tokio::task::JoinHandle<()>>>,
+    pre_subscribe_buffer: Arc<Mutex<Option<VecDeque<String>>>>,
+}
+
+pub(crate) struct ConnectionRegistry {
+    pub server: Arc<AcpServer>,
+    connections: RwLock<HashMap<String, Arc<Connection>>>,
+}
+
+impl ConnectionRegistry {
+    pub fn new(server: Arc<AcpServer>) -> 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<Connection>)> {
+        let (to_agent_tx, to_agent_rx) = mpsc::channel::<String>(256);
+        let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::<String>();
+        let (outbound_tx, _) = broadcast::channel::<String>(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<Arc<Connection>> {
+        self.connections.read().await.get(connection_id).cloned()
+    }
+
+    pub async fn remove(&self, connection_id: &str) -> Option<Arc<Connection>> {
+        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<Self>) {
+        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<String>, broadcast::Receiver<String>) {
+        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<Connection>, mpsc::UnboundedSender<String>) {
+        let (to_agent_tx, _to_agent_rx) = mpsc::channel::<String>(256);
+        let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::<String>();
+        let (outbound_tx, _) = broadcast::channel::<String>(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;
+    }
+}

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<AcpServer>,
-    // Keyed by acp_session_id: a connection-scoped UUID serving many Goose sessions.
-    sessions: RwLock<HashMap<String, TransportSession>>,
-}
-
-impl HttpState {
-    pub fn new(server: Arc<AcpServer>) -> Self {
-        Self {
-            server,
-            sessions: RwLock::new(HashMap::new()),
-        }
-    }
-
-    async fn create_session(&self) -> Result<String, StatusCode> {
-        let (to_agent_tx, to_agent_rx) = mpsc::channel::<String>(256);
-        let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::<String>();
-
-        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<Arc<Mutex<mpsc::UnboundedReceiver<String>>>, 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<Mutex<mpsc::UnboundedReceiver<String>>>,
-    cleanup: Option<(Arc<HttpState>, String)>,
-) -> Sse<impl futures::Stream<Item = Result<axum::response::sse::Event, Infallible>>> {
-    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<HttpState>, 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<HttpState>,
-    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<HttpState>,
-    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<Arc<HttpState>>,
+    State(registry): State<Arc<ConnectionRegistry>>,
     request: Request<Body>,
 ) -> 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<HttpState>, request: Request<Body>) -> Response {
+async fn handle_initialize(registry: Arc<ConnectionRegistry>, 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<ConnectionRegistry>,
+    request: Request<Body>,
+) -> 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<HttpState>, request: Request<Body>) ->
             .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<Connection>,
+    replay: Vec<String>,
+    mut receiver: broadcast::Receiver<String>,
+) -> Sse<impl futures::Stream<Item = Result<axum::response::sse::Event, Infallible>>> {
     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<Arc<HttpState>>,
-    request: Request<Body>,
-) -> 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<Arc<ConnectionRegistry>>,
+    request: Request<Body>,
+) -> 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()
 }

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<String>,
-    pub from_agent_rx: Arc<Mutex<mpsc::UnboundedReceiver<String>>>,
-    pub handle: tokio::task::JoinHandle<()>,
-}
-
 pub(crate) fn accepts_mime_type(request: &Request<Body>, mime_type: &str) -> bool {
     request
         .headers()
@@ -38,16 +33,6 @@ pub(crate) fn accepts_mime_type(request: &Request<Body>, mime_type: &str) -> boo
         .is_some_and(|accept| accept.contains(mime_type))
 }
 
-pub(crate) fn accepts_json_and_sse(request: &Request<Body>) -> 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<Body>) -> bool {
     request
         .headers()
@@ -56,15 +41,15 @@ pub(crate) fn content_type_is_json(request: &Request<Body>) -> bool {
         .is_some_and(|ct| ct.starts_with(JSON_MIME_TYPE))
 }
 
-pub(crate) fn get_session_id(request: &Request<Body>) -> Option<String> {
+pub(crate) fn header_value(request: &Request<Body>, name: &str) -> Option<String> {
     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<WebSocketUpgrade, WebSocketUpgradeRejection>,
-    State(state): State<(Arc<http::HttpState>, Arc<websocket::WsState>)>,
+    State(state): State<Arc<connection::ConnectionRegistry>>,
     request: Request<Body>,
 ) -> 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<AcpServer>) -> 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<AcpServer>) -> 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)
 }

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<AcpServer>,
-    // Keyed by acp_session_id: a connection-scoped UUID serving many Goose sessions.
-    sessions: RwLock<HashMap<String, TransportSession>>,
-}
-
-impl WsState {
-    pub fn new(server: Arc<AcpServer>) -> Self {
-        Self {
-            server,
-            sessions: RwLock::new(HashMap::new()),
-        }
-    }
-
-    async fn create_connection(&self) -> Result<String> {
-        let (to_agent_tx, to_agent_rx) = mpsc::channel::<String>(256);
-        let (from_agent_tx, from_agent_rx) = mpsc::unbounded_channel::<String>();
-
-        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<WsState>, 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<ConnectionRegistry>,
+    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<WsState>, 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<WsState>, acp_session_id: String) {
+async fn run_ws(
+    socket: WebSocket,
+    registry: Arc<ConnectionRegistry>,
+    connection_id: String,
+    connection: Arc<super::connection::Connection>,
+) {
     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;
+    }
 }

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<string, string | undefined> };
+  };
+  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<string>([
+  "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<void> {
-    if (incoming.length > 0) return Promise.resolve();
-    return new Promise<void>((r) => waiters.push(r));
+  function waitForInbox(): Promise<void> {
+    if (inbox.length > 0) return Promise.resolve();
+    return new Promise<void>((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<Uint8Array>) {
+    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<string, string> = {
+      "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<AnyMessage>({
     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<AnyMessage>({
     async write(msg) {
-      const isRequest =
-        "method" in msg &&
-        "id" in msg &&
-        msg.id !== undefined &&
-        msg.id !== null;
-
-      const headers: Record<string, string> = {
-        "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();
+      }
     },
   });