cmux-tui: agent-session-adopt binds another chat of this machine to an agent tab (1/3 of #22871)

cmux-tui/crates/cmux-tui-core/src/server.rs

@@ -131,9 +131,9 @@ mod agent_inventory;
 mod agent_session_attach;
 #[cfg(unix)]
 pub use agent_session_attach::{
-    AGENT_SESSION_ATTACH_CAPABILITY, AGENT_SESSION_START_CAPABILITY,
-    AGENT_SESSION_START_KEYED_REASON_CAPABILITY, AGENT_SESSION_TRUST_CAPABILITY, AcpmuxStarter,
-    AgentStartHost, AgentStartHostSource,
+    AGENT_SESSION_ADOPT_CAPABILITY, AGENT_SESSION_ATTACH_CAPABILITY,
+    AGENT_SESSION_START_CAPABILITY, AGENT_SESSION_START_KEYED_REASON_CAPABILITY,
+    AGENT_SESSION_TRUST_CAPABILITY, AcpmuxStarter, AgentStartHost, AgentStartHostSource,
 };
 #[cfg(unix)]
 pub mod agent_transfer;

cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs

@@ -0,0 +1,147 @@
+//! Remote agent session adopt (`agent-session-adopt-v1`): a trusted local
+//! client (an app on another Mac, through the SSH or Cloud carrier) opens a
+//! chat that already runs in THIS machine's acpmux (an agent or `cmux agent
+//! open` started it here) in an agent tab of this daemon's store, and the
+//! daemon binds it to the tab. The app then shows it through
+//! `agent-session-attach-v1`, as for a started chat.
+//!
+//! `agent-session-adopt {surface, session}` answers `{surface, session,
+//! conversation, family, mode, asking}`.
+//!
+//! Scope: the tab is checked as for `agent-session-start` (an unbound
+//! `agent_session` tab of this store), and the session must be one this
+//! machine's own acpmux lists (`_acpmux/sessions`): never a chat mirrored
+//! from a peer (a session never leaves its machine, cx-k21v), never the Home
+//! Chief's, never an id acpmux does not know (`agent_session.unknown_session`).
+//! acpmux is never started for an adopt: nothing listening runs no chat.
+//!
+//! Permissions: the session keeps its mode. A chat in a mode that does not ask
+//! before acting (bypass) is adopted too, only on a host the start's host gate
+//! allows (a Mac, an SSH host or an owner-bound Cloud VM the person enrolled;
+//! never a team VM or a Cloud VM of unknown kind), and the answer says so
+//! (`asking: false`), which the app shows as a "bypass" badge on the
+//! workspace. The daemon changes no mode.
+
+use std::path::PathBuf;
+use std::sync::Arc;
+
+use serde::Deserialize;
+use serde_json::{Value, json};
+
+use super::agent_session_link::AcpmuxLink;
+use super::agent_session_start::{
+    AgentSessionStarts, CALL_TIMEOUT, THREAD_STACK_BYTES, checked_mode, is_chief, startable,
+};
+use super::{MessageWriter, Refusal, SurfaceId, ok, refuse};
+use crate::mux::Mux;
+use crate::state::conversation_tabs_store::{validate_harness, validate_session};
+
+pub const AGENT_SESSION_ADOPT_CAPABILITY: &str = "agent-session-adopt-v1";
+
+#[derive(Deserialize)]
+#[serde(deny_unknown_fields)]
+pub(super) struct AdoptParams {
+    surface: SurfaceId,
+    session: String,
+}
+
+/// `agent-session-adopt` of a trusted local connection (the parent checked
+/// trust and the acpmux socket).
+pub(super) fn adopt(
+    mux: &Arc<Mux>,
+    client: u64,
+    id: Option<Value>,
+    socket: PathBuf,
+    params: AdoptParams,
+    writer: &MessageWriter,
+) -> bool {
+    if validate_session(&params.session).is_err() {
+        return refuse(writer, id, Refusal::BadRequest);
+    }
+    if let Err(refusal) = mux.control_clients.agent_sessions.starts.host_gate() {
+        return refuse(writer, id, refusal);
+    }
+    if let Err(refusal) = startable(mux, params.surface) {
+        return refuse(writer, id, refusal);
+    }
+    let Some(slot) = AgentSessionStarts::reserve(mux, params.surface) else {
+        return refuse(writer, id, Refusal::Limit);
+    };
+    let worker_mux = mux.clone();
+    let worker_writer = writer.clone();
+    let worker_id = id.clone();
+    let spawned = std::thread::Builder::new()
+        .name("mux-agent-adopt".into())
+        .stack_size(THREAD_STACK_BYTES)
+        .spawn(move || {
+            let result = run(&worker_mux, client, &socket, &params);
+            drop(slot);
+            match result {
+                Ok(data) => ok(&worker_writer, worker_id, data),
+                Err(refusal) => refuse(&worker_writer, worker_id, refusal),
+            };
+        });
+    // A failed spawn drops the closure and with it the slot.
+    if spawned.is_err() {
+        return refuse(writer, id, Refusal::Limit);
+    }
+    true
+}
+
+fn run(
+    mux: &Arc<Mux>,
+    client: u64,
+    socket: &std::path::Path,
+    params: &AdoptParams,
+) -> Result<Value, Refusal> {
+    let link = AcpmuxLink::connect(socket, |_| {}).map_err(|_| Refusal::AcpmuxUnavailable)?;
+    let result = adopt_and_bind(mux, client, &link, params);
+    link.close();
+    result
+}
+
+fn adopt_and_bind(
+    mux: &Arc<Mux>,
+    client: u64,
+    link: &AcpmuxLink,
+    params: &AdoptParams,
+) -> Result<Value, Refusal> {
+    let listed = link.call("_acpmux/sessions", json!({}), CALL_TIMEOUT).map_err(Refusal::of)?;
+    let summary = own_session(&listed, &params.session).ok_or(Refusal::UnknownSession)?;
+    let table = link
+        .call("_acpmux/web_modes", json!({"sessionId": params.session}), CALL_TIMEOUT)
+        .map_err(Refusal::of)?;
+    let facts = checked_mode(&params.session, &table)?;
+    let family = facts["family"].as_str().unwrap_or_default();
+    // The tab's brand mark: the agent acpmux runs (a profile by its last
+    // part), else its family.
+    let harness = [summary.get("harness").and_then(Value::as_str), Some(family)]
+        .into_iter()
+        .flatten()
+        .filter_map(|harness| harness.rsplit('/').next())
+        .find(|harness| validate_harness(harness).is_ok());
+    // The tab may have closed, or a start or adopt bound it, meanwhile.
+    let actor = super::super::origin_gate::connection_actor(mux, client);
+    let (record, _) = mux
+        .bind_conversation_tab_session_as(&actor, params.surface, &params.session, None, harness)
+        .map_err(|_| startable(mux, params.surface).err().unwrap_or(Refusal::Refused))?;
+    Ok(json!({
+        "surface": params.surface,
+        "session": params.session,
+        "conversation": record.wire(),
+        "family": family,
+        "mode": facts["mode"],
+        "asking": facts["asking"],
+    }))
+}
+
+/// `session`'s summary in a `_acpmux/sessions` answer when it is a chat of
+/// this machine: not mirrored from a peer and not the Home Chief's.
+fn own_session<'a>(listed: &'a Value, session: &str) -> Option<&'a Value> {
+    let list = listed.get("sessions").unwrap_or(listed).as_array()?;
+    list.iter().find(|summary| {
+        summary.get("sessionId").and_then(Value::as_str) == Some(session)
+            && summary.get("peer").is_none_or(Value::is_null)
+            && !is_chief(summary)
+    })
+}

cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs

@@ -15,6 +15,8 @@ use std::time::Duration;
 use serde::Deserialize;
 use serde_json::{Value, json};
 
+#[path = "agent_session_adopt.rs"]
+mod agent_session_adopt;
 #[path = "agent_session_children.rs"]
 pub(super) mod agent_session_children;
 #[path = "agent_session_link.rs"]
@@ -30,6 +32,7 @@ mod inventory_socket;
 use super::{MessageWriter, OutboundStream, Response, SurfaceId, send_response};
 use crate::mux::Mux;
 use crate::state::conversation_tabs_store::ConversationTabRecord;
+pub use agent_session_adopt::AGENT_SESSION_ADOPT_CAPABILITY;
 use agent_session_link::{AcpmuxLink, Inbound, LinkError};
 pub use agent_session_start::{
     AGENT_SESSION_START_CAPABILITY, AGENT_SESSION_START_KEYED_REASON_CAPABILITY, AcpmuxStarter,
@@ -335,6 +338,8 @@ pub(crate) enum Refusal {
     UnknownPermission,
     /// `agent-session-start`: the tab already has a session.
     Bound,
+    /// `agent-session-adopt`: this machine's acpmux lists no such chat of its own.
+    UnknownSession,
     /// `agent-session-start`: acpmux is not running and could not be started.
     AcpmuxUnavailable,
     /// `agent-session-start`: the agent has no mode that asks before acting.
@@ -369,6 +374,7 @@ impl Refusal {
             Self::Refused => "agent_session.refused",
             Self::UnknownPermission => "agent_session.unknown_permission",
             Self::Bound => "agent_session.bound",
+            Self::UnknownSession => "agent_session.unknown_session",
             Self::AcpmuxUnavailable => "agent_session.acpmux_unavailable",
             Self::NotAsking => "agent_session.not_asking",
             Self::UntrustedFolder => "agent_session.untrusted_folder",
@@ -394,6 +400,7 @@ impl Refusal {
             Self::Refused => "acpmux refused the request",
             Self::UnknownPermission => "no such pending permission request on this attachment",
             Self::Bound => "the agent tab already has a session",
+            Self::UnknownSession => "this machine runs no such chat of its own",
             Self::AcpmuxUnavailable => {
                 "acpmux is not running on this machine and could not be started"
             }
@@ -456,6 +463,9 @@ pub(super) fn try_handle(
     if cmd == "agent-session-trust" {
         return Some(trust(mux, client, id, object, writer));
     }
+    if cmd == "agent-session-adopt" {
+        return Some(adopt(mux, client, id, object, writer));
+    }
     let Some(command) = AgentSessionCommand::parse(&cmd, Value::Object(object)) else {
         return Some(refuse(writer, id, Refusal::BadRequest));
     };
@@ -508,6 +518,26 @@ fn start(
     agent_session_start::start(mux, client, id, socket, params, writer)
 }
 
+/// `agent-session-adopt` (agent_session_adopt.rs), with the same checks.
+fn adopt(
+    mux: &Arc<Mux>,
+    client: u64,
+    id: Option<Value>,
+    params: serde_json::Map<String, Value>,
+    writer: &MessageWriter,
+) -> bool {
+    let Ok(params) = serde_json::from_value(Value::Object(params)) else {
+        return refuse(writer, id, Refusal::BadRequest);
+    };
+    if !mux.control_clients.is_unix(client) || mux.is_remote_client(client) {
+        return refuse(writer, id, Refusal::NotTrusted);
+    }
+    let Some(socket) = mux.control_clients.agent_sessions.socket() else {
+        return refuse(writer, id, Refusal::Unavailable);
+    };
+    agent_session_adopt::adopt(mux, client, id, socket, params, writer)
+}
+
 /// `agent-session-trust` (agent_session_trust.rs), with the same checks.
 fn trust(
     mux: &Arc<Mux>,

cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs

@@ -81,8 +81,8 @@ const MAX_STARTS: usize = 8;
 const MAX_CWD_BYTES: usize = 4096;
 /// A harness spawns and answers `session/new` (a cold agent can take long).
 const NEW_TIMEOUT: Duration = Duration::from_secs(60);
-const CALL_TIMEOUT: Duration = Duration::from_secs(5);
-const THREAD_STACK_BYTES: usize = 256 * 1024;
+pub(super) const CALL_TIMEOUT: Duration = Duration::from_secs(5);
+pub(super) const THREAD_STACK_BYTES: usize = 256 * 1024;
 
 /// Reads this machine's [`AgentStartHost`] (a few file checks), at each start.
 pub type AgentStartHostSource = Arc<dyn Fn() -> AgentStartHost + Send + Sync>;
@@ -109,7 +109,7 @@ impl AgentSessionStarts {
     /// Ok on an allowed host; the refusal otherwise (also when unset). The
     /// kind is read again for every start, so a bind or team enroll that
     /// lands after the daemon started counts at once.
-    fn host_gate(&self) -> Result<(), Refusal> {
+    pub(super) fn host_gate(&self) -> Result<(), Refusal> {
         let source = self.host.lock().unwrap_or_else(|e| e.into_inner()).clone();
         match source.map(|read| read()) {
             Some(AgentStartHost::Allowed) => Ok(()),
@@ -124,7 +124,7 @@ impl AgentSessionStarts {
 
     /// Takes `surface`'s start slot; None while it starts or at the cap.
     /// The slot is given back when the returned guard drops (also on a panic).
-    fn reserve(mux: &Arc<Mux>, surface: SurfaceId) -> Option<StartSlot> {
+    pub(super) fn reserve(mux: &Arc<Mux>, surface: SurfaceId) -> Option<StartSlot> {
         let starts = &mux.control_clients.agent_sessions.starts;
         let mut starting = starts.starting.lock().unwrap_or_else(|e| e.into_inner());
         (starting.len() < MAX_STARTS && starting.insert(surface))
@@ -133,7 +133,7 @@ impl AgentSessionStarts {
 }
 
 /// A taken start slot.
-struct StartSlot {
+pub(super) struct StartSlot {
     mux: Arc<Mux>,
     surface: SurfaceId,
 }
@@ -313,7 +313,7 @@ pub(crate) fn keyed_restart_reason(status: Option<&Value>) -> Value {
 }
 
 /// Ok when `surface` is an unbound agent tab whose record names this store.
-fn startable(mux: &Mux, surface: SurfaceId) -> Result<(), Refusal> {
+pub(super) fn startable(mux: &Mux, surface: SurfaceId) -> Result<(), Refusal> {
     let runtime = mux.surface(surface).ok_or(Refusal::UnknownTab)?;
     match mux.conversation_tab_of(&runtime) {
         Some(ConversationTabRecord::AgentSession { host, session, .. })
@@ -499,6 +499,59 @@ fn refused_family(table: &Value, family: &str) -> bool {
         .is_some_and(|refused| refused.iter().any(|name| name.as_str() == Some(family)))
 }
 
+/// Whether a `_acpmux/web_modes {sessionId}` answer's session asks before
+/// acting (its family is not refused and its mode is one of the family's
+/// asking modes; no mode counts as asking), and the family's asking default.
+pub(super) fn asks(table: &Value) -> (bool, Option<String>) {
+    let family = table.pointer("/session/family").and_then(Value::as_str).unwrap_or_default();
+    let modes: Vec<&str> = table
+        .pointer(&format!("/families/{}", family.replace('~', "~0").replace('/', "~1")))
+        .and_then(Value::as_array)
+        .map(|modes| modes.iter().filter_map(Value::as_str).collect())
+        .unwrap_or_default();
+    let default = modes.first().map(|mode| mode.to_string());
+    if family.is_empty() || refused_family(table, family) {
+        return (false, None);
+    }
+    let asks = match table.pointer("/session/mode").and_then(Value::as_str) {
+        None => true,
+        Some(mode) => modes.contains(&mode),
+    };
+    (asks, default)
+}
+
+/// `session`'s mode in a `_acpmux/web_modes {sessionId}` answer, as the adopt
+/// reply carries it: its family, its mode and whether it asks before acting
+/// ([`asks`]); the app shows `asking: false` as the bypass badge.
+fn mode_facts(session: &str, table: &Value) -> Value {
+    let (asking, _) = asks(table);
+    json!({
+        "session": session,
+        "family": table.pointer("/session/family").and_then(Value::as_str).unwrap_or_default(),
+        "mode": table.pointer("/session/mode").and_then(Value::as_str),
+        "asking": asking,
+    })
+}
+
+/// [`mode_facts`] of `session` in `table`, only when acpmux knows that
+/// session by exactly this id (an unknown or purged one has no `session`, and
+/// a prefix resolves to another id): never facts for a chat that is not there.
+pub(super) fn checked_mode(session: &str, table: &Value) -> Result<Value, Refusal> {
+    if table.pointer("/session/sessionId").and_then(Value::as_str) != Some(session) {
+        return Err(Refusal::UnknownSession);
+    }
+    Ok(mode_facts(session, table))
+}
+
+/// A Home Chief session (acpmux `hub/route_switch.rs` is_chief, the app's
+/// `AcpmuxSessionCensus.isChief`), from a `_acpmux/sessions` summary.
+pub(in crate::server) fn is_chief(summary: &Value) -> bool {
+    let text = |key: &str| summary[key].as_str().unwrap_or_default();
+    summary.field("tags").get("cmux.chief").is_some()
+        || text("name").starts_with("optchat-")
+        || text("harness").starts_with("optchat-compact-")
+}
+
 /// acpmux's remote rule for a new session: its family is not refused and
 /// its mode asks (no mode counts as asking), else it moves to the family's
 /// asking default; anything else is refused. Returns the session's family.
@@ -507,23 +560,6 @@ fn settle_asking_mode(link: &AcpmuxLink, session: &str) -> Result<String, Refusa
         link.call("_acpmux/web_modes", json!({"sessionId": session}), CALL_TIMEOUT)
             .map_err(Refusal::of)
     };
-    let asks = |table: &Value| -> (bool, Option<String>) {
-        let family = table.pointer("/session/family").and_then(Value::as_str).unwrap_or_default();
-        let modes: Vec<&str> = table
-            .pointer(&format!("/families/{}", family.replace('~', "~0").replace('/', "~1")))
-            .and_then(Value::as_array)
-            .map(|modes| modes.iter().filter_map(Value::as_str).collect())
-            .unwrap_or_default();
-        let default = modes.first().map(|mode| mode.to_string());
-        if family.is_empty() || refused_family(table, family) {
-            return (false, None);
-        }
-        let asks = match table.pointer("/session/mode").and_then(Value::as_str) {
-            None => true,
-            Some(mode) => modes.contains(&mode),
-        };
-        (asks, default)
-    };
     let table = read()?;
     let family =
         table.pointer("/session/family").and_then(Value::as_str).unwrap_or_default().to_string();

cmux-tui/crates/cmux-tui-core/src/server/agent_transfer/agent_update.rs

@@ -348,7 +348,9 @@ fn threads(sessions: &Value) -> (usize, usize) {
     let list = sessions.get("sessions").unwrap_or(sessions);
     let (mut live, mut running) = (0, 0);
     for summary in list.as_array().into_iter().flatten() {
-        if summary.get("peer").is_some_and(|peer| !peer.is_null()) || is_chief(summary) {
+        if summary.get("peer").is_some_and(|peer| !peer.is_null())
+            || agent_session_start::is_chief(summary)
+        {
             continue;
         }
         match summary.field("status").as_str() {
@@ -363,15 +365,6 @@ fn threads(sessions: &Value) -> (usize, usize) {
     (live, running)
 }
 
-/// A Home Chief session (acpmux `hub/route_switch.rs` is_chief, the app's
-/// `AcpmuxSessionCensus.isChief`).
-fn is_chief(summary: &Value) -> bool {
-    let text = |key: &str| summary[key].as_str().unwrap_or_default();
-    summary.field("tags").get("cmux.chief").is_some()
-        || text("name").starts_with("optchat-")
-        || text("harness").starts_with("optchat-compact-")
-}
-
 // MARK: Apply
 
 fn apply(mux: &Arc<Mux>, socket: &Path, params: ApplyParams) -> Result<Value, Failure> {