From fcd4f270b133ba661d9fce16fed42e89ce9537e6 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 23 Oct 2026 09:02:59 -0700 Subject: [PATCH 1/4] cmux-tui: agent-session-adopt binds another chat of this machine's acpmux to an agent tab Split out of #22871 (the daemon half). agent-session-adopt {surface, session} (agent-session-adopt-v1) binds a session of this machine's own acpmux to an unbound agent tab of this store, with agent-session-start's gates (a trusted local non-remote client, the host gate that refuses team VMs and unverified Cloud VMs, a tab of this store with no session yet). It refuses with agent_session.unknown_session any id this machine's _acpmux/sessions does not list as its own (a peer's mirror, the Home Chief's, an unknown id), and answers {surface, session, conversation, family, mode, asking}. A new agent-session-mode verb reads a bound tab's mode; the attach reply carries it, and an attachment with mode_events hears agent-session-mode pushes when it changes (one read at a time, released with an AcqRel swap so a change during a read is never lost). is_chief moves to agent_session_start and is shared with agent_update. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16040-fd0b-7dd0-9487-1a3a198d9f5c&host=cosmic-gray --- cmux-tui/crates/cmux-tui-core/src/server.rs | 6 +- .../src/server/agent_session_adopt.rs | 227 ++++++++++++++++++ .../src/server/agent_session_attach.rs | 66 +++++ .../src/server/agent_session_attach_tests.rs | 19 +- .../src/server/agent_session_start.rs | 91 +++++-- .../src/server/agent_session_stream.rs | 80 +++++- .../src/server/agent_transfer/agent_update.rs | 13 +- .../cmux-tui-core/src/server/capabilities.rs | 1 + .../src/server/conversation_tabs_wire.rs | 11 +- .../cmux-tui/tests/cli/agent_session_adopt.rs | 176 ++++++++++++++ .../cmux-tui/tests/cli/agent_session_start.rs | 2 + cmux-tui/spec/commands.md | 44 ++++ 12 files changed, 689 insertions(+), 47 deletions(-) create mode 100644 cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs create mode 100644 cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs diff --git a/cmux-tui/crates/cmux-tui-core/src/server.rs b/cmux-tui/crates/cmux-tui-core/src/server.rs index 2b56d3deab2e..ada3e6af9e2e 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server.rs +++ b/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; diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs new file mode 100644 index 000000000000..0e136304716e --- /dev/null +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs @@ -0,0 +1,227 @@ +//! 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 std::sync::atomic::{AtomicUsize, Ordering}; + +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, read_mode, + 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 ModeParams { + surface: SurfaceId, +} + +/// At most [`MAX_MODE_READS`] `agent-session-mode` reads run at once, each on +/// its own thread and acpmux connection; one more is `agent_session.limit`. +const MAX_MODE_READS: usize = 8; +static MODE_READS: AtomicUsize = AtomicUsize::new(0); + +/// One running mode read; dropping it frees the slot. +struct ModeRead; + +impl ModeRead { + fn admit() -> Option { + MODE_READS + .fetch_update(Ordering::AcqRel, Ordering::Acquire, |n| { + (n < MAX_MODE_READS).then_some(n + 1) + }) + .ok() + .map(|_| Self) + } +} + +impl Drop for ModeRead { + fn drop(&mut self) { + MODE_READS.fetch_sub(1, Ordering::AcqRel); + } +} + +/// `agent-session-mode {surface}` of a trusted local connection (the parent +/// checked trust and the acpmux socket): the mode of the chat the tab is bound +/// to, `{surface, session, family, mode, asking}`, read from this machine's +/// acpmux. Read-only: the app reads it once for each bound tab it lists at a +/// connect (a restored tab it does not show), so the bypass badge never +/// depends on what this app remembers. +pub(super) fn mode( + mux: &Arc, + id: Option, + socket: PathBuf, + params: ModeParams, + writer: &MessageWriter, +) -> bool { + let Some(session) = super::tab_session(mux, params.surface) else { + return refuse(writer, id, Refusal::UnknownTab); + }; + let Some(slot) = ModeRead::admit() else { + return refuse(writer, id, Refusal::Limit); + }; + let worker_writer = writer.clone(); + let worker_id = id.clone(); + let spawned = std::thread::Builder::new() + .name("mux-agent-mode".into()) + .stack_size(THREAD_STACK_BYTES) + .spawn(move || { + let _slot = slot; + let read = AcpmuxLink::connect(&socket, |_| {}) + .map_err(|_| Refusal::Unavailable) + .and_then(|link| { + let read = read_mode(&link, &session); + link.close(); + read + }); + match read { + Ok(mut facts) => { + facts["surface"] = json!(params.surface); + ok(&worker_writer, worker_id, facts) + } + Err(refusal) => refuse(&worker_writer, worker_id, refusal), + }; + }); + if spawned.is_err() { + return refuse(writer, id, Refusal::Limit); + } + true +} + +#[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, + client: u64, + id: Option, + socket: PathBuf, + params: AdoptParams, + writer: &MessageWriter, +) -> bool { + if validate_session(¶ms.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, ¶ms); + 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, + client: u64, + socket: &std::path::Path, + params: &AdoptParams, +) -> Result { + 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, + client: u64, + link: &AcpmuxLink, + params: &AdoptParams, +) -> Result { + let listed = link.call("_acpmux/sessions", json!({}), CALL_TIMEOUT).map_err(Refusal::of)?; + let summary = own_session(&listed, ¶ms.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(¶ms.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, ¶ms.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) + }) +} diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs index 4b3138755bc1..39bd7cfc96cd 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs +++ b/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, @@ -164,6 +167,12 @@ struct Attachment { permissions: Mutex>, /// Prompt ids waiting for `_acpmux/prompt_accepted`, with their reply slot. prompts: Mutex>>, + /// The attach asked for `agent-session-mode` pushes (`mode_events`). + mode_events: bool, + /// A mode read runs on a worker thread, and whether another change came + /// since it began (it reads again): one reader at a time, in order. + mode_reading: AtomicBool, + mode_again: AtomicBool, /// The session's descendants this attachment follows /// (`agent-session-children-v1`): their changes are streamed too. children: Mutex>, @@ -237,6 +246,11 @@ struct PageParams { limit: Option, #[serde(default)] kinds: Option>, + /// Attach only: the attachment also pushes `agent-session-mode` when the + /// chat's mode changes (an app that reads it asks; any other client would + /// not know the event). + #[serde(default)] + mode_events: bool, } enum AgentSessionCommand { @@ -335,6 +349,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 +385,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 +411,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 +474,12 @@ 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)); + } + if cmd == "agent-session-mode" { + return Some(mode(mux, client, id, object, writer)); + } let Some(command) = AgentSessionCommand::parse(&cmd, Value::Object(object)) else { return Some(refuse(writer, id, Refusal::BadRequest)); }; @@ -508,6 +532,46 @@ 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, + client: u64, + id: Option, + params: serde_json::Map, + 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-mode` (agent_session_adopt.rs), with the attach verbs' checks. +fn mode( + mux: &Arc, + client: u64, + id: Option, + params: serde_json::Map, + 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::mode(mux, id, socket, params, writer) +} + /// `agent-session-trust` (agent_session_trust.rs), with the same checks. fn trust( mux: &Arc, @@ -627,6 +691,7 @@ fn attach( let worker_writer = writer.clone(); let worker_id = id.clone(); let surface = params.surface; + let mode_events = params.mode_events; let spawned = std::thread::Builder::new() .name("mux-agent-attach".into()) .stack_size(THREAD_STACK_BYTES) @@ -640,6 +705,7 @@ fn attach( record_session, acpmux_params, read_only, + mode_events, worker_writer, ); }); diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach_tests.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach_tests.rs index b05511089806..d9cd39da659b 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach_tests.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach_tests.rs @@ -58,12 +58,12 @@ impl FakeAcpmux { Self { directory, socket, frames, connections, current: None } } - /// The next frame the daemon sent to acpmux. - /// The next frame the daemon sent to acpmux (its session watch skipped). + /// The next frame the daemon sent to acpmux (its session watch and the + /// attach's mode read skipped: the daemon's own reads, never the client's). pub(super) fn frame(&self) -> Value { loop { let frame = self.frames.recv_timeout(WAIT).expect("a frame to acpmux"); - if frame["method"] != json!("_acpmux/watch") { + if !daemon_read(&frame) { return frame; } } @@ -73,7 +73,7 @@ impl FakeAcpmux { let deadline = Instant::now() + Duration::from_millis(200); while let Some(left) = deadline.checked_duration_since(Instant::now()) { match self.frames.recv_timeout(left) { - Ok(frame) if frame["method"] == json!("_acpmux/watch") => continue, + Ok(frame) if daemon_read(&frame) => continue, Ok(frame) => panic!("the daemon must not reach acpmux: {frame}"), Err(_) => return, } @@ -112,6 +112,12 @@ impl Drop for FakeAcpmux { } } +/// A read the daemon makes for itself on an attach: the session watch, and +/// the chat's mode for the attach reply's `agentMode` (`agent-session-adopt-v1`). +fn daemon_read(frame: &Value) -> bool { + matches!(frame["method"].as_str(), Some("_acpmux/watch" | "_acpmux/web_modes")) +} + fn serve(stream: UnixStream, frames: Sender) { let mut writer = stream.try_clone().unwrap(); for line in BufReader::new(stream).lines() { @@ -140,6 +146,11 @@ fn serve(stream: UnixStream, frames: Sender) { // (agent_session_children_tests.rs). "_acpmux/sessions" => Some(machine_sessions()), "_acpmux/events" => Some(json!({"events": [], "hasMore": false, "lastSeq": 1})), + // The attached chat asks before acting (its mode is its family's asking default). + "_acpmux/web_modes" => Some(json!({ + "families": {"claude": ["default"]}, + "session": {"sessionId": params["sessionId"], "family": "claude", "mode": "default"}, + })), "_acpmux/permission_respond" => Some(json!({})), // `acpmux.transfer.v1` with the guarded receive (agent_transfer_tests.rs). method if method.starts_with("_acpmux/transfer_") => transfer_reply(method, params), diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs index d1563309e919..2fad871e95fc 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs +++ b/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 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, surface: SurfaceId) -> Option { + pub(super) fn reserve(mux: &Arc, surface: SurfaceId) -> Option { 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, 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,68 @@ 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) { + 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, the attach reply's `agentMode` and `agent-session-mode` carry 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 { + if table.pointer("/session/sessionId").and_then(Value::as_str) != Some(session) { + return Err(Refusal::UnknownSession); + } + Ok(mode_facts(session, table)) +} + +/// [`checked_mode`] of `session`, read on `link`. +pub(super) fn read_mode(link: &AcpmuxLink, session: &str) -> Result { + let table = link + .call("_acpmux/web_modes", json!({"sessionId": session}), CALL_TIMEOUT) + .map_err(Refusal::of)?; + checked_mode(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 +569,6 @@ fn settle_asking_mode(link: &AcpmuxLink, session: &str) -> Result (bool, Option) { - 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(); diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs index 98e47701fa5c..108738bb81d1 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs @@ -12,6 +12,7 @@ pub(super) fn connect_and_attach( record_session: String, params: Value, read_only: bool, + mode_events: bool, writer: MessageWriter, ) { let sessions = &mux.control_clients.agent_sessions; @@ -42,7 +43,7 @@ pub(super) fn connect_and_attach( return; } }; - let page = match link.call("_acpmux/attach", params, CALL_TIMEOUT) { + let mut page = match link.call("_acpmux/attach", params, CALL_TIMEOUT) { Ok(page) => bound_page(page, newest_first), Err(error) => { release(sessions); @@ -70,6 +71,11 @@ pub(super) fn connect_and_attach( return; } slot.lock().unwrap_or_else(|e| e.into_inner()).session = Some(resolved.clone()); + // The chat's mode, for the app's bypass badge; an acpmux that cannot + // tell leaves it out. + if let Ok(facts) = agent_session_start::read_mode(&link, &resolved) { + page["agentMode"] = facts; + } // Session status changes (`_acpmux/session_changed`) come only to // watchers; the reader keeps only this session's and its descendants'. // The answer (every session's summary) never reaches the client: only @@ -89,6 +95,9 @@ pub(super) fn connect_and_attach( ended: AtomicBool::new(false), permissions: Mutex::default(), prompts: Mutex::default(), + mode_events, + mode_reading: AtomicBool::new(false), + mode_again: AtomicBool::new(false), children: Mutex::default(), }); if let Ok(watched) = &watched { @@ -162,7 +171,7 @@ pub(crate) fn bound_page(mut page: Value, newest_first: bool) -> Value { } /// One notification or the close from the attachment's acpmux link. -pub(super) fn deliver(attachment: &Attachment, inbound: Inbound) { +pub(super) fn deliver(attachment: &Arc, inbound: Inbound) { if attachment.ended.load(Ordering::Acquire) { return; } @@ -177,6 +186,9 @@ pub(super) fn deliver(attachment: &Attachment, inbound: Inbound) { } match method.as_str() { "_acpmux/event" if same_session => { + if mode_update(¶ms) { + reread_mode(attachment); + } let event = json!({"event": "agent-session-record", "surface": attachment.surface, "record": params}); // Never blocks: a client that does not keep up overflows its // stream, which ends with the `overflow` close (the stream's own @@ -200,6 +212,12 @@ pub(super) fn deliver(attachment: &Attachment, inbound: Inbound) { if !same_session && !attachment.tab_current() { return attachment.end(Some("detached")); } + // Only the session's own mode change re-reads its mode (a descendant's never does). + if same_session + && matches!(params.get("kind").and_then(Value::as_str), Some("mode" | "config")) + { + reread_mode(attachment); + } let event = json!({"event": "agent-session-changed", "surface": attachment.surface, "change": params}); if attachment.writer.send_stream(&event, &attachment.outbound).is_err() { attachment.end(None); @@ -216,3 +234,61 @@ pub(super) fn deliver(attachment: &Attachment, inbound: Inbound) { _ => {} } } + +/// A record of the agent changing the chat's mode by itself (a harness +/// `current_mode_update` or `config_option_update`). A mode a client set +/// comes as `_acpmux/session_changed` kind `mode` or `config` instead. +fn mode_update(record: &Value) -> bool { + record.get("dir").and_then(Value::as_str) == Some("in") + && matches!( + record.pointer("/msg/params/update/sessionUpdate").and_then(Value::as_str), + Some("current_mode_update" | "config_option_update") + ) +} + +/// Reads the chat's mode again off the reader thread and pushes it as +/// `agent-session-mode` (only for an attach that asked, `mode_events`). One +/// read runs at a time; a change during it makes it read once more, so the +/// last push is the newest mode. +fn reread_mode(attachment: &Arc) { + if !attachment.mode_events { + return; + } + attachment.mode_again.store(true, Ordering::Release); + if attachment.mode_reading.swap(true, Ordering::AcqRel) { + return; + } + let worker = attachment.clone(); + let spawned = std::thread::Builder::new() + .name("mux-agent-mode".into()) + .stack_size(THREAD_STACK_BYTES) + .spawn(move || { + loop { + worker.mode_again.store(false, Ordering::Release); + if worker.ended.load(Ordering::Acquire) { + return; + } + if let Ok(facts) = agent_session_start::read_mode(&worker.link, &worker.session) { + let event = json!({"event": "agent-session-mode", "surface": worker.surface, "mode": facts}); + if worker.writer.send_stream(&event, &worker.outbound).is_err() { + return worker.end(None); + } + } + // A swap, not a store: an event's `swap(true)` that came first in + // `mode_reading`'s order is read here (acquire), so its + // `mode_again = true` is seen below; one that comes later reads + // false and starts a worker itself. A plain store left both + // sides free to miss each other (store buffering) and drop the + // change. + worker.mode_reading.swap(false, Ordering::AcqRel); + if !worker.mode_again.load(Ordering::Acquire) + || worker.mode_reading.swap(true, Ordering::AcqRel) + { + return; + } + } + }); + if spawned.is_err() { + attachment.mode_reading.store(false, Ordering::Release); + } +} diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_transfer/agent_update.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_transfer/agent_update.rs index f01c3548b2d4..285bedb609d8 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_transfer/agent_update.rs +++ b/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, socket: &Path, params: ApplyParams) -> Result { diff --git a/cmux-tui/crates/cmux-tui-core/src/server/capabilities.rs b/cmux-tui/crates/cmux-tui-core/src/server/capabilities.rs index 73a3effbaf03..498950f8c5f0 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/capabilities.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/capabilities.rs @@ -61,6 +61,7 @@ pub(super) fn identify_capabilities(mux: &Mux) -> Vec<&'static str> { #[cfg(unix)] if mux.serves_agent_session_start() { capabilities.push(AGENT_SESSION_START_CAPABILITY); + capabilities.push(AGENT_SESSION_ADOPT_CAPABILITY); capabilities.push(AGENT_SESSION_TRUST_CAPABILITY); capabilities.push(AGENT_SESSION_START_KEYED_REASON_CAPABILITY); } diff --git a/cmux-tui/crates/cmux-tui-core/src/server/conversation_tabs_wire.rs b/cmux-tui/crates/cmux-tui-core/src/server/conversation_tabs_wire.rs index b18e9adec84d..2f5df6ded07a 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/conversation_tabs_wire.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/conversation_tabs_wire.rs @@ -152,7 +152,7 @@ pub(super) struct AgentSessionParams { const MACHINE_HOST_PREFIX: &str = "registry:"; /// A session for a `registry:` tab from anywhere but that machine's own -/// `agent-session-start` (cx-k21v): a session never leaves its machine, so +/// `agent-session-start` or `agent-session-adopt` (cx-k21v): a session never leaves its machine, so /// another machine's chat (a Mac's acpmux session or an outside chat it /// resumes) is never bound into a tab of this one. Two exceptions, both chats /// of this machine's own acpmux ([`landed_here`]): a chat this daemon itself @@ -231,8 +231,8 @@ fn record_of( } (None, None, Some(AgentSessionParams { host, session, harness, host_name }), None) => { // A machine's own tab is created without a session; only that - // machine's agent-session-start binds one, or it names a chat - // this daemon landed. + // machine's agent-session-start or agent-session-adopt binds one, + // or it names a chat this daemon landed. if host.starts_with(MACHINE_HOST_PREFIX) && let Some(session) = &session && !landed_here(mux, &host, session) @@ -341,8 +341,9 @@ pub(super) fn bind( params: BindSessionParams, ) -> anyhow::Result { let BindSessionParams { surface, session, expected_session, harness } = params; - // A machine's own tab binds only through agent-session-start (which - // binds in the store directly, never through this command). + // A machine's own tab binds only through agent-session-start or + // agent-session-adopt (which bind in the store directly, never through + // this command). let machine_tab = mux.surface(surface).and_then(|runtime| mux.conversation_tab_of(&runtime)); if let Some(ConversationTabRecord::AgentSession { host, .. }) = machine_tab && host.starts_with(MACHINE_HOST_PREFIX) diff --git a/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs new file mode 100644 index 000000000000..ea86098866cb --- /dev/null +++ b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs @@ -0,0 +1,176 @@ +//! `agent-session-adopt` (agent-session-adopt-v1): the app opens a chat that +//! already runs in this machine's acpmux (one an agent or `cmux agent open` +//! started there) in an agent tab of this machine, as a workspace of its +//! sidebar. The daemon binds the session only when this machine's own acpmux +//! lists it: never a chat mirrored from a peer, never the Home Chief's, never +//! an id acpmux does not know. A session in a mode that does not ask is +//! adopted too, on an enrolled host only (the start's host gate), and the +//! answer says so (`asking: false`) for the app's bypass badge. The badge +//! follows the chat's mode afterwards: `agent-session-mode` reads it for a +//! tab, the attach reply carries it, and an attachment that asked for +//! `mode_events` hears `agent-session-mode` when the mode changes. + +use std::os::unix::net::{UnixListener, UnixStream}; +use std::sync::{Arc, Mutex}; + +use serde_json::{Value, json}; + +use super::*; + +/// The acpmux stand-in's state: own-1's mode, and the connections that +/// watch session changes. +#[derive(Clone)] +struct FakeAcpmux { + mode: Arc>, + watchers: Arc>>, +} + +impl FakeAcpmux { + /// own-1 switches to `mode`; watchers hear it as acpmux sends a mux + /// `mode` record (`_acpmux/session_changed` kind `mode`). + fn switch_mode(&self, mode: &'static str) { + *self.mode.lock().unwrap() = mode; + let notice = json!({"jsonrpc": "2.0", "method": "_acpmux/session_changed", "params": { + "sessionId": "own-1", "kind": "mode", "recordKind": "mode", + "session": {"sessionId": "own-1"}}}); + for watcher in self.watchers.lock().unwrap().iter_mut() { + let _ = watcher.write_all(format!("{notice}\n").as_bytes()); + } + } +} + +/// An acpmux stand-in that lists one chat of this machine in a bypass mode, +/// one mirrored from a peer and one of the Home Chief. +fn listing_acpmux(socket: PathBuf) -> FakeAcpmux { + let listener = UnixListener::bind(&socket).unwrap(); + let fake = + FakeAcpmux { mode: Arc::new(Mutex::new("bypassPermissions")), watchers: Arc::default() }; + let state = fake.clone(); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(stream) = stream else { return }; + let state = state.clone(); + std::thread::spawn(move || { + let mut writer = stream.try_clone().unwrap(); + for line in BufReader::new(stream).lines() { + let Ok(line) = line else { return }; + let Ok(message) = serde_json::from_str::(&line) else { continue }; + let Some(id) = message.get("id").cloned() else { continue }; + let result = match message["method"].as_str().unwrap_or_default() { + "_acpmux/sessions" => json!({"sessions": [ + {"sessionId": "own-1", "status": "ready", "peer": null, + "harness": "fake", "name": "lane", "tags": {}}, + {"sessionId": "peer-1", "status": "ready", "peer": "mac-a", + "harness": "fake", "tags": {}}, + {"sessionId": "chief-1", "status": "ready", "peer": null, + "harness": "fake", "tags": {"cmux.chief": "home"}}, + ]}), + "_acpmux/web_modes" => json!({ + "refusedFamilies": [], "families": {"fake": ["ask"]}, + "session": {"sessionId": "own-1", "family": "fake", + "mode": *state.mode.lock().unwrap()}, + }), + "_acpmux/attach" => json!({"session": { + "sessionId": "own-1", "name": "lane", "pending": []}, "events": []}), + "_acpmux/watch" => { + state.watchers.lock().unwrap().push(writer.try_clone().unwrap()); + json!({"sessions": []}) + } + _ => json!({}), + }; + let reply = json!({"jsonrpc": "2.0", "id": id, "result": result}); + if writer.write_all(format!("{reply}\n").as_bytes()).is_err() { + return; + } + } + }); + } + }); + fake +} + +impl Daemon { + fn adopt(&self, surface: u64, session: &str) -> Value { + self.rpc(json!({"id": 21, "cmd": "agent-session-adopt", "surface": surface, + "session": session})) + } + + fn mode(&self, surface: u64) -> Value { + self.rpc(json!({"id": 22, "cmd": "agent-session-mode", "surface": surface})) + } +} + +impl Conn { + /// The next pushed `event` on this connection. + fn next_event(&mut self, event: &str) -> Value { + loop { + let mut line = String::new(); + assert_ne!(self.reader.read_line(&mut line).unwrap(), 0, "the daemon closed"); + let message: Value = serde_json::from_str(&line).unwrap(); + if message["event"] == event { + return message; + } + } + } +} + +#[test] +fn agent_session_adopt_binds_a_chat_of_this_machine_and_names_its_bypass_mode() { + let daemon = Daemon::start_with("adopt", None, None); + listing_acpmux(daemon.dir.join("acpmux/acpmux.sock")); + let identity = daemon.rpc(json!({"id": 1, "cmd": "identify"})); + assert!(identity["data"]["capabilities"].to_string().contains("agent-session-adopt-v1")); + let surface = daemon.own_agent_tab(); + for session in ["peer-1", "chief-1", "nope-1"] { + let refused = daemon.adopt(surface, session); + assert_eq!(refused["error_code"], "agent_session.unknown_session", "{session}: {refused}"); + } + let adopted = daemon.adopt(surface, "own-1"); + assert_eq!(adopted["ok"], true, "{adopted}"); + assert_eq!(adopted["data"]["session"], "own-1", "{adopted}"); + assert_eq!(adopted["data"]["asking"], false, "{adopted}"); + assert_eq!(adopted["data"]["mode"], "bypassPermissions", "{adopted}"); + assert_eq!(adopted["data"]["conversation"]["agent_session"]["session"], "own-1", "{adopted}"); + let again = daemon.adopt(surface, "own-1"); + assert_eq!(again["error_code"], "agent_session.bound", "{again}"); + let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); +} + +#[test] +fn agent_session_adopt_is_refused_on_a_team_vm() { + let daemon = Daemon::start_with("adopt-team", Some("team-vm"), None); + listing_acpmux(daemon.dir.join("acpmux/acpmux.sock")); + let surface = daemon.own_agent_tab(); + let refused = daemon.adopt(surface, "own-1"); + assert_eq!(refused["error_code"], "agent_session.team_vm_blocked", "{refused}"); + let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); +} + +#[test] +fn agent_session_mode_follows_an_adopted_chat_through_reads_attach_and_changes() { + let daemon = Daemon::start_with("adopt-mode", None, None); + let acpmux = listing_acpmux(daemon.dir.join("acpmux/acpmux.sock")); + let surface = daemon.own_agent_tab(); + let unbound = daemon.mode(surface); + assert_eq!(unbound["error_code"], "agent_session.unknown_tab", "{unbound}"); + assert_eq!(daemon.adopt(surface, "own-1")["ok"], true); + // A restored tab the app does not show: one read. + let read = daemon.mode(surface); + assert_eq!(read["data"]["session"], "own-1", "{read}"); + assert_eq!(read["data"]["mode"], "bypassPermissions", "{read}"); + assert_eq!(read["data"]["asking"], false, "{read}"); + // A shown tab: the attach reply, then each change. + let mut app = daemon.client(); + let attached = + app.rpc(json!({"cmd": "agent-session-attach", "surface": surface, "mode_events": true})); + assert_eq!(attached["ok"], true, "{attached}"); + assert_eq!(attached["data"]["agentMode"]["asking"], false, "{attached}"); + assert_eq!(attached["data"]["agentMode"]["session"], "own-1", "{attached}"); + acpmux.switch_mode("ask"); + let changed = app.next_event("agent-session-mode"); + assert_eq!(changed["surface"], surface, "{changed}"); + assert_eq!(changed["mode"]["session"], "own-1", "{changed}"); + assert_eq!(changed["mode"]["mode"], "ask", "{changed}"); + assert_eq!(changed["mode"]["asking"], true, "{changed}"); + let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); +} diff --git a/cmux-tui/crates/cmux-tui/tests/cli/agent_session_start.rs b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_start.rs index 0051410874c1..ee2ec5b5e2d5 100644 --- a/cmux-tui/crates/cmux-tui/tests/cli/agent_session_start.rs +++ b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_start.rs @@ -23,6 +23,8 @@ mod agent_focus_cli; mod agent_home_reopen; #[path = "agent_input_hint.rs"] mod agent_input_hint; +#[path = "agent_session_adopt.rs"] +mod agent_session_adopt; #[path = "agents_chat_tools.rs"] mod agents_chat_tools; #[path = "agents_cli_receipt.rs"] diff --git a/cmux-tui/spec/commands.md b/cmux-tui/spec/commands.md index 3ebca6b1b52b..c4cb317e081b 100644 --- a/cmux-tui/spec/commands.md +++ b/cmux-tui/spec/commands.md @@ -9308,6 +9308,50 @@ key itself (and only on `safe_to_restart`, or after the person's click), then starts again; this check does not close the race with an agent that starts between the check and a restart (cx-ofowe). +### agent-session-adopt + +`agent-session-adopt-v1`: `{id, cmd:"agent-session-adopt", surface, session}` +binds a chat that already runs in this machine's own acpmux (one an agent or +`cmux agent open` started here) to an agent tab of this store, then answers +`{surface, session, conversation, family, mode, asking}`. The client then +shows it with `agent-session-attach`. The tab is checked as for +`agent-session-start` (`agent_session.unknown_tab`, `agent_session.bound`), +and so is the machine: a team VM answers `agent_session.team_vm_blocked`, a +Cloud VM of unknown kind `agent_session.host_unverified`. `session` (1 to 128 +letters, digits or `_`, `.`, `:`, `-`) must be listed by this machine's +`_acpmux/sessions` as its own: a chat mirrored from a peer (`peer` set), the +Home Chief's (`tags.cmux.chief`, an `optchat-` name or harness) and an id +acpmux does not list are `agent_session.unknown_session`. The daemon never +starts acpmux for an adopt (`agent_session.acpmux_unavailable` when nothing +listens). The session keeps its mode: `family` and `mode` are its +`_acpmux/web_modes` answer, and `asking` is false when that mode does not ask +before acting (or the family is one the remote table refuses). Such a chat is +adopted too, on the hosts the machine check allows, and the app shows it with +a "bypass" badge; the daemon changes no mode. The bind records the harness +acpmux lists for the session (a profile by its last part), else its family. +Starts and adopts share the per-daemon and per-tab limit +(`agent_session.limit`). Trust is the attach verbs': trusted local (Unix) +connections only. + +The same capability makes the mode readable afterwards, so the badge follows +the chat and never rests on what one app remembers: + +- `{id, cmd:"agent-session-mode", surface}` answers `{surface, session, + family, mode, asking}` for the session the tab's record names, read from + this machine's `_acpmux/web_modes` (`agent_session.unknown_tab` for a tab + with no session, `agent_session.unknown_session` when acpmux no longer knows + that exact session id, `agent_session.limit` past 8 reads at once). + Read-only; the app reads each bound tab it lists once per connect. Adopt + checks the same way that acpmux still knows the session. +- An `agent-session-attach` reply carries `agentMode: {session, family, mode, + asking}` when acpmux answers `_acpmux/web_modes`. +- An attach with `mode_events: true` also pushes `{event: + "agent-session-mode", surface, mode: {session, family, mode, asking}}` + after the mode changes: a mode a client set (`_acpmux/session_changed` + kind `mode` or `config`) or one the agent changed itself (a harness + `current_mode_update` or `config_option_update` record the attachment + receives). Reads run one at a time, so the last push is the newest mode. + ### agent-session-trust `agent-session-trust-v1`: `{id, cmd:"agent-session-trust", surface, cwd, level?}` From e77111ac4f0be753c2e28d23978747709c122144 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 23 Oct 2026 10:17:22 -0700 Subject: [PATCH 2/4] cmux-tui: mode reads follow a re-attach, agent mode changes under transcript kinds, and a record's full name Review (codex xhigh) at fcd4f27: a re-attach on the same connection left out agentMode; mode_events missed current_mode_update when the attach's kinds were transcript plus config_option_update (acpmux filters it out of transcript); agent-session-mode refused a tab whose record holds the chat's full name. The re-attach reads the mode as the first attach does. An attach with mode_events subscribes to the mode record kinds its kinds leave out and keeps them out of the client's page and stream. Mode facts are read for the record's value (an id, or a full name acpmux lists for the resolved id) and keyed by it. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16040-fd0b-7dd0-9487-1a3a198d9f5c&host=cosmic-gray --- .../src/server/agent_session_adopt.rs | 4 +- .../src/server/agent_session_attach.rs | 41 +++- .../src/server/agent_session_start.rs | 37 +++- .../src/server/agent_session_stream.rs | 51 ++++- .../cmux-tui/tests/cli/agent_session_adopt.rs | 178 ++++++++++++++++-- cmux-tui/spec/commands.md | 15 +- 6 files changed, 293 insertions(+), 33 deletions(-) diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs index 0e136304716e..f11b123424b0 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs @@ -31,7 +31,7 @@ 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, read_mode, + AgentSessionStarts, CALL_TIMEOUT, THREAD_STACK_BYTES, checked_mode, is_chief, read_record_mode, startable, }; use super::{MessageWriter, Refusal, SurfaceId, ok, refuse}; @@ -100,7 +100,7 @@ pub(super) fn mode( let read = AcpmuxLink::connect(&socket, |_| {}) .map_err(|_| Refusal::Unavailable) .and_then(|link| { - let read = read_mode(&link, &session); + let read = read_record_mode(&link, &session); link.close(); read }); diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs index 39bd7cfc96cd..6e656ff7048a 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs @@ -169,6 +169,9 @@ struct Attachment { prompts: Mutex>>, /// The attach asked for `agent-session-mode` pushes (`mode_events`). mode_events: bool, + /// The mode record kinds acpmux streams only for the daemon's mode pushes + /// ([`stream::hidden_mode_kinds`]); set again by a re-attach. + hidden_kinds: Mutex>, /// A mode read runs on a worker thread, and whether another change came /// since it began (it reads again): one reader at a time, in order. mode_reading: AtomicBool, @@ -624,8 +627,13 @@ fn tab_session(mux: &Mux, surface: SurfaceId) -> Option { } /// acpmux page params from the client's, with the session pinned and the -/// limit and kinds bounded. -fn page_params(session: &str, params: &PageParams, attach: bool) -> Result { +/// limit and kinds bounded; an attach also subscribes to `hidden` kinds. +fn page_params( + session: &str, + params: &PageParams, + attach: bool, + hidden: &[&'static str], +) -> Result { let mut out = json!({"sessionId": session, "limit": params.limit.unwrap_or(200).clamp(1, MAX_PAGE)}); if let Some(after) = params.after_seq { @@ -643,6 +651,10 @@ fn page_params(session: &str, params: &PageParams, attach: bool) -> Result value, Err(refusal) => return refuse(writer, id, refusal), }; @@ -706,6 +719,7 @@ fn attach( acpmux_params, read_only, mode_events, + hidden_kinds, worker_writer, ); }); @@ -764,7 +778,7 @@ impl ReaderSlot { #[path = "agent_session_stream.rs"] mod stream; pub(crate) use stream::bound_page; -use stream::connect_and_attach; +use stream::{connect_and_attach, strip_kinds}; /// Runs a call that waits on acpmux off the connection thread. Calls of one /// attachment may run side by side; the link bounds how many wait. @@ -831,13 +845,26 @@ fn call(attachment: &Attachment, command: AgentSessionCommand) -> Result { let newest_first = params.after_seq.is_none() || params.before_seq.is_some(); - let params = page_params(session, ¶ms, true)?; + // The re-attach replaces the link's subscription (and its kinds). + let hidden = stream::hidden_mode_kinds(params.kinds.as_deref(), attachment.mode_events); + let params = page_params(session, ¶ms, true, &hidden)?; + *attachment.hidden_kinds.lock().unwrap_or_else(|e| e.into_inner()) = hidden.clone(); let page = attachment.link.call("_acpmux/attach", params, CALL_TIMEOUT); - page.map(|page| bound_page(page, newest_first)).map_err(Refusal::of) + let mut page = + strip_kinds(bound_page(page.map_err(Refusal::of)?, newest_first), &hidden); + // As the first attach's reply: the chat's mode, for the bypass badge. + if let Ok(facts) = agent_session_start::read_mode( + &attachment.link, + session, + &attachment.record_session, + ) { + page["agentMode"] = facts; + } + Ok(page) } AgentSessionCommand::Events(params) => { let newest_first = params.before_seq.is_some(); - let params = page_params(session, ¶ms, false)?; + let params = page_params(session, ¶ms, false, &[])?; let page = attachment.link.call("_acpmux/events", params, CALL_TIMEOUT); page.map(|page| bound_page(page, newest_first)).map_err(Refusal::of) } diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs index 2fad871e95fc..7e95b6d320cf 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs @@ -544,12 +544,43 @@ pub(super) fn checked_mode(session: &str, table: &Value) -> Result Result { +/// [`checked_mode`] of the session acpmux knows by the id `session`, read on +/// `link`, keyed by `record` (the value the tab's record holds, that id or the +/// session's full name), as the app files it. +pub(super) fn read_mode(link: &AcpmuxLink, session: &str, record: &str) -> Result { let table = link .call("_acpmux/web_modes", json!({"sessionId": session}), CALL_TIMEOUT) .map_err(Refusal::of)?; - checked_mode(session, &table) + let mut facts = checked_mode(session, &table)?; + facts["session"] = json!(record); + Ok(facts) +} + +/// The mode of the chat a tab's record names, read on `link`. The record holds +/// the session's id or its full name (attach accepts either, never a prefix); +/// the facts carry the record's value as `session`, the key the app files the +/// bypass badge under. A record acpmux resolves to another session (a prefix, +/// or a name that now belongs to no chat) is `agent_session.unknown_session`. +pub(super) fn read_record_mode(link: &AcpmuxLink, record: &str) -> Result { + let table = link + .call("_acpmux/web_modes", json!({"sessionId": record}), CALL_TIMEOUT) + .map_err(Refusal::of)?; + let resolved = table + .pointer("/session/sessionId") + .and_then(Value::as_str) + .ok_or(Refusal::UnknownSession)?; + if resolved != record { + let listed = link.call("_acpmux/sessions", json!({}), CALL_TIMEOUT).map_err(Refusal::of)?; + let list = listed.get("sessions").unwrap_or(&listed).as_array(); + let named = list.into_iter().flatten().any(|summary| { + summary.get("sessionId").and_then(Value::as_str) == Some(resolved) + && summary.get("name").and_then(Value::as_str) == Some(record) + }); + if !named { + return Err(Refusal::UnknownSession); + } + } + Ok(mode_facts(record, &table)) } /// A Home Chief session (acpmux `hub/route_switch.rs` is_chief, the app's diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs index 108738bb81d1..faba7a886a63 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs @@ -13,6 +13,7 @@ pub(super) fn connect_and_attach( params: Value, read_only: bool, mode_events: bool, + hidden_kinds: Vec<&'static str>, writer: MessageWriter, ) { let sessions = &mux.control_clients.agent_sessions; @@ -44,7 +45,7 @@ pub(super) fn connect_and_attach( } }; let mut page = match link.call("_acpmux/attach", params, CALL_TIMEOUT) { - Ok(page) => bound_page(page, newest_first), + Ok(page) => strip_kinds(bound_page(page, newest_first), &hidden_kinds), Err(error) => { release(sessions); link.close(); @@ -73,7 +74,7 @@ pub(super) fn connect_and_attach( slot.lock().unwrap_or_else(|e| e.into_inner()).session = Some(resolved.clone()); // The chat's mode, for the app's bypass badge; an acpmux that cannot // tell leaves it out. - if let Ok(facts) = agent_session_start::read_mode(&link, &resolved) { + if let Ok(facts) = agent_session_start::read_mode(&link, &resolved, &record_session) { page["agentMode"] = facts; } // Session status changes (`_acpmux/session_changed`) come only to @@ -96,6 +97,7 @@ pub(super) fn connect_and_attach( permissions: Mutex::default(), prompts: Mutex::default(), mode_events, + hidden_kinds: Mutex::new(hidden_kinds), mode_reading: AtomicBool::new(false), mode_again: AtomicBool::new(false), children: Mutex::default(), @@ -170,6 +172,40 @@ pub(crate) fn bound_page(mut page: Value, newest_first: bool) -> Value { page } +/// The records of the agent changing the chat's mode ([`mode_update`]). +const MODE_KINDS: [&str; 2] = ["current_mode_update", "config_option_update"]; + +/// The mode record kinds an attach with `mode_events` subscribes to although +/// the client's `kinds` leave them out (none for a client that takes every +/// record): the daemon needs them to push `agent-session-mode` when the agent +/// changes the mode by itself, and they never reach that client +/// ([`strip_kinds`], [`deliver`]). `transcript` and `wire` leave both out. +pub(super) fn hidden_mode_kinds(kinds: Option<&[String]>, mode_events: bool) -> Vec<&'static str> { + let Some(kinds) = kinds.filter(|_| mode_events) else { return Vec::new() }; + if kinds.iter().any(|kind| kind == "all") { + return Vec::new(); + } + MODE_KINDS.into_iter().filter(|mode| !kinds.iter().any(|kind| kind == mode)).collect() +} + +/// `page` without its records of the `hidden` kinds. A page with more before +/// it that holds nothing else keeps them: an empty page would leave the +/// client no record to page on from. +pub(crate) fn strip_kinds(mut page: Value, hidden: &[&str]) -> Value { + if hidden.is_empty() { + return page; + } + let has_more = page.get("hasMore").and_then(Value::as_bool).unwrap_or(false); + let Some(events) = page.get_mut("events").and_then(Value::as_array_mut) else { return page }; + let shown = |event: &Value| { + !event.get("kind").and_then(Value::as_str).is_some_and(|kind| hidden.contains(&kind)) + }; + if !has_more || events.iter().any(shown) { + events.retain(shown); + } + page +} + /// One notification or the close from the attachment's acpmux link. pub(super) fn deliver(attachment: &Arc, inbound: Inbound) { if attachment.ended.load(Ordering::Acquire) { @@ -189,6 +225,11 @@ pub(super) fn deliver(attachment: &Arc, inbound: Inbound) { if mode_update(¶ms) { reread_mode(attachment); } + // A mode record only the daemon subscribed to (`hidden_mode_kinds`). + let kind = params.get("kind").and_then(Value::as_str).unwrap_or_default(); + if attachment.hidden_kinds.lock().unwrap_or_else(|e| e.into_inner()).contains(&kind) { + return; + } let event = json!({"event": "agent-session-record", "surface": attachment.surface, "record": params}); // Never blocks: a client that does not keep up overflows its // stream, which ends with the `overflow` close (the stream's own @@ -268,7 +309,11 @@ fn reread_mode(attachment: &Arc) { if worker.ended.load(Ordering::Acquire) { return; } - if let Ok(facts) = agent_session_start::read_mode(&worker.link, &worker.session) { + if let Ok(facts) = agent_session_start::read_mode( + &worker.link, + &worker.session, + &worker.record_session, + ) { let event = json!({"event": "agent-session-mode", "surface": worker.surface, "mode": facts}); if worker.writer.send_stream(&event, &worker.outbound).is_err() { return worker.end(None); diff --git a/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs index ea86098866cb..129b5cf48c80 100644 --- a/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs +++ b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs @@ -10,6 +10,7 @@ //! tab, the attach reply carries it, and an attachment that asked for //! `mode_events` hears `agent-session-mode` when the mode changes. +use std::collections::HashMap; use std::os::unix::net::{UnixListener, UnixStream}; use std::sync::{Arc, Mutex}; @@ -17,12 +18,29 @@ use serde_json::{Value, json}; use super::*; -/// The acpmux stand-in's state: own-1's mode, and the connections that -/// watch session changes. +/// The acpmux stand-in's state: own-1's mode, the connections that watch +/// session changes, and each attached connection with its `kinds`. #[derive(Clone)] struct FakeAcpmux { mode: Arc>, watchers: Arc>>, + attached: Arc>)>>>, +} + +/// own-1's record of `kind` at `seq`: an agent `session/update`. +fn record(seq: u64, kind: &str) -> Value { + json!({"sessionId": "own-1", "seq": seq, "dir": "in", "kind": kind, + "msg": {"method": "session/update", "params": {"sessionId": "own-1", + "update": {"sessionUpdate": kind, "currentModeId": "ask"}}}}) +} + +/// Whether acpmux's `kinds` filter passes a record of `kind` (`hub/paging.rs`: +/// `transcript` leaves the mode updates out, `wire` every live update). +fn passes(kinds: &Option>, kind: &str) -> bool { + let mode = matches!(kind, "current_mode_update" | "config_option_update"); + kinds.as_ref().is_none_or(|kinds| { + kinds.iter().any(|k| k == "all" || k == kind || (k == "transcript" && !mode)) + }) } impl FakeAcpmux { @@ -37,17 +55,38 @@ impl FakeAcpmux { let _ = watcher.write_all(format!("{notice}\n").as_bytes()); } } + + /// The agent records `kind` for own-1 (a harness `current_mode_update` + /// when it changes the mode by itself, which is no `session_changed`); + /// each attached connection whose kinds pass it hears it. + fn agent_records(&self, seq: u64, kind: &str) { + let event = + json!({"jsonrpc": "2.0", "method": "_acpmux/event", "params": record(seq, kind)}); + for (stream, kinds) in self.attached.lock().unwrap().values_mut() { + if passes(kinds, kind) { + let _ = stream.write_all(format!("{event}\n").as_bytes()); + } + } + } + + /// The `kinds` each attached connection subscribed with. + fn attach_kinds(&self) -> Vec>> { + self.attached.lock().unwrap().values().map(|(_, kinds)| kinds.clone()).collect() + } } /// An acpmux stand-in that lists one chat of this machine in a bypass mode, /// one mirrored from a peer and one of the Home Chief. fn listing_acpmux(socket: PathBuf) -> FakeAcpmux { let listener = UnixListener::bind(&socket).unwrap(); - let fake = - FakeAcpmux { mode: Arc::new(Mutex::new("bypassPermissions")), watchers: Arc::default() }; + let fake = FakeAcpmux { + mode: Arc::new(Mutex::new("bypassPermissions")), + watchers: Arc::default(), + attached: Arc::default(), + }; let state = fake.clone(); std::thread::spawn(move || { - for stream in listener.incoming() { + for (connection, stream) in listener.incoming().enumerate() { let Ok(stream) = stream else { return }; let state = state.clone(); std::thread::spawn(move || { @@ -65,13 +104,33 @@ fn listing_acpmux(socket: PathBuf) -> FakeAcpmux { {"sessionId": "chief-1", "status": "ready", "peer": null, "harness": "fake", "tags": {"cmux.chief": "home"}}, ]}), - "_acpmux/web_modes" => json!({ - "refusedFamilies": [], "families": {"fake": ["ask"]}, - "session": {"sessionId": "own-1", "family": "fake", - "mode": *state.mode.lock().unwrap()}, - }), - "_acpmux/attach" => json!({"session": { - "sessionId": "own-1", "name": "lane", "pending": []}, "events": []}), + // acpmux resolves an id, a name or a unique prefix of either. + "_acpmux/web_modes" => { + let key = message["params"]["sessionId"].as_str().unwrap_or(""); + let known = + !key.is_empty() && ("own-1".starts_with(key) || key == "lane"); + let mut table = + json!({"refusedFamilies": [], "families": {"fake": ["ask"]}}); + if known { + table["session"] = json!({"sessionId": "own-1", "family": "fake", + "mode": *state.mode.lock().unwrap()}); + } + table + } + "_acpmux/attach" => { + let kinds: Option> = + serde_json::from_value(message["params"]["kinds"].clone()).unwrap(); + let events: Vec = + [(1, "agent_message_chunk"), (2, "current_mode_update")] + .into_iter() + .filter(|(_, kind)| passes(&kinds, kind)) + .map(|(seq, kind)| record(seq, kind)) + .collect(); + let stream = writer.try_clone().unwrap(); + state.attached.lock().unwrap().insert(connection, (stream, kinds)); + json!({"session": {"sessionId": "own-1", "name": "lane", "pending": []}, + "events": events, "hasMore": false}) + } "_acpmux/watch" => { state.watchers.lock().unwrap().push(writer.try_clone().unwrap()); json!({"sessions": []}) @@ -103,17 +162,31 @@ impl Daemon { impl Conn { /// The next pushed `event` on this connection. fn next_event(&mut self, event: &str) -> Value { + self.events_until(event).pop().unwrap() + } + + /// The pushed events on this connection up to the next `event`, with it. + fn events_until(&mut self, event: &str) -> Vec { + let mut seen = Vec::new(); loop { let mut line = String::new(); assert_ne!(self.reader.read_line(&mut line).unwrap(), 0, "the daemon closed"); let message: Value = serde_json::from_str(&line).unwrap(); - if message["event"] == event { - return message; + let done = message["event"] == event; + seen.push(message); + if done { + return seen; } } } } +/// The record kinds of an attach reply's page. +fn page_kinds(reply: &Value) -> Vec { + let events = reply["data"]["events"].as_array().cloned().unwrap_or_default(); + events.iter().map(|event| event["kind"].as_str().unwrap_or_default().to_owned()).collect() +} + #[test] fn agent_session_adopt_binds_a_chat_of_this_machine_and_names_its_bypass_mode() { let daemon = Daemon::start_with("adopt", None, None); @@ -174,3 +247,80 @@ fn agent_session_mode_follows_an_adopted_chat_through_reads_attach_and_changes() assert_eq!(changed["mode"]["asking"], true, "{changed}"); let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); } + +/// The app's attach asks for the transcript and config records only, yet the +/// agent changing the mode by itself (a `current_mode_update`, which acpmux +/// announces as no `session_changed`) still moves the badge: the daemon +/// subscribes to the mode records for its `agent-session-mode` pushes and +/// keeps them out of the client's page and stream. A second attach on the same +/// connection carries the mode too. +#[test] +fn agent_session_mode_pushes_an_agent_mode_change_under_transcript_kinds() { + let daemon = Daemon::start_with("adopt-kinds", None, None); + let acpmux = listing_acpmux(daemon.dir.join("acpmux/acpmux.sock")); + let surface = daemon.own_agent_tab(); + assert_eq!(daemon.adopt(surface, "own-1")["ok"], true); + let mut app = daemon.client(); + let attach = json!({"cmd": "agent-session-attach", "surface": surface, "mode_events": true, + "kinds": ["transcript", "config_option_update"]}); + let attached = app.rpc(attach.clone()); + assert_eq!(attached["ok"], true, "{attached}"); + assert_eq!(page_kinds(&attached), ["agent_message_chunk"], "{attached}"); + let subscribed = acpmux.attach_kinds(); + assert!( + subscribed.iter().flatten().any(|kinds| kinds.iter().any(|k| k == "current_mode_update")), + "{subscribed:?}" + ); + let again = app.rpc(attach); + assert_eq!(again["ok"], true, "{again}"); + assert_eq!(page_kinds(&again), ["agent_message_chunk"], "{again}"); + assert_eq!(again["data"]["agentMode"]["session"], "own-1", "{again}"); + assert_eq!(again["data"]["agentMode"]["asking"], false, "{again}"); + *acpmux.mode.lock().unwrap() = "ask"; + acpmux.agent_records(3, "current_mode_update"); + let seen = app.events_until("agent-session-mode"); + let changed = seen.last().unwrap(); + assert_eq!(changed["mode"]["mode"], "ask", "{changed}"); + assert_eq!(changed["mode"]["asking"], true, "{changed}"); + acpmux.agent_records(4, "agent_message_chunk"); + let seen = [seen, app.events_until("agent-session-record")].concat(); + let records: Vec<&Value> = + seen.iter().filter(|event| event["event"] == "agent-session-record").collect(); + assert_eq!(records.len(), 1, "{seen:?}"); + assert_eq!(records[0]["record"]["kind"], "agent_message_chunk", "{seen:?}"); + let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); +} + +/// A tab whose record holds the chat's full name (attach accepts it) reads +/// its mode, keyed by that name as the app files it; a record acpmux only +/// resolves as a prefix of another chat is refused. +#[test] +fn agent_session_mode_reads_a_tab_that_records_the_chats_full_name() { + let daemon = Daemon::start_with("adopt-name", None, None); + listing_acpmux(daemon.dir.join("acpmux/acpmux.sock")); + let tab = |session: &str| { + let tab = daemon.rpc(json!({"id": 30, "cmd": "new-conversation-tab", + "agent_session": {"host": "install:mac-test"}})); + let surface = tab["data"]["surface"].as_u64().unwrap(); + let bound = daemon.rpc(json!({"id": 31, "cmd": "bind-conversation-tab-session", + "surface": surface, "session": session, + "expected_session": null})); + assert_eq!(bound["ok"], true, "{bound}"); + surface + }; + assert_eq!(daemon.rpc(json!({"id": 29, "cmd": "new-workspace"}))["ok"], true); + let named = tab("lane"); + let read = daemon.mode(named); + assert_eq!(read["ok"], true, "{read}"); + assert_eq!(read["data"]["session"], "lane", "{read}"); + assert_eq!(read["data"]["asking"], false, "{read}"); + let mut app = daemon.client(); + let attached = + app.rpc(json!({"cmd": "agent-session-attach", "surface": named, "mode_events": true})); + assert_eq!(attached["ok"], true, "{attached}"); + assert_eq!(attached["data"]["agentMode"]["session"], "lane", "{attached}"); + let prefix = tab("own"); + let refused = daemon.mode(prefix); + assert_eq!(refused["error_code"], "agent_session.unknown_session", "{refused}"); + let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); +} diff --git a/cmux-tui/spec/commands.md b/cmux-tui/spec/commands.md index c4cb317e081b..d5d1c3362a43 100644 --- a/cmux-tui/spec/commands.md +++ b/cmux-tui/spec/commands.md @@ -9340,17 +9340,24 @@ the chat and never rests on what one app remembers: family, mode, asking}` for the session the tab's record names, read from this machine's `_acpmux/web_modes` (`agent_session.unknown_tab` for a tab with no session, `agent_session.unknown_session` when acpmux no longer knows - that exact session id, `agent_session.limit` past 8 reads at once). + that exact session id or full name, as attach accepts it, never a prefix; + `agent_session.limit` past 8 reads at once). `session` is the record's + value (the id or the name), the key the app files the badge under; the + attach reply and the pushes use the same key. Read-only; the app reads each bound tab it lists once per connect. Adopt checks the same way that acpmux still knows the session. - An `agent-session-attach` reply carries `agentMode: {session, family, mode, - asking}` when acpmux answers `_acpmux/web_modes`. + asking}` when acpmux answers `_acpmux/web_modes`, a repeat attach on the + same connection too. - An attach with `mode_events: true` also pushes `{event: "agent-session-mode", surface, mode: {session, family, mode, asking}}` after the mode changes: a mode a client set (`_acpmux/session_changed` kind `mode` or `config`) or one the agent changed itself (a harness - `current_mode_update` or `config_option_update` record the attachment - receives). Reads run one at a time, so the last push is the newest mode. + `current_mode_update` or `config_option_update` record). When the attach's + `kinds` leave those records out (`transcript` does), the daemon subscribes + to them anyway and keeps them out of the client's page and stream (a page + with more before it that holds nothing else keeps them, so the client can + page on). Reads run one at a time, so the last push is the newest mode. ### agent-session-trust From 5024de4e08bba2ff54000111af50587725b500d8 Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 23 Oct 2026 10:18:50 -0700 Subject: [PATCH 3/4] cmux-tui: keep this PR to agent-session-adopt; the mode reads move to their own PR The adopt verb alone stays here; agent-session-mode, the attach reply's agentMode and the mode_events pushes land in a PR stacked on this one, so each stays under about 800 changed lines. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16040-fd0b-7dd0-9487-1a3a198d9f5c&host=cosmic-gray --- .../src/server/agent_session_adopt.rs | 82 +----- .../src/server/agent_session_attach.rs | 77 +----- .../src/server/agent_session_attach_tests.rs | 19 +- .../src/server/agent_session_start.rs | 44 +-- .../src/server/agent_session_stream.rs | 127 +-------- .../cmux-tui/tests/cli/agent_session_adopt.rs | 252 +----------------- cmux-tui/spec/commands.md | 26 -- 7 files changed, 26 insertions(+), 601 deletions(-) diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs index f11b123424b0..cda789d70ad6 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_adopt.rs @@ -24,15 +24,13 @@ use std::path::PathBuf; use std::sync::Arc; -use std::sync::atomic::{AtomicUsize, Ordering}; 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, read_record_mode, - startable, + AgentSessionStarts, CALL_TIMEOUT, THREAD_STACK_BYTES, checked_mode, is_chief, startable, }; use super::{MessageWriter, Refusal, SurfaceId, ok, refuse}; use crate::mux::Mux; @@ -40,84 +38,6 @@ 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 ModeParams { - surface: SurfaceId, -} - -/// At most [`MAX_MODE_READS`] `agent-session-mode` reads run at once, each on -/// its own thread and acpmux connection; one more is `agent_session.limit`. -const MAX_MODE_READS: usize = 8; -static MODE_READS: AtomicUsize = AtomicUsize::new(0); - -/// One running mode read; dropping it frees the slot. -struct ModeRead; - -impl ModeRead { - fn admit() -> Option { - MODE_READS - .fetch_update(Ordering::AcqRel, Ordering::Acquire, |n| { - (n < MAX_MODE_READS).then_some(n + 1) - }) - .ok() - .map(|_| Self) - } -} - -impl Drop for ModeRead { - fn drop(&mut self) { - MODE_READS.fetch_sub(1, Ordering::AcqRel); - } -} - -/// `agent-session-mode {surface}` of a trusted local connection (the parent -/// checked trust and the acpmux socket): the mode of the chat the tab is bound -/// to, `{surface, session, family, mode, asking}`, read from this machine's -/// acpmux. Read-only: the app reads it once for each bound tab it lists at a -/// connect (a restored tab it does not show), so the bypass badge never -/// depends on what this app remembers. -pub(super) fn mode( - mux: &Arc, - id: Option, - socket: PathBuf, - params: ModeParams, - writer: &MessageWriter, -) -> bool { - let Some(session) = super::tab_session(mux, params.surface) else { - return refuse(writer, id, Refusal::UnknownTab); - }; - let Some(slot) = ModeRead::admit() else { - return refuse(writer, id, Refusal::Limit); - }; - let worker_writer = writer.clone(); - let worker_id = id.clone(); - let spawned = std::thread::Builder::new() - .name("mux-agent-mode".into()) - .stack_size(THREAD_STACK_BYTES) - .spawn(move || { - let _slot = slot; - let read = AcpmuxLink::connect(&socket, |_| {}) - .map_err(|_| Refusal::Unavailable) - .and_then(|link| { - let read = read_record_mode(&link, &session); - link.close(); - read - }); - match read { - Ok(mut facts) => { - facts["surface"] = json!(params.surface); - ok(&worker_writer, worker_id, facts) - } - Err(refusal) => refuse(&worker_writer, worker_id, refusal), - }; - }); - if spawned.is_err() { - return refuse(writer, id, Refusal::Limit); - } - true -} - #[derive(Deserialize)] #[serde(deny_unknown_fields)] pub(super) struct AdoptParams { diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs index 6e656ff7048a..adfe323b82ee 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach.rs @@ -167,15 +167,6 @@ struct Attachment { permissions: Mutex>, /// Prompt ids waiting for `_acpmux/prompt_accepted`, with their reply slot. prompts: Mutex>>, - /// The attach asked for `agent-session-mode` pushes (`mode_events`). - mode_events: bool, - /// The mode record kinds acpmux streams only for the daemon's mode pushes - /// ([`stream::hidden_mode_kinds`]); set again by a re-attach. - hidden_kinds: Mutex>, - /// A mode read runs on a worker thread, and whether another change came - /// since it began (it reads again): one reader at a time, in order. - mode_reading: AtomicBool, - mode_again: AtomicBool, /// The session's descendants this attachment follows /// (`agent-session-children-v1`): their changes are streamed too. children: Mutex>, @@ -249,11 +240,6 @@ struct PageParams { limit: Option, #[serde(default)] kinds: Option>, - /// Attach only: the attachment also pushes `agent-session-mode` when the - /// chat's mode changes (an app that reads it asks; any other client would - /// not know the event). - #[serde(default)] - mode_events: bool, } enum AgentSessionCommand { @@ -480,9 +466,6 @@ pub(super) fn try_handle( if cmd == "agent-session-adopt" { return Some(adopt(mux, client, id, object, writer)); } - if cmd == "agent-session-mode" { - return Some(mode(mux, client, id, object, writer)); - } let Some(command) = AgentSessionCommand::parse(&cmd, Value::Object(object)) else { return Some(refuse(writer, id, Refusal::BadRequest)); }; @@ -555,26 +538,6 @@ fn adopt( agent_session_adopt::adopt(mux, client, id, socket, params, writer) } -/// `agent-session-mode` (agent_session_adopt.rs), with the attach verbs' checks. -fn mode( - mux: &Arc, - client: u64, - id: Option, - params: serde_json::Map, - 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::mode(mux, id, socket, params, writer) -} - /// `agent-session-trust` (agent_session_trust.rs), with the same checks. fn trust( mux: &Arc, @@ -627,13 +590,8 @@ fn tab_session(mux: &Mux, surface: SurfaceId) -> Option { } /// acpmux page params from the client's, with the session pinned and the -/// limit and kinds bounded; an attach also subscribes to `hidden` kinds. -fn page_params( - session: &str, - params: &PageParams, - attach: bool, - hidden: &[&'static str], -) -> Result { +/// limit and kinds bounded. +fn page_params(session: &str, params: &PageParams, attach: bool) -> Result { let mut out = json!({"sessionId": session, "limit": params.limit.unwrap_or(200).clamp(1, MAX_PAGE)}); if let Some(after) = params.after_seq { @@ -651,10 +609,6 @@ fn page_params( if !valid { return Err(Refusal::BadRequest); } - let mut kinds = kinds.clone(); - if attach { - kinds.extend(hidden.iter().map(|kind| kind.to_string())); - } out.insert("kinds", json!(kinds)); } if attach { @@ -684,8 +638,7 @@ fn attach( return refuse(writer, id, Refusal::UnknownTab); }; let record_session = session.clone(); - let hidden_kinds = stream::hidden_mode_kinds(params.kinds.as_deref(), params.mode_events); - let acpmux_params = match page_params(&session, ¶ms, true, &hidden_kinds) { + let acpmux_params = match page_params(&session, ¶ms, true) { Ok(value) => value, Err(refusal) => return refuse(writer, id, refusal), }; @@ -704,7 +657,6 @@ fn attach( let worker_writer = writer.clone(); let worker_id = id.clone(); let surface = params.surface; - let mode_events = params.mode_events; let spawned = std::thread::Builder::new() .name("mux-agent-attach".into()) .stack_size(THREAD_STACK_BYTES) @@ -718,8 +670,6 @@ fn attach( record_session, acpmux_params, read_only, - mode_events, - hidden_kinds, worker_writer, ); }); @@ -778,7 +728,7 @@ impl ReaderSlot { #[path = "agent_session_stream.rs"] mod stream; pub(crate) use stream::bound_page; -use stream::{connect_and_attach, strip_kinds}; +use stream::connect_and_attach; /// Runs a call that waits on acpmux off the connection thread. Calls of one /// attachment may run side by side; the link bounds how many wait. @@ -845,26 +795,13 @@ fn call(attachment: &Attachment, command: AgentSessionCommand) -> Result { let newest_first = params.after_seq.is_none() || params.before_seq.is_some(); - // The re-attach replaces the link's subscription (and its kinds). - let hidden = stream::hidden_mode_kinds(params.kinds.as_deref(), attachment.mode_events); - let params = page_params(session, ¶ms, true, &hidden)?; - *attachment.hidden_kinds.lock().unwrap_or_else(|e| e.into_inner()) = hidden.clone(); + let params = page_params(session, ¶ms, true)?; let page = attachment.link.call("_acpmux/attach", params, CALL_TIMEOUT); - let mut page = - strip_kinds(bound_page(page.map_err(Refusal::of)?, newest_first), &hidden); - // As the first attach's reply: the chat's mode, for the bypass badge. - if let Ok(facts) = agent_session_start::read_mode( - &attachment.link, - session, - &attachment.record_session, - ) { - page["agentMode"] = facts; - } - Ok(page) + page.map(|page| bound_page(page, newest_first)).map_err(Refusal::of) } AgentSessionCommand::Events(params) => { let newest_first = params.before_seq.is_some(); - let params = page_params(session, ¶ms, false, &[])?; + let params = page_params(session, ¶ms, false)?; let page = attachment.link.call("_acpmux/events", params, CALL_TIMEOUT); page.map(|page| bound_page(page, newest_first)).map_err(Refusal::of) } diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach_tests.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach_tests.rs index d9cd39da659b..b05511089806 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach_tests.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_attach_tests.rs @@ -58,12 +58,12 @@ impl FakeAcpmux { Self { directory, socket, frames, connections, current: None } } - /// The next frame the daemon sent to acpmux (its session watch and the - /// attach's mode read skipped: the daemon's own reads, never the client's). + /// The next frame the daemon sent to acpmux. + /// The next frame the daemon sent to acpmux (its session watch skipped). pub(super) fn frame(&self) -> Value { loop { let frame = self.frames.recv_timeout(WAIT).expect("a frame to acpmux"); - if !daemon_read(&frame) { + if frame["method"] != json!("_acpmux/watch") { return frame; } } @@ -73,7 +73,7 @@ impl FakeAcpmux { let deadline = Instant::now() + Duration::from_millis(200); while let Some(left) = deadline.checked_duration_since(Instant::now()) { match self.frames.recv_timeout(left) { - Ok(frame) if daemon_read(&frame) => continue, + Ok(frame) if frame["method"] == json!("_acpmux/watch") => continue, Ok(frame) => panic!("the daemon must not reach acpmux: {frame}"), Err(_) => return, } @@ -112,12 +112,6 @@ impl Drop for FakeAcpmux { } } -/// A read the daemon makes for itself on an attach: the session watch, and -/// the chat's mode for the attach reply's `agentMode` (`agent-session-adopt-v1`). -fn daemon_read(frame: &Value) -> bool { - matches!(frame["method"].as_str(), Some("_acpmux/watch" | "_acpmux/web_modes")) -} - fn serve(stream: UnixStream, frames: Sender) { let mut writer = stream.try_clone().unwrap(); for line in BufReader::new(stream).lines() { @@ -146,11 +140,6 @@ fn serve(stream: UnixStream, frames: Sender) { // (agent_session_children_tests.rs). "_acpmux/sessions" => Some(machine_sessions()), "_acpmux/events" => Some(json!({"events": [], "hasMore": false, "lastSeq": 1})), - // The attached chat asks before acting (its mode is its family's asking default). - "_acpmux/web_modes" => Some(json!({ - "families": {"claude": ["default"]}, - "session": {"sessionId": params["sessionId"], "family": "claude", "mode": "default"}, - })), "_acpmux/permission_respond" => Some(json!({})), // `acpmux.transfer.v1` with the guarded receive (agent_transfer_tests.rs). method if method.starts_with("_acpmux/transfer_") => transfer_reply(method, params), diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs index 7e95b6d320cf..5aedcdd5e5ce 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_start.rs @@ -521,9 +521,8 @@ pub(super) fn asks(table: &Value) -> (bool, Option) { } /// `session`'s mode in a `_acpmux/web_modes {sessionId}` answer, as the adopt -/// reply, the attach reply's `agentMode` and `agent-session-mode` carry it: -/// its family, its mode and whether it asks before acting ([`asks`]); the app -/// shows `asking: false` as the bypass badge. +/// 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!({ @@ -544,45 +543,6 @@ pub(super) fn checked_mode(session: &str, table: &Value) -> Result Result { - let table = link - .call("_acpmux/web_modes", json!({"sessionId": session}), CALL_TIMEOUT) - .map_err(Refusal::of)?; - let mut facts = checked_mode(session, &table)?; - facts["session"] = json!(record); - Ok(facts) -} - -/// The mode of the chat a tab's record names, read on `link`. The record holds -/// the session's id or its full name (attach accepts either, never a prefix); -/// the facts carry the record's value as `session`, the key the app files the -/// bypass badge under. A record acpmux resolves to another session (a prefix, -/// or a name that now belongs to no chat) is `agent_session.unknown_session`. -pub(super) fn read_record_mode(link: &AcpmuxLink, record: &str) -> Result { - let table = link - .call("_acpmux/web_modes", json!({"sessionId": record}), CALL_TIMEOUT) - .map_err(Refusal::of)?; - let resolved = table - .pointer("/session/sessionId") - .and_then(Value::as_str) - .ok_or(Refusal::UnknownSession)?; - if resolved != record { - let listed = link.call("_acpmux/sessions", json!({}), CALL_TIMEOUT).map_err(Refusal::of)?; - let list = listed.get("sessions").unwrap_or(&listed).as_array(); - let named = list.into_iter().flatten().any(|summary| { - summary.get("sessionId").and_then(Value::as_str) == Some(resolved) - && summary.get("name").and_then(Value::as_str) == Some(record) - }); - if !named { - return Err(Refusal::UnknownSession); - } - } - Ok(mode_facts(record, &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 { diff --git a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs index faba7a886a63..98e47701fa5c 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/agent_session_stream.rs @@ -12,8 +12,6 @@ pub(super) fn connect_and_attach( record_session: String, params: Value, read_only: bool, - mode_events: bool, - hidden_kinds: Vec<&'static str>, writer: MessageWriter, ) { let sessions = &mux.control_clients.agent_sessions; @@ -44,8 +42,8 @@ pub(super) fn connect_and_attach( return; } }; - let mut page = match link.call("_acpmux/attach", params, CALL_TIMEOUT) { - Ok(page) => strip_kinds(bound_page(page, newest_first), &hidden_kinds), + let page = match link.call("_acpmux/attach", params, CALL_TIMEOUT) { + Ok(page) => bound_page(page, newest_first), Err(error) => { release(sessions); link.close(); @@ -72,11 +70,6 @@ pub(super) fn connect_and_attach( return; } slot.lock().unwrap_or_else(|e| e.into_inner()).session = Some(resolved.clone()); - // The chat's mode, for the app's bypass badge; an acpmux that cannot - // tell leaves it out. - if let Ok(facts) = agent_session_start::read_mode(&link, &resolved, &record_session) { - page["agentMode"] = facts; - } // Session status changes (`_acpmux/session_changed`) come only to // watchers; the reader keeps only this session's and its descendants'. // The answer (every session's summary) never reaches the client: only @@ -96,10 +89,6 @@ pub(super) fn connect_and_attach( ended: AtomicBool::new(false), permissions: Mutex::default(), prompts: Mutex::default(), - mode_events, - hidden_kinds: Mutex::new(hidden_kinds), - mode_reading: AtomicBool::new(false), - mode_again: AtomicBool::new(false), children: Mutex::default(), }); if let Ok(watched) = &watched { @@ -172,42 +161,8 @@ pub(crate) fn bound_page(mut page: Value, newest_first: bool) -> Value { page } -/// The records of the agent changing the chat's mode ([`mode_update`]). -const MODE_KINDS: [&str; 2] = ["current_mode_update", "config_option_update"]; - -/// The mode record kinds an attach with `mode_events` subscribes to although -/// the client's `kinds` leave them out (none for a client that takes every -/// record): the daemon needs them to push `agent-session-mode` when the agent -/// changes the mode by itself, and they never reach that client -/// ([`strip_kinds`], [`deliver`]). `transcript` and `wire` leave both out. -pub(super) fn hidden_mode_kinds(kinds: Option<&[String]>, mode_events: bool) -> Vec<&'static str> { - let Some(kinds) = kinds.filter(|_| mode_events) else { return Vec::new() }; - if kinds.iter().any(|kind| kind == "all") { - return Vec::new(); - } - MODE_KINDS.into_iter().filter(|mode| !kinds.iter().any(|kind| kind == mode)).collect() -} - -/// `page` without its records of the `hidden` kinds. A page with more before -/// it that holds nothing else keeps them: an empty page would leave the -/// client no record to page on from. -pub(crate) fn strip_kinds(mut page: Value, hidden: &[&str]) -> Value { - if hidden.is_empty() { - return page; - } - let has_more = page.get("hasMore").and_then(Value::as_bool).unwrap_or(false); - let Some(events) = page.get_mut("events").and_then(Value::as_array_mut) else { return page }; - let shown = |event: &Value| { - !event.get("kind").and_then(Value::as_str).is_some_and(|kind| hidden.contains(&kind)) - }; - if !has_more || events.iter().any(shown) { - events.retain(shown); - } - page -} - /// One notification or the close from the attachment's acpmux link. -pub(super) fn deliver(attachment: &Arc, inbound: Inbound) { +pub(super) fn deliver(attachment: &Attachment, inbound: Inbound) { if attachment.ended.load(Ordering::Acquire) { return; } @@ -222,14 +177,6 @@ pub(super) fn deliver(attachment: &Arc, inbound: Inbound) { } match method.as_str() { "_acpmux/event" if same_session => { - if mode_update(¶ms) { - reread_mode(attachment); - } - // A mode record only the daemon subscribed to (`hidden_mode_kinds`). - let kind = params.get("kind").and_then(Value::as_str).unwrap_or_default(); - if attachment.hidden_kinds.lock().unwrap_or_else(|e| e.into_inner()).contains(&kind) { - return; - } let event = json!({"event": "agent-session-record", "surface": attachment.surface, "record": params}); // Never blocks: a client that does not keep up overflows its // stream, which ends with the `overflow` close (the stream's own @@ -253,12 +200,6 @@ pub(super) fn deliver(attachment: &Arc, inbound: Inbound) { if !same_session && !attachment.tab_current() { return attachment.end(Some("detached")); } - // Only the session's own mode change re-reads its mode (a descendant's never does). - if same_session - && matches!(params.get("kind").and_then(Value::as_str), Some("mode" | "config")) - { - reread_mode(attachment); - } let event = json!({"event": "agent-session-changed", "surface": attachment.surface, "change": params}); if attachment.writer.send_stream(&event, &attachment.outbound).is_err() { attachment.end(None); @@ -275,65 +216,3 @@ pub(super) fn deliver(attachment: &Arc, inbound: Inbound) { _ => {} } } - -/// A record of the agent changing the chat's mode by itself (a harness -/// `current_mode_update` or `config_option_update`). A mode a client set -/// comes as `_acpmux/session_changed` kind `mode` or `config` instead. -fn mode_update(record: &Value) -> bool { - record.get("dir").and_then(Value::as_str) == Some("in") - && matches!( - record.pointer("/msg/params/update/sessionUpdate").and_then(Value::as_str), - Some("current_mode_update" | "config_option_update") - ) -} - -/// Reads the chat's mode again off the reader thread and pushes it as -/// `agent-session-mode` (only for an attach that asked, `mode_events`). One -/// read runs at a time; a change during it makes it read once more, so the -/// last push is the newest mode. -fn reread_mode(attachment: &Arc) { - if !attachment.mode_events { - return; - } - attachment.mode_again.store(true, Ordering::Release); - if attachment.mode_reading.swap(true, Ordering::AcqRel) { - return; - } - let worker = attachment.clone(); - let spawned = std::thread::Builder::new() - .name("mux-agent-mode".into()) - .stack_size(THREAD_STACK_BYTES) - .spawn(move || { - loop { - worker.mode_again.store(false, Ordering::Release); - if worker.ended.load(Ordering::Acquire) { - return; - } - if let Ok(facts) = agent_session_start::read_mode( - &worker.link, - &worker.session, - &worker.record_session, - ) { - let event = json!({"event": "agent-session-mode", "surface": worker.surface, "mode": facts}); - if worker.writer.send_stream(&event, &worker.outbound).is_err() { - return worker.end(None); - } - } - // A swap, not a store: an event's `swap(true)` that came first in - // `mode_reading`'s order is read here (acquire), so its - // `mode_again = true` is seen below; one that comes later reads - // false and starts a worker itself. A plain store left both - // sides free to miss each other (store buffering) and drop the - // change. - worker.mode_reading.swap(false, Ordering::AcqRel); - if !worker.mode_again.load(Ordering::Acquire) - || worker.mode_reading.swap(true, Ordering::AcqRel) - { - return; - } - } - }); - if spawned.is_err() { - attachment.mode_reading.store(false, Ordering::Release); - } -} diff --git a/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs index 129b5cf48c80..31da49db34c5 100644 --- a/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs +++ b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs @@ -5,90 +5,21 @@ //! lists it: never a chat mirrored from a peer, never the Home Chief's, never //! an id acpmux does not know. A session in a mode that does not ask is //! adopted too, on an enrolled host only (the start's host gate), and the -//! answer says so (`asking: false`) for the app's bypass badge. The badge -//! follows the chat's mode afterwards: `agent-session-mode` reads it for a -//! tab, the attach reply carries it, and an attachment that asked for -//! `mode_events` hears `agent-session-mode` when the mode changes. +//! answer says so (`asking: false`) for the app's bypass badge. -use std::collections::HashMap; -use std::os::unix::net::{UnixListener, UnixStream}; -use std::sync::{Arc, Mutex}; +use std::os::unix::net::UnixListener; use serde_json::{Value, json}; use super::*; -/// The acpmux stand-in's state: own-1's mode, the connections that watch -/// session changes, and each attached connection with its `kinds`. -#[derive(Clone)] -struct FakeAcpmux { - mode: Arc>, - watchers: Arc>>, - attached: Arc>)>>>, -} - -/// own-1's record of `kind` at `seq`: an agent `session/update`. -fn record(seq: u64, kind: &str) -> Value { - json!({"sessionId": "own-1", "seq": seq, "dir": "in", "kind": kind, - "msg": {"method": "session/update", "params": {"sessionId": "own-1", - "update": {"sessionUpdate": kind, "currentModeId": "ask"}}}}) -} - -/// Whether acpmux's `kinds` filter passes a record of `kind` (`hub/paging.rs`: -/// `transcript` leaves the mode updates out, `wire` every live update). -fn passes(kinds: &Option>, kind: &str) -> bool { - let mode = matches!(kind, "current_mode_update" | "config_option_update"); - kinds.as_ref().is_none_or(|kinds| { - kinds.iter().any(|k| k == "all" || k == kind || (k == "transcript" && !mode)) - }) -} - -impl FakeAcpmux { - /// own-1 switches to `mode`; watchers hear it as acpmux sends a mux - /// `mode` record (`_acpmux/session_changed` kind `mode`). - fn switch_mode(&self, mode: &'static str) { - *self.mode.lock().unwrap() = mode; - let notice = json!({"jsonrpc": "2.0", "method": "_acpmux/session_changed", "params": { - "sessionId": "own-1", "kind": "mode", "recordKind": "mode", - "session": {"sessionId": "own-1"}}}); - for watcher in self.watchers.lock().unwrap().iter_mut() { - let _ = watcher.write_all(format!("{notice}\n").as_bytes()); - } - } - - /// The agent records `kind` for own-1 (a harness `current_mode_update` - /// when it changes the mode by itself, which is no `session_changed`); - /// each attached connection whose kinds pass it hears it. - fn agent_records(&self, seq: u64, kind: &str) { - let event = - json!({"jsonrpc": "2.0", "method": "_acpmux/event", "params": record(seq, kind)}); - for (stream, kinds) in self.attached.lock().unwrap().values_mut() { - if passes(kinds, kind) { - let _ = stream.write_all(format!("{event}\n").as_bytes()); - } - } - } - - /// The `kinds` each attached connection subscribed with. - fn attach_kinds(&self) -> Vec>> { - self.attached.lock().unwrap().values().map(|(_, kinds)| kinds.clone()).collect() - } -} - /// An acpmux stand-in that lists one chat of this machine in a bypass mode, /// one mirrored from a peer and one of the Home Chief. -fn listing_acpmux(socket: PathBuf) -> FakeAcpmux { +fn listing_acpmux(socket: PathBuf) { let listener = UnixListener::bind(&socket).unwrap(); - let fake = FakeAcpmux { - mode: Arc::new(Mutex::new("bypassPermissions")), - watchers: Arc::default(), - attached: Arc::default(), - }; - let state = fake.clone(); std::thread::spawn(move || { - for (connection, stream) in listener.incoming().enumerate() { + for stream in listener.incoming() { let Ok(stream) = stream else { return }; - let state = state.clone(); std::thread::spawn(move || { let mut writer = stream.try_clone().unwrap(); for line in BufReader::new(stream).lines() { @@ -104,37 +35,11 @@ fn listing_acpmux(socket: PathBuf) -> FakeAcpmux { {"sessionId": "chief-1", "status": "ready", "peer": null, "harness": "fake", "tags": {"cmux.chief": "home"}}, ]}), - // acpmux resolves an id, a name or a unique prefix of either. - "_acpmux/web_modes" => { - let key = message["params"]["sessionId"].as_str().unwrap_or(""); - let known = - !key.is_empty() && ("own-1".starts_with(key) || key == "lane"); - let mut table = - json!({"refusedFamilies": [], "families": {"fake": ["ask"]}}); - if known { - table["session"] = json!({"sessionId": "own-1", "family": "fake", - "mode": *state.mode.lock().unwrap()}); - } - table - } - "_acpmux/attach" => { - let kinds: Option> = - serde_json::from_value(message["params"]["kinds"].clone()).unwrap(); - let events: Vec = - [(1, "agent_message_chunk"), (2, "current_mode_update")] - .into_iter() - .filter(|(_, kind)| passes(&kinds, kind)) - .map(|(seq, kind)| record(seq, kind)) - .collect(); - let stream = writer.try_clone().unwrap(); - state.attached.lock().unwrap().insert(connection, (stream, kinds)); - json!({"session": {"sessionId": "own-1", "name": "lane", "pending": []}, - "events": events, "hasMore": false}) - } - "_acpmux/watch" => { - state.watchers.lock().unwrap().push(writer.try_clone().unwrap()); - json!({"sessions": []}) - } + "_acpmux/web_modes" => json!({ + "refusedFamilies": [], "families": {"fake": ["ask"]}, + "session": {"sessionId": "own-1", "family": "fake", + "mode": "bypassPermissions"}, + }), _ => json!({}), }; let reply = json!({"jsonrpc": "2.0", "id": id, "result": result}); @@ -145,7 +50,6 @@ fn listing_acpmux(socket: PathBuf) -> FakeAcpmux { }); } }); - fake } impl Daemon { @@ -153,38 +57,6 @@ impl Daemon { self.rpc(json!({"id": 21, "cmd": "agent-session-adopt", "surface": surface, "session": session})) } - - fn mode(&self, surface: u64) -> Value { - self.rpc(json!({"id": 22, "cmd": "agent-session-mode", "surface": surface})) - } -} - -impl Conn { - /// The next pushed `event` on this connection. - fn next_event(&mut self, event: &str) -> Value { - self.events_until(event).pop().unwrap() - } - - /// The pushed events on this connection up to the next `event`, with it. - fn events_until(&mut self, event: &str) -> Vec { - let mut seen = Vec::new(); - loop { - let mut line = String::new(); - assert_ne!(self.reader.read_line(&mut line).unwrap(), 0, "the daemon closed"); - let message: Value = serde_json::from_str(&line).unwrap(); - let done = message["event"] == event; - seen.push(message); - if done { - return seen; - } - } - } -} - -/// The record kinds of an attach reply's page. -fn page_kinds(reply: &Value) -> Vec { - let events = reply["data"]["events"].as_array().cloned().unwrap_or_default(); - events.iter().map(|event| event["kind"].as_str().unwrap_or_default().to_owned()).collect() } #[test] @@ -218,109 +90,3 @@ fn agent_session_adopt_is_refused_on_a_team_vm() { assert_eq!(refused["error_code"], "agent_session.team_vm_blocked", "{refused}"); let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); } - -#[test] -fn agent_session_mode_follows_an_adopted_chat_through_reads_attach_and_changes() { - let daemon = Daemon::start_with("adopt-mode", None, None); - let acpmux = listing_acpmux(daemon.dir.join("acpmux/acpmux.sock")); - let surface = daemon.own_agent_tab(); - let unbound = daemon.mode(surface); - assert_eq!(unbound["error_code"], "agent_session.unknown_tab", "{unbound}"); - assert_eq!(daemon.adopt(surface, "own-1")["ok"], true); - // A restored tab the app does not show: one read. - let read = daemon.mode(surface); - assert_eq!(read["data"]["session"], "own-1", "{read}"); - assert_eq!(read["data"]["mode"], "bypassPermissions", "{read}"); - assert_eq!(read["data"]["asking"], false, "{read}"); - // A shown tab: the attach reply, then each change. - let mut app = daemon.client(); - let attached = - app.rpc(json!({"cmd": "agent-session-attach", "surface": surface, "mode_events": true})); - assert_eq!(attached["ok"], true, "{attached}"); - assert_eq!(attached["data"]["agentMode"]["asking"], false, "{attached}"); - assert_eq!(attached["data"]["agentMode"]["session"], "own-1", "{attached}"); - acpmux.switch_mode("ask"); - let changed = app.next_event("agent-session-mode"); - assert_eq!(changed["surface"], surface, "{changed}"); - assert_eq!(changed["mode"]["session"], "own-1", "{changed}"); - assert_eq!(changed["mode"]["mode"], "ask", "{changed}"); - assert_eq!(changed["mode"]["asking"], true, "{changed}"); - let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); -} - -/// The app's attach asks for the transcript and config records only, yet the -/// agent changing the mode by itself (a `current_mode_update`, which acpmux -/// announces as no `session_changed`) still moves the badge: the daemon -/// subscribes to the mode records for its `agent-session-mode` pushes and -/// keeps them out of the client's page and stream. A second attach on the same -/// connection carries the mode too. -#[test] -fn agent_session_mode_pushes_an_agent_mode_change_under_transcript_kinds() { - let daemon = Daemon::start_with("adopt-kinds", None, None); - let acpmux = listing_acpmux(daemon.dir.join("acpmux/acpmux.sock")); - let surface = daemon.own_agent_tab(); - assert_eq!(daemon.adopt(surface, "own-1")["ok"], true); - let mut app = daemon.client(); - let attach = json!({"cmd": "agent-session-attach", "surface": surface, "mode_events": true, - "kinds": ["transcript", "config_option_update"]}); - let attached = app.rpc(attach.clone()); - assert_eq!(attached["ok"], true, "{attached}"); - assert_eq!(page_kinds(&attached), ["agent_message_chunk"], "{attached}"); - let subscribed = acpmux.attach_kinds(); - assert!( - subscribed.iter().flatten().any(|kinds| kinds.iter().any(|k| k == "current_mode_update")), - "{subscribed:?}" - ); - let again = app.rpc(attach); - assert_eq!(again["ok"], true, "{again}"); - assert_eq!(page_kinds(&again), ["agent_message_chunk"], "{again}"); - assert_eq!(again["data"]["agentMode"]["session"], "own-1", "{again}"); - assert_eq!(again["data"]["agentMode"]["asking"], false, "{again}"); - *acpmux.mode.lock().unwrap() = "ask"; - acpmux.agent_records(3, "current_mode_update"); - let seen = app.events_until("agent-session-mode"); - let changed = seen.last().unwrap(); - assert_eq!(changed["mode"]["mode"], "ask", "{changed}"); - assert_eq!(changed["mode"]["asking"], true, "{changed}"); - acpmux.agent_records(4, "agent_message_chunk"); - let seen = [seen, app.events_until("agent-session-record")].concat(); - let records: Vec<&Value> = - seen.iter().filter(|event| event["event"] == "agent-session-record").collect(); - assert_eq!(records.len(), 1, "{seen:?}"); - assert_eq!(records[0]["record"]["kind"], "agent_message_chunk", "{seen:?}"); - let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); -} - -/// A tab whose record holds the chat's full name (attach accepts it) reads -/// its mode, keyed by that name as the app files it; a record acpmux only -/// resolves as a prefix of another chat is refused. -#[test] -fn agent_session_mode_reads_a_tab_that_records_the_chats_full_name() { - let daemon = Daemon::start_with("adopt-name", None, None); - listing_acpmux(daemon.dir.join("acpmux/acpmux.sock")); - let tab = |session: &str| { - let tab = daemon.rpc(json!({"id": 30, "cmd": "new-conversation-tab", - "agent_session": {"host": "install:mac-test"}})); - let surface = tab["data"]["surface"].as_u64().unwrap(); - let bound = daemon.rpc(json!({"id": 31, "cmd": "bind-conversation-tab-session", - "surface": surface, "session": session, - "expected_session": null})); - assert_eq!(bound["ok"], true, "{bound}"); - surface - }; - assert_eq!(daemon.rpc(json!({"id": 29, "cmd": "new-workspace"}))["ok"], true); - let named = tab("lane"); - let read = daemon.mode(named); - assert_eq!(read["ok"], true, "{read}"); - assert_eq!(read["data"]["session"], "lane", "{read}"); - assert_eq!(read["data"]["asking"], false, "{read}"); - let mut app = daemon.client(); - let attached = - app.rpc(json!({"cmd": "agent-session-attach", "surface": named, "mode_events": true})); - assert_eq!(attached["ok"], true, "{attached}"); - assert_eq!(attached["data"]["agentMode"]["session"], "lane", "{attached}"); - let prefix = tab("own"); - let refused = daemon.mode(prefix); - assert_eq!(refused["error_code"], "agent_session.unknown_session", "{refused}"); - let _ = fs::remove_file(daemon.dir.join("acpmux/acpmux.sock")); -} diff --git a/cmux-tui/spec/commands.md b/cmux-tui/spec/commands.md index d5d1c3362a43..53bd09feb300 100644 --- a/cmux-tui/spec/commands.md +++ b/cmux-tui/spec/commands.md @@ -9333,32 +9333,6 @@ Starts and adopts share the per-daemon and per-tab limit (`agent_session.limit`). Trust is the attach verbs': trusted local (Unix) connections only. -The same capability makes the mode readable afterwards, so the badge follows -the chat and never rests on what one app remembers: - -- `{id, cmd:"agent-session-mode", surface}` answers `{surface, session, - family, mode, asking}` for the session the tab's record names, read from - this machine's `_acpmux/web_modes` (`agent_session.unknown_tab` for a tab - with no session, `agent_session.unknown_session` when acpmux no longer knows - that exact session id or full name, as attach accepts it, never a prefix; - `agent_session.limit` past 8 reads at once). `session` is the record's - value (the id or the name), the key the app files the badge under; the - attach reply and the pushes use the same key. - Read-only; the app reads each bound tab it lists once per connect. Adopt - checks the same way that acpmux still knows the session. -- An `agent-session-attach` reply carries `agentMode: {session, family, mode, - asking}` when acpmux answers `_acpmux/web_modes`, a repeat attach on the - same connection too. -- An attach with `mode_events: true` also pushes `{event: - "agent-session-mode", surface, mode: {session, family, mode, asking}}` - after the mode changes: a mode a client set (`_acpmux/session_changed` - kind `mode` or `config`) or one the agent changed itself (a harness - `current_mode_update` or `config_option_update` record). When the attach's - `kinds` leave those records out (`transcript` does), the daemon subscribes - to them anyway and keeps them out of the client's page and stream (a page - with more before it that holds nothing else keeps them, so the client can - page on). Reads run one at a time, so the last push is the newest mode. - ### agent-session-trust `agent-session-trust-v1`: `{id, cmd:"agent-session-trust", surface, cwd, level?}` From d07af2f19224e9a0587bf34e6e09566529cd1f1c Mon Sep 17 00:00:00 2001 From: Leo Li Date: Fri, 23 Oct 2026 10:19:28 -0700 Subject: [PATCH 4/4] cmux-tui: advertise agent-session-adopt-v1 only with the mode reads An app that sees the capability sends mode_events and agent-session-mode, which this PR does not serve yet; the stacked mode PR advertises it. Co-Authored-By: Claude Opus 5.5 Agent-Session: https://cmux.com/deeplink/agent?harness=claude&session=01a16040-fd0b-7dd0-9487-1a3a198d9f5c&host=cosmic-gray --- cmux-tui/crates/cmux-tui-core/src/server/capabilities.rs | 1 - cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs | 4 +++- cmux-tui/spec/commands.md | 5 ++++- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/cmux-tui/crates/cmux-tui-core/src/server/capabilities.rs b/cmux-tui/crates/cmux-tui-core/src/server/capabilities.rs index 498950f8c5f0..73a3effbaf03 100644 --- a/cmux-tui/crates/cmux-tui-core/src/server/capabilities.rs +++ b/cmux-tui/crates/cmux-tui-core/src/server/capabilities.rs @@ -61,7 +61,6 @@ pub(super) fn identify_capabilities(mux: &Mux) -> Vec<&'static str> { #[cfg(unix)] if mux.serves_agent_session_start() { capabilities.push(AGENT_SESSION_START_CAPABILITY); - capabilities.push(AGENT_SESSION_ADOPT_CAPABILITY); capabilities.push(AGENT_SESSION_TRUST_CAPABILITY); capabilities.push(AGENT_SESSION_START_KEYED_REASON_CAPABILITY); } diff --git a/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs index 31da49db34c5..54680440e0ae 100644 --- a/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs +++ b/cmux-tui/crates/cmux-tui/tests/cli/agent_session_adopt.rs @@ -63,8 +63,10 @@ impl Daemon { fn agent_session_adopt_binds_a_chat_of_this_machine_and_names_its_bypass_mode() { let daemon = Daemon::start_with("adopt", None, None); listing_acpmux(daemon.dir.join("acpmux/acpmux.sock")); + // Not advertised until the mode reads land with it (the app sends + // `mode_events` and `agent-session-mode` to a daemon that advertises it). let identity = daemon.rpc(json!({"id": 1, "cmd": "identify"})); - assert!(identity["data"]["capabilities"].to_string().contains("agent-session-adopt-v1")); + assert!(!identity["data"]["capabilities"].to_string().contains("agent-session-adopt-v1")); let surface = daemon.own_agent_tab(); for session in ["peer-1", "chief-1", "nope-1"] { let refused = daemon.adopt(surface, session); diff --git a/cmux-tui/spec/commands.md b/cmux-tui/spec/commands.md index 53bd09feb300..53be6da58b30 100644 --- a/cmux-tui/spec/commands.md +++ b/cmux-tui/spec/commands.md @@ -9331,7 +9331,10 @@ a "bypass" badge; the daemon changes no mode. The bind records the harness acpmux lists for the session (a profile by its last part), else its family. Starts and adopts share the per-daemon and per-tab limit (`agent_session.limit`). Trust is the attach verbs': trusted local (Unix) -connections only. +connections only. The daemon does not advertise `agent-session-adopt-v1` in +`identify` yet: it does once the mode reads the app relies on with it +(`agent-session-mode`, the attach reply's `agentMode`, `mode_events`) are +served too. ### agent-session-trust