diff --git a/crates/acp-tunnel/src/config.rs b/crates/acp-tunnel/src/config.rs index 620e8d1..22c5965 100644 --- a/crates/acp-tunnel/src/config.rs +++ b/crates/acp-tunnel/src/config.rs @@ -83,6 +83,163 @@ impl RemoteConfig { } } +/// One agent endpoint in the registry (`[[agent]]` in `agents.toml`): a named +/// `/acp` connection plus a `management` policy flag. The connection fields are +/// the same shape as [`RemoteConfig`] (a legacy `remote.toml` maps to a single +/// `management = true` entry), so the dial path and validation are shared. +/// +/// Fields are spelled out rather than `#[serde(flatten)]`-ing a `RemoteConfig` +/// because `toml` serialization of a flattened struct is order-fragile; the +/// [`AgentEndpoint::conn`] accessor rebuilds the `RemoteConfig` the transport +/// dials with, keeping one source of truth for connection validation. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct AgentEndpoint { + /// Stable identity — the map key in `RemoteState` and the label in the + /// selector. Required (a nameless entry can't be addressed). + #[serde(default)] + pub name: String, + /// The `/acp` WSS endpoint. + #[serde(default)] + pub url: String, + /// The `/acp` bearer token (secret; never logged). + #[serde(default)] + pub token: String, + /// The session `cwd` sent on `session/new` (defaults to `/`). + #[serde(default = "default_cwd")] + pub cwd: String, + /// This entry backs the **management console**: Studio publishes its + /// reverse-MCP `oab` fleet-control tools to this agent (the whole point of + /// the management binding). **Off by default** — an ordinary agent console + /// chats with and configures an agent *without* granting it fleet control + /// (least privilege, ADR agent-consoles Part A). + #[serde(default)] + pub management: bool, +} + +impl Default for AgentEndpoint { + fn default() -> Self { + AgentEndpoint { + name: String::new(), + url: String::new(), + token: String::new(), + cwd: default_cwd(), + management: false, + } + } +} + +impl AgentEndpoint { + /// The connection view the transport dials with — one source of truth for + /// the `/acp` handshake fields and their validation. + pub fn conn(&self) -> RemoteConfig { + RemoteConfig { + url: self.url.clone(), + token: self.token.clone(), + cwd: self.cwd.clone(), + } + } + + /// Enough is set to attempt a connection (a name **and** a configured conn). + pub fn is_configured(&self) -> bool { + !self.name.trim().is_empty() && self.conn().is_configured() + } + + /// Validate before dialing: a name is required (it addresses the endpoint), + /// then the connection fields validate as a [`RemoteConfig`]. + pub fn validate(&self) -> Result<(), String> { + if self.name.trim().is_empty() { + return Err("agent name is required (it addresses the endpoint)".to_string()); + } + self.conn().validate() + } +} + +/// The per-agent endpoint registry (`~/.config/oab-studio/agents.toml`): a list +/// of `[[agent]]` entries. Generalizes the single `remote.toml` so Studio can +/// reach N agents — one carries `management = true` (the management console + its +/// reverse-MCP grant); all are selectable as agent consoles. A legacy +/// `remote.toml` is adopted via [`AgentRegistry::from_legacy`] as one management +/// entry, so existing setups keep working (ADR agent-consoles Part B). +#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)] +pub struct AgentRegistry { + /// The endpoints, one per `[[agent]]` table. Empty ⇒ "not configured". + #[serde(default, rename = "agent")] + pub agents: Vec, +} + +impl AgentRegistry { + /// Parse `agents.toml`. An empty file yields an empty registry (not an error), + /// so a fresh install is "not configured" rather than broken. + pub fn parse(text: &str) -> Result { + toml::from_str(text) + } + + /// Serialize back to `agents.toml` text (used when the app writes structured + /// changes; the editor otherwise round-trips the operator's raw text). + pub fn to_toml(&self) -> String { + toml::to_string_pretty(self).unwrap_or_default() + } + + /// Adopt a legacy single `remote.toml` as one `management = true` entry, so a + /// pre-registry setup keeps working while the registry is rolled out. The + /// entry is named `name` (the app passes a stable default like `"management"`). + pub fn from_legacy(cfg: RemoteConfig, name: &str) -> Self { + AgentRegistry { + agents: vec![AgentEndpoint { + name: name.to_string(), + url: cfg.url, + token: cfg.token, + cwd: cfg.cwd, + management: true, + }], + } + } + + /// Look an endpoint up by name (the `RemoteState` key). + pub fn get(&self, name: &str) -> Option<&AgentEndpoint> { + self.agents.iter().find(|a| a.name == name) + } + + /// The management endpoint (the one carrying `management = true`), if any. + /// Backs the top-level console and its reverse-MCP `oab` grant; also the + /// default target for the legacy single-endpoint commands. + pub fn management(&self) -> Option<&AgentEndpoint> { + self.agents.iter().find(|a| a.management) + } + + /// No endpoints configured. + pub fn is_empty(&self) -> bool { + self.agents.is_empty() + } + + /// Structural validation for a file save (mirrors how the fleets.toml editor + /// rejects a bad file without writing): every entry is named, names are + /// unique (they key connections), and **at most one** entry is `management` + /// (exactly one binding carries the reverse-MCP grant). Per-endpoint + /// connection completeness is checked at dial time, not here — so a + /// half-filled entry can still be saved, like an empty `remote.toml`. + pub fn validate(&self) -> Result<(), String> { + let mut seen = std::collections::HashSet::new(); + let mut management = 0usize; + for a in &self.agents { + let name = a.name.trim(); + if name.is_empty() { + return Err("every [[agent]] needs a name".to_string()); + } + if !seen.insert(name) { + return Err(format!("duplicate agent name {name:?} — names must be unique")); + } + if a.management { + management += 1; + } + } + if management > 1 { + return Err("at most one agent may be management = true".to_string()); + } + Ok(()) + } +} + #[cfg(test)] mod tests { use super::*; @@ -152,4 +309,131 @@ token = "t""#) }; assert_eq!(RemoteConfig::parse(&cfg.to_toml()).unwrap(), cfg); } + + // ---- AgentRegistry (per-agent endpoint registry) -------------------------- + + #[test] + fn registry_parses_multiple_agents() { + let reg = AgentRegistry::parse( + r#" +[[agent]] +name = "orca" +url = "wss://orca/acp" +token = "s1" +cwd = "/home/node" +management = true + +[[agent]] +name = "mira" +url = "wss://mira/acp" +token = "s2" +"#, + ) + .expect("parse"); + assert_eq!(reg.agents.len(), 2); + let orca = reg.get("orca").expect("orca present"); + assert_eq!(orca.cwd, "/home/node"); + assert!(orca.management); + assert!(orca.is_configured()); + // cwd defaults to "/" when omitted, mirroring RemoteConfig. + let mira = reg.get("mira").expect("mira present"); + assert_eq!(mira.cwd, "/"); + assert!(!mira.management); // management defaults off (least privilege) + // management() returns the one flagged entry. + assert_eq!(reg.management().map(|a| a.name.as_str()), Some("orca")); + } + + #[test] + fn empty_registry_is_not_configured() { + let reg = AgentRegistry::parse("").expect("empty parses"); + assert!(reg.is_empty()); + assert!(reg.management().is_none()); + assert_eq!(reg, AgentRegistry::default()); + assert!(reg.validate().is_ok()); + } + + #[test] + fn from_legacy_remote_becomes_one_management_entry() { + let cfg = RemoteConfig { + url: "wss://gw/acp".into(), + token: "sek".into(), + cwd: "/work".into(), + }; + let reg = AgentRegistry::from_legacy(cfg, "management"); + assert_eq!(reg.agents.len(), 1); + let e = reg.management().expect("has management"); + assert_eq!(e.name, "management"); + assert_eq!(e.cwd, "/work"); + assert!(e.management); + // The adopted entry dials with the same conn the legacy file did. + assert!(e.conn().validate().is_ok()); + } + + #[test] + fn validate_rejects_dup_names_missing_names_and_two_managements() { + assert!(AgentRegistry { + agents: vec![AgentEndpoint { name: "a".into(), ..Default::default() }, + AgentEndpoint { name: "a".into(), ..Default::default() }], + } + .validate() + .unwrap_err() + .contains("unique")); + + assert!(AgentRegistry { + agents: vec![AgentEndpoint { name: " ".into(), ..Default::default() }], + } + .validate() + .is_err()); + + assert!(AgentRegistry { + agents: vec![ + AgentEndpoint { name: "a".into(), management: true, ..Default::default() }, + AgentEndpoint { name: "b".into(), management: true, ..Default::default() }, + ], + } + .validate() + .unwrap_err() + .contains("management")); + } + + #[test] + fn endpoint_validate_requires_name_and_conn() { + // Missing name → error even with a good conn. + assert!(AgentEndpoint { + name: "".into(), + url: "wss://x/acp".into(), + token: "t".into(), + ..Default::default() + } + .validate() + .unwrap_err() + .contains("name")); + // Named but unconfigured conn → the RemoteConfig validation fires. + assert!(AgentEndpoint { name: "x".into(), ..Default::default() } + .validate() + .is_err()); + } + + #[test] + fn registry_to_toml_round_trips() { + let reg = AgentRegistry { + agents: vec![ + AgentEndpoint { + name: "orca".into(), + url: "wss://orca/acp".into(), + token: "s1".into(), + cwd: "/home/node".into(), + management: true, + }, + AgentEndpoint { + name: "mira".into(), + url: "wss://mira/acp".into(), + token: "s2".into(), + cwd: "/".into(), + management: false, + }, + ], + }; + assert_eq!(AgentRegistry::parse(®.to_toml()).unwrap(), reg); + } } diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 3db42f9..6ab2b31 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -338,7 +338,20 @@ async fn remote_view(remote: &remote::Remote) -> Result { let path = remote::config_path() .map(|p| p.display().to_string()) .unwrap_or_default(); - let status = remote.0.lock().await.status.clone(); + // Status of the management connection — the one this legacy panel controls. + // `RemoteState` is now keyed per agent, so look it up under the management + // endpoint's name (the legacy `remote.toml` adopts the name "management"). + let mgmt_name = remote::load_registry() + .ok() + .and_then(|r| r.management().map(|e| e.name.clone())) + .unwrap_or_else(|| remote::LEGACY_MANAGEMENT_NAME.to_string()); + let status = remote + .0 + .lock() + .await + .get(&mgmt_name) + .map(|st| st.status.clone()) + .unwrap_or_default(); Ok(json!({ "path": path, "text": text, @@ -354,6 +367,37 @@ async fn remote_config(remote: tauri::State<'_, remote::Remote>) -> Result) -> Result { + let reg = remote::load_registry()?; + let guard = remote.0.lock().await; + let agents: Vec = reg + .agents + .iter() + .map(|a| { + let status = guard + .get(&a.name) + .map(|st| st.status.clone()) + .filter(|s| !s.is_empty()) + .unwrap_or_else(|| "disconnected".to_string()); + json!({ + "name": a.name, + "url": a.url, + "cwd": a.cwd, + "management": a.management, + "configured": a.is_configured(), + "status": status, + }) + }) + .collect(); + Ok(json!({ "agents": agents })) +} + /// Persist the edited `remote.toml` (validates it parses before writing) and /// return the refreshed view. #[tauri::command] @@ -373,6 +417,7 @@ async fn remote_connect( app: tauri::AppHandle, core: tauri::State<'_, Core>, remote: tauri::State<'_, remote::Remote>, + agent: Option, ) -> Result<(), String> { let client = { let guard = core.0.lock().await; @@ -381,34 +426,47 @@ async fn remote_connect( .cloned() .ok_or_else(|| "core not started yet — start the core before connecting".to_string())? }; - remote::connect(app, &remote, client).await + // `None` ⇒ the management endpoint (legacy single-console behaviour); `Some` + // ⇒ a specific agent console from the registry. + let endpoint = remote::resolve_endpoint(agent.as_deref())?; + remote::connect(app, &remote, client, endpoint).await } -/// Deactivate the remote connection. +/// Deactivate a connection. `None` targets the management endpoint. #[tauri::command] async fn remote_disconnect( app: tauri::AppHandle, remote: tauri::State<'_, remote::Remote>, + agent: Option, ) -> Result<(), String> { - remote::disconnect(&app, &remote).await; + let name = remote::resolve_name(agent.as_deref())?; + remote::disconnect(&app, &remote, &name).await; Ok(()) } -/// Send a chat turn to the connected agent (ADR *agent-chat-panel*): pushes a -/// `session/prompt` onto the live `/acp` session. The reply streams back as -/// `agent-update` events. Errors if no session is active. +/// Send a chat turn to an agent (ADR *agent-chat-panel*): pushes a `session/prompt` +/// onto that agent's live `/acp` session. The reply streams back as `agent-update` +/// events tagged with the agent name. `None` targets the management endpoint. +/// Errors if that agent's session is not active. #[tauri::command] async fn agent_prompt( remote: tauri::State<'_, remote::Remote>, + agent: Option, text: String, ) -> Result<(), String> { - remote.send_prompt(text).await + let name = remote::resolve_name(agent.as_deref())?; + remote.send_prompt(&name, text).await } -/// Abandon the in-flight chat turn (`session/cancel`). Best-effort. +/// Abandon an agent's in-flight chat turn (`session/cancel`). Best-effort. +/// `None` targets the management endpoint. #[tauri::command] -async fn agent_cancel(remote: tauri::State<'_, remote::Remote>) -> Result<(), String> { - remote.send_cancel().await +async fn agent_cancel( + remote: tauri::State<'_, remote::Remote>, + agent: Option, +) -> Result<(), String> { + let name = remote::resolve_name(agent.as_deref())?; + remote.send_cancel(&name).await } /// What the frontend needs to render the "update available" state: the version @@ -490,6 +548,7 @@ pub fn run() { fleet_config_write, deploy_scale, remote_config, + remote_agents, remote_config_write, remote_connect, remote_disconnect, @@ -507,7 +566,7 @@ pub fn run() { // Disconnect button already does this; this covers Cmd-Q / window close. if let tauri::RunEvent::ExitRequested { .. } = event { let remote = app_handle.state::(); - tauri::async_runtime::block_on(remote::disconnect(app_handle, &remote)); + tauri::async_runtime::block_on(remote::disconnect_all(app_handle, &remote)); } }); } diff --git a/src-tauri/src/remote.rs b/src-tauri/src/remote.rs index 9db6868..e5b081c 100644 --- a/src-tauri/src/remote.rs +++ b/src-tauri/src/remote.rs @@ -10,13 +10,14 @@ //! are validated against the live gateway once the §5 endpoint exists; this //! module is structurally complete and compiles under `desktop.yml`. +use std::collections::HashMap; use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant}; use acp_tunnel as acp; -use acp_tunnel::config::RemoteConfig; +use acp_tunnel::config::{AgentEndpoint, AgentRegistry, RemoteConfig}; use acp_tunnel::{DisconnectReason, Inbound, Session}; use futures_util::{Sink, SinkExt, StreamExt}; use serde_json::{json, Value}; @@ -42,10 +43,14 @@ const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(20); /// katashiro's `ACP_PROMPT_TIMEOUT_MS`. const PROMPT_TIMEOUT: Duration = Duration::from_secs(600); -/// Managed state: the running connection task (abort to disconnect) plus the last -/// status string the UI renders. +/// Managed state: a map of **per-agent** connections keyed by the endpoint name +/// (ADR agent-consoles Part B — `RemoteState` is no longer a singleton). Each +/// agent console dials its own endpoint; the management console is just the entry +/// keyed by the `management` endpoint's name. The map grows on connect and each +/// entry carries its own task/status/reconnect, so opening one console never +/// disturbs another. #[derive(Default)] -pub struct Remote(pub AsyncMutex); +pub struct Remote(pub AsyncMutex>); #[derive(Default)] pub struct RemoteState { @@ -76,21 +81,22 @@ pub enum OutMsg { } impl Remote { - /// Send a chat turn to the connected agent. Errors if no session is live. - pub async fn send_prompt(&self, text: String) -> Result<(), String> { - self.push(OutMsg::Prompt(text)).await + /// Send a chat turn to the named agent. Errors if that agent's session is not + /// live (each agent console has its own connection, so the target is explicit). + pub async fn send_prompt(&self, agent: &str, text: String) -> Result<(), String> { + self.push(agent, OutMsg::Prompt(text)).await } - /// Cancel the in-flight turn (best-effort). - pub async fn send_cancel(&self) -> Result<(), String> { - self.push(OutMsg::Cancel).await + /// Cancel the named agent's in-flight turn (best-effort). + pub async fn send_cancel(&self, agent: &str) -> Result<(), String> { + self.push(agent, OutMsg::Cancel).await } - async fn push(&self, msg: OutMsg) -> Result<(), String> { + async fn push(&self, agent: &str, msg: OutMsg) -> Result<(), String> { let guard = self.0.lock().await; let tx = guard - .prompt_tx - .as_ref() + .get(agent) + .and_then(|st| st.prompt_tx.as_ref()) .ok_or_else(|| "not connected — activate the remote connection first".to_string())?; // A live `prompt_tx` whose receiver has gone means the connection is // tearing down (the socket closed and `run_reconnecting` is about to @@ -103,13 +109,81 @@ impl Remote { } } -/// `~/.config/oab-studio/remote.toml` — beside `fleets.toml`. +/// Legacy single-endpoint config `~/.config/oab-studio/remote.toml` — beside +/// `fleets.toml`. Still the file the current management-console editor writes; the +/// registry adopts it as one `management = true` entry when `agents.toml` is +/// absent (ADR agent-consoles Part B back-compat). pub fn config_path() -> Result { dirs::config_dir() .map(|d| d.join("oab-studio").join("remote.toml")) .ok_or_else(|| "no config directory resolved".to_string()) } +/// The per-agent endpoint registry `~/.config/oab-studio/agents.toml`. When +/// present it is the source of truth; the legacy `remote.toml` is the fallback. +pub fn registry_path() -> Result { + dirs::config_dir() + .map(|d| d.join("oab-studio").join("agents.toml")) + .ok_or_else(|| "no config directory resolved".to_string()) +} + +/// The stable name given to a legacy `remote.toml` when it is adopted into the +/// registry as the single management entry. +pub const LEGACY_MANAGEMENT_NAME: &str = "management"; + +/// Load the endpoint registry: prefer `agents.toml`; if it is missing or empty, +/// adopt the legacy `remote.toml` as one `management = true` entry so existing +/// single-endpoint setups keep working untouched. +pub fn load_registry() -> Result { + let rp = registry_path()?; + match std::fs::read_to_string(&rp) { + Ok(s) if !s.trim().is_empty() => { + return AgentRegistry::parse(&s).map_err(|e| format!("invalid agents.toml: {e}")) + } + Ok(_) => {} // present but empty → fall through to legacy + Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} + Err(e) => return Err(format!("read {}: {e}", rp.display())), + } + let legacy = load_config()?; + if legacy.is_configured() { + Ok(AgentRegistry::from_legacy(legacy, LEGACY_MANAGEMENT_NAME)) + } else { + // Nothing configured anywhere → empty registry (the app shows + // "not configured" rather than erroring). + Ok(AgentRegistry::default()) + } +} + +/// Resolve a command's optional `agent` argument to just the connection **key** +/// (the map name). `Some` is already the key; `None` resolves to the management +/// endpoint's name. Cheaper than [`resolve_endpoint`] when only the key is needed +/// (disconnect / prompt / cancel target an already-open connection). +pub fn resolve_name(agent: Option<&str>) -> Result { + match agent { + Some(n) => Ok(n.to_string()), + None => load_registry()? + .management() + .map(|e| e.name.clone()) + .ok_or_else(|| "no management agent configured".to_string()), + } +} + +/// Resolve a command's optional `agent` argument to a concrete endpoint. `None` +/// means the legacy single-endpoint commands — resolve to the management entry. +pub fn resolve_endpoint(agent: Option<&str>) -> Result { + let reg = load_registry()?; + match agent { + Some(name) => reg + .get(name) + .cloned() + .ok_or_else(|| format!("no agent named {name:?} in the registry")), + None => reg + .management() + .cloned() + .ok_or_else(|| "no management agent configured".to_string()), + } +} + /// Raw file text for the editor; a missing file is empty ("not configured"). pub fn read_config_text() -> Result { let p = config_path()?; @@ -135,34 +209,46 @@ fn load_config() -> Result { RemoteConfig::parse(&read_config_text()?).map_err(|e| format!("invalid remote.toml: {e}")) } -fn emit_status(app: &AppHandle, status: &str) { - let _ = app.emit("remote-status", json!({ "status": status })); +/// Emit a connection-status change for a specific agent. The `agent` field lets +/// the UI route the update to the matching console (the management console keys +/// off the management endpoint's name); a single-agent UI can ignore it. +fn emit_status(app: &AppHandle, agent: &str, status: &str) { + let _ = app.emit("remote-status", json!({ "agent": agent, "status": status })); } -/// Activate the remote connection: validate config, spawn the connection task. -/// Idempotent — a no-op if already connected. +/// Activate a named agent's connection: validate the endpoint, spawn its +/// connection task keyed under the endpoint name. Idempotent — a no-op if that +/// agent is already connecting/connected. Reverse-MCP `oab` fleet-control tools +/// are published **only** when the endpoint is `management` (least privilege): an +/// ordinary agent console dials, chats, and (later) edits files without granting +/// the agent fleet control. pub async fn connect( app: AppHandle, remote: &Remote, client: McpClient, + endpoint: AgentEndpoint, ) -> Result<(), String> { - let cfg = load_config()?; - cfg.validate()?; + endpoint.validate()?; + let agent = endpoint.name.clone(); + let cfg = endpoint.conn(); + let management = endpoint.management; let mut guard = remote.0.lock().await; - if guard.task.is_some() { + let st = guard.entry(agent.clone()).or_default(); + if st.task.is_some() { return Ok(()); } let app_task = app.clone(); + let agent_task = agent.clone(); let stop = Arc::new(AtomicBool::new(false)); let stop_task = stop.clone(); let task = tauri::async_runtime::spawn(async move { - run_reconnecting(app_task, cfg, client, stop_task).await; + run_reconnecting(app_task, agent_task, cfg, management, client, stop_task).await; }); - guard.task = Some(task); - guard.stop = Some(stop); - guard.status = "connecting".to_string(); - emit_status(&app, "connecting"); + st.task = Some(task); + st.stop = Some(stop); + st.status = "connecting".to_string(); + emit_status(&app, &agent, "connecting"); Ok(()) } @@ -171,15 +257,18 @@ pub async fn connect( /// frame) so the gateway releases the session slot immediately rather than holding /// it for a resume that will never come (until its TTL / liveness reaper fires). /// Safe to call on app teardown and from the Disconnect button; a no-op if idle. -pub async fn disconnect(app: &AppHandle, remote: &Remote) { +pub async fn disconnect(app: &AppHandle, remote: &Remote, agent: &str) { // Flag the reconnect loop to stop, and grab the pieces we need to tear down // outside the lock (so the loop can take the lock to retract `prompt_tx`). let (task, tx) = { let mut guard = remote.0.lock().await; - if let Some(stop) = guard.stop.take() { + let Some(st) = guard.get_mut(agent) else { + return; // never connected — nothing to tear down + }; + if let Some(stop) = st.stop.take() { stop.store(true, Ordering::SeqCst); } - (guard.task.take(), guard.prompt_tx.take()) + (st.task.take(), st.prompt_tx.take()) }; // Ask the running connection to flush a graceful close. if let Some(tx) = tx { @@ -192,16 +281,38 @@ pub async fn disconnect(app: &AppHandle, remote: &Remote) { tokio::time::sleep(Duration::from_millis(400)).await; t.abort(); } - let mut guard = remote.0.lock().await; - guard.status = "disconnected".to_string(); - guard.prompt_tx = None; - emit_status(app, "disconnected"); + if let Some(st) = remote.0.lock().await.get_mut(agent) { + st.status = "disconnected".to_string(); + st.prompt_tx = None; + } + emit_status(app, agent, "disconnected"); let _ = app.emit( "app-log", - json!({ "level": "info", "msg": "remote: disconnected by user" }), + json!({ "level": "info", "msg": format!("remote: {agent} disconnected by user") }), ); } +/// Disconnect **every** live agent connection (app teardown): close each socket +/// cleanly so the gateway frees all held session slots at once. +pub async fn disconnect_all(app: &AppHandle, remote: &Remote) { + let agents: Vec = remote.0.lock().await.keys().cloned().collect(); + for agent in agents { + disconnect(app, remote, &agent).await; + } +} + +/// The reverse-MCP `oab` server declaration for a connection, **only** when this +/// endpoint is the management binding. A non-management agent console declares no +/// servers, so the gateway never tunnels Studio's fleet-control tools to it +/// (least privilege, ADR agent-consoles Part A). +fn servers_for(management: bool, conn_id: &str) -> Vec { + if management { + vec![acp::oab_server(conn_id)] + } else { + vec![] + } +} + /// Reconnect loop: one attempt, then back off and retry until the task is /// aborted (by [`disconnect`]). /// @@ -213,11 +324,16 @@ pub async fn disconnect(app: &AppHandle, remote: &Remote) { /// a full `initialize` handshake. async fn run_reconnecting( app: AppHandle, + agent: String, cfg: RemoteConfig, + management: bool, client: McpClient, stop: Arc, ) { - let mut session = Session::new(vec![acp::oab_server(&uuid::Uuid::new_v4().to_string())]); + let mut session = Session::new(servers_for( + management, + &uuid::Uuid::new_v4().to_string(), + )); // Consecutive-failure counter driving the reconnect backoff. Reset to 0 once an // attempt has held a live connection for a while (see below), so a long-running // session that blips reconnects promptly instead of at the capped delay. @@ -228,20 +344,24 @@ async fn run_reconnecting( } let conn_id = uuid::Uuid::new_v4().to_string(); // Fresh per-connection server id + phase reset; keeps `session_id` so - // `run_once` picks the resume path when one exists. - session.redeclare(vec![acp::oab_server(&conn_id)]); + // `run_once` picks the resume path when one exists. `servers_for` gates the + // `oab` declaration on `management`, so a non-management console re-attaches + // without ever republishing fleet-control tools. + session.redeclare(servers_for(management, &conn_id)); if session.session_id().is_some() { let _ = app.emit( "app-log", - json!({ "level": "info", "msg": "remote: reconnecting — will resume the existing session" }), + json!({ "level": "info", "msg": format!("remote: {agent} reconnecting — will resume the existing session") }), ); } let started = Instant::now(); - let result = run_once(&app, &cfg, &client, &mut session, &conn_id).await; - // The socket is gone — retract the outbound-chat channel so a prompt - // between attempts fails fast rather than dropping into a dead sink. - app.state::().0.lock().await.prompt_tx = None; + let result = run_once(&app, &agent, &cfg, management, &client, &mut session, &conn_id).await; + // The socket is gone — retract this agent's outbound-chat channel so a + // prompt between attempts fails fast rather than dropping into a dead sink. + if let Some(st) = app.state::().0.lock().await.get_mut(&agent) { + st.prompt_tx = None; + } // User-initiated disconnect: stop here instead of reconnecting (and skip // the misleading "reconnecting…" log). if stop.load(Ordering::SeqCst) { @@ -257,10 +377,10 @@ async fn run_reconnecting( // Classify so the status line says *why* (network / auth rejected / // server at capacity / protocol) instead of a raw error blob. let reason = DisconnectReason::classify(&e); - emit_status(&app, &format!("error: {}", reason.label())); + emit_status(&app, &agent, &format!("error: {}", reason.label())); let _ = app.emit( "app-log", - json!({ "level": "error", "msg": format!("remote: {} — {e}", reason.label()) }), + json!({ "level": "error", "msg": format!("remote: {agent} {} — {e}", reason.label()) }), ); } // Exponential backoff (capped 30s) + per-connection jitter, so a flapping @@ -269,10 +389,10 @@ async fn run_reconnecting( let salt = conn_id.as_bytes().first().copied().unwrap_or(0); let delay = acp::backoff_delay(attempt, salt); attempt = attempt.saturating_add(1); - emit_status(&app, "connecting"); + emit_status(&app, &agent, "connecting"); let _ = app.emit( "app-log", - json!({ "level": "info", "msg": format!("remote: reconnecting in {}s…", delay.as_secs()) }), + json!({ "level": "info", "msg": format!("remote: {agent} reconnecting in {}s…", delay.as_secs()) }), ); tokio::time::sleep(delay).await; } @@ -284,7 +404,9 @@ async fn run_reconnecting( /// is dead and returns `Err` so [`run_reconnecting`] reconnects. async fn run_once( app: &AppHandle, + agent: &str, cfg: &RemoteConfig, + management: bool, client: &McpClient, session: &mut Session, conn_id: &str, @@ -318,7 +440,7 @@ async fn run_once( // otherwise invisible in Activity until it succeeds or errors. let _ = app.emit( "app-log", - json!({ "level": "info", "msg": format!("remote: dialing {}…", cfg.url) }), + json!({ "level": "info", "msg": format!("remote: {agent} dialing {}…", cfg.url) }), ); let (ws, _resp) = tokio_tungstenite::connect_async(req) .await @@ -578,7 +700,7 @@ async fn run_once( let _ = app.emit( "app-log", json!({ "level": "warn", "msg": format!( - "remote: session/resume rejected ({emsg}) — opening a fresh session" + "remote: {agent} session/resume rejected ({emsg}) — opening a fresh session" ) }), ); session.forget_session(); @@ -600,15 +722,30 @@ async fn run_once( .unwrap_or_default() .to_string(); session.on_session_created(sid); - // Now a turn can be sent — publish the outbound channel. - app.state::().0.lock().await.prompt_tx = Some(out_tx.clone()); - emit_status(app, "connected"); - let msg = if resume_attempted { - "remote: session resumed — oab tools republished" + // Now a turn can be sent — publish this agent's outbound channel. + if let Some(st) = + app.state::().0.lock().await.get_mut(agent) + { + st.prompt_tx = Some(out_tx.clone()); + st.status = "connected".to_string(); + } + emit_status(app, agent, "connected"); + // Only the management binding publishes `oab` tools; an + // agent console runs chat-only (least privilege). + let tools = if management { + if resume_attempted { + " — oab tools republished" + } else { + " — oab tools published" + } } else { - "remote: session active — oab tools published" + "" }; - let _ = app.emit("app-log", json!({ "level": "info", "msg": msg })); + let verb = if resume_attempted { "resumed" } else { "active" }; + let _ = app.emit( + "app-log", + json!({ "level": "info", "msg": format!("remote: {agent} session {verb}{tools}") }), + ); } } acp::Phase::SessionActive => { @@ -625,7 +762,7 @@ async fn run_once( .to_string(); let _ = app.emit( "agent-update", - json!({ "kind": "turn_end", "stopReason": stop }), + json!({ "agent": agent, "kind": "turn_end", "stopReason": stop }), ); } // Any other method-less frame (e.g. a heartbeat probe's @@ -669,7 +806,7 @@ async fn run_once( } // A piece of the agent's chat reply → forward to the panel. Inbound::AgentChunk { text } => { - let _ = app.emit("agent-update", json!({ "kind": "chunk", "text": text })); + let _ = app.emit("agent-update", json!({ "agent": agent, "kind": "chunk", "text": text })); } Inbound::Cancel { .. } | Inbound::Other => {} } @@ -685,7 +822,7 @@ async fn run_once( if pending_prompt.is_some() { let _ = app.emit( "app-log", - json!({ "level": "warn", "msg": "remote: connection dropped with a turn in flight — turn abandoned" }), + json!({ "level": "warn", "msg": format!("remote: {agent} connection dropped with a turn in flight — turn abandoned") }), ); } outcome