From ebbf4d9f8e3d2b687323e0b7d90d8a40c30ca46a Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 21:24:14 +0000 Subject: [PATCH 01/26] test(acpmux): cover peer lifecycle routing and ambiguous names Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- .../acpmux/tests/session_control_peer_cli.rs | 445 ++++++++++++++++++ 1 file changed, 445 insertions(+) create mode 100644 cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs new file mode 100644 index 000000000000..3aef7cfaf8bd --- /dev/null +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -0,0 +1,445 @@ +//! Session lifecycle commands resolve once, retain the owning peer in the id, and +//! refuse a name that names more than one session. +#![cfg(unix)] + +#[path = "support/daemon_env.rs"] +mod daemon_env; +const HERMETIC: daemon_env::Process = daemon_env::Process::stand_in() + .with_env(&[("ACPMUX_AGENT_HOSTS", "0"), ("ACPMUX_LOGIN_ENV", "0")]); + +use acpmux::config::{Config, StoreMode}; +use acpmux::hub::Hub; +use acpmux::rpc::{Message, method}; +use acpmux::server::{ + WsAuth, bind_unix, bind_ws, peer_auth::PeerAuth, serve_connection, serve_unix, serve_ws_with, +}; +use futures_util::{SinkExt, StreamExt}; +use serde_json::{Value, json}; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; + +const PERSON_KEY: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; + +fn config(root: &Path) -> Config { + let fake = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/fake_agent.py"); + let mut cfg: Config = serde_json::from_value(json!({ + "harnesses": {"fake": {"argv": ["python3", fake]}}, + "defaultHarness": "fake", + "webRoots": [root], + "webAskingModes": {"fake": ["normal"]}, + })) + .unwrap(); + cfg.store.mode = StoreMode::Memory; + cfg +} + +fn hub(root: &Path) -> Arc { + let cfg = config(root); + let store = acpmux::store::open(&cfg.store, root).unwrap(); + Hub::new(cfg, store) +} + +struct Rpc { + tx: mpsc::Sender, + rx: mpsc::Receiver, + next: i64, +} + +impl Rpc { + async fn call(&mut self, method: &str, params: Value) -> Value { + self.next += 1; + let id = self.next; + self.tx.send(Message::request(id, method, params).to_line()).await.unwrap(); + loop { + let line = tokio::time::timeout(Duration::from_secs(20), self.rx.recv()) + .await + .expect("rpc timeout") + .expect("rpc closed"); + if let Message::Response { id: got, result, error } = Message::parse(&line).unwrap() + && got == id + { + if let Some(error) = error { + panic!("{method} failed: {}", error.message); + } + return result.unwrap_or(Value::Null); + } + } + } + + async fn send(&mut self, method: &str, params: Value) -> i64 { + self.next += 1; + let id = self.next; + self.tx.send(Message::request(id, method, params).to_line()).await.unwrap(); + id + } + + async fn wait(&mut self, method: &str, matches: impl Fn(&Value) -> bool) -> Value { + loop { + let line = tokio::time::timeout(Duration::from_secs(20), self.rx.recv()) + .await + .expect("notification timeout") + .expect("rpc closed"); + if let Message::Notification { method: got, params } = Message::parse(&line).unwrap() { + let params = params.unwrap_or(Value::Null); + if got == method && matches(¶ms) { + return params; + } + } + } + } +} + +async fn client(hub: Arc) -> Rpc { + let _ = hub.person.install_spawn_key(PERSON_KEY); + let (tx, rx) = mpsc::channel(64); + let (out_tx, out_rx) = mpsc::channel(4096); + tokio::spawn(serve_connection(hub, rx, out_tx)); + let mut client = Rpc { tx, rx: out_rx, next: 0 }; + let init = client + .call( + method::INITIALIZE, + json!({ + "protocolVersion": 1, + "clientInfo": {"name": "session-control-peer-cli"} + }), + ) + .await; + let challenge = &init["_meta"]["acpmux"]["personChallenge"]; + let proof = acpmux::hub::person::person_proof( + PERSON_KEY, + acpmux::hub::person::TRANSPORT_UNIX, + challenge["nonce"].as_str().unwrap(), + challenge["connection"].as_str().unwrap(), + ) + .unwrap(); + client.call("_acpmux/person_prove", json!({"proof": proof})).await; + client.call(method::MUX_WATCH, json!({"enabled": true})).await; + client +} + +async fn listen_peer( + hub: Arc, + root: &Path, + tag: &str, +) -> (u16, String, tokio::task::JoinHandle>) { + let auth_root = root.join(tag); + let auth = PeerAuth::create(&auth_root).unwrap(); + let peer_token = + std::fs::read_to_string(acpmux::server::peer_auth::token_path(&auth_root)).unwrap(); + let listener = bind_ws("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let task = tokio::spawn(serve_ws_with( + hub, + listener, + "test-token".into(), + WsAuth { local_app: None, peer: Some(Arc::new(auth)) }, + )); + (port, peer_token, task) +} + +/// A tiny peer that deliberately reuses a local raw session id. +async fn lying_peer(session_id: &str) -> (u16, mpsc::UnboundedReceiver) { + use tokio_tungstenite::tungstenite::Message as Ws; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let session_id = session_id.to_owned(); + let (calls, received) = mpsc::unbounded_channel(); + tokio::spawn(async move { + while let Ok((tcp, _)) = listener.accept().await { + let Ok(mut ws) = tokio_tungstenite::accept_async(tcp).await else { continue }; + let id = session_id.clone(); + let calls = calls.clone(); + tokio::spawn(async move { + while let Some(Ok(Ws::Text(text))) = ws.next().await { + let Ok(Message::Request { id: request_id, method: m, .. }) = + Message::parse(&text) + else { + continue; + }; + calls.send(m.clone()).unwrap(); + let detail = json!({"sessionId": id, "name": "remote-name", "status": "idle", "policy": "ask", "harness": "fake", "queue": [], "turn": null}); + let result = match m.as_str() { + method::INITIALIZE => { + json!({"protocolVersion": 1, "_meta": {"acpmux": {"version": "0", "build": "lying", "origin": "peer"}}}) + } + method::MUX_WATCH => { + json!({"sessions": [{"sessionId": id, "name": "remote-name", "harness": "fake", "status": "idle"}]}) + } + method::MUX_INFO | method::SESSION_NEW => detail, + _ => json!({}), + }; + let line = Message::ok(request_id, result).to_line(); + if ws.send(Ws::Text(line.trim_end().to_owned().into())).await.is_err() { + return; + } + } + }); + } + }); + (port, received) +} + +async fn new_session(client: &mut Rpc, root: &Path, name: &str) -> String { + client + .call( + method::SESSION_NEW, + json!({ + "cwd": root, + "mcpServers": [], + "_meta": {"acpmux": {"name": name}} + }), + ) + .await["sessionId"] + .as_str() + .unwrap() + .to_owned() +} + +async fn cli(root: &Path, socket: &Path, args: &[&str]) -> (bool, String, String) { + let mut command = daemon_env::tokio_command(root, daemon_env::Agents::Deny); + command.env("ACPMUX_SOCKET", socket).args(args); + let out = tokio::time::timeout(Duration::from_secs(30), command.kill_on_drop(true).output()) + .await + .expect("CLI command timeout") + .unwrap(); + ( + out.status.success(), + String::from_utf8_lossy(&out.stdout).into_owned(), + String::from_utf8_lossy(&out.stderr).into_owned(), + ) +} + +#[tokio::test] +async fn lifecycle_commands_use_peer_stable_ids_and_refuse_ambiguous_names() { + let root = PathBuf::from("/tmp").join(format!("amx-scp-{}", std::process::id())); + std::fs::create_dir_all(&root).unwrap(); + let b = hub(&root); + let c = hub(&root); + let mut cb = client(b.clone()).await; + let mut cc = client(c.clone()).await; + let stop_id = new_session(&mut cb, &root, "remote-stop").await; + let resume_id = new_session(&mut cb, &root, "remote-resume").await; + let retry_id = new_session(&mut cb, &root, "remote-retry").await; + let _same_local = new_session(&mut cb, &root, "local-ambiguous").await; + let _peer_ambiguous = new_session(&mut cb, &root, "peer-ambiguous").await; + let _ = new_session(&mut cc, &root, "peer-ambiguous").await; + + // Make a held queue and a failed turn before mirroring the peer rows. + cb.send( + method::SESSION_PROMPT, + json!({"sessionId": resume_id, "prompt": [{"type": "text", "text": "until-cancel"}]}), + ) + .await; + cb.wait(method::SESSION_UPDATE, |p| p["update"]["content"]["text"] == "running").await; + for (prompt_id, text) in [("p-f", "fail-streamed: boom"), ("p-a", "queued")] { + cb.send(method::SESSION_PROMPT, json!({"sessionId": resume_id, "prompt": [{"type": "text", "text": text}], "_meta": {"acpmux": {"promptId": prompt_id}}})).await; + cb.wait("_acpmux/prompt_accepted", |p| p["promptId"] == prompt_id && p["queued"] == true) + .await; + } + cb.send(method::SESSION_CANCEL, json!({"sessionId": resume_id})).await; + cb.wait(method::MUX_EVENT, |p| p["sessionId"] == resume_id && p["kind"] == "queue_paused") + .await; + assert_eq!( + cb.call(method::MUX_INFO, json!({"sessionId": resume_id})).await["queuePaused"], + true + ); + cb.send(method::SESSION_PROMPT, json!({"sessionId": retry_id, "prompt": [{"type": "text", "text": "fail-once: retry"}], "_meta": {"acpmux": {"promptId": "retry"}}})).await; + cb.wait(method::MUX_EVENT, |p| p["sessionId"] == retry_id && p["kind"] == "turn_error").await; + + let (port_b, token_b, b_server) = listen_peer(b.clone(), &root, "b").await; + let (port_c, token_c, c_server) = listen_peer(c.clone(), &root, "c").await; + let a = hub(&root); + a.add_peer( + "b", + &format!("ws://127.0.0.1:{port_b}"), + Some("test-token".into()), + Some(token_b), + true, + ) + .await + .unwrap(); + a.add_peer( + "c", + &format!("ws://127.0.0.1:{port_c}"), + Some("test-token".into()), + Some(token_c), + true, + ) + .await + .unwrap(); + let local_id = { + let mut ca = client(a.clone()).await; + new_session(&mut ca, &root, "local-ambiguous").await + }; + assert!(!local_id.is_empty()); + let socket = root.join("a.sock"); + let local_server = tokio::spawn(serve_unix(a.clone(), bind_unix(&socket).await.unwrap())); + + // `new --host` carries a permitted start policy to the target peer. + let (ok, stdout, stderr) = cli( + &root, + &socket, + &[ + "--json", + "new", + "--host", + "b", + "--name", + "remote-start", + "--model", + "fake", + "--policy", + "ask", + ], + ) + .await; + assert!(ok, "{stdout}{stderr}"); + let started: Value = serde_json::from_str(&stdout).unwrap(); + assert_eq!(started["policy"], "ask", "{started}"); + + let (ok, stdout, stderr) = cli( + &root, + &socket, + &[ + "--json", + "new", + "--host", + "b", + "--name", + "remote-prompt", + "--model", + "fake", + "--policy", + "ask", + "hello", + ], + ) + .await; + assert!(ok, "{stdout}{stderr}"); + let prompted: Value = serde_json::from_str(&stdout).unwrap(); + assert_eq!(prompted["reply"], "echo: hello", "{prompted}"); + + let (ok, stdout, stderr) = cli(&root, &socket, &["--json", "session", "stop", &stop_id]).await; + assert!(ok, "{stdout}{stderr}"); + let stopped = cb.call(method::MUX_INFO, json!({"sessionId": stop_id})).await; + assert_eq!(stopped["status"], "closed"); + + let (ok, stdout, stderr) = + cli(&root, &socket, &["--json", "session", "resume", "remote-resume"]).await; + assert!(ok, "{stdout}{stderr}"); + let resumed: Value = serde_json::from_str(&stdout).unwrap(); + assert_eq!(resumed["resumed"], true, "{stdout}{stderr}"); + cb.wait(method::MUX_EVENT, |p| p["sessionId"] == resume_id && p["kind"] == "queue_resumed") + .await; + assert_eq!( + cb.call(method::MUX_INFO, json!({"sessionId": resume_id})).await["queuePaused"], + false + ); + + // The same resume target is addressable by stable id after its queue drains. + let (ok, stdout, stderr) = + cli(&root, &socket, &["--json", "session", "resume", &resume_id]).await; + assert!(ok, "{stdout}{stderr}"); + let (ok, stdout, stderr) = + cli(&root, &socket, &["--json", "session", "stop", "remote-start"]).await; + assert!(ok, "{stdout}{stderr}"); + let (ok, stdout, stderr) = + cli(&root, &socket, &["--json", "session", "stop", "definitely-missing"]).await; + assert!(!ok && format!("{stdout}{stderr}").contains("no_session"), "{stdout}{stderr}"); + + let (ok, stdout, stderr) = cli(&root, &socket, &["--json", "retry", &retry_id]).await; + assert!(ok, "{stdout}{stderr}"); + let retried: Value = serde_json::from_str(&stdout).unwrap(); + assert!( + retried["promptId"].as_str().is_some_and(|id| id.contains("retry~resume-1")), + "{stdout}{stderr}" + ); + cb.wait(method::MUX_EVENT, |p| p["sessionId"] == retry_id && p["kind"] == "turn_result").await; + + // A local and a peer session with the same name, and two peers with the same + // name, must both refuse a lifecycle operation rather than pick a winner. + for name in ["local-ambiguous", "peer-ambiguous"] { + let (ok, stdout, stderr) = cli(&root, &socket, &["--json", "session", "stop", name]).await; + assert!(!ok, "{name}: {stdout}{stderr}"); + assert!( + stdout.contains("matches") + || stderr.contains("matches") + || stdout.contains("ambiguous") + || stderr.contains("ambiguous"), + "{name}: {stdout}{stderr}" + ); + } + + local_server.abort(); + b_server.abort(); + c_server.abort(); + b.end_agents_at_shutdown(Default::default()).unwrap(); + c.end_agents_at_shutdown(Default::default()).unwrap(); + a.end_agents_at_shutdown(Default::default()).unwrap(); + b.shutdown_all().await; + c.shutdown_all().await; + a.shutdown_all().await; + std::fs::remove_dir_all(root).unwrap(); +} + +#[tokio::test] +async fn lifecycle_commands_keep_a_peer_owner_when_raw_ids_collide() { + let root = PathBuf::from("/tmp").join(format!("amx-scp-collision-{}", std::process::id())); + std::fs::create_dir_all(&root).unwrap(); + let a = hub(&root); + let mut ca = client(a.clone()).await; + let local_id = new_session(&mut ca, &root, "local-owner").await; + let qualified = format!("lying/{local_id}"); + let shadow_id = new_session(&mut ca, &root, &qualified).await; + let (port, mut calls) = lying_peer(&local_id).await; + a.add_peer("lying", &format!("ws://127.0.0.1:{port}"), Some("tok".into()), None, true) + .await + .unwrap(); + let socket = root.join("a.sock"); + let server = tokio::spawn(serve_unix(a.clone(), bind_unix(&socket).await.unwrap())); + + let (ok, stdout, stderr) = cli( + &root, + &socket, + &[ + "--json", + "new", + "--host", + "lying", + "--name", + "remote-start", + "--model", + "fake", + "--policy", + "ask", + ], + ) + .await; + assert!(ok, "{stdout}{stderr}"); + let started: Value = serde_json::from_str(&stdout).unwrap(); + assert_eq!(started["name"], "remote-name", "{started}"); + + let (ok, stdout, stderr) = + cli(&root, &socket, &["--json", "session", "stop", "lying/remote-name"]).await; + assert!(ok, "{stdout}{stderr}"); + let local = ca.call(method::MUX_INFO, json!({"sessionId": local_id})).await; + assert_ne!(local["status"], "closed", "a peer stop killed the local owner: {local}"); + let shadow = ca.call(method::MUX_INFO, json!({"sessionId": shadow_id})).await; + assert_ne!(shadow["status"], "closed", "a display name hijacked the peer id: {shadow}"); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if calls.recv().await.as_deref() == Some(method::MUX_KILL) { + break; + } + } + }) + .await + .expect("the peer received stop"); + + server.abort(); + a.end_agents_at_shutdown(Default::default()).unwrap(); + a.shutdown_all().await; + std::fs::remove_dir_all(root).unwrap(); +} From a0e2a04de037efb4522d73ac415b2b367102691c Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 21:24:25 +0000 Subject: [PATCH 02/26] fix(acpmux): retain peer ownership for session lifecycle commands Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/README.md | 18 ++++++ cmux-tui/crates/acpmux/src/cli/output.rs | 12 ++-- cmux-tui/crates/acpmux/src/cli/resolve.rs | 60 ++++++++++++++++++- cmux-tui/crates/acpmux/src/cli/retry.rs | 6 +- cmux-tui/crates/acpmux/src/cli/run/extra.rs | 6 +- .../crates/acpmux/src/cli/run/standard.rs | 27 ++++++--- cmux-tui/crates/acpmux/src/hub/locate.rs | 36 ++++++++--- cmux-tui/crates/acpmux/src/hub/own_host.rs | 5 ++ 8 files changed, 142 insertions(+), 28 deletions(-) diff --git a/cmux-tui/crates/acpmux/README.md b/cmux-tui/crates/acpmux/README.md index cb36c422dd30..2c44ff460d89 100644 --- a/cmux-tui/crates/acpmux/README.md +++ b/cmux-tui/crates/acpmux/README.md @@ -176,6 +176,24 @@ work but are hidden from help. `--json` on any command prints the raw response. +Session controls work across configured peers without a shell on the target host: + +```sh +acpmux new --host worker -m codex -n review -d +acpmux resolve worker/review +acpmux session resume worker/review # release held queued prompts +acpmux retry worker/review # retry its failed last turn +acpmux session cancel worker/review # cancel its running turn +acpmux session stop worker/review # stop its agent; keep the session +``` + +Use the stable `sessionId` from `resolve`, a unique name or prefix, or `HOST/NAME`. +Resume, retry, cancel and stop refuse ambiguous matches, including a local and remote +session with the same name, and list qualified IDs to select from. Once resolved, a control +keeps the owning peer and session ID even if another host has a copy of that session. +The target daemon decides which start policy and controls the caller may use; a peer link +does not grant app-only permissions. + ## Orchestrating agents from scripts Every command takes `--json` before it for machine output; errors then go to stderr as one diff --git a/cmux-tui/crates/acpmux/src/cli/output.rs b/cmux-tui/crates/acpmux/src/cli/output.rs index bfc18ba71f18..d0d89582b84d 100644 --- a/cmux-tui/crates/acpmux/src/cli/output.rs +++ b/cmux-tui/crates/acpmux/src/cli/output.rs @@ -304,7 +304,10 @@ pub(crate) async fn stream_prompt( ) -> Result<()> { let mut notes = client.notifications().await.ok_or_else(|| anyhow!("notifications already taken"))?; - client.request(method::MUX_ATTACH, json!({"sessionId": id, "limit": 0})).await?; + let attached = client.request(method::MUX_ATTACH, json!({"sessionId": id, "limit": 0})).await?; + // Requests retain the peer route; streamed notifications carry the owning daemon's + // unqualified id, as the attach reply identifies it. + let event_id = attached["session"]["sessionId"].as_str().unwrap_or(id); let c = client.clone(); let params = prompt.params(id, text, steer, None); let turn = tokio::spawn(async move { c.request(method::SESSION_PROMPT, params).await }); @@ -366,7 +369,7 @@ pub(crate) async fn stream_prompt( let Message::Notification { method: m, params } = m else { continue }; if m == method::MUX_DISCONNECTED { return Err(prompt.closed(&client, id, "streaming the session")); } let p = params.unwrap_or(Value::Null); - if p.get("sessionId").and_then(Value::as_str) != Some(id) { continue; } + if p.get("sessionId").and_then(Value::as_str) != Some(event_id) { continue; } if let Some(answer) = steered.heard(&m, &p) { break answer; } if agent_activity(&m, &p) { activity = true; @@ -656,7 +659,8 @@ async fn collect_once( ) -> Result { let mut notes = client.notifications().await.ok_or_else(|| anyhow!("notifications already taken"))?; - client.request(method::MUX_ATTACH, json!({"sessionId": id, "limit": 0})).await?; + let attached = client.request(method::MUX_ATTACH, json!({"sessionId": id, "limit": 0})).await?; + let event_id = attached["session"]["sessionId"].as_str().unwrap_or(id); let c = client.clone(); let params = prompt.params(id, text, opts.steer, None); let mut turn = tokio::spawn(async move { c.request(method::SESSION_PROMPT, params).await }); @@ -709,7 +713,7 @@ async fn collect_once( let Message::Notification { method: m, params } = m else { continue }; if m == method::MUX_DISCONNECTED { return Err(prompt.closed(&client, id, "streaming the session")); } let p = params.unwrap_or(Value::Null); - if p.get("sessionId").and_then(Value::as_str) != Some(id) { continue; } + if p.get("sessionId").and_then(Value::as_str) != Some(event_id) { continue; } if let Some(answer) = steered.heard(&m, &p) { break answer; } if agent_activity(&m, &p) { activity = true; diff --git a/cmux-tui/crates/acpmux/src/cli/resolve.rs b/cmux-tui/crates/acpmux/src/cli/resolve.rs index e5184e9f6688..0b3d14580925 100644 --- a/cmux-tui/crates/acpmux/src/cli/resolve.rs +++ b/cmux-tui/crates/acpmux/src/cli/resolve.rs @@ -2,12 +2,70 @@ //! `cmux.com/deeplink/agent?...` of a PR footer), a session id or a name names, on this daemon //! or a peer, with its host, ids, harness, model, folder, lane, pull requests and last turn. -use anyhow::Result; +use anyhow::{Result, anyhow}; use serde_json::{Value, json}; +use crate::cli::errors::{AppError, Code}; use crate::cli::output::{age, print_json, short}; use crate::rpc::method; +/// Resolve a lifecycle operation once, before sending it to the owning daemon. A name is +/// only a lookup key; every subsequent request carries the stable id and its peer route. +pub(crate) async fn control_id(client: &crate::client::Client, key: &str) -> Result { + let query = crate::cli::orchestrate::expand_session_key(key)?; + let reply = + client.request(method::MUX_RESOLVE, json!({"query": query})).await.map_err(|e| { + if e.downcast_ref::().is_some_and(|e| e.0.code == -32002) { + AppError::new(Code::NoSession, "no_session", e.to_string()).into() + } else { + e + } + })?; + let matches = reply["matches"] + .as_array() + .ok_or_else(|| anyhow!("resolve returned no session matches for {query:?}"))?; + if matches.is_empty() { + return Err(AppError::no_session(&query).into()); + } + let [session] = matches.as_slice() else { + let candidates = matches + .iter() + .map(|s| { + format!( + "{} ({})", + s["name"].as_str().unwrap_or("?"), + s["qualifiedSessionId"].as_str().unwrap_or("?") + ) + }) + .collect::>() + .join(", "); + return Err(AppError::new( + Code::Usage, + "ambiguous_session", + format!( + "session {query:?} is ambiguous ({} matches); use a qualified session id: {candidates}", + matches.len() + ), + ) + .into()); + }; + routed_id(session) +} + +/// A create/resolve reply's stable id, qualified for subsequent requests on this daemon. +pub(crate) fn routed_id(session: &Value) -> Result { + let id = session["sessionId"] + .as_str() + .filter(|id| !id.is_empty()) + .ok_or_else(|| anyhow!("the daemon returned no stable session id"))?; + // Local replies retain the unqualified id. For a peer, never lose its route by looking + // up the bare id a second time: another daemon may hold a copy of the same session. + Ok(match session["peer"].as_str() { + Some(peer) => format!("{peer}/{id}"), + None => id.to_owned(), + }) +} + pub async fn run(query: String, json_out: bool) -> Result<()> { let query = crate::cli::orchestrate::expand_session_key(&query)?; let client = crate::daemon::connect(true).await?; diff --git a/cmux-tui/crates/acpmux/src/cli/retry.rs b/cmux-tui/crates/acpmux/src/cli/retry.rs index 61a7c031afea..43ede7824416 100644 --- a/cmux-tui/crates/acpmux/src/cli/retry.rs +++ b/cmux-tui/crates/acpmux/src/cli/retry.rs @@ -6,7 +6,7 @@ //! instead (`hub/subagent_retry_manual.rs`): `retry.unknown_subagent` when the chat has none. use crate::cli::output::print_json; -use crate::cli::run::resolve_id; +use crate::cli::resolve::control_id; use crate::daemon::connect; use crate::rpc::method; use anyhow::Result; @@ -14,7 +14,7 @@ use serde_json::json; #[derive(clap::Args)] pub struct RetryArgs { - /// The session's name, id or unique prefix. + /// The session's stable id, name or unique prefix, optionally qualified as HOST/NAME. pub session: String, /// One of the session's failed subagents, by its id or agent id: resumed in the session. #[arg(long)] @@ -23,7 +23,7 @@ pub struct RetryArgs { pub async fn run(args: RetryArgs, json_out: bool) -> Result<()> { let client = connect(true).await?; - let id = resolve_id(&client, &args.session).await?; + let id = control_id(&client, &args.session).await?; let mut params = json!({"sessionId": id}); if let Some(subagent) = args.subagent { params["subagentId"] = json!(subagent); diff --git a/cmux-tui/crates/acpmux/src/cli/run/extra.rs b/cmux-tui/crates/acpmux/src/cli/run/extra.rs index c15713973bc9..fb56eb24f5e3 100644 --- a/cmux-tui/crates/acpmux/src/cli/run/extra.rs +++ b/cmux-tui/crates/acpmux/src/cli/run/extra.rs @@ -31,7 +31,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, _suppress_reads: bool) -> } Command::Cancel { session } => { let client = connect(true).await?; - let id = resolve_id(&client, &session).await?; + let id = crate::cli::resolve::control_id(&client, &session).await?; // As a request, the reply says the daemon handled the cancel. client.request(method::SESSION_CANCEL, json!({"sessionId": id})).await?; if json_out { @@ -43,7 +43,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, _suppress_reads: bool) -> } Command::Kill { session, purge } => { let client = connect(true).await?; - let id = resolve_id(&client, &session).await?; + let id = crate::cli::resolve::control_id(&client, &session).await?; let v = client.request(method::MUX_KILL, json!({"sessionId": id, "purge": purge})).await?; if json_out { @@ -78,7 +78,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, _suppress_reads: bool) -> } Command::ResumeQueue { session } => { let client = connect(true).await?; - let id = resolve_id(&client, &session).await?; + let id = crate::cli::resolve::control_id(&client, &session).await?; let info = client.request(method::MUX_INFO, json!({"sessionId": id})).await?; // `_acpmux/queue_edit` `resume` names the queue by any of its prompts: a held one. let held = info["queue"] diff --git a/cmux-tui/crates/acpmux/src/cli/run/standard.rs b/cmux-tui/crates/acpmux/src/cli/run/standard.rs index b543d68748bb..6910896f92ef 100644 --- a/cmux-tui/crates/acpmux/src/cli/run/standard.rs +++ b/cmux-tui/crates/acpmux/src/cli/run/standard.rs @@ -486,7 +486,8 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R } }, }; - let id = v.get("sessionId").and_then(Value::as_str).unwrap_or("").to_owned(); + let id = crate::cli::resolve::routed_id(&v)?; + let session_id = v["sessionId"].as_str().unwrap_or_default(); // Started past this host's lane cap: a warning, never a refusal (`hub/admission.rs`). if reused.is_none() && let Some(w) = @@ -498,7 +499,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R .pointer("/_meta/acpmux/name") .or_else(|| v.get("name")) .and_then(Value::as_str) - .unwrap_or(&id) + .unwrap_or(session_id) .to_owned(); // A failed sidebar step only warns: the prompt is still sent. let show = || { @@ -507,7 +508,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R // Generated names can be identical on independent peers. let title = launch.name.as_ref().map(|_| launch.title.as_deref().unwrap_or(&name)); - if let Err(e) = launch::show_in_group(&id, title, group) { + if let Err(e) = launch::show_in_group(session_id, title, group) { eprintln!("acpmux: {e:#}"); } } @@ -539,13 +540,13 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R } else if first.is_some() { println!( "reused {name} ({}): it had no prompt yet, so its first prompt is queued", - &id[..8.min(id.len())] + &session_id[..8.min(session_id.len())] ); } else { println!( "reused {name} ({}): it already runs, so the prompt was not sent again; \ use `acpmux send {name}` to give it more", - &id[..8.min(id.len())] + &session_id[..8.min(session_id.len())] ); } return Ok(()); @@ -565,8 +566,12 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R print_json(&info); } else if !one_shot { let folder = v.pointer("/_meta/acpmux/cwd").and_then(Value::as_str); - let line = - crate::cli::session_folder::created_line(&name, &id, host.as_deref(), folder); + let line = crate::cli::session_folder::created_line( + &name, + session_id, + host.as_deref(), + folder, + ); println!("{line}"); if let Some(w) = v.pointer("/_meta/acpmux/worktree") { let g = |k: &str| w.get(k).and_then(Value::as_str).unwrap_or_default(); @@ -600,7 +605,11 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R } let r = outcome?; if json_out { - let mut out = json!({"sessionId": id, "name": name, "reply": r.reply, "stopReason": r.stop_reason, "permissions": r.permissions_asked, "permissionsDenied": r.permissions_denied, "ephemeral": ephemeral}); + let mut out = json!({"sessionId": session_id, "name": name, "reply": r.reply, "stopReason": r.stop_reason, "permissions": r.permissions_asked, "permissionsDenied": r.permissions_denied, "ephemeral": ephemeral}); + if v["peer"].is_string() { + out["peer"] = v["peer"].clone(); + out["qualifiedSessionId"] = json!(id); + } if let Some(a) = v.pointer("/_meta/acpmux/admission") { out["admission"] = a.clone(); } @@ -645,7 +654,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R if args.detach || json_out { return Ok(()); } - crate::tui::run(client, Some(id)).await + crate::tui::run(client, Some(session_id.to_owned())).await } Command::Send { session, diff --git a/cmux-tui/crates/acpmux/src/hub/locate.rs b/cmux-tui/crates/acpmux/src/hub/locate.rs index fbc6a20cc267..4d3878ae388c 100644 --- a/cmux-tui/crates/acpmux/src/hub/locate.rs +++ b/cmux-tui/crates/acpmux/src/hub/locate.rs @@ -150,14 +150,14 @@ fn by_link<'a>(summaries: &'a [Value], link: &AgentLink, own: &str) -> Vec<&'a V found } -/// The sessions `key` names, by the first rule that finds any: an id (acpmux's, qualified or -/// the harness's) or whole name; a bead id the session carries (`beads.rs`); `HOST/NAME` on -/// that host (this machine's name included); a bare name on any host. +/// The sessions `key` names: ids take precedence over names. A bare name includes every +/// host, so a local chat cannot hide a peer's chat with the same name from control callers. +/// Qualified names select their host; bead ids still find the sessions that carry them. fn by_key<'a>(summaries: &'a [Value], key: &str, own: &str) -> Vec<&'a Value> { let exact: Vec<&Value> = summaries .iter() .filter(|s| { - ["sessionId", "qualifiedSessionId", "agentSessionId", "name"] + ["sessionId", "qualifiedSessionId", "agentSessionId"] .iter() .any(|f| s[*f].as_str() == Some(key)) }) @@ -165,6 +165,13 @@ fn by_key<'a>(summaries: &'a [Value], key: &str, own: &str) -> Vec<&'a Value> { if !exact.is_empty() { return exact; } + let named: Vec<&Value> = summaries + .iter() + .filter(|s| s["name"].as_str() == Some(key) || bare_name(s) == key) + .collect(); + if !named.is_empty() { + return named; + } if crate::beads::is_bead_id(key) { let carry = by_bead(summaries, key); if !carry.is_empty() { @@ -172,13 +179,26 @@ fn by_key<'a>(summaries: &'a [Value], key: &str, own: &str) -> Vec<&'a Value> { } } if let Some((host, rest)) = key.split_once('/') { - return summaries + let on_host: Vec<&Value> = + summaries.iter().filter(|s| host_of(s, own).eq_ignore_ascii_case(host)).collect(); + let exact: Vec<&Value> = on_host .iter() - .filter(|s| host_of(s, own).eq_ignore_ascii_case(host)) + .copied() .filter(|s| bare_name(s) == rest || s["sessionId"].as_str() == Some(rest)) .collect(); + if !exact.is_empty() { + return exact; + } + return on_host.into_iter().filter(|s| key_prefix(s, rest)).collect(); } - summaries.iter().filter(|s| bare_name(s) == key).collect() + summaries.iter().filter(|s| key_prefix(s, key)).collect() +} + +/// Keep the CLI's unique id/name prefixes, collecting all candidates before a control +/// caller chooses one. A prefix never wins over an exact id or name. +fn key_prefix(summary: &Value, key: &str) -> bool { + summary["sessionId"].as_str().is_some_and(|id| id.starts_with(key)) + || bare_name(summary).starts_with(key) } /// The sessions that carry bead `id` (`beads.rs`). @@ -307,7 +327,7 @@ mod tests { let rows = rows(); assert_eq!(ids(&by_key(&rows, "hb/m2", "big-red")), ["m2"]); assert_eq!(ids(&by_key(&rows, "a2", "big-red")), ["m2"]); - assert_eq!(ids(&by_key(&rows, "review", "big-red")), ["m1"]); + assert_eq!(ids(&by_key(&rows, "review", "big-red")), ["m1", "m3"]); assert_eq!(ids(&by_key(&rows, "big-red/review", "big-red")), ["m1"]); assert_eq!(ids(&by_key(&rows, "builder", "big-red")), ["m2"]); assert!(by_key(&rows, "nobody", "big-red").is_empty()); diff --git a/cmux-tui/crates/acpmux/src/hub/own_host.rs b/cmux-tui/crates/acpmux/src/hub/own_host.rs index b5aef929867b..6197c131f198 100644 --- a/cmux-tui/crates/acpmux/src/hub/own_host.rs +++ b/cmux-tui/crates/acpmux/src/hub/own_host.rs @@ -11,6 +11,11 @@ impl Hub { /// that name (its session is the one meant), so it can only reach a session the bare `NAME` /// names exactly. pub fn resolve(&self, key: &str) -> Result, RpcError> { + // A known peer's qualified stable id is authoritative, even when a local chat's + // display name happens to be that entire string. Leave it to peer forwarding. + if key.contains('/') && self.remote_sessions.lock().unwrap().contains_key(key) { + return Err(RpcError::not_found(format!("no local session with the id {key:?}"))); + } let found = self.resolve_here(key); if found.as_ref().is_err_and(|e| e.code == RpcError::not_found("").code) && let Some((host, rest)) = key.split_once('/') From 1bdd4f978f07a314603542533707633c8372744c Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 21:32:22 +0000 Subject: [PATCH 03/26] test(acpmux): resume held peer queues by both name and id Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- .../acpmux/tests/session_control_peer_cli.rs | 47 +++++++++++-------- 1 file changed, 28 insertions(+), 19 deletions(-) diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs index 3aef7cfaf8bd..61aefff76433 100644 --- a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -221,30 +221,37 @@ async fn lifecycle_commands_use_peer_stable_ids_and_refuse_ambiguous_names() { let mut cc = client(c.clone()).await; let stop_id = new_session(&mut cb, &root, "remote-stop").await; let resume_id = new_session(&mut cb, &root, "remote-resume").await; + let resume_by_id = new_session(&mut cb, &root, "resume-by-id").await; let retry_id = new_session(&mut cb, &root, "remote-retry").await; let _same_local = new_session(&mut cb, &root, "local-ambiguous").await; let _peer_ambiguous = new_session(&mut cb, &root, "peer-ambiguous").await; let _ = new_session(&mut cc, &root, "peer-ambiguous").await; - // Make a held queue and a failed turn before mirroring the peer rows. - cb.send( - method::SESSION_PROMPT, - json!({"sessionId": resume_id, "prompt": [{"type": "text", "text": "until-cancel"}]}), - ) - .await; - cb.wait(method::SESSION_UPDATE, |p| p["update"]["content"]["text"] == "running").await; - for (prompt_id, text) in [("p-f", "fail-streamed: boom"), ("p-a", "queued")] { - cb.send(method::SESSION_PROMPT, json!({"sessionId": resume_id, "prompt": [{"type": "text", "text": text}], "_meta": {"acpmux": {"promptId": prompt_id}}})).await; - cb.wait("_acpmux/prompt_accepted", |p| p["promptId"] == prompt_id && p["queued"] == true) + // Hold both queues before mirroring the peer rows: exercise real resume by name and id. + for id in [&resume_id, &resume_by_id] { + cb.send( + method::SESSION_PROMPT, + json!({"sessionId": id.as_str(), "prompt": [{"type": "text", "text": "until-cancel"}]}), + ) + .await; + cb.wait(method::SESSION_UPDATE, |p| p["update"]["content"]["text"] == "running").await; + for (prompt_id, text) in [("p-f", "fail-streamed: boom"), ("p-a", "queued")] { + cb.send(method::SESSION_PROMPT, json!({"sessionId": id.as_str(), "prompt": [{"type": "text", "text": text}], "_meta": {"acpmux": {"promptId": prompt_id}}})).await; + cb.wait("_acpmux/prompt_accepted", |p| { + p["promptId"] == prompt_id && p["queued"] == true + }) .await; - } - cb.send(method::SESSION_CANCEL, json!({"sessionId": resume_id})).await; - cb.wait(method::MUX_EVENT, |p| p["sessionId"] == resume_id && p["kind"] == "queue_paused") + } + cb.send(method::SESSION_CANCEL, json!({"sessionId": id.as_str()})).await; + cb.wait(method::MUX_EVENT, |p| { + p["sessionId"] == id.as_str() && p["kind"] == "queue_paused" + }) .await; - assert_eq!( - cb.call(method::MUX_INFO, json!({"sessionId": resume_id})).await["queuePaused"], - true - ); + assert_eq!( + cb.call(method::MUX_INFO, json!({"sessionId": id.as_str()})).await["queuePaused"], + true + ); + } cb.send(method::SESSION_PROMPT, json!({"sessionId": retry_id, "prompt": [{"type": "text", "text": "fail-once: retry"}], "_meta": {"acpmux": {"promptId": "retry"}}})).await; cb.wait(method::MUX_EVENT, |p| p["sessionId"] == retry_id && p["kind"] == "turn_error").await; @@ -338,10 +345,12 @@ async fn lifecycle_commands_use_peer_stable_ids_and_refuse_ambiguous_names() { false ); - // The same resume target is addressable by stable id after its queue drains. + // A stable id resumes another held queue on the same peer. let (ok, stdout, stderr) = - cli(&root, &socket, &["--json", "session", "resume", &resume_id]).await; + cli(&root, &socket, &["--json", "session", "resume", &resume_by_id]).await; assert!(ok, "{stdout}{stderr}"); + let resumed: Value = serde_json::from_str(&stdout).unwrap(); + assert_eq!(resumed["resumed"], true, "{stdout}{stderr}"); let (ok, stdout, stderr) = cli(&root, &socket, &["--json", "session", "stop", "remote-start"]).await; assert!(ok, "{stdout}{stderr}"); From 53f3aa71ff3ab9d0dd8bbdc3f6b26fde63966634 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 21:36:49 +0000 Subject: [PATCH 04/26] fix(acpmux): retain qualified peer routes after row removal Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/hub/own_host.rs | 8 +++++--- cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs | 6 ++++++ 2 files changed, 11 insertions(+), 3 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/hub/own_host.rs b/cmux-tui/crates/acpmux/src/hub/own_host.rs index 6197c131f198..6c2159b5b7bb 100644 --- a/cmux-tui/crates/acpmux/src/hub/own_host.rs +++ b/cmux-tui/crates/acpmux/src/hub/own_host.rs @@ -11,9 +11,11 @@ impl Hub { /// that name (its session is the one meant), so it can only reach a session the bare `NAME` /// names exactly. pub fn resolve(&self, key: &str) -> Result, RpcError> { - // A known peer's qualified stable id is authoritative, even when a local chat's - // display name happens to be that entire string. Leave it to peer forwarding. - if key.contains('/') && self.remote_sessions.lock().unwrap().contains_key(key) { + // A configured peer owns its namespace even when its mirrored row disappeared + // after lookup. A local display name must never intercept the resolved route. + if let Some((host, _)) = key.split_once('/') + && self.peer(host).is_some() + { return Err(RpcError::not_found(format!("no local session with the id {key:?}"))); } let found = self.resolve_here(key); diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs index 61aefff76433..5ab682a8181d 100644 --- a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -447,6 +447,12 @@ async fn lifecycle_commands_keep_a_peer_owner_when_raw_ids_collide() { .await .expect("the peer received stop"); + // A vanished mirrored row must not turn the already resolved peer id into a + // local name. The configured peer still owns this namespace. + let peer = a.peer("lying").unwrap(); + a.forget_remote(("lying", peer.generation), &local_id, None); + assert!(a.resolve(&qualified).is_err(), "a vanished peer row exposed a local name"); + server.abort(); a.end_agents_at_shutdown(Default::default()).unwrap(); a.shutdown_all().await; From 442e155c828e00bc6ed1f7d89b0fdf32959f25b0 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 21:41:39 +0000 Subject: [PATCH 05/26] fix(acpmux): keep peer identity when opening a session group Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/cli/run/launch.rs | 4 +- .../crates/acpmux/src/cli/run/standard.rs | 2 +- .../crates/acpmux/tests/remote_group_cli.rs | 3 +- .../acpmux/tests/session_control_peer_cli.rs | 38 ++++++++++++++++++- 4 files changed, 42 insertions(+), 5 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/cli/run/launch.rs b/cmux-tui/crates/acpmux/src/cli/run/launch.rs index 158de6d5cc2d..662724fd5a45 100644 --- a/cmux-tui/crates/acpmux/src/cli/run/launch.rs +++ b/cmux-tui/crates/acpmux/src/cli/run/launch.rs @@ -318,8 +318,8 @@ fn workspace_key(id: &str) -> String { /// app's `agent open` action (`cmux agent open`), optionally named `title`. /// Unnamed launches omit the title so a peer's generated session name cannot /// accidentally match another session's workspace in the same group. -/// A remote create returns the ID mirrored by the local daemon. Open that -/// ID here, without `--host`: the workspace belongs to this Mac and its +/// A remote create's ID stays qualified with its owning peer. Open that +/// qualified ID here, without `--host`: the workspace belongs to this Mac and its /// acpmux forwards the chat to the peer. This also avoids a name lookup in /// the app's peer inventory, which may not have seen the new session yet. pub(super) fn show_in_group(id: &str, title: Option<&str>, group: &str) -> Result<()> { diff --git a/cmux-tui/crates/acpmux/src/cli/run/standard.rs b/cmux-tui/crates/acpmux/src/cli/run/standard.rs index 6910896f92ef..ccf25c92ce48 100644 --- a/cmux-tui/crates/acpmux/src/cli/run/standard.rs +++ b/cmux-tui/crates/acpmux/src/cli/run/standard.rs @@ -508,7 +508,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R // Generated names can be identical on independent peers. let title = launch.name.as_ref().map(|_| launch.title.as_deref().unwrap_or(&name)); - if let Err(e) = launch::show_in_group(session_id, title, group) { + if let Err(e) = launch::show_in_group(&id, title, group) { eprintln!("acpmux: {e:#}"); } } diff --git a/cmux-tui/crates/acpmux/tests/remote_group_cli.rs b/cmux-tui/crates/acpmux/tests/remote_group_cli.rs index 1574eb73e718..47f61af871a5 100644 --- a/cmux-tui/crates/acpmux/tests/remote_group_cli.rs +++ b/cmux-tui/crates/acpmux/tests/remote_group_cli.rs @@ -113,7 +113,8 @@ async fn remote_group_cli_opens_the_remote_id_locally_and_reuses_its_row() { assert!(uuid::Uuid::parse_str(args[key_index + 1]).is_ok()); let id = args[2]; let remote = &remotes[usize::from(peer == "other")]; - assert!(remote.resolve(id).is_ok(), "the session belongs to the remote hub"); + let raw_id = id.strip_prefix(&format!("{peer}/")).expect("the group keeps its peer route"); + assert!(remote.resolve(raw_id).is_ok(), "the session belongs to the remote hub"); assert!(local.resolve(id).is_err(), "no local agent was created"); // The ID handed to cmux is usable through the controller daemon, without --host. let client = acpmux::client::Client::connect(&socket).await.unwrap(); diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs index 5ab682a8181d..c6830cecea06 100644 --- a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -15,6 +15,7 @@ use acpmux::server::{ }; use futures_util::{SinkExt, StreamExt}; use serde_json::{Value, json}; +use std::os::unix::fs::PermissionsExt; use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; @@ -199,7 +200,11 @@ async fn new_session(client: &mut Rpc, root: &Path, name: &str) -> String { async fn cli(root: &Path, socket: &Path, args: &[&str]) -> (bool, String, String) { let mut command = daemon_env::tokio_command(root, daemon_env::Agents::Deny); - command.env("ACPMUX_SOCKET", socket).args(args); + command + .env("ACPMUX_SOCKET", socket) + .env("PATH", format!("{}:{}", root.join("bin").display(), daemon_env::bare_path(&[]))) + .env("GROUP_CAPTURE", root.join("group-args")) + .args(args); let out = tokio::time::timeout(Duration::from_secs(30), command.kill_on_drop(true).output()) .await .expect("CLI command timeout") @@ -222,6 +227,7 @@ async fn lifecycle_commands_use_peer_stable_ids_and_refuse_ambiguous_names() { let stop_id = new_session(&mut cb, &root, "remote-stop").await; let resume_id = new_session(&mut cb, &root, "remote-resume").await; let resume_by_id = new_session(&mut cb, &root, "resume-by-id").await; + let cancel_id = new_session(&mut cb, &root, "remote-cancel").await; let retry_id = new_session(&mut cb, &root, "remote-retry").await; let _same_local = new_session(&mut cb, &root, "local-ambiguous").await; let _peer_ambiguous = new_session(&mut cb, &root, "peer-ambiguous").await; @@ -328,6 +334,27 @@ async fn lifecycle_commands_use_peer_stable_ids_and_refuse_ambiguous_names() { let prompted: Value = serde_json::from_str(&stdout).unwrap(); assert_eq!(prompted["reply"], "echo: hello", "{prompted}"); + for key in ["remote-cancel", cancel_id.as_str()] { + cb.send( + method::SESSION_PROMPT, + json!({"sessionId": cancel_id, + "prompt": [{"type": "text", "text": "until-cancel"}]}), + ) + .await; + cb.wait(method::SESSION_UPDATE, |p| { + p["sessionId"] == cancel_id && p["update"]["content"]["text"] == "running" + }) + .await; + let (ok, stdout, stderr) = cli(&root, &socket, &["--json", "session", "cancel", key]).await; + assert!(ok, "{stdout}{stderr}"); + cb.wait(method::MUX_EVENT, |p| { + p["sessionId"] == cancel_id + && p["kind"] == "turn_result" + && p["msg"]["stopReason"] == "cancelled" + }) + .await; + } + let (ok, stdout, stderr) = cli(&root, &socket, &["--json", "session", "stop", &stop_id]).await; assert!(ok, "{stdout}{stderr}"); let stopped = cb.call(method::MUX_INFO, json!({"sessionId": stop_id})).await; @@ -408,6 +435,10 @@ async fn lifecycle_commands_keep_a_peer_owner_when_raw_ids_collide() { .unwrap(); let socket = root.join("a.sock"); let server = tokio::spawn(serve_unix(a.clone(), bind_unix(&socket).await.unwrap())); + std::fs::create_dir(root.join("bin")).unwrap(); + let cmux = root.join("bin/cmux"); + std::fs::write(&cmux, "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$GROUP_CAPTURE\"\n").unwrap(); + std::fs::set_permissions(&cmux, std::fs::Permissions::from_mode(0o700)).unwrap(); let (ok, stdout, stderr) = cli( &root, @@ -417,6 +448,9 @@ async fn lifecycle_commands_keep_a_peer_owner_when_raw_ids_collide() { "new", "--host", "lying", + "--group", + "test-group", + "--detach", "--name", "remote-start", "--model", @@ -429,6 +463,8 @@ async fn lifecycle_commands_keep_a_peer_owner_when_raw_ids_collide() { assert!(ok, "{stdout}{stderr}"); let started: Value = serde_json::from_str(&stdout).unwrap(); assert_eq!(started["name"], "remote-name", "{started}"); + let opened = std::fs::read_to_string(root.join("group-args")).unwrap(); + assert_eq!(opened.lines().nth(2), Some(qualified.as_str()), "{opened}"); let (ok, stdout, stderr) = cli(&root, &socket, &["--json", "session", "stop", "lying/remote-name"]).await; From f085ce4a74e278a960965cfcbfe4491bc92a810e Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 21:45:21 +0000 Subject: [PATCH 06/26] test(acpmux): avoid private peer helper in collision fixture Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs | 6 ------ 1 file changed, 6 deletions(-) diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs index c6830cecea06..cdfa148d7742 100644 --- a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -483,12 +483,6 @@ async fn lifecycle_commands_keep_a_peer_owner_when_raw_ids_collide() { .await .expect("the peer received stop"); - // A vanished mirrored row must not turn the already resolved peer id into a - // local name. The configured peer still owns this namespace. - let peer = a.peer("lying").unwrap(); - a.forget_remote(("lying", peer.generation), &local_id, None); - assert!(a.resolve(&qualified).is_err(), "a vanished peer row exposed a local name"); - server.abort(); a.end_agents_at_shutdown(Default::default()).unwrap(); a.shutdown_all().await; From a0c79af2862eb8978e10c592c2f3037f06888429 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 22:50:55 +0000 Subject: [PATCH 07/26] fix(acpmux): preserve raw peer request ids until mirroring Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- .../crates/acpmux/src/cli/run/standard.rs | 20 ++++++++++++------- .../acpmux/tests/session_control_peer_cli.rs | 7 ++++++- 2 files changed, 19 insertions(+), 8 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/cli/run/standard.rs b/cmux-tui/crates/acpmux/src/cli/run/standard.rs index ccf25c92ce48..434deba3c9ad 100644 --- a/cmux-tui/crates/acpmux/src/cli/run/standard.rs +++ b/cmux-tui/crates/acpmux/src/cli/run/standard.rs @@ -526,7 +526,8 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R .filter(|_| !lost_race && launch::never_prompted(info)); if let Some(text) = &first { let prompt = PromptId::new(args.prompt_id.clone()); - let accepted = queue_prompt(client, &id, text, false, None, &prompt).await?; + let accepted = + queue_prompt(client, &session_id, text, false, None, &prompt).await?; if !json_out { warn_host_pressure(&accepted); } @@ -554,7 +555,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R let first = launch.first_prompt(&args, &v); if json_out && !one_shot { let mut info = client - .request(method::MUX_INFO, json!({"sessionId": id})) + .request(method::MUX_INFO, json!({"sessionId": session_id})) .await .unwrap_or(v.clone()); if let Some(w) = v.pointer("/_meta/acpmux/worktree") { @@ -580,7 +581,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R // Before it works: an allow it asks for is the app's alone. if on_permission == OnPermission::Wait && let Ok(info) = - client.request(method::MUX_INFO, json!({"sessionId": id})).await + client.request(method::MUX_INFO, json!({"sessionId": session_id})).await && let Some(note) = crate::question_answer::app_allow_note(&info, &name) { eprintln!("{note}"); @@ -597,10 +598,14 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R steer: false, }; show(); - let outcome = collect_reply(client.clone(), &id, &text, opts, &prompt).await; + let outcome = + collect_reply(client.clone(), &session_id, &text, opts, &prompt).await; if ephemeral { let _ = client - .request(method::MUX_KILL, json!({"sessionId": id, "purge": true})) + .request( + method::MUX_KILL, + json!({"sessionId": session_id, "purge": true}), + ) .await; } let r = outcome?; @@ -622,7 +627,8 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R if args.detach { // The prompt is queued before the sidebar step, so a // session never runs without its instructions. - let accepted = queue_prompt(client, &id, &text, false, None, &prompt).await?; + let accepted = + queue_prompt(client, &session_id, &text, false, None, &prompt).await?; if !json_out { warn_host_pressure(&accepted); } @@ -639,7 +645,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R show(); return stream_prompt( client, - &id, + &session_id, &text, false, false, diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs index cdfa148d7742..72d3dc648532 100644 --- a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -346,7 +346,12 @@ async fn lifecycle_commands_use_peer_stable_ids_and_refuse_ambiguous_names() { }) .await; let (ok, stdout, stderr) = cli(&root, &socket, &["--json", "session", "cancel", key]).await; - assert!(ok, "{stdout}{stderr}"); + assert!(!ok, "remote cancel must retain the local-origin ownership rule: {stdout}{stderr}"); + assert!( + format!("{stdout}{stderr}").contains("remote.session_cancel_ownership"), + "{stdout}{stderr}" + ); + cb.send(method::SESSION_CANCEL, json!({"sessionId": cancel_id})).await; cb.wait(method::MUX_EVENT, |p| { p["sessionId"] == cancel_id && p["kind"] == "turn_result" From 564819f54402bdce40339f7ade54a96e92d4b537 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 22:55:24 +0000 Subject: [PATCH 08/26] fix(acpmux): normalize already-qualified peer session ids Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/cli/resolve.rs | 1 + cmux-tui/crates/acpmux/src/cli/run/standard.rs | 4 ++++ 2 files changed, 5 insertions(+) diff --git a/cmux-tui/crates/acpmux/src/cli/resolve.rs b/cmux-tui/crates/acpmux/src/cli/resolve.rs index 0b3d14580925..2254c47efd5c 100644 --- a/cmux-tui/crates/acpmux/src/cli/resolve.rs +++ b/cmux-tui/crates/acpmux/src/cli/resolve.rs @@ -61,6 +61,7 @@ pub(crate) fn routed_id(session: &Value) -> Result { // Local replies retain the unqualified id. For a peer, never lose its route by looking // up the bare id a second time: another daemon may hold a copy of the same session. Ok(match session["peer"].as_str() { + Some(peer) if id.strip_prefix(&format!("{peer}/")).is_some() => id.to_owned(), Some(peer) => format!("{peer}/{id}"), None => id.to_owned(), }) diff --git a/cmux-tui/crates/acpmux/src/cli/run/standard.rs b/cmux-tui/crates/acpmux/src/cli/run/standard.rs index 434deba3c9ad..7e9a5f3e5bbe 100644 --- a/cmux-tui/crates/acpmux/src/cli/run/standard.rs +++ b/cmux-tui/crates/acpmux/src/cli/run/standard.rs @@ -488,6 +488,10 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R }; let id = crate::cli::resolve::routed_id(&v)?; let session_id = v["sessionId"].as_str().unwrap_or_default(); + let session_id = v["peer"] + .as_str() + .and_then(|peer| session_id.strip_prefix(&format!("{peer}/"))) + .unwrap_or(session_id); // Started past this host's lane cap: a warning, never a refusal (`hub/admission.rs`). if reused.is_none() && let Some(w) = From 1eb87134904bf0a866c6a00bb09b6f7c783a4ac3 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 23:05:03 +0000 Subject: [PATCH 09/26] test(acpmux): make remote group CLI fixture executable Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/tests/remote_group_cli.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/cmux-tui/crates/acpmux/tests/remote_group_cli.rs b/cmux-tui/crates/acpmux/tests/remote_group_cli.rs index 47f61af871a5..109359923938 100644 --- a/cmux-tui/crates/acpmux/tests/remote_group_cli.rs +++ b/cmux-tui/crates/acpmux/tests/remote_group_cli.rs @@ -70,6 +70,7 @@ async fn remote_group_cli_opens_the_remote_id_locally_and_reuses_its_row() { let cmux = root.join("bin/cmux"); std::fs::write(&cmux, "#!/bin/sh\nprintf '%s\\n' \"$@\" > \"$GROUP_CAPTURE\"\n").unwrap(); std::fs::set_permissions(&cmux, std::fs::Permissions::from_mode(0o700)).unwrap(); + std::fs::set_permissions(&cmux, std::fs::Permissions::from_mode(0o700)).unwrap(); let mut previous = None; let mut unnamed = None; for (attempt, (peer, named)) in From 17ee78cda4c73df05e763576d2c85b5db8262578 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 23:13:43 +0000 Subject: [PATCH 10/26] fix(acpmux): keep qualified route after remote start Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- .../crates/acpmux/src/cli/run/standard.rs | 20 +++++++++++-------- 1 file changed, 12 insertions(+), 8 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/cli/run/standard.rs b/cmux-tui/crates/acpmux/src/cli/run/standard.rs index 7e9a5f3e5bbe..a48afe1cf160 100644 --- a/cmux-tui/crates/acpmux/src/cli/run/standard.rs +++ b/cmux-tui/crates/acpmux/src/cli/run/standard.rs @@ -492,6 +492,10 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R .as_str() .and_then(|peer| session_id.strip_prefix(&format!("{peer}/"))) .unwrap_or(session_id); + // Keep the host-qualified route for every follow-up request. A peer may + // legitimately return the same raw id as a local session; using the raw id + // here would send prompts, info and lifecycle controls to the local chat. + let route_id = id.as_str(); // Started past this host's lane cap: a warning, never a refusal (`hub/admission.rs`). if reused.is_none() && let Some(w) = @@ -531,7 +535,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R if let Some(text) = &first { let prompt = PromptId::new(args.prompt_id.clone()); let accepted = - queue_prompt(client, &session_id, text, false, None, &prompt).await?; + queue_prompt(client, route_id, text, false, None, &prompt).await?; if !json_out { warn_host_pressure(&accepted); } @@ -559,7 +563,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R let first = launch.first_prompt(&args, &v); if json_out && !one_shot { let mut info = client - .request(method::MUX_INFO, json!({"sessionId": session_id})) + .request(method::MUX_INFO, json!({"sessionId": route_id})) .await .unwrap_or(v.clone()); if let Some(w) = v.pointer("/_meta/acpmux/worktree") { @@ -585,7 +589,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R // Before it works: an allow it asks for is the app's alone. if on_permission == OnPermission::Wait && let Ok(info) = - client.request(method::MUX_INFO, json!({"sessionId": session_id})).await + client.request(method::MUX_INFO, json!({"sessionId": route_id})).await && let Some(note) = crate::question_answer::app_allow_note(&info, &name) { eprintln!("{note}"); @@ -603,12 +607,12 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R }; show(); let outcome = - collect_reply(client.clone(), &session_id, &text, opts, &prompt).await; + collect_reply(client.clone(), route_id, &text, opts, &prompt).await; if ephemeral { let _ = client .request( method::MUX_KILL, - json!({"sessionId": session_id, "purge": true}), + json!({"sessionId": route_id, "purge": true}), ) .await; } @@ -632,7 +636,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R // The prompt is queued before the sidebar step, so a // session never runs without its instructions. let accepted = - queue_prompt(client, &session_id, &text, false, None, &prompt).await?; + queue_prompt(client, route_id, &text, false, None, &prompt).await?; if !json_out { warn_host_pressure(&accepted); } @@ -649,7 +653,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R show(); return stream_prompt( client, - &session_id, + route_id, &text, false, false, @@ -664,7 +668,7 @@ pub(super) async fn run(cmd: Command, json_out: bool, suppress_reads: bool) -> R if args.detach || json_out { return Ok(()); } - crate::tui::run(client, Some(session_id.to_owned())).await + crate::tui::run(client, Some(route_id.to_owned())).await } Command::Send { session, From 05507d37facad569d595573d27a9fecdb8223454 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 23:19:19 +0000 Subject: [PATCH 11/26] fix(acpmux): treat qualified reuse misses as absent Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/cli/run/launch.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/cmux-tui/crates/acpmux/src/cli/run/launch.rs b/cmux-tui/crates/acpmux/src/cli/run/launch.rs index 662724fd5a45..7fb33ac00d4e 100644 --- a/cmux-tui/crates/acpmux/src/cli/run/launch.rs +++ b/cmux-tui/crates/acpmux/src/cli/run/launch.rs @@ -248,7 +248,10 @@ pub(super) async fn existing(client: &Client, key: &str) -> Result Ok(_) => Ok(None), Err(e) => { let m = e.to_string().to_lowercase(); - let none = m.contains("no session") || m.contains("not found") || m.contains("matches"); + let none = m.contains("no session") + || m.contains("no local session") + || m.contains("not found") + || m.contains("matches"); if none { Ok(None) } else { Err(e) } } } From 8daf383cb2e875ad2a13a4ba2f84e26b0eb0239c Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 23:27:25 +0000 Subject: [PATCH 12/26] fix(acpmux): qualify remote child tags Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/client.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/cmux-tui/crates/acpmux/src/client.rs b/cmux-tui/crates/acpmux/src/client.rs index a035306fcfa9..a66cf42cd4b5 100644 --- a/cmux-tui/crates/acpmux/src/client.rs +++ b/cmux-tui/crates/acpmux/src/client.rs @@ -216,7 +216,9 @@ impl Client { if !reply_lacks_lineage(reply) { return; } - let Some(id) = reply.get("sessionId").and_then(Value::as_str) else { return }; + // A peer may return the same raw id as a local session. Keep the owner + // namespace on the tag request just like prompts and lifecycle calls. + let Some(id) = crate::cli::resolve::routed_id(reply).ok() else { return }; let set = json!({ crate::hub::views::SPAWNED_BY_TAG: parent }); let _ = self.send_request(method::MUX_TAG, json!({"sessionId": id, "set": set})).await; } From 8465cced0d3e9329e6881e78bd1bff235699b4d9 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 23:33:31 +0000 Subject: [PATCH 13/26] fix(acpmux): select qualified remote sessions in tui Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/tui/run.rs | 21 ++++++++++++++++++++- 1 file changed, 20 insertions(+), 1 deletion(-) diff --git a/cmux-tui/crates/acpmux/src/tui/run.rs b/cmux-tui/crates/acpmux/src/tui/run.rs index 6075d10fb1eb..2572797a5551 100644 --- a/cmux-tui/crates/acpmux/src/tui/run.rs +++ b/cmux-tui/crates/acpmux/src/tui/run.rs @@ -48,7 +48,7 @@ pub async fn run(client: Arc, initial: Option) -> Result<()> { if let Some(i) = app .sessions .iter() - .position(|s| s.get("sessionId").and_then(Value::as_str) == Some(&id)) + .position(|s| initial_session_matches(s, &id)) { app.select(i); } @@ -229,6 +229,25 @@ pub async fn run(client: Arc, initial: Option) -> Result<()> { result } +fn initial_session_matches(session: &Value, id: &str) -> bool { + session.get("sessionId").and_then(Value::as_str) == Some(id) + || session.get("qualifiedSessionId").and_then(Value::as_str) == Some(id) +} + +#[cfg(test)] +mod tests { + use super::initial_session_matches; + use serde_json::json; + + #[test] + fn initial_selection_accepts_a_peer_qualified_id() { + let session = json!({"sessionId": "raw", "qualifiedSessionId": "peer/raw"}); + assert!(initial_session_matches(&session, "peer/raw")); + assert!(initial_session_matches(&session, "raw")); + assert!(!initial_session_matches(&session, "other/raw")); + } +} + pub(super) fn make_app( client: Arc, tx: mpsc::UnboundedSender, From 254e0926d803b48c825feb8400ae1473dc27d87b Mon Sep 17 00:00:00 2001 From: Leo Li Date: Wed, 28 Oct 2026 23:44:18 +0000 Subject: [PATCH 14/26] fix(acpmux): retain peer route in tui selection Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/tui/mod.rs | 28 ++++++++++++++++++++++++--- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/tui/mod.rs b/cmux-tui/crates/acpmux/src/tui/mod.rs index 2b8d7045b012..4136fcfbce6b 100644 --- a/cmux-tui/crates/acpmux/src/tui/mod.rs +++ b/cmux-tui/crates/acpmux/src/tui/mod.rs @@ -416,9 +416,7 @@ impl App { true } pub(super) fn selected_id(&self) -> Option { - self.selected_session() - .and_then(|s| s.get("sessionId").and_then(Value::as_str)) - .map(str::to_owned) + self.selected_session().and_then(session_route_id) } pub(super) fn selected_name(&self) -> String { if let Some(d) = self.draft() { @@ -711,6 +709,30 @@ impl App { } } +fn session_route_id(session: &Value) -> Option { + session + .get("peer") + .and_then(Value::as_str) + .and_then(|_| session.get("qualifiedSessionId").and_then(Value::as_str)) + .or_else(|| session.get("sessionId").and_then(Value::as_str)) + .map(str::to_owned) +} + +#[cfg(test)] +mod session_route_tests { + use super::session_route_id; + use serde_json::json; + + #[test] + fn peer_rows_keep_the_qualified_route_for_tui_requests() { + assert_eq!( + session_route_id(&json!({"sessionId": "raw", "peer": "box", "qualifiedSessionId": "box/raw"})), + Some("box/raw".into()) + ); + assert_eq!(session_route_id(&json!({"sessionId": "local"})), Some("local".into())); + } +} + #[cfg(all(test, unix))] pub(super) mod interaction_tests { use super::*; From b5f0d4f9205abd2142a0507fba05f85f5f05640f Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 04:55:13 +0000 Subject: [PATCH 15/26] test(acpmux): cover peer tui events and selection identity Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/tui/events.rs | 44 ++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/cmux-tui/crates/acpmux/src/tui/events.rs b/cmux-tui/crates/acpmux/src/tui/events.rs index 5ea9d041d119..7b429c995cb1 100644 --- a/cmux-tui/crates/acpmux/src/tui/events.rs +++ b/cmux-tui/crates/acpmux/src/tui/events.rs @@ -301,3 +301,47 @@ impl App { } } } + +#[cfg(all(test, unix))] +mod peer_route_tests { + use super::*; + + #[tokio::test] + async fn peer_events_update_selected_transcript_and_preserve_selection_when_sorted() { + let (mut app, _requests) = interaction_tests::app().await; + app.drafts.clear(); + let local = json!({"sessionId": "raw", "name": "local", "updatedAt": 2}); + let remote = json!({"sessionId": "raw", "peer": "worker", "qualifiedSessionId": "worker/raw", "name": "worker/remote", "updatedAt": 1}); + app.sessions = vec![local.clone(), remote.clone()]; + app.selected = 1; + app.transcripts.insert("worker/raw".into(), Transcript::default()); + app.on_notification(method::SESSION_UPDATE, json!({ + "sessionId": "raw", "peer": "worker", + "update": {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "remote reply"}} + })); + assert!( + !app.transcripts["worker/raw"].items.is_empty(), + "remote update must reach the selected transcript" + ); + assert!( + !app.transcripts.contains_key("raw"), + "remote output must not enter a local transcript" + ); + let mut changed = remote.clone(); + changed["updatedAt"] = json!(3); + app.on_notification( + method::MUX_SESSION_CHANGED, + json!({"sessionId": "raw", "peer": "worker", "session": changed, "kind": "turn_start"}), + ); + assert_eq!(app.sessions.len(), 2); + assert_eq!(app.selected, 0); + assert_eq!(app.selected_id().as_deref(), Some("worker/raw")); + assert_eq!(app.sessions[1], local); + // The returned list may already have a new order before sort_sessions runs. + app.on_msg(AppMsg::Sessions(vec![local.clone(), remote])); + assert_eq!(app.selected_id().as_deref(), Some("worker/raw")); + app.on_notification(method::MUX_SESSION_CHANGED, json!({"sessionId": "raw", "peer": "worker", "session": {"sessionId": "raw"}, "kind": "purged"})); + assert_eq!(app.sessions, vec![local]); + assert!(!app.transcripts.contains_key("worker/raw")); + } +} From 7d8631e70ac3f17c160b3b644d951827dacead31 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 04:56:46 +0000 Subject: [PATCH 16/26] fix(acpmux): keep peer tui events and selection on one route Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/server/mod.rs | 19 ++++++++--- cmux-tui/crates/acpmux/src/tui/actions.rs | 2 +- cmux-tui/crates/acpmux/src/tui/events.rs | 33 +++++++++---------- cmux-tui/crates/acpmux/src/tui/menu.rs | 6 ++-- cmux-tui/crates/acpmux/src/tui/mod.rs | 13 +++----- .../crates/acpmux/src/tui/render/sidebar.rs | 5 +-- cmux-tui/crates/acpmux/src/tui/run.rs | 14 +++----- cmux-tui/crates/acpmux/src/tui/session_ops.rs | 6 ++-- .../acpmux/tests/session_control_peer_cli.rs | 18 ++++++++++ 9 files changed, 66 insertions(+), 50 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/server/mod.rs b/cmux-tui/crates/acpmux/src/server/mod.rs index 04cfd1f0db39..dfd6184d33e6 100644 --- a/cmux-tui/crates/acpmux/src/server/mod.rs +++ b/cmux-tui/crates/acpmux/src/server/mod.rs @@ -1492,6 +1492,17 @@ fn deliver(hub: &Hub, conn: &Conn, ev: HubEvent) { chats::push_activity(hub, conn, &ev); processes::push_update(conn, &ev); let rec = &ev.record; + // Attached peer updates retain the raw wire id, but also identify its owner so + // clients can key live transcripts by the same route as their session rows. + let routed = |mut params: Value| { + if let Some(obj) = params.as_object_mut() { + obj.remove("peer"); + if let Some(remote) = &ev.remote { + obj.insert("peer".into(), json!(remote.peer)); + } + } + params + }; let watching = conn.watch_all.load(Ordering::SeqCst); let sub = conn.sub_opts(&ev.session_id); let attached = sub.is_some(); @@ -1505,7 +1516,7 @@ fn deliver(hub: &Hub, conn: &Conn, ev: HubEvent) { if sub.filter.matches(rec) { conn.send(&Message::notification( method::MUX_EVENT, - event_value(&ev.session_id, rec), + routed(event_value(&ev.session_id, rec)), )); } } else { @@ -1521,19 +1532,19 @@ fn deliver(hub: &Hub, conn: &Conn, ev: HubEvent) { &mut params, json!({"seq": rec.seq, "at": rec.at, "kind": rec.kind}), ); - conn.send(&Message::notification(method::SESSION_UPDATE, params)); + conn.send(&Message::notification(method::SESSION_UPDATE, routed(params))); } if rec.dir == "mux" { conn.send(&Message::notification( method::MUX_EVENT, - event_value(&ev.session_id, rec), + routed(event_value(&ev.session_id, rec)), )); } } if rec.dir == "mux" && is_permission_ask(&rec.kind) { conn.send(&Message::notification( method::MUX_PERMISSION_PENDING, - permission_pending(&ev.session_id, rec, "attach"), + routed(permission_pending(&ev.session_id, rec, "attach")), )); } } diff --git a/cmux-tui/crates/acpmux/src/tui/actions.rs b/cmux-tui/crates/acpmux/src/tui/actions.rs index 20beb87725eb..fa556b1fd56f 100644 --- a/cmux-tui/crates/acpmux/src/tui/actions.rs +++ b/cmux-tui/crates/acpmux/src/tui/actions.rs @@ -218,7 +218,7 @@ impl App { note: String::new(), }); for s in &self.sessions { - let id = s.get("sessionId").and_then(Value::as_str).unwrap_or("").to_owned(); + let id = session_route_id(s).unwrap_or_default(); let title = render::session_title(s); let cwd = s.get("cwd").and_then(Value::as_str).unwrap_or(""); let project = match s.get("peer").and_then(Value::as_str) { diff --git a/cmux-tui/crates/acpmux/src/tui/events.rs b/cmux-tui/crates/acpmux/src/tui/events.rs index 7b429c995cb1..3bc72b32147d 100644 --- a/cmux-tui/crates/acpmux/src/tui/events.rs +++ b/cmux-tui/crates/acpmux/src/tui/events.rs @@ -6,7 +6,7 @@ impl App { // --------------------------------------------------------- inbound pub(super) fn on_notification(&mut self, m: &str, p: Value) { - let sid = p.get("sessionId").and_then(Value::as_str).map(str::to_owned); + let sid = session_route_id(&p); match m { method::SESSION_UPDATE => { if let Some(id) = sid { @@ -27,7 +27,7 @@ impl App { let who = self .sessions .iter() - .find(|s| s.get("sessionId").and_then(Value::as_str) == Some(&id)) + .find(|s| session_route_id(s).as_ref() == Some(&id)) .and_then(|s| s.get("name").and_then(Value::as_str)) .unwrap_or("session") .to_owned(); @@ -46,19 +46,16 @@ impl App { } method::MUX_SESSION_CHANGED => { if let Some(s) = p.get("session") { - let id = s.get("sessionId").and_then(Value::as_str).unwrap_or("").to_owned(); + let Some(id) = sid.clone().or_else(|| session_route_id(s)) else { return }; if p.get("kind").and_then(Value::as_str) == Some("purged") { - self.sessions - .retain(|x| x.get("sessionId").and_then(Value::as_str) != Some(&id)); + self.sessions.retain(|x| session_route_id(x).as_ref() != Some(&id)); self.transcripts.remove(&id); self.attention.remove(&id); self.selected = self.selected.min(self.row_count().saturating_sub(1)); return; } - if let Some(slot) = self - .sessions - .iter_mut() - .find(|x| x.get("sessionId").and_then(Value::as_str) == Some(&id)) + if let Some(slot) = + self.sessions.iter_mut().find(|x| session_route_id(x).as_ref() == Some(&id)) { *slot = s.clone(); } else { @@ -87,9 +84,7 @@ impl App { let who = sid .as_deref() .and_then(|id| { - self.sessions - .iter() - .find(|s| s.get("sessionId").and_then(Value::as_str) == Some(id)) + self.sessions.iter().find(|s| session_route_id(s).as_deref() == Some(id)) }) .and_then(|s| s.get("name").and_then(Value::as_str)) .unwrap_or("?"); @@ -128,17 +123,18 @@ impl App { } pub(super) fn sort_sessions(&mut self) { - let selected_id = self.selected_id(); + self.sort_sessions_preserving(self.selected_id()); + } + + fn sort_sessions_preserving(&mut self, selected_id: Option) { self.sessions.sort_by(|a, b| { let ua = a.get("updatedAt").and_then(Value::as_u64).unwrap_or(0); let ub = b.get("updatedAt").and_then(Value::as_u64).unwrap_or(0); ub.cmp(&ua) }); if let Some(id) = selected_id - && let Some(i) = self - .sessions - .iter() - .position(|s| s.get("sessionId").and_then(Value::as_str) == Some(&id)) + && let Some(i) = + self.sessions.iter().position(|s| session_route_id(s).as_ref() == Some(&id)) { self.selected = i + self.drafts.len(); } @@ -148,8 +144,9 @@ impl App { match msg { AppMsg::Sessions(list) => { let had = !self.sessions.is_empty(); + let selected_id = self.selected_id(); self.sessions = list; - self.sort_sessions(); + self.sort_sessions_preserving(selected_id); if !had && !self.sessions.is_empty() && self.drafts.is_empty() { self.select(0); } diff --git a/cmux-tui/crates/acpmux/src/tui/menu.rs b/cmux-tui/crates/acpmux/src/tui/menu.rs index d32d10f998f1..6b36f30ebd14 100644 --- a/cmux-tui/crates/acpmux/src/tui/menu.rs +++ b/cmux-tui/crates/acpmux/src/tui/menu.rs @@ -284,10 +284,8 @@ impl App { match action { MenuAction::Run(a) => self.run_action(a, &[]), MenuAction::Session(id, op) => { - if let Some(i) = self - .sessions - .iter() - .position(|s| s.get("sessionId").and_then(Value::as_str) == Some(&id)) + if let Some(i) = + self.sessions.iter().position(|s| session_route_id(s).as_ref() == Some(&id)) { self.select(i + self.drafts.len()); } diff --git a/cmux-tui/crates/acpmux/src/tui/mod.rs b/cmux-tui/crates/acpmux/src/tui/mod.rs index 4136fcfbce6b..d39dd5ca01bc 100644 --- a/cmux-tui/crates/acpmux/src/tui/mod.rs +++ b/cmux-tui/crates/acpmux/src/tui/mod.rs @@ -581,7 +581,7 @@ impl App { if (m == method::SESSION_NEW || m == method::SESSION_FORK || m == method::MUX_IMPORT) - && let Some(id) = v.get("sessionId").and_then(Value::as_str) + && let Some(id) = session_route_id(&v) { let _ = tx.send(AppMsg::Created(id.to_owned())); } @@ -710,12 +710,7 @@ impl App { } fn session_route_id(session: &Value) -> Option { - session - .get("peer") - .and_then(Value::as_str) - .and_then(|_| session.get("qualifiedSessionId").and_then(Value::as_str)) - .or_else(|| session.get("sessionId").and_then(Value::as_str)) - .map(str::to_owned) + crate::cli::resolve::routed_id(session).ok() } #[cfg(test)] @@ -726,7 +721,9 @@ mod session_route_tests { #[test] fn peer_rows_keep_the_qualified_route_for_tui_requests() { assert_eq!( - session_route_id(&json!({"sessionId": "raw", "peer": "box", "qualifiedSessionId": "box/raw"})), + session_route_id( + &json!({"sessionId": "raw", "peer": "box", "qualifiedSessionId": "box/raw"}) + ), Some("box/raw".into()) ); assert_eq!(session_route_id(&json!({"sessionId": "local"})), Some("local".into())); diff --git a/cmux-tui/crates/acpmux/src/tui/render/sidebar.rs b/cmux-tui/crates/acpmux/src/tui/render/sidebar.rs index b5bb87dae255..e77edf6f6846 100644 --- a/cmux-tui/crates/acpmux/src/tui/render/sidebar.rs +++ b/cmux-tui/crates/acpmux/src/tui/render/sidebar.rs @@ -6,6 +6,7 @@ //! line per session with a status mark at the right edge, and "Add host" //! pinned to the bottom. No vertical rule: the shade is the edge. +use super::super::session_route_id; use super::*; enum Entry { @@ -224,8 +225,8 @@ pub(super) fn draw_sidebar(f: &mut ratatui::Frame, area: Rect, app: &mut App) { let name = name_owned.as_str(); let status = s.get("status").and_then(Value::as_str).unwrap_or(""); let pending = s.get("pendingPermissions").and_then(Value::as_u64).unwrap_or(0) > 0; - let sid = s.get("sessionId").and_then(Value::as_str).unwrap_or(""); - let dot = app.attention.get(sid).copied().unwrap_or(0); + let sid = session_route_id(s).unwrap_or_default(); + let dot = app.attention.get(&sid).copied().unwrap_or(0); let hot = hovered_row(app.hover, y); let row_style = if is_sel { ground.bg(c.sidebar_selected_bg).fg(c.sidebar_selected_fg) diff --git a/cmux-tui/crates/acpmux/src/tui/run.rs b/cmux-tui/crates/acpmux/src/tui/run.rs index 2572797a5551..1fe8608180fc 100644 --- a/cmux-tui/crates/acpmux/src/tui/run.rs +++ b/cmux-tui/crates/acpmux/src/tui/run.rs @@ -45,11 +45,7 @@ pub async fn run(client: Arc, initial: Option) -> Result<()> { }); } if let Some(id) = initial { - if let Some(i) = app - .sessions - .iter() - .position(|s| initial_session_matches(s, &id)) - { + if let Some(i) = app.sessions.iter().position(|s| initial_session_matches(s, &id)) { app.select(i); } } else if !app.sessions.is_empty() { @@ -165,7 +161,7 @@ pub async fn run(client: Arc, initial: Option) -> Result<()> { match n { Some(Message::Notification { method: m, params }) if m != method::MUX_DISCONNECTED => { let p = params.unwrap_or(Value::Null); - let selected = p.get("sessionId").and_then(Value::as_str) == app.selected_id().as_deref(); + let selected = session_route_id(&p) == app.selected_id(); let visible = match m.as_str() { method::SESSION_UPDATE => selected, method::MUX_EVENT => selected || matches!(p.get("kind").and_then(Value::as_str), Some("turn_end" | "turn_error")), @@ -197,10 +193,8 @@ pub async fn run(client: Arc, initial: Option) -> Result<()> { } } if let Some(id) = app.pending_select.take() { - if let Some(i) = app - .sessions - .iter() - .position(|s| s.get("sessionId").and_then(Value::as_str) == Some(&id)) + if let Some(i) = + app.sessions.iter().position(|s| session_route_id(s).as_ref() == Some(&id)) { app.drafts.retain(|d| !d.creating); app.select(i + app.drafts.len()); diff --git a/cmux-tui/crates/acpmux/src/tui/session_ops.rs b/cmux-tui/crates/acpmux/src/tui/session_ops.rs index eceb73595ab8..5aeaea0af97a 100644 --- a/cmux-tui/crates/acpmux/src/tui/session_ops.rs +++ b/cmux-tui/crates/acpmux/src/tui/session_ops.rs @@ -200,7 +200,7 @@ impl App { tokio::spawn(async move { match client.request(method::SESSION_NEW, params).await { Ok(v) => { - if let Some(id) = v.get("sessionId").and_then(Value::as_str) { + if let Some(id) = session_route_id(&v) { let _ = tx.send(AppMsg::Created(id.to_owned())); let id = id.to_owned(); let c = client.clone(); @@ -271,7 +271,7 @@ impl App { tokio::spawn(async move { match client.request(method::SESSION_NEW, params).await { Ok(v) => { - if let Some(id) = v.get("sessionId").and_then(Value::as_str) { + if let Some(id) = session_route_id(&v) { let _ = tx.send(AppMsg::Created(id.to_owned())); if !first.is_empty() { let id = id.to_owned(); @@ -904,7 +904,7 @@ impl App { if let Some(i) = self .sessions .iter() - .position(|s| s.get("sessionId").and_then(Value::as_str) == Some(id)) + .position(|s| session_route_id(s).as_deref() == Some(id)) { self.select(i + ndrafts); self.focus = Focus::Input; diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs index 72d3dc648532..c7466f71a343 100644 --- a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -312,6 +312,24 @@ async fn lifecycle_commands_use_peer_stable_ids_and_refuse_ambiguous_names() { let started: Value = serde_json::from_str(&stdout).unwrap(); assert_eq!(started["policy"], "ask", "{started}"); + // The TUI keys its selection by route; attached live events must name that owner. + let mut observer = client(a.clone()).await; + let route = format!("b/{}", started["sessionId"].as_str().unwrap()); + observer.call(method::MUX_ATTACH, json!({"sessionId": route, "limit": 0})).await; + observer + .send( + method::SESSION_PROMPT, + json!({"sessionId": route, "prompt": [{"type": "text", "text": "owner event"}]}), + ) + .await; + let update = observer + .wait(method::SESSION_UPDATE, |p| p["update"]["content"]["text"] == "echo: owner event") + .await; + assert_eq!(update["peer"], "b"); + assert_eq!(update["sessionId"], started["sessionId"]); + let end = observer.wait(method::MUX_EVENT, |p| p["kind"] == "turn_end").await; + assert_eq!(end["peer"], "b"); + let (ok, stdout, stderr) = cli( &root, &socket, From 46586af1537ccbc83a2a0d21d16de67bfba40e45 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 05:07:10 +0000 Subject: [PATCH 17/26] fix(acpmux): keep tui test module after production items Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/tui/run.rs | 28 +++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/tui/run.rs b/cmux-tui/crates/acpmux/src/tui/run.rs index 1fe8608180fc..b38a4269696f 100644 --- a/cmux-tui/crates/acpmux/src/tui/run.rs +++ b/cmux-tui/crates/acpmux/src/tui/run.rs @@ -228,20 +228,6 @@ fn initial_session_matches(session: &Value, id: &str) -> bool { || session.get("qualifiedSessionId").and_then(Value::as_str) == Some(id) } -#[cfg(test)] -mod tests { - use super::initial_session_matches; - use serde_json::json; - - #[test] - fn initial_selection_accepts_a_peer_qualified_id() { - let session = json!({"sessionId": "raw", "qualifiedSessionId": "peer/raw"}); - assert!(initial_session_matches(&session, "peer/raw")); - assert!(initial_session_matches(&session, "raw")); - assert!(!initial_session_matches(&session, "other/raw")); - } -} - pub(super) fn make_app( client: Arc, tx: mpsc::UnboundedSender, @@ -397,3 +383,17 @@ async fn reconnect(app: &mut App) -> Result } } } + +#[cfg(test)] +mod tests { + use super::initial_session_matches; + use serde_json::json; + + #[test] + fn initial_selection_accepts_a_peer_qualified_id() { + let session = json!({"sessionId": "raw", "qualifiedSessionId": "peer/raw"}); + assert!(initial_session_matches(&session, "peer/raw")); + assert!(initial_session_matches(&session, "raw")); + assert!(!initial_session_matches(&session, "other/raw")); + } +} From 4320818e191b2924230c9d8af232e6de3f705a62 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 05:28:42 +0000 Subject: [PATCH 18/26] test(acpmux): isolate cli peer streams on qualified ids Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- .../acpmux/tests/session_control_peer_cli.rs | 175 ++++++++++++++++++ 1 file changed, 175 insertions(+) diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs index c7466f71a343..c2947fa20841 100644 --- a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -17,8 +17,10 @@ use futures_util::{SinkExt, StreamExt}; use serde_json::{Value, json}; use std::os::unix::fs::PermissionsExt; use std::path::{Path, PathBuf}; +use std::process::Stdio; use std::sync::Arc; use std::time::Duration; +use tokio::io::AsyncReadExt; use tokio::sync::mpsc; const PERSON_KEY: &str = "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"; @@ -182,6 +184,89 @@ async fn lying_peer(session_id: &str) -> (u16, mpsc::UnboundedReceiver) (port, received) } +/// A peer that reuses a local raw id and emits a live transcript after every attach. +/// The delayed event lets a colliding local session publish a notification at the +/// same time, so CLI output filters must use the peer-qualified route. +async fn event_peer(session_id: &str) -> u16 { + use tokio_tungstenite::tungstenite::Message as Ws; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let session_id = session_id.to_owned(); + tokio::spawn(async move { + while let Ok((tcp, _)) = listener.accept().await { + let Ok(mut ws) = tokio_tungstenite::accept_async(tcp).await else { continue }; + let id = session_id.clone(); + tokio::spawn(async move { + while let Some(Ok(Ws::Text(text))) = ws.next().await { + let Ok(Message::Request { id: request_id, method: m, .. }) = + Message::parse(&text) + else { + continue; + }; + let detail = json!({ + "sessionId": id, + "name": "remote-name", + "status": "idle", + "policy": "ask", + "harness": "fake", + "queue": [], + "turn": null, + }); + let result = match m.as_str() { + method::INITIALIZE => { + json!({"protocolVersion": 1, "_meta": {"acpmux": {"version": "0", "build": "event-lying", "origin": "peer"}}}) + } + method::MUX_WATCH => { + json!({"sessions": [{"sessionId": id, "name": "remote-name", "harness": "fake", "status": "idle"}]}) + } + method::MUX_INFO => detail.clone(), + method::MUX_ATTACH => { + json!({"session": detail, "events": [], "hasMore": false}) + } + method::SESSION_PROMPT => { + json!({"stopReason": "steered", "_meta": {"acpmux": {"turnId": "remote-turn"}}}) + } + _ => json!({}), + }; + let line = Message::ok(request_id, result).to_line(); + if ws.send(Ws::Text(line.trim_end().to_owned().into())).await.is_err() { + return; + } + if m != method::MUX_ATTACH { + continue; + } + tokio::time::sleep(Duration::from_millis(350)).await; + let update = Message::notification( + method::SESSION_UPDATE, + json!({ + "sessionId": id, + "update": {"sessionUpdate": "agent_message_chunk", "content": {"type": "text", "text": "REMOTE-STREAM"}}, + "_meta": {"acpmux": {"seq": 1, "at": 1}}, + }), + ); + let end = Message::notification( + method::MUX_EVENT, + json!({ + "sessionId": id, + "seq": 2, + "at": 2, + "dir": "mux", + "kind": "turn_result", + "msg": {"turnId": "remote-turn", "status": "completed", "stopReason": "end_turn"}, + }), + ); + for note in [update, end] { + if ws.send(Ws::Text(note.to_line().trim_end().to_owned().into())).await.is_err() { + return; + } + } + } + }); + } + }); + port +} + async fn new_session(client: &mut Rpc, root: &Path, name: &str) -> String { client .call( @@ -216,6 +301,37 @@ async fn cli(root: &Path, socket: &Path, args: &[&str]) -> (bool, String, String ) } +async fn cli_until(root: &Path, socket: &Path, args: &[&str], wait: Duration) -> (bool, String, String) { + let mut command = daemon_env::tokio_command(root, daemon_env::Agents::Deny); + command + .env("ACPMUX_SOCKET", socket) + .env("PATH", format!("{}:{}", root.join("bin").display(), daemon_env::bare_path(&[]))) + .env("GROUP_CAPTURE", root.join("group-args")) + .args(args) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + let mut child = command.spawn().expect("start CLI"); + let mut stdout = child.stdout.take().expect("CLI stdout"); + let mut stderr = child.stderr.take().expect("CLI stderr"); + let stdout_task = tokio::spawn(async move { + let mut bytes = Vec::new(); + stdout.read_to_end(&mut bytes).await.unwrap(); + bytes + }); + let stderr_task = tokio::spawn(async move { + let mut bytes = Vec::new(); + stderr.read_to_end(&mut bytes).await.unwrap(); + bytes + }); + tokio::time::sleep(wait).await; + let _ = child.kill().await; + let status = child.wait().await.unwrap(); + let stdout = String::from_utf8_lossy(&stdout_task.await.unwrap()).into_owned(); + let stderr = String::from_utf8_lossy(&stderr_task.await.unwrap()).into_owned(); + (status.success(), stdout, stderr) +} + #[tokio::test] async fn lifecycle_commands_use_peer_stable_ids_and_refuse_ambiguous_names() { let root = PathBuf::from("/tmp").join(format!("amx-scp-{}", std::process::id())); @@ -443,6 +559,65 @@ async fn lifecycle_commands_use_peer_stable_ids_and_refuse_ambiguous_names() { std::fs::remove_dir_all(root).unwrap(); } +#[tokio::test] +async fn cli_output_ignores_local_events_when_a_peer_reuses_the_raw_id() { + let root = PathBuf::from("/tmp").join(format!("amx-scp-output-{}", std::process::id())); + std::fs::create_dir_all(&root).unwrap(); + let a = hub(&root); + let mut ca = client(a.clone()).await; + let local_id = new_session(&mut ca, &root, "local-output").await; + ca.call(method::MUX_ATTACH, json!({"sessionId": local_id, "limit": 0})).await; + let port = event_peer(&local_id).await; + a.add_peer("lying", &format!("ws://127.0.0.1:{port}"), Some("tok".into()), None, true) + .await + .unwrap(); + let socket = root.join("a.sock"); + let server = tokio::spawn(serve_unix(a.clone(), bind_unix(&socket).await.unwrap())); + + for args in [ + vec!["send", "lying/remote-name", "remote prompt", "--stall", "5"], + vec!["send", "--quiet", "lying/remote-name", "remote prompt"], + ] { + ca.send( + method::SESSION_PROMPT, + json!({"sessionId": local_id, "prompt": [{"type": "text", "text": "ticking:4:300"}]}), + ) + .await; + ca.wait(method::SESSION_UPDATE, |p| { + p["sessionId"] == local_id && p["update"]["content"]["text"] == "tick0 " + }) + .await; + let (ok, stdout, stderr) = cli(&root, &socket, &args).await; + assert!(ok, "{stdout}{stderr}"); + assert!(stdout.contains("REMOTE-STREAM"), "{stdout}{stderr}"); + assert!(!stdout.contains("tick"), "local event leaked into peer output: {stdout}{stderr}"); + } + + ca.send( + method::SESSION_PROMPT, + json!({"sessionId": local_id, "prompt": [{"type": "text", "text": "ticking:4:300"}]}), + ) + .await; + ca.wait(method::SESSION_UPDATE, |p| { + p["sessionId"] == local_id && p["update"]["content"]["text"] == "tick0 " + }) + .await; + let (_, stdout, stderr) = cli_until( + &root, + &socket, + &["attach", "--plain", "lying/remote-name"], + Duration::from_millis(900), + ) + .await; + assert!(stdout.contains("REMOTE-STREAM"), "{stdout}{stderr}"); + assert!(!stdout.contains("tick"), "local event leaked into plain attach: {stdout}{stderr}"); + + server.abort(); + a.end_agents_at_shutdown(Default::default()).unwrap(); + a.shutdown_all().await; + std::fs::remove_dir_all(root).unwrap(); +} + #[tokio::test] async fn lifecycle_commands_keep_a_peer_owner_when_raw_ids_collide() { let root = PathBuf::from("/tmp").join(format!("amx-scp-collision-{}", std::process::id())); From 6ab2915dfb7cf045d3e619541fe4724ee8c3ced9 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 05:31:50 +0000 Subject: [PATCH 19/26] fix(acpmux): filter cli streams by peer-qualified ids Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/cli/orchestrate.rs | 2 +- cmux-tui/crates/acpmux/src/cli/output.rs | 14 +++++--------- cmux-tui/crates/acpmux/src/cli/run.rs | 2 +- .../acpmux/tests/session_control_peer_cli.rs | 13 +++++++++++-- 4 files changed, 18 insertions(+), 13 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/cli/orchestrate.rs b/cmux-tui/crates/acpmux/src/cli/orchestrate.rs index 3593a4abe450..c66fea0a9227 100644 --- a/cmux-tui/crates/acpmux/src/cli/orchestrate.rs +++ b/cmux-tui/crates/acpmux/src/cli/orchestrate.rs @@ -804,7 +804,7 @@ pub(crate) async fn tail( return Err(client.closed("following the session")); } let p = params.unwrap_or(Value::Null); - if p.get("sessionId").and_then(Value::as_str) != Some(id.as_str()) { + if crate::cli::resolve::routed_id(&p).ok().as_deref() != Some(id.as_str()) { continue; } let seq = if m == method::MUX_EVENT { diff --git a/cmux-tui/crates/acpmux/src/cli/output.rs b/cmux-tui/crates/acpmux/src/cli/output.rs index d0d89582b84d..42393f18866c 100644 --- a/cmux-tui/crates/acpmux/src/cli/output.rs +++ b/cmux-tui/crates/acpmux/src/cli/output.rs @@ -304,10 +304,7 @@ pub(crate) async fn stream_prompt( ) -> Result<()> { let mut notes = client.notifications().await.ok_or_else(|| anyhow!("notifications already taken"))?; - let attached = client.request(method::MUX_ATTACH, json!({"sessionId": id, "limit": 0})).await?; - // Requests retain the peer route; streamed notifications carry the owning daemon's - // unqualified id, as the attach reply identifies it. - let event_id = attached["session"]["sessionId"].as_str().unwrap_or(id); + client.request(method::MUX_ATTACH, json!({"sessionId": id, "limit": 0})).await?; let c = client.clone(); let params = prompt.params(id, text, steer, None); let turn = tokio::spawn(async move { c.request(method::SESSION_PROMPT, params).await }); @@ -369,7 +366,7 @@ pub(crate) async fn stream_prompt( let Message::Notification { method: m, params } = m else { continue }; if m == method::MUX_DISCONNECTED { return Err(prompt.closed(&client, id, "streaming the session")); } let p = params.unwrap_or(Value::Null); - if p.get("sessionId").and_then(Value::as_str) != Some(event_id) { continue; } + if crate::cli::resolve::routed_id(&p).ok().as_deref() != Some(id) { continue; } if let Some(answer) = steered.heard(&m, &p) { break answer; } if agent_activity(&m, &p) { activity = true; @@ -505,7 +502,7 @@ pub(crate) async fn plain_attach(client: Arc, id: &str) -> Result<()> { return Err(client.closed("streaming the session")); } let p = params.unwrap_or(Value::Null); - if p.get("sessionId").and_then(Value::as_str) != Some(id) { + if crate::cli::resolve::routed_id(&p).ok().as_deref() != Some(id) { continue; } match m.as_str() { @@ -659,8 +656,7 @@ async fn collect_once( ) -> Result { let mut notes = client.notifications().await.ok_or_else(|| anyhow!("notifications already taken"))?; - let attached = client.request(method::MUX_ATTACH, json!({"sessionId": id, "limit": 0})).await?; - let event_id = attached["session"]["sessionId"].as_str().unwrap_or(id); + client.request(method::MUX_ATTACH, json!({"sessionId": id, "limit": 0})).await?; let c = client.clone(); let params = prompt.params(id, text, opts.steer, None); let mut turn = tokio::spawn(async move { c.request(method::SESSION_PROMPT, params).await }); @@ -713,7 +709,7 @@ async fn collect_once( let Message::Notification { method: m, params } = m else { continue }; if m == method::MUX_DISCONNECTED { return Err(prompt.closed(&client, id, "streaming the session")); } let p = params.unwrap_or(Value::Null); - if p.get("sessionId").and_then(Value::as_str) != Some(event_id) { continue; } + if crate::cli::resolve::routed_id(&p).ok().as_deref() != Some(id) { continue; } if let Some(answer) = steered.heard(&m, &p) { break answer; } if agent_activity(&m, &p) { activity = true; diff --git a/cmux-tui/crates/acpmux/src/cli/run.rs b/cmux-tui/crates/acpmux/src/cli/run.rs index 57b9db9dfbd5..0d06b2b589ce 100644 --- a/cmux-tui/crates/acpmux/src/cli/run.rs +++ b/cmux-tui/crates/acpmux/src/cli/run.rs @@ -39,5 +39,5 @@ pub(crate) async fn resolve_id(client: &Client, key: &str) -> Result { e } })?; - Ok(v.get("sessionId").and_then(Value::as_str).unwrap_or(&key).to_owned()) + crate::cli::resolve::routed_id(&v) } diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs index c2947fa20841..85476a34a19e 100644 --- a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -256,7 +256,11 @@ async fn event_peer(session_id: &str) -> u16 { }), ); for note in [update, end] { - if ws.send(Ws::Text(note.to_line().trim_end().to_owned().into())).await.is_err() { + if ws + .send(Ws::Text(note.to_line().trim_end().to_owned().into())) + .await + .is_err() + { return; } } @@ -301,7 +305,12 @@ async fn cli(root: &Path, socket: &Path, args: &[&str]) -> (bool, String, String ) } -async fn cli_until(root: &Path, socket: &Path, args: &[&str], wait: Duration) -> (bool, String, String) { +async fn cli_until( + root: &Path, + socket: &Path, + args: &[&str], + wait: Duration, +) -> (bool, String, String) { let mut command = daemon_env::tokio_command(root, daemon_env::Agents::Deny); command .env("ACPMUX_SOCKET", socket) From 3c8d15dfafa9e0f3d3e41575bee57d87cae3fe9e Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 06:41:27 +0000 Subject: [PATCH 20/26] test(acpmux): cover qualified peer match waits and tails Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- .../acpmux/tests/session_control_peer_cli.rs | 41 +++++++++++-------- 1 file changed, 23 insertions(+), 18 deletions(-) diff --git a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs index 85476a34a19e..96724b11b2cd 100644 --- a/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs +++ b/cmux-tui/crates/acpmux/tests/session_control_peer_cli.rs @@ -586,6 +586,8 @@ async fn cli_output_ignores_local_events_when_a_peer_reuses_the_raw_id() { for args in [ vec!["send", "lying/remote-name", "remote prompt", "--stall", "5"], vec!["send", "--quiet", "lying/remote-name", "remote prompt"], + vec!["wait", "lying/remote-name", "--match", "REMOTE-STREAM", "--timeout", "2"], + vec!["wait", "lying/remote-name", "--regex", "REMOTE-STREAM|tick", "--timeout", "2"], ] { ca.send( method::SESSION_PROMPT, @@ -602,24 +604,27 @@ async fn cli_output_ignores_local_events_when_a_peer_reuses_the_raw_id() { assert!(!stdout.contains("tick"), "local event leaked into peer output: {stdout}{stderr}"); } - ca.send( - method::SESSION_PROMPT, - json!({"sessionId": local_id, "prompt": [{"type": "text", "text": "ticking:4:300"}]}), - ) - .await; - ca.wait(method::SESSION_UPDATE, |p| { - p["sessionId"] == local_id && p["update"]["content"]["text"] == "tick0 " - }) - .await; - let (_, stdout, stderr) = cli_until( - &root, - &socket, - &["attach", "--plain", "lying/remote-name"], - Duration::from_millis(900), - ) - .await; - assert!(stdout.contains("REMOTE-STREAM"), "{stdout}{stderr}"); - assert!(!stdout.contains("tick"), "local event leaked into plain attach: {stdout}{stderr}"); + for args in [ + vec!["attach", "--plain", "lying/remote-name"], + vec!["tail", "lying/remote-name", "--follow", "--since", "0"], + ] { + ca.send( + method::SESSION_PROMPT, + json!({"sessionId": local_id, "prompt": [{"type": "text", "text": "ticking:4:300"}]}), + ) + .await; + ca.wait(method::SESSION_UPDATE, |p| { + p["sessionId"] == local_id && p["update"]["content"]["text"] == "tick0 " + }) + .await; + let (_, stdout, stderr) = + cli_until(&root, &socket, &args, Duration::from_millis(900)).await; + assert!(stdout.contains("REMOTE-STREAM"), "{stdout}{stderr}"); + assert!( + !stdout.contains("tick"), + "local event leaked into following output: {stdout}{stderr}" + ); + } server.abort(); a.end_agents_at_shutdown(Default::default()).unwrap(); From 1cd7700e3c5d502fdceb4a42e5cc08479792f1dc Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 06:42:00 +0000 Subject: [PATCH 21/26] fix(acpmux): match waits against qualified peer events Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/cli/orchestrate.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/cmux-tui/crates/acpmux/src/cli/orchestrate.rs b/cmux-tui/crates/acpmux/src/cli/orchestrate.rs index c66fea0a9227..03c9bcb7303d 100644 --- a/cmux-tui/crates/acpmux/src/cli/orchestrate.rs +++ b/cmux-tui/crates/acpmux/src/cli/orchestrate.rs @@ -365,7 +365,7 @@ async fn wait_match( return Err(client.closed("following the session events")); } let p = params.unwrap_or(Value::Null); - let sid = p.get("sessionId").and_then(Value::as_str).unwrap_or("").to_owned(); + let Ok(sid) = crate::cli::resolve::routed_id(&p) else { continue }; let Some(i) = transcripts.iter().position(|(_, id, _)| *id == sid) else { continue; }; From 7539ba60c9efe73f54a84b06a322d98019f4e6f7 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 07:10:14 +0000 Subject: [PATCH 22/26] fix(protocol): refresh compatibility and phone field snapshots Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- .../DaemonLane/DaemonLanePolicy+Fields.generated.swift | 2 +- cmux-tui/spec/api-compat.surface.json | 8 ++++++++ 2 files changed, 9 insertions(+), 1 deletion(-) diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextMobile/DaemonLane/DaemonLanePolicy+Fields.generated.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextMobile/DaemonLane/DaemonLanePolicy+Fields.generated.swift index 7956a758b6b2..b5997e45245d 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextMobile/DaemonLane/DaemonLanePolicy+Fields.generated.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextMobile/DaemonLane/DaemonLanePolicy+Fields.generated.swift @@ -73,7 +73,7 @@ extension DaemonLanePolicy { "new-browser-tab": ["cols", "pane", "rows", "url"], "new-conversation-tab": ["agent_session", "cols", "conversation", "mutation_id", "origin", "owner", "page", "pane", "route", "rows", "split", "transaction", "workspace"], "new-pane": ["cols", "keep", "pane", "pane_id", "rows", "tab_id", "terminal_id"], - "new-pane-right": ["cols", "keep", "kind", "pane", "pane_id", "rows", "tab_id", "terminal_id", "url", "width"], + "new-pane-right": ["cols", "keep", "kind", "pane", "pane_id", "rows", "side", "tab_id", "terminal_id", "url", "width"], "new-screen": ["color", "cols", "group", "icon", "index", "pinned", "rows", "screen_name", "terminal_id", "workspace"], "new-tab": ["cols", "keep", "pane", "rows", "terminal_id"], "new-workspace": ["cols", "name", "rows"], diff --git a/cmux-tui/spec/api-compat.surface.json b/cmux-tui/spec/api-compat.surface.json index 324f97fd5b09..08deaaab38a3 100644 --- a/cmux-tui/spec/api-compat.surface.json +++ b/cmux-tui/spec/api-compat.surface.json @@ -10016,6 +10016,14 @@ "/shell_args/*": { "ty": "string" }, + "/side": { + "ty": "enum", + "nullable": true, + "values": [ + "left", + "right" + ] + }, "/tab_id": { "ty": "string", "nullable": true From 88ff53dd9801bb9b542c9184dca12fae4bc92963 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 07:24:57 +0000 Subject: [PATCH 23/26] test(acpmux): preserve local owner during initial tui selection Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/tui/run.rs | 16 ++++++++++++++-- 1 file changed, 14 insertions(+), 2 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/tui/run.rs b/cmux-tui/crates/acpmux/src/tui/run.rs index b38a4269696f..213b1a1c4cd4 100644 --- a/cmux-tui/crates/acpmux/src/tui/run.rs +++ b/cmux-tui/crates/acpmux/src/tui/run.rs @@ -391,9 +391,21 @@ mod tests { #[test] fn initial_selection_accepts_a_peer_qualified_id() { - let session = json!({"sessionId": "raw", "qualifiedSessionId": "peer/raw"}); + let session = json!({"sessionId": "raw", "peer": "peer", "qualifiedSessionId": "peer/raw"}); assert!(initial_session_matches(&session, "peer/raw")); - assert!(initial_session_matches(&session, "raw")); + assert!(!initial_session_matches(&session, "raw")); assert!(!initial_session_matches(&session, "other/raw")); } + + #[test] + fn initial_selection_keeps_the_local_owner_when_a_peer_reuses_its_id() { + // The more recent peer row precedes the requested local row in the sidebar. + let sessions = [ + json!({"sessionId": "raw", "peer": "peer", "qualifiedSessionId": "peer/raw"}), + json!({"sessionId": "raw", "qualifiedSessionId": "local/raw"}), + ]; + let initial = |id| sessions.iter().position(|s| initial_session_matches(s, id)); + assert_eq!(initial("raw"), Some(1), "a peer must not capture the local attach target"); + assert_eq!(initial("peer/raw"), Some(0)); + } } From 7749e2b36820d29737fc52f28e3bf341d4bdb323 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 07:27:24 +0000 Subject: [PATCH 24/26] test(acpmux): retain relay fallback for missing peer sessions Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/cli/run.rs | 44 +++++++++++++++++++++++++++ 1 file changed, 44 insertions(+) diff --git a/cmux-tui/crates/acpmux/src/cli/run.rs b/cmux-tui/crates/acpmux/src/cli/run.rs index 0d06b2b589ce..fd6116950a7c 100644 --- a/cmux-tui/crates/acpmux/src/cli/run.rs +++ b/cmux-tui/crates/acpmux/src/cli/run.rs @@ -41,3 +41,47 @@ pub(crate) async fn resolve_id(client: &Client, key: &str) -> Result { })?; crate::cli::resolve::routed_id(&v) } + +#[cfg(all(test, unix))] +mod tests { + use super::*; + use crate::rpc::{Message, RpcError}; + use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; + + #[tokio::test] + async fn missing_peer_session_allows_relay_but_other_rpc_errors_do_not() { + for (error, should_relay) in [ + (RpcError::not_found("no local session with the id \"peer/name\""), true), + (RpcError::invalid_params("ambiguous session"), false), + (RpcError::internal("credentials not found"), false), + ] { + let dir = tempfile::tempdir().unwrap(); + let socket = dir.path().join("daemon.sock"); + let listener = tokio::net::UnixListener::bind(&socket).unwrap(); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.unwrap(); + let (rd, mut wr) = stream.into_split(); + let mut lines = BufReader::new(rd).lines(); + for expected in [method::INITIALIZE, method::MUX_INFO] { + let line = lines.next_line().await.unwrap().unwrap(); + let Message::Request { id, method, params } = Message::parse(&line).unwrap() + else { + panic!("expected request"); + }; + assert_eq!(method, expected); + let response = if expected == method::INITIALIZE { + Message::ok(id, json!({})) + } else { + assert_eq!(params.unwrap()["sessionId"], "peer/name"); + Message::err(id, error.clone()) + }; + wr.write_all(response.to_line().as_bytes()).await.unwrap(); + } + }); + let client = Client::connect(&socket).await.unwrap(); + let error = resolve_id(&client, "peer/name").await.unwrap_err(); + server.await.unwrap(); + assert_eq!(crate::cli::send_relay::no_such_session(&error), should_relay, "{error}"); + } + } +} From 2af6de39f21cb28447910ba5df6893532427f9ac Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 07:28:11 +0000 Subject: [PATCH 25/26] fix(acpmux): preserve owner selection and typed relay fallback Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- cmux-tui/crates/acpmux/src/cli/run.rs | 5 +++-- cmux-tui/crates/acpmux/src/tui/run.rs | 3 +-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/cmux-tui/crates/acpmux/src/cli/run.rs b/cmux-tui/crates/acpmux/src/cli/run.rs index fd6116950a7c..9cdc05a06bce 100644 --- a/cmux-tui/crates/acpmux/src/cli/run.rs +++ b/cmux-tui/crates/acpmux/src/cli/run.rs @@ -32,8 +32,9 @@ pub(crate) async fn run_client(cmd: Command, json_out: bool, suppress_reads: boo pub(crate) async fn resolve_id(client: &Client, key: &str) -> Result { let key = orchestrate::expand_session_key(key)?; let v = client.request(method::MUX_INFO, json!({"sessionId": key})).await.map_err(|e| { - let m = e.to_string().to_lowercase(); - if m.contains("no session") || m.contains("not found") { + if e.downcast_ref::() + .is_some_and(|e| e.0.code == crate::rpc::RpcError::not_found("").code) + { anyhow::Error::from(errors::AppError::no_session(&key)) } else { e diff --git a/cmux-tui/crates/acpmux/src/tui/run.rs b/cmux-tui/crates/acpmux/src/tui/run.rs index 213b1a1c4cd4..7ad9034cba6b 100644 --- a/cmux-tui/crates/acpmux/src/tui/run.rs +++ b/cmux-tui/crates/acpmux/src/tui/run.rs @@ -224,8 +224,7 @@ pub async fn run(client: Arc, initial: Option) -> Result<()> { } fn initial_session_matches(session: &Value, id: &str) -> bool { - session.get("sessionId").and_then(Value::as_str) == Some(id) - || session.get("qualifiedSessionId").and_then(Value::as_str) == Some(id) + session_route_id(session).as_deref() == Some(id) } pub(super) fn make_app( From 97fe72c9ec3734e4f39183796bda53967d4e3092 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Thu, 29 Oct 2026 07:50:29 +0000 Subject: [PATCH 26/26] chore(acpmux): leave phone field regeneration to base repair Agent-Session: https://cmux.com/deeplink/agent?id=aid-ca805d5a919ddc878a43 --- .../DaemonLane/DaemonLanePolicy+Fields.generated.swift | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Packages/macOS/CmuxNext/Sources/CmuxNextMobile/DaemonLane/DaemonLanePolicy+Fields.generated.swift b/Packages/macOS/CmuxNext/Sources/CmuxNextMobile/DaemonLane/DaemonLanePolicy+Fields.generated.swift index b5997e45245d..7956a758b6b2 100644 --- a/Packages/macOS/CmuxNext/Sources/CmuxNextMobile/DaemonLane/DaemonLanePolicy+Fields.generated.swift +++ b/Packages/macOS/CmuxNext/Sources/CmuxNextMobile/DaemonLane/DaemonLanePolicy+Fields.generated.swift @@ -73,7 +73,7 @@ extension DaemonLanePolicy { "new-browser-tab": ["cols", "pane", "rows", "url"], "new-conversation-tab": ["agent_session", "cols", "conversation", "mutation_id", "origin", "owner", "page", "pane", "route", "rows", "split", "transaction", "workspace"], "new-pane": ["cols", "keep", "pane", "pane_id", "rows", "tab_id", "terminal_id"], - "new-pane-right": ["cols", "keep", "kind", "pane", "pane_id", "rows", "side", "tab_id", "terminal_id", "url", "width"], + "new-pane-right": ["cols", "keep", "kind", "pane", "pane_id", "rows", "tab_id", "terminal_id", "url", "width"], "new-screen": ["color", "cols", "group", "icon", "index", "pinned", "rows", "screen_name", "terminal_id", "workspace"], "new-tab": ["cols", "keep", "pane", "rows", "terminal_id"], "new-workspace": ["cols", "name", "rows"],