From 10f3baa1fe2f64a5d0c6993c94b5eb94654caab3 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 30 Oct 2026 08:18:23 +0000 Subject: [PATCH 1/3] test(acpmux): a peer link hears no change of a row mirrored from another peer Red on base: every change of a mirrored row goes out on every peer link, which drops it (peer.rs), so with 25 peers each change went out 24 times. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-c11d09c9f41988fc1d93 --- .../acpmux/src/server/mirror_relay_tests.rs | 120 ++++++++++++++++++ cmux-tui/crates/acpmux/src/server/mod.rs | 4 + 2 files changed, 124 insertions(+) create mode 100644 cmux-tui/crates/acpmux/src/server/mirror_relay_tests.rs diff --git a/cmux-tui/crates/acpmux/src/server/mirror_relay_tests.rs b/cmux-tui/crates/acpmux/src/server/mirror_relay_tests.rs new file mode 100644 index 000000000000..de73624f426f --- /dev/null +++ b/cmux-tui/crates/acpmux/src/server/mirror_relay_tests.rs @@ -0,0 +1,120 @@ +//! Another daemon's link hears no change of a row this daemon mirrors from a peer. The link +//! drops every mirrored row (`peer.rs`), so with 25 peers (one daemon's, 2026-10-30) each change +//! of one peer's chat went out on the 24 other links for nothing, with its tools. + +use super::{Origin, serve_connection_with}; +use crate::config::{Config, StoreMode}; +use crate::peer::{Peer, PeerNotice}; +use crate::rpc::{Message, method}; +use serde_json::{Value, json}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; + +const OWN: &str = "own-chat"; + +/// A hub with one chat of its own and one (`m1`) mirrored from its peer `box`, and the peer. +async fn hub_with_a_mirrored_row() -> (Arc, Arc) { + let mut config = Config::default(); + config.store.mode = StoreMode::Memory; + let store = crate::store::open(&config.store, std::path::Path::new("/nonexistent")).unwrap(); + let meta = serde_json::from_value(json!({ + "schema": crate::store::META_SCHEMA, "id": OWN, "name": OWN, + "harness": "fake", "cwd": "/tmp", "status": "idle", "createdAt": 1, "updatedAt": 1 + })) + .unwrap(); + store.save(&meta).unwrap(); + let hub = crate::hub::Hub::new(config, store); + let (tx, _rx) = mpsc::channel(8); + let peer = Peer::new("box", "ws://127.0.0.1:9", None, None, tx); + hub.peers.lock().unwrap().insert("box".to_owned(), peer.clone()); + let row = json!({"sessionId": "m1", "name": "lane", "status": "idle", "updatedAt": 5}); + let mut settled = peer.settled(); + for notice in [PeerNotice::Sessions(vec![row]), PeerNotice::Connected] { + hub.peer_notices.send(("box".to_owned(), peer.generation, notice)).await.unwrap(); + } + tokio::time::timeout(Duration::from_secs(5), settled.wait_for(|n| *n >= 1)) + .await + .expect("the hub applies the notices") + .unwrap(); + assert_eq!(hub.remote_sessions().len(), 1); + (hub, peer) +} + +/// A connection of `origin` named `client` that watches the list, and its outgoing lines. +async fn watching( + hub: &Arc, + origin: Origin, + client: &str, +) -> (mpsc::Sender, mpsc::Receiver) { + let (input, incoming) = mpsc::channel(64); + let (output, mut outgoing) = mpsc::channel(64); + tokio::spawn(serve_connection_with(hub.clone(), incoming, output, origin)); + let hello = json!({"protocolVersion": 1, "clientCapabilities": {}, + "clientInfo": {"name": client, "version": "0"}}); + for (id, m, params) in [ + (1, method::INITIALIZE, hello), + (2, method::MUX_WATCH, json!({"enabled": true, "withTools": true})), + ] { + input.send(Message::request(id, m, params).to_line()).await.unwrap(); + let reply = until(&mut outgoing, |line| line["id"] == id).await; + assert!(reply.get("error").is_none(), "{m}: {reply}"); + } + (input, outgoing) +} + +/// The next line `outgoing` sends that `stop` takes, and the `session_changed` notices before it. +async fn until_noting( + outgoing: &mut mpsc::Receiver, + stop: impl Fn(&Value) -> bool, +) -> (Value, Vec) { + let mut changes = Vec::new(); + let line = tokio::time::timeout(Duration::from_secs(5), async { + loop { + let line: Value = serde_json::from_str(&outgoing.recv().await.unwrap()).unwrap(); + if stop(&line) { + return line; + } + if line["method"] == method::MUX_SESSION_CHANGED { + changes.push(line["params"].clone()); + } + } + }) + .await + .expect("the line comes"); + (line, changes) +} + +async fn until(outgoing: &mut mpsc::Receiver, stop: impl Fn(&Value) -> bool) -> Value { + until_noting(outgoing, stop).await.0 +} + +fn changed(id: &str) -> impl Fn(&Value) -> bool + '_ { + move |line| line["method"] == method::MUX_SESSION_CHANGED && line["params"]["sessionId"] == id +} + +#[tokio::test] +async fn a_peer_link_hears_no_change_of_a_row_mirrored_from_another_peer() { + let (hub, peer) = hub_with_a_mirrored_row().await; + let (cli, mut cli_out) = watching(&hub, Origin::Local, "cli").await; + // A peer's link, served as Peer or (added with the dashboard token) as Web. + let (_link, mut link_out) = watching(&hub, Origin::Peer, crate::peer::CLIENT_NAME).await; + let (_web_link, mut web_link_out) = watching(&hub, Origin::Web, crate::peer::CLIENT_NAME).await; + + let row = json!({"sessionId": "m1", "name": "lane", "status": "running", "updatedAt": 6}); + let notice = PeerNotice::SessionChanged { session: row, kind: "status".into(), seq: 7 }; + hub.peer_notices.send(("box".to_owned(), peer.generation, notice)).await.unwrap(); + // A watcher that is not a link hears it, with the peer it is from. + let heard = until(&mut cli_out, changed("m1")).await; + assert_eq!(heard["params"]["peer"], "box"); + + // Then a change of this daemon's own chat, which every watcher hears after the one before: + // a link that heard the mirrored row's change heard it first. + let rename = json!({"sessionId": OWN, "newName": "renamed"}); + cli.send(Message::request(3, method::MUX_RENAME, rename).to_line()).await.unwrap(); + for out in [&mut link_out, &mut web_link_out] { + let (_, before) = until_noting(out, changed(OWN)).await; + let mirrored: Vec<&Value> = before.iter().filter(|c| c["sessionId"] == "m1").collect(); + assert!(mirrored.is_empty(), "a link heard a mirrored row's change: {mirrored:?}"); + } +} diff --git a/cmux-tui/crates/acpmux/src/server/mod.rs b/cmux-tui/crates/acpmux/src/server/mod.rs index 8a185cdae79a..cdc8dee1dc3c 100644 --- a/cmux-tui/crates/acpmux/src/server/mod.rs +++ b/cmux-tui/crates/acpmux/src/server/mod.rs @@ -1746,6 +1746,10 @@ pub(crate) mod wait; mod windows_socket; use requests::handle_notification; +#[cfg(test)] +#[path = "mirror_relay_tests.rs"] +mod mirror_relay_tests; + #[cfg(test)] mod request_admission_tests { use super::{ From 8f1a1fdc4a2799fb32192422e014fcdff6fae360 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 30 Oct 2026 08:26:50 +0000 Subject: [PATCH 2/3] acpmux: a peer link hears no change of a row mirrored from another peer Every change of a row this daemon mirrors from a peer went out as _acpmux/session_changed on every other peer's link too, where the link drops it (peer.rs: a mirrored row or a notice naming a peer). With 25 peers that was 24 copies of each change, tools included. A link now hears only this daemon's own rows' changes, as its watch list since #28105 lists only them. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-c11d09c9f41988fc1d93 --- cmux-tui/crates/acpmux/src/server/mod.rs | 7 +++++++ cmux-tui/crates/acpmux/src/server/view.rs | 2 +- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/cmux-tui/crates/acpmux/src/server/mod.rs b/cmux-tui/crates/acpmux/src/server/mod.rs index cdc8dee1dc3c..9f4cdd77c92d 100644 --- a/cmux-tui/crates/acpmux/src/server/mod.rs +++ b/cmux-tui/crates/acpmux/src/server/mod.rs @@ -1613,6 +1613,13 @@ fn deliver(hub: &Hub, conn: &Conn, ev: HubEvent) { } if watching { if let Some(remote) = &ev.remote { + // Another daemon's link hears no change of a row this daemon mirrors: the link + // drops every such row (`peer.rs` `watch_sessions`), so each change of one peer's + // row went out on every other peer's link for nothing (24 copies of every change + // with one daemon's 25 peers, 2026-10-30), and the row's origin hears it from its own. + if view::a_link(conn) { + return; + } // A peer's prompt the person must answer, for a unix socket or app // watcher that did not attach the session (`hub/person.rs` (c)): // never relayed on to a Web client or another peer daemon. diff --git a/cmux-tui/crates/acpmux/src/server/view.rs b/cmux-tui/crates/acpmux/src/server/view.rs index d86bf226dc69..d4b048975a16 100644 --- a/cmux-tui/crates/acpmux/src/server/view.rs +++ b/cmux-tui/crates/acpmux/src/server/view.rs @@ -48,7 +48,7 @@ fn open(conn: &Conn) -> bool { /// (`hub/presence.rs`): served as Peer, or, added with the dashboard token, served as Web but /// named `acpmux-peer` in `initialize` (`peer.rs`). Any client may take that name; it only /// keeps that client out of presence. -fn a_link(conn: &Conn) -> bool { +pub(super) fn a_link(conn: &Conn) -> bool { conn.origin == Origin::Peer || *conn.name.lock_or_recover() == crate::peer::CLIENT_NAME } From b64ee747b64cc9d822d0de7231af4e311cd39ea3 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 30 Oct 2026 08:43:24 +0000 Subject: [PATCH 3/3] acpmux: a link still hears a peer's pending prompt; only session_changed is skipped Review of 8f1a1fdc: the link check returned before the peer permission branch, and a link is any client named acpmux-peer, so a unix-socket watcher with that name lost a peer's _acpmux/permission_pending. The check now guards only the session_changed send; the permission branch is as on base. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?id=aid-c11d09c9f41988fc1d93 --- cmux-tui/crates/acpmux/src/server/mod.rs | 15 ++++++--------- 1 file changed, 6 insertions(+), 9 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/server/mod.rs b/cmux-tui/crates/acpmux/src/server/mod.rs index 9f4cdd77c92d..dd2394619044 100644 --- a/cmux-tui/crates/acpmux/src/server/mod.rs +++ b/cmux-tui/crates/acpmux/src/server/mod.rs @@ -1613,13 +1613,6 @@ fn deliver(hub: &Hub, conn: &Conn, ev: HubEvent) { } if watching { if let Some(remote) = &ev.remote { - // Another daemon's link hears no change of a row this daemon mirrors: the link - // drops every such row (`peer.rs` `watch_sessions`), so each change of one peer's - // row went out on every other peer's link for nothing (24 copies of every change - // with one daemon's 25 peers, 2026-10-30), and the row's origin hears it from its own. - if view::a_link(conn) { - return; - } // A peer's prompt the person must answer, for a unix socket or app // watcher that did not attach the session (`hub/person.rs` (c)): // never relayed on to a Web client or another peer daemon. @@ -1632,8 +1625,12 @@ fn deliver(hub: &Hub, conn: &Conn, ev: HubEvent) { } return; } - // Peers already filter to the interesting kinds and send a summary. - if rec.dir == "peer" { + // Peers already filter to the interesting kinds and send a summary. Another + // daemon's link hears no change of a row this daemon mirrors: the link drops every + // such row (`peer.rs` `watch_sessions`), so each change of one peer's row went out on + // every other peer's link for nothing (24 copies of every change with one daemon's + // 25 peers, 2026-10-30), and the row's origin hears it from its own. + if rec.dir == "peer" && !view::a_link(conn) { let sid = remote .summary .get("sessionId")