acpmux: a peer handshake survives a watch list too large to send; a peer link's watch lists only own sessions

cmux-tui/crates/acpmux/src/hub/list_guard.rs

@@ -115,6 +115,14 @@ impl Snapshot {
     }
 
     /// Lazily serialize the same snapshot with tools for the ACP list and peer bootstrap.
+    /// This daemon's own sessions only, no peer's row, serialized as a list reply: what a
+    /// peer's link reads in its watch (`server/list_reply.rs` `watch`).
+    pub(crate) fn own_whole(hub: &Hub, with_tools: bool) -> String {
+        let mut rows = Rows::default();
+        hub.visit_own_session_summaries(with_tools, |row| rows.push(&row));
+        rows.finish().0
+    }
+
     pub(crate) fn full_whole(&self, hub: &Hub) -> Arc<String> {
         let mut full = self.full.lock().unwrap_or_else(|e| e.into_inner());
         full.get_or_insert_with(|| {

cmux-tui/crates/acpmux/src/hub/peers.rs

@@ -1039,15 +1039,24 @@ impl Hub {
         }
     }
 
-    /// Each summary once, local before remote, without retaining the JSON trees.
-    pub(super) fn visit_session_summaries(&self, with_tools: bool, mut visit: impl FnMut(Value)) {
+    /// This daemon's own sessions' summaries, without retaining the JSON trees.
+    pub(super) fn visit_own_session_summaries(
+        &self,
+        with_tools: bool,
+        mut visit: impl FnMut(Value),
+    ) {
         for session in self.sessions() {
             visit(if with_tools {
                 self.session_summary(&session)
             } else {
                 self.session_list_summary(&session)
             });
         }
+    }
+
+    /// Each summary once, local before remote, without retaining the JSON trees.
+    pub(super) fn visit_session_summaries(&self, with_tools: bool, mut visit: impl FnMut(Value)) {
+        self.visit_own_session_summaries(with_tools, &mut visit);
         self.visit_remote_rows(
             |_| true,
             |row| {

cmux-tui/crates/acpmux/src/peer.rs

@@ -33,6 +33,10 @@ mod state_tests;
 #[path = "peer_restart_tests.rs"]
 mod restart_tests;
 
+#[cfg(test)]
+#[path = "peer_handshake_tests.rs"]
+mod handshake_tests;
+
 /// Source of `Peer::generation`.
 static NEXT_GENERATION: AtomicU64 = AtomicU64::new(1);
 
@@ -1158,7 +1162,7 @@ impl Peer {
         // Handshake in a task so the read loop below can serve the responses.
         let me = self.clone();
         let (stale_tx, mut stale_rx) = oneshot::channel::<String>();
-        let handshake = tokio::spawn(async move {
+        let mut handshake = tokio::spawn(async move {
             let mut hello = json!({"protocolVersion": 1, "clientCapabilities": {}, "clientInfo": {"name": CLIENT_NAME, "version": crate::hub::VERSION}});
             // This Mac lends the peer its GitHub login (`hub/github_owner.rs` on the peer).
             if me.github_broker() {
@@ -1211,12 +1215,20 @@ impl Peer {
             let auth = lists(crate::hub::session_auth::FEATURE);
             me.session_auth.store(auth, Ordering::SeqCst);
             // The reader published the reply's list (`watch_key`), so Connected follows it.
-            me.link_request_on(
-                method::MUX_WATCH,
-                json!({"enabled": true, "withTools": true}),
-                Some(link),
-            )
-            .await?;
+            let watch = |tools| json!({"enabled": true, "withTools": tools});
+            match me.link_request_on(method::MUX_WATCH, watch(true), Some(link)).await {
+                // A list too large to send with tools (big-red's, 17 MB with 5241 rows, on
+                // 2026-10-30) is read without them: the link connects with its rows' tools
+                // missing rather than never. A peer on this build leaves its peers' rows out
+                // of a link's list (`server/list_reply.rs` `watch`), so its list stays small.
+                Err(e) if too_large(&e) => {
+                    tracing::warn!(peer = %me.name, "the watch list is too large with tools ({}); reading it without them", e.message);
+                    me.link_request_on(method::MUX_WATCH, watch(false), Some(link)).await?;
+                }
+                answer => {
+                    answer?;
+                }
+            }
             // Connected only now, and ready with it, so its state note follows `connected`.
             // `run` moves the epoch on before it clears `connected` and takes the link down, so
             // a link that dropped meanwhile is seen here: it is not made ready again, and never
@@ -1256,6 +1268,7 @@ impl Peer {
         ping.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
         let mut last_frame = tokio::time::Instant::now();
         let mut stale_settled = false;
+        let mut handshake_ended = false;
         let mut ticks: u64 = 0;
         let handshake_deadline = tokio::time::sleep(timings.handshake);
         tokio::pin!(handshake_deadline);
@@ -1304,6 +1317,25 @@ impl Peer {
                         break Err(why);
                     }
                 }
+                // A handshake that ended without making the link ready ends the link now, with
+                // why: a step the peer refused (or that failed on this link), a stale token, a
+                // link that dropped meanwhile, or a task that panicked. The deadline below would
+                // only say the handshake timed out.
+                ended = &mut handshake, if !handshake_ended => {
+                    handshake_ended = true;
+                    match ended {
+                        Ok(Err(e)) => break Err(format!("handshake failed: {}", e.message)),
+                        Err(e) => break Err(format!("handshake failed: {e}")),
+                        Ok(Ok(())) if !self.link().is_ready() => {
+                            // The stale-token path sends its reason before it returns.
+                            if !stale_settled && let Ok(why) = stale_rx.try_recv() {
+                                break Err(why);
+                            }
+                            break Err("handshake ended before the link was ready".to_owned());
+                        }
+                        Ok(Ok(())) => {}
+                    }
+                }
                 _ = &mut handshake_deadline, if !self.link().is_ready() => {
                     break Err(format!("handshake timed out after {}s", timings.handshake.as_secs()));
                 }
@@ -1477,6 +1509,12 @@ fn watch_sessions(reply: &Value) -> Vec<Value> {
         .collect()
 }
 
+/// Whether `e` is a reply too large to send (`RpcError::too_large`).
+fn too_large(e: &RpcError) -> bool {
+    e.data.as_ref().and_then(|d| d.get("reason")).and_then(Value::as_str)
+        == Some("message.too_large")
+}
+
 /// The waits before each reread of a remote that has no peer token yet
 /// (`Peer::remote_tokens_settled`, `Peer::stale_token`).
 const TOKEN_REREADS: [Duration; 3] =

cmux-tui/crates/acpmux/src/peer_handshake_tests.rs

@@ -0,0 +1,157 @@
+//! The handshake's own failures: a watch list too large to send, and any step the peer
+//! refuses. huge-blue's link to big-red (2026-10-30) sat 90 s on every try, `handshaking`
+//! with no error, because big-red's `_acpmux/watch` reply (17 MB with tools, 5241 rows) was
+//! over the 15 MiB message limit and came back as `message.too_large`.
+
+use super::*;
+use tokio_tungstenite::WebSocketStream;
+use tokio_tungstenite::tungstenite::{Message as Frame, protocol::Role};
+
+/// A fake peer on `server` that answers `initialize` and every other request with `{}`, and
+/// each `_acpmux/watch` with `watch(withTools)`. Sends each watch's `withTools` on `asked`.
+fn fake_peer(
+    mut server: WebSocketStream<tokio::io::DuplexStream>,
+    watch: fn(bool) -> Result<Value, RpcError>,
+    asked: mpsc::UnboundedSender<bool>,
+) -> tokio::task::JoinHandle<()> {
+    tokio::spawn(async move {
+        while let Some(Ok(frame)) = server.next().await {
+            let Frame::Text(text) = frame else { continue };
+            let msg: Value = serde_json::from_str(&text).unwrap();
+            let Some(id) = msg.get("id").cloned() else { continue };
+            let reply = if msg["method"] == method::MUX_WATCH {
+                let tools = msg["params"]["withTools"] == true;
+                let _ = asked.send(tools);
+                watch(tools)
+            } else {
+                Ok(json!({}))
+            };
+            let reply = match reply {
+                Ok(result) => Message::ok(id, result),
+                Err(error) => Message::err(id, error),
+            };
+            let line = reply.to_line();
+            if server.send(Frame::Text(line.trim_end().to_owned().into())).await.is_err() {
+                return;
+            }
+        }
+    })
+}
+
+/// A watch list too large to send with tools is asked for again without them: the link
+/// connects, with the rows of that list, instead of timing out.
+#[tokio::test]
+async fn a_watch_list_too_large_with_tools_is_read_without_them() {
+    let (tx, mut notices) = mpsc::channel(64);
+    let peer = Peer::new("big", "ws://127.0.0.1:9", None, None, tx);
+    let (ours, theirs) = tokio::io::duplex(1 << 16);
+    let client = WebSocketStream::from_raw_socket(ours, Role::Client, None).await;
+    let server = WebSocketStream::from_raw_socket(theirs, Role::Server, None).await;
+    let (asked_tx, mut asked) = mpsc::unbounded_channel();
+    let fake = fake_peer(
+        server,
+        |tools| {
+            if tools {
+                Err(RpcError::too_large(17_181_757))
+            } else {
+                Ok(json!({"sessions": [{"sessionId": "m1", "name": "lane", "updatedAt": 5}]}))
+            }
+        },
+        asked_tx,
+    );
+    let mut out_rx = peer.out_rx.lock().await.take().unwrap();
+    let link = peer.clone();
+    let serving = tokio::spawn(async move { link.serve(client, &mut out_rx).await });
+
+    let mut listed = None;
+    loop {
+        let (_, _, notice) = tokio::time::timeout(Duration::from_secs(5), notices.recv())
+            .await
+            .expect("the link connects instead of waiting out the handshake deadline")
+            .expect("the link stays open");
+        match notice {
+            PeerNotice::Sessions(rows) => listed = Some(rows),
+            PeerNotice::Connected => break,
+            _ => {}
+        }
+    }
+    let rows = listed.expect("the list came before Connected");
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0]["sessionId"], "m1");
+    assert!(peer.link().is_ready());
+    assert_eq!(asked.recv().await, Some(true));
+    assert_eq!(asked.recv().await, Some(false));
+    serving.abort();
+    fake.abort();
+}
+
+/// A handshake step the peer refuses ends the link at once with the peer's error, not with
+/// `handshake timed out after 90s` and no error in between.
+#[tokio::test]
+async fn a_refused_handshake_step_ends_the_link_with_its_error() {
+    let (tx, _notices) = mpsc::channel(64);
+    let peer = Peer::new("big", "ws://127.0.0.1:9", None, None, tx);
+    let (ours, theirs) = tokio::io::duplex(1 << 16);
+    let client = WebSocketStream::from_raw_socket(ours, Role::Client, None).await;
+    let server = WebSocketStream::from_raw_socket(theirs, Role::Server, None).await;
+    let (asked_tx, _asked) = mpsc::unbounded_channel();
+    let fake =
+        fake_peer(server, |_| Err(RpcError::internal("the watch list broke".to_owned())), asked_tx);
+    let mut out_rx = peer.out_rx.lock().await.take().unwrap();
+    let ended = tokio::time::timeout(Duration::from_secs(5), peer.serve(client, &mut out_rx))
+        .await
+        .expect("the link ends when its handshake fails");
+    let why = ended.expect_err("a failed handshake is an error");
+    assert!(why.contains("the watch list broke"), "{why}");
+    assert!(!peer.link().is_ready());
+    fake.abort();
+}
+
+/// A handshake whose link went down meanwhile (its epoch moved on) returns without making the
+/// link ready: the link ends then, not at the deadline.
+#[tokio::test]
+async fn a_handshake_that_ends_without_a_ready_link_ends_the_link() {
+    let (tx, _notices) = mpsc::channel(64);
+    let peer = Peer::new("big", "ws://127.0.0.1:9", None, None, tx);
+    let (ours, theirs) = tokio::io::duplex(1 << 16);
+    let client = WebSocketStream::from_raw_socket(ours, Role::Client, None).await;
+    let mut server = WebSocketStream::from_raw_socket(theirs, Role::Server, None).await;
+    let (asked_tx, asked) = tokio::sync::oneshot::channel::<()>();
+    let (go_tx, go) = tokio::sync::oneshot::channel::<()>();
+    // Answers `initialize` at once; holds the watch's answer until the test lets it go.
+    let fake = tokio::spawn(async move {
+        let (mut asked_tx, mut go) = (Some(asked_tx), Some(go));
+        while let Some(Ok(frame)) = server.next().await {
+            let Frame::Text(text) = frame else { continue };
+            let msg: Value = serde_json::from_str(&text).unwrap();
+            let Some(id) = msg.get("id").cloned() else { continue };
+            let result = if msg["method"] == method::MUX_WATCH {
+                let _ = asked_tx.take().map(|tx| tx.send(()));
+                if let Some(go) = go.take() {
+                    let _ = go.await;
+                }
+                json!({"sessions": []})
+            } else {
+                json!({})
+            };
+            let line = Message::ok(id, result).to_line();
+            if server.send(Frame::Text(line.trim_end().to_owned().into())).await.is_err() {
+                return;
+            }
+        }
+    });
+    let mut out_rx = peer.out_rx.lock().await.take().unwrap();
+    let link = peer.clone();
+    let serving = tokio::spawn(async move { link.serve(client, &mut out_rx).await });
+    tokio::time::timeout(Duration::from_secs(5), asked).await.unwrap().unwrap();
+    // The link's epoch moves on before the watch is answered.
+    peer.link_ended();
+    go_tx.send(()).unwrap();
+    let ended = tokio::time::timeout(Duration::from_secs(5), serving)
+        .await
+        .expect("the link ends when its handshake ends without a ready link")
+        .unwrap();
+    assert!(ended.is_err(), "{ended:?}");
+    assert!(!peer.link().is_ready());
+    fake.abort();
+}

cmux-tui/crates/acpmux/src/server/list_reply.rs

@@ -132,12 +132,33 @@ pub(super) async fn sessions(hub: &Hub, conn: &Conn, params: &Value) -> Result<V
 }
 
 /// `_acpmux/watch`: turns the watch on before it reads the list, so no change is missed.
+/// A connection served as a peer (it proved this launch's peer token) reads only this
+/// daemon's own sessions: the link drops every row mirrored from a peer (`peer.rs`
+/// `watch_sessions`), and with them big-red's list was 17 MB with tools on 2026-10-30, past
+/// the message limit, so no peer's handshake there finished. Built from the own sessions
+/// alone, never from the whole list, once per snapshot.
 pub(super) async fn watch(hub: &Hub, conn: &Conn, params: &Value) -> Result<Value, RpcError> {
     crate::hub::session_pages::refuse_watch_filters(params)?;
     let on = params.get("enabled").and_then(Value::as_bool).unwrap_or(true);
     let tools = with_tools(params)?;
     conn.watch_tools.store(on && tools, Ordering::SeqCst);
     conn.watch_all.store(on, Ordering::SeqCst);
+    if conn.origin == super::Origin::Peer && whole(params) {
+        let raw = conn.raw_text;
+        return reply(hub, conn, None, |list| {
+            let name = match (tools, raw) {
+                (true, true) => "own tools",
+                (true, false) => "own tools masked",
+                (false, true) => "own",
+                (false, false) => "own masked",
+            };
+            Ok(list.made(name, |_| {
+                let own = Snapshot::own_whole(hub, tools);
+                if raw { own } else { masked(serde_json::from_str(&own).unwrap_or_default()) }
+            }))
+        })
+        .await;
+    }
     sessions(hub, conn, params).await
 }
 
@@ -187,6 +208,10 @@ pub(super) fn reply_line(id: &Value, result: &str) -> String {
     line
 }
 
+#[cfg(test)]
+#[path = "list_reply_tests.rs"]
+mod link_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;