fix(acpmux): preserve peer ownership for session lifecycle commands

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

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;
         };
@@ -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 {

cmux-tui/crates/acpmux/src/cli/output.rs

@@ -366,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(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;
@@ -502,7 +502,7 @@ pub(crate) async fn plain_attach(client: Arc<Client>, 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() {
@@ -709,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(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;

cmux-tui/crates/acpmux/src/cli/resolve.rs

@@ -2,12 +2,71 @@
 //! `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<String> {
+    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::<crate::client::DaemonError>().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::<Vec<_>>()
+            .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<String> {
+    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) if id.strip_prefix(&format!("{peer}/")).is_some() => id.to_owned(),
+        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?;

cmux-tui/crates/acpmux/src/cli/retry.rs

@@ -6,15 +6,15 @@
 //! 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;
 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);

cmux-tui/crates/acpmux/src/cli/run.rs

@@ -32,12 +32,57 @@ 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<String> {
     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::<crate::client::DaemonError>()
+            .is_some_and(|e| e.0.code == crate::rpc::RpcError::not_found("").code)
+        {
             anyhow::Error::from(errors::AppError::no_session(&key))
         } else {
             e
         }
     })?;
-    Ok(v.get("sessionId").and_then(Value::as_str).unwrap_or(&key).to_owned())
+    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}");
+        }
+    }
 }

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"]

cmux-tui/crates/acpmux/src/cli/run/launch.rs

@@ -248,7 +248,10 @@ pub(super) async fn existing(client: &Client, key: &str) -> Result<Option<Value>
         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) }
         }
     }
@@ -318,8 +321,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<()> {