Make roaming an opt-in goose-cli feature (default-off, stays in-tree) (#11516)
Signed-off-by: Michael Neale <michael.neale@gmail.com> Co-authored-by: Michael Neale <14976+michaelneale@users.noreply.github.com> Co-authored-by: pstayets <philip@vulturelabs.com> Co-authored-by: pstayet <philipstayetski2416@gmail.com> Co-authored-by: Colin Gauvin <colin@gauvin.id>
This commit is contained in:
Michael Neale
@@ -18,6 +18,8 @@ use crate::commands::configure::handle_configure;
|
||||
use crate::commands::info::handle_info;
|
||||
use crate::commands::plugin::{handle_plugin_install, handle_plugin_update};
|
||||
use crate::commands::recipe::{handle_deeplink, handle_list, handle_open, handle_validate};
|
||||
#[cfg(feature = "roaming")]
|
||||
use crate::commands::roam::{handle_roam_command, RoamCommand};
|
||||
use crate::commands::term::{
|
||||
handle_term_info, handle_term_init, handle_term_log, handle_term_run, Shell,
|
||||
};
|
||||
@@ -840,6 +842,14 @@ enum Command {
|
||||
enable_scheduler: bool,
|
||||
},
|
||||
|
||||
/// Share or connect to agents peer-to-peer over iroh
|
||||
#[cfg(feature = "roaming")]
|
||||
#[command(about = "Share or connect to agents peer-to-peer (roaming)")]
|
||||
Roam {
|
||||
#[command(subcommand)]
|
||||
command: RoamCommand,
|
||||
},
|
||||
|
||||
/// Start ACP server over HTTP and WebSocket
|
||||
#[command(about = "Start ACP server over HTTP and WebSocket")]
|
||||
Serve {
|
||||
@@ -887,6 +897,14 @@ enum Command {
|
||||
|
||||
#[arg(long, help = "Enable scheduled recipe execution")]
|
||||
enable_scheduler: bool,
|
||||
|
||||
/// Also expose this server over goose roam (p2p) so paired devices can connect remotely
|
||||
#[cfg(feature = "roaming")]
|
||||
#[arg(
|
||||
long,
|
||||
help = "Also expose this server over goose roam (p2p) so paired devices can connect remotely"
|
||||
)]
|
||||
roam: bool,
|
||||
},
|
||||
|
||||
/// Start or resume interactive chat sessions
|
||||
@@ -1356,6 +1374,8 @@ fn get_command_name(command: &Option<Command>) -> &'static str {
|
||||
Some(Command::Info { .. }) => "info",
|
||||
Some(Command::Mcp { .. }) => "mcp",
|
||||
Some(Command::Acp { .. }) => "acp",
|
||||
#[cfg(feature = "roaming")]
|
||||
Some(Command::Roam { .. }) => "roam",
|
||||
Some(Command::Serve { .. }) => "serve",
|
||||
Some(Command::Session { .. }) => "session",
|
||||
Some(Command::Run { .. }) => "run",
|
||||
@@ -1606,6 +1626,127 @@ struct ServeCommandArgs {
|
||||
dangerously_unauthenticated: bool,
|
||||
allowed_origins: Vec<String>,
|
||||
enable_scheduler: bool,
|
||||
#[cfg(feature = "roaming")]
|
||||
roam: bool,
|
||||
}
|
||||
|
||||
#[cfg(feature = "roaming")]
|
||||
type RoamShareSlot =
|
||||
std::sync::Arc<tokio::sync::RwLock<Option<std::sync::Arc<goose_roaming::RoamingNode>>>>;
|
||||
|
||||
#[cfg(feature = "roaming")]
|
||||
fn spawn_roam_share(
|
||||
server: std::sync::Arc<goose::acp::server_factory::AcpServer>,
|
||||
) -> RoamShareSlot {
|
||||
use crate::commands::roam::try_acquire_roam_lock_owner;
|
||||
|
||||
let slot = RoamShareSlot::default();
|
||||
let task_slot = slot.clone();
|
||||
tokio::spawn(async move {
|
||||
let mut standing_by = false;
|
||||
loop {
|
||||
match try_acquire_roam_lock_owner() {
|
||||
Ok(Some(lock)) => match start_roam_share(server.clone()).await {
|
||||
Ok(node) => {
|
||||
*task_slot.write().await = Some(node);
|
||||
let _lock = lock;
|
||||
std::future::pending::<()>().await;
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::error!("roam share failed to start: {error}");
|
||||
drop(lock);
|
||||
}
|
||||
},
|
||||
Ok(None) => {
|
||||
if !standing_by {
|
||||
standing_by = true;
|
||||
eprintln!(
|
||||
"another goose process owns the roaming endpoint; standing by to take over if it exits"
|
||||
);
|
||||
}
|
||||
}
|
||||
// A real failure (unwritable data dir, filesystem error) will
|
||||
// not fix itself: surface it and stop instead of retrying as
|
||||
// if the endpoint were merely busy.
|
||||
Err(error) => {
|
||||
eprintln!("roaming disabled: cannot acquire the endpoint lock: {error:#}");
|
||||
return;
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(std::time::Duration::from_secs(3)).await;
|
||||
}
|
||||
});
|
||||
slot
|
||||
}
|
||||
|
||||
#[cfg(feature = "roaming")]
|
||||
async fn start_roam_share(
|
||||
server: std::sync::Arc<goose::acp::server_factory::AcpServer>,
|
||||
) -> Result<std::sync::Arc<goose_roaming::RoamingNode>> {
|
||||
use crate::commands::roam::{
|
||||
directory_path, load_identity, resolve_relay_settings, trust_path,
|
||||
};
|
||||
use crate::commands::roam_full_bridge::FullAcpBridge;
|
||||
use goose::config::paths::Paths;
|
||||
use goose_roaming::{RoamingConfig, RoamingNode, TrustBook};
|
||||
use std::sync::Arc;
|
||||
|
||||
let status_path = Paths::data_dir().join("roam/serve.json");
|
||||
let _ = std::fs::remove_file(&status_path);
|
||||
|
||||
let identity = load_identity()?;
|
||||
let node = RoamingNode::bind(RoamingConfig {
|
||||
identity,
|
||||
relay: resolve_relay_settings()?,
|
||||
trust: TrustBook::new(),
|
||||
trust_path: Some(trust_path()),
|
||||
directory: goose_roaming::Directory::persistent_owned(directory_path()),
|
||||
bind_addr: None,
|
||||
relay_tls: None,
|
||||
})
|
||||
.await?;
|
||||
|
||||
let agent_id = node.endpoint_id().to_string();
|
||||
// Roaming sessions run where `goose serve` was started: the connector's
|
||||
// machine-local path is meaningless on this host, and the serve-wide
|
||||
// server keeps `session_cwd: None` for local ACP clients.
|
||||
let session_cwd =
|
||||
std::env::current_dir().map_err(|e| anyhow::anyhow!("could not determine cwd: {e}"))?;
|
||||
node.share(Arc::new(FullAcpBridge::new(server, agent_id, session_cwd)))
|
||||
.await?;
|
||||
|
||||
if !node.wait_online(std::time::Duration::from_secs(15)).await {
|
||||
tracing::warn!(
|
||||
"roaming endpoint did not come online; the card may lack a reachable address"
|
||||
);
|
||||
}
|
||||
|
||||
let card = node.card();
|
||||
let card_encoded = card.encode()?;
|
||||
let started_at = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_millis() as u64)
|
||||
.unwrap_or(0);
|
||||
let status = serde_json::json!({
|
||||
"card": card_encoded,
|
||||
"endpointId": card.endpoint_id.to_string(),
|
||||
"fingerprint": card.fingerprint(),
|
||||
"startedAt": started_at,
|
||||
});
|
||||
if let Some(dir) = status_path.parent() {
|
||||
std::fs::create_dir_all(dir)?;
|
||||
}
|
||||
let tmp_path = status_path.with_extension("json.tmp");
|
||||
std::fs::write(&tmp_path, serde_json::to_vec_pretty(&status)?)?;
|
||||
std::fs::rename(&tmp_path, &status_path)?;
|
||||
|
||||
eprintln!("roam is enabled for this server");
|
||||
eprintln!(" endpoint id : {}", card.endpoint_id);
|
||||
eprintln!(" fingerprint : {}", card.fingerprint());
|
||||
eprintln!("your connection card (share with a peer so it can reach you):");
|
||||
eprintln!("{card_encoded}");
|
||||
|
||||
Ok(node)
|
||||
}
|
||||
|
||||
async fn handle_serve_command(args: ServeCommandArgs) -> Result<()> {
|
||||
@@ -1629,6 +1770,8 @@ async fn handle_serve_command(args: ServeCommandArgs) -> Result<()> {
|
||||
dangerously_unauthenticated,
|
||||
allowed_origins,
|
||||
enable_scheduler,
|
||||
#[cfg(feature = "roaming")]
|
||||
roam,
|
||||
} = args;
|
||||
|
||||
let builtins = AcpBuiltinSelection::from_requested(builtins);
|
||||
@@ -1651,6 +1794,7 @@ async fn handle_serve_command(args: ServeCommandArgs) -> Result<()> {
|
||||
config_dir: Paths::config_dir(),
|
||||
goose_platform: platform.into(),
|
||||
additional_source_roots,
|
||||
session_cwd: None,
|
||||
enable_scheduler,
|
||||
}));
|
||||
let env_secret = std::env::var(GOOSE_SERVER_SECRET_KEY_ENV)
|
||||
@@ -1684,6 +1828,12 @@ async fn handle_serve_command(args: ServeCommandArgs) -> Result<()> {
|
||||
if let Err(error) = server.start_scheduler().await {
|
||||
warn!("Scheduler failed to start; scheduled jobs will not run until a client connects: {error}");
|
||||
}
|
||||
#[cfg(feature = "roaming")]
|
||||
let roam_share = if roam {
|
||||
Some(spawn_roam_share(server.clone()))
|
||||
} else {
|
||||
None
|
||||
};
|
||||
let router = create_router(
|
||||
server,
|
||||
secret_key,
|
||||
@@ -1741,6 +1891,13 @@ async fn handle_serve_command(args: ServeCommandArgs) -> Result<()> {
|
||||
.await?;
|
||||
}
|
||||
|
||||
#[cfg(feature = "roaming")]
|
||||
if let Some(slot) = roam_share {
|
||||
if let Some(node) = slot.write().await.take() {
|
||||
let _ = node.shutdown().await;
|
||||
}
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -2652,6 +2809,8 @@ pub async fn cli() -> anyhow::Result<()> {
|
||||
builtins,
|
||||
enable_scheduler,
|
||||
}) => goose::acp::server::run(builtins, enable_scheduler).await,
|
||||
#[cfg(feature = "roaming")]
|
||||
Some(Command::Roam { command }) => handle_roam_command(command).await,
|
||||
Some(Command::Serve {
|
||||
host,
|
||||
port,
|
||||
@@ -2663,6 +2822,8 @@ pub async fn cli() -> anyhow::Result<()> {
|
||||
dangerously_unauthenticated,
|
||||
allowed_origins,
|
||||
enable_scheduler,
|
||||
#[cfg(feature = "roaming")]
|
||||
roam,
|
||||
}) => {
|
||||
handle_serve_command(ServeCommandArgs {
|
||||
host,
|
||||
@@ -2675,6 +2836,8 @@ pub async fn cli() -> anyhow::Result<()> {
|
||||
dangerously_unauthenticated,
|
||||
allowed_origins,
|
||||
enable_scheduler,
|
||||
#[cfg(feature = "roaming")]
|
||||
roam,
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
@@ -5,6 +5,14 @@ pub mod info;
|
||||
pub mod plugin;
|
||||
pub mod recipe;
|
||||
pub mod review;
|
||||
#[cfg(feature = "roaming")]
|
||||
pub mod roam;
|
||||
#[cfg(feature = "roaming")]
|
||||
pub mod roam_client;
|
||||
#[cfg(feature = "roaming")]
|
||||
pub mod roam_full_bridge;
|
||||
#[cfg(feature = "roaming")]
|
||||
pub mod roam_proxy;
|
||||
pub mod schedule;
|
||||
pub mod session;
|
||||
pub mod skills;
|
||||
|
||||
@@ -0,0 +1,972 @@
|
||||
//! `goose roam` — peer-to-peer agent access over iroh.
|
||||
//!
|
||||
//! The model is deliberately infrastructural: roaming is just an authenticated
|
||||
//! p2p ACP transport. Each node has one identity and produces a **connection
|
||||
//! card** (`roam id`) — a non-secret string carrying its public key and how to
|
||||
//! reach it. You swap cards with another node and each side chooses to **accept**
|
||||
//! the other's key. A connection only succeeds when the host has accepted the
|
||||
//! dialer's key; there is no bearer token that grants access by possession. An
|
||||
//! accepted peer gets goose's full ACP surface.
|
||||
//!
|
||||
//! Subcommands:
|
||||
//! * `id` — print this node's connection card (share it with a peer).
|
||||
//! * `share` — serve this node's agent to accepted peers over ACP.
|
||||
//! * `peers` — manage saved peer cards and which keys you accept.
|
||||
//! * `connect` / `delegate` / `bridge` — reach a peer that has accepted you.
|
||||
//! * `connections` (alias `list`) — show live/observed connections.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use clap::Subcommand;
|
||||
use goose::acp::server::AcpBuiltinSelection;
|
||||
use goose::acp::server_factory::{AcpServer, AcpServerFactoryConfig};
|
||||
use goose::agents::GoosePlatform;
|
||||
use goose::config::paths::Paths;
|
||||
use goose::config::{Config, ConfigError};
|
||||
use goose_roaming::{
|
||||
default_key_path, parse_endpoint_id, ConnectionCard, Directory, EndpointId, RelayEntry,
|
||||
RelaySettings, RoamingConfig, RoamingIdentity, RoamingNode, TrustBook,
|
||||
};
|
||||
|
||||
use crate::commands::roam_full_bridge::FullAcpBridge;
|
||||
|
||||
const CARD_SCHEME: &str = "goose+roam://";
|
||||
|
||||
pub(crate) fn directory_path() -> std::path::PathBuf {
|
||||
Paths::state_dir().join("roaming_directory.json")
|
||||
}
|
||||
|
||||
pub(crate) fn trust_path() -> std::path::PathBuf {
|
||||
Paths::config_dir().join("roaming_trust.json")
|
||||
}
|
||||
|
||||
fn peerbook_path() -> std::path::PathBuf {
|
||||
Paths::config_dir().join("roaming_peers.json")
|
||||
}
|
||||
|
||||
/// Config key (or `GOOSE_ROAM_RELAYS` env) overriding the relay URLs the
|
||||
/// roaming endpoint uses. When unset, roaming uses the managed relays below —
|
||||
/// never iroh's shared public n0 relays.
|
||||
const CONFIG_ROAM_RELAYS_KEY: &str = "GOOSE_ROAM_RELAYS";
|
||||
/// Optional shared bearer token (secret) presented to every *explicitly
|
||||
/// configured* relay (for gated / `AccessConfig::Restricted` relays). Not
|
||||
/// applied to the default managed relays, which register open.
|
||||
const CONFIG_ROAM_RELAY_TOKEN_KEY: &str = "GOOSE_ROAM_RELAY_TOKEN";
|
||||
|
||||
/// Default managed iroh relays — the same four dedicated relays mesh-llm uses
|
||||
/// (provisioned via services.iroh.computer on the `iroh.link` domain), one per
|
||||
/// region for global coverage. They register open (`AccessConfig::Everyone`, no
|
||||
/// auth token), so a browser client can reach them over `wss://` with no auth
|
||||
/// header. We default to these rather than iroh's public n0 relays so roaming
|
||||
/// never depends on the shared, rate-limited, no-SLA public relays.
|
||||
///
|
||||
/// Override with `GOOSE_ROAM_RELAYS` to point at other relays (e.g. a Block-run
|
||||
/// deployment).
|
||||
const DEFAULT_ROAM_RELAYS: &[&str] = &[
|
||||
"https://usw1-2.relay.michaelneale.mesh-llm.iroh.link./", // US West
|
||||
"https://use1-1.relay.michaelneale.mesh-llm.iroh.link./", // US East
|
||||
"https://euc1-1.relay.michaelneale.mesh-llm.iroh.link./", // EU Central
|
||||
"https://aps1-1.relay.michaelneale.mesh-llm.iroh.link./", // Asia-Pacific South
|
||||
];
|
||||
|
||||
/// Resolve the relay settings for a roaming endpoint.
|
||||
///
|
||||
/// Uses `GOOSE_ROAM_RELAYS` (env or config file) when set — optionally
|
||||
/// authenticated with the `GOOSE_ROAM_RELAY_TOKEN` secret applied to each —
|
||||
/// otherwise the default managed relays. Never iroh's public n0 relays.
|
||||
///
|
||||
/// Fails when `GOOSE_ROAM_RELAYS` is set but unreadable: a deployment that
|
||||
/// configured private relays must not silently fall back to the managed ones.
|
||||
pub(crate) fn resolve_relay_settings() -> Result<RelaySettings> {
|
||||
let config = Config::global();
|
||||
let urls: Vec<String> = match config.get_param::<Vec<String>>(CONFIG_ROAM_RELAYS_KEY) {
|
||||
Ok(urls) => urls,
|
||||
Err(ConfigError::NotFound(_)) => Vec::new(),
|
||||
Err(error) => {
|
||||
return Err(anyhow::anyhow!(
|
||||
"{CONFIG_ROAM_RELAYS_KEY} is set but could not be read as a list of relay \
|
||||
URLs; refusing to fall back to the default relays: {error}"
|
||||
))
|
||||
}
|
||||
};
|
||||
// Only explicit relay URLs use a token; skip the secret-store read (and
|
||||
// any keyring stall it may involve) entirely on the default-relay path.
|
||||
if urls.iter().all(|u| u.trim().is_empty()) {
|
||||
return Ok(build_relay_settings(urls, None));
|
||||
}
|
||||
let token = match config.get_secret::<String>(CONFIG_ROAM_RELAY_TOKEN_KEY) {
|
||||
Ok(token) => Some(token),
|
||||
Err(ConfigError::NotFound(_)) => None,
|
||||
// A configured token that cannot be read (keyring timeout, corrupt
|
||||
// store) must not silently become unauthenticated relay dials.
|
||||
Err(error) => {
|
||||
return Err(anyhow::anyhow!(
|
||||
"{CONFIG_ROAM_RELAY_TOKEN_KEY} could not be read; refusing to contact \
|
||||
relays without the configured token: {error}"
|
||||
))
|
||||
}
|
||||
};
|
||||
Ok(build_relay_settings(urls, token))
|
||||
}
|
||||
|
||||
/// Pure relay-settings builder. With no configured URLs, use the default
|
||||
/// managed relays (open, no token). With configured URLs (empty entries
|
||||
/// filtered out), use those, applying a non-empty token to each.
|
||||
fn build_relay_settings(urls: Vec<String>, token: Option<String>) -> RelaySettings {
|
||||
let urls: Vec<String> = urls.into_iter().filter(|u| !u.trim().is_empty()).collect();
|
||||
if urls.is_empty() {
|
||||
let entries = DEFAULT_ROAM_RELAYS
|
||||
.iter()
|
||||
.map(|url| RelayEntry::new(*url))
|
||||
.collect();
|
||||
return RelaySettings::Custom(entries);
|
||||
}
|
||||
let token = token.filter(|t| !t.is_empty());
|
||||
let entries = urls
|
||||
.into_iter()
|
||||
.map(|url| match &token {
|
||||
Some(token) => RelayEntry::with_auth(url, token.clone()),
|
||||
None => RelayEntry::new(url),
|
||||
})
|
||||
.collect();
|
||||
RelaySettings::Custom(entries)
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
pub enum RoamCommand {
|
||||
/// Print this node's connection card — the non-secret string you share with
|
||||
/// a peer so it can find and identify this node. Nothing in it is a secret;
|
||||
/// a peer must still be accepted (`roam peers accept`) before it can connect.
|
||||
#[command(visible_alias = "card")]
|
||||
Id {
|
||||
/// Also render the card as a QR code in the terminal (scan from a phone).
|
||||
#[arg(long)]
|
||||
qr: bool,
|
||||
},
|
||||
|
||||
/// Pair a new device interactively: shows this node's card as a QR code,
|
||||
/// then reads the device's card from stdin and accepts it in one step
|
||||
/// (the equivalent of `roam peers add` + `roam peers accept`).
|
||||
Pair {
|
||||
/// Nickname to save the device under (defaults to `device-<short id>`).
|
||||
#[arg(long)]
|
||||
name: Option<String>,
|
||||
},
|
||||
|
||||
/// Serve this node's agent to accepted peers over ACP.
|
||||
///
|
||||
/// Only peers whose key you have accepted (`roam peers accept`) can connect.
|
||||
/// Each connected peer gets goose's full ACP surface — it drives its own
|
||||
/// sessions (new/list/load/prompt) backed by this node's session store.
|
||||
Share {
|
||||
/// Builtin extensions to load into the hosted agent.
|
||||
#[arg(long = "with-builtin", value_delimiter = ',')]
|
||||
builtins: Vec<String>,
|
||||
|
||||
/// Working directory the hosted agent runs in. Defaults to the directory
|
||||
/// `roam share` was started in. The connecting client's own path is
|
||||
/// always ignored.
|
||||
#[arg(long)]
|
||||
cwd: Option<std::path::PathBuf>,
|
||||
|
||||
/// Also render the connection card as a QR code in the terminal.
|
||||
#[arg(long)]
|
||||
qr: bool,
|
||||
},
|
||||
|
||||
/// Open a quick interactive REPL against a remote agent (debug/peek).
|
||||
///
|
||||
/// This is a minimal built-in chat loop, handy for a quick sanity check. For
|
||||
/// real work, prefer `bridge` (drive the remote agent from Zed or any other
|
||||
/// ACP client) or `delegate` (scriptable one-shot tasks).
|
||||
Connect {
|
||||
/// A saved peer nickname (see `roam peers`) or a `goose+roam://...` card.
|
||||
target: String,
|
||||
|
||||
/// Optional label reported to the host's directory.
|
||||
#[arg(long)]
|
||||
label: Option<String>,
|
||||
},
|
||||
|
||||
/// Delegate a one-shot task to a remote agent and print its response.
|
||||
///
|
||||
/// This is a thin ACP client: it connects, opens a session (new, or the one
|
||||
/// named by `--session`), sends the task as a prompt, prints the reply, and
|
||||
/// exits. Session enumeration and resume are plain ACP (`session/list` /
|
||||
/// `session/load`) served by the remote's full ACP surface.
|
||||
Delegate {
|
||||
/// A saved peer nickname (see `roam peers`) or a `goose+roam://...` card.
|
||||
target: String,
|
||||
/// The task/question to send to the remote agent. Omit when using
|
||||
/// `--list-sessions`.
|
||||
task: Option<String>,
|
||||
/// Run the task against an existing remote session id (via `session/load`)
|
||||
/// instead of a fresh session. List ids with `--list-sessions`.
|
||||
#[arg(long, value_name = "SESSION_ID")]
|
||||
session: Option<String>,
|
||||
/// List the remote agent's sessions (`session/list`) and exit.
|
||||
#[arg(long)]
|
||||
list_sessions: bool,
|
||||
},
|
||||
|
||||
/// Expose a remote agent as a local ACP endpoint that any ACP client can drive.
|
||||
///
|
||||
/// Unlike `connect` (which has its own terminal UI), `bridge` runs no UI and
|
||||
/// no agent: it transparently proxies ACP between a local transport and the
|
||||
/// remote agent. Point Zed or any other ACP client at it
|
||||
/// and the remote agent behaves as if it were running locally.
|
||||
///
|
||||
/// Defaults to stdio (for a client that spawns `goose roam bridge ...` as a
|
||||
/// subprocess). Use `--listen` to accept a single TCP connection instead.
|
||||
Bridge {
|
||||
/// A saved peer nickname (see `roam peers`) or a `goose+roam://...` card.
|
||||
target: String,
|
||||
|
||||
/// Listen for one ACP client on this TCP address (e.g. `127.0.0.1:8900`)
|
||||
/// instead of using stdio. Loopback only: the TCP side carries the
|
||||
/// remote agent's full ACP surface with no authentication of its own.
|
||||
#[arg(long, value_name = "ADDR")]
|
||||
listen: Option<String>,
|
||||
|
||||
/// Allow `--listen` on a non-loopback address. Anyone who can reach
|
||||
/// the socket gets the remote agent — put real authentication or a
|
||||
/// private network in front of it.
|
||||
#[arg(long, requires = "listen")]
|
||||
allow_remote_clients: bool,
|
||||
|
||||
/// Optional label reported to the host's directory.
|
||||
#[arg(long)]
|
||||
label: Option<String>,
|
||||
},
|
||||
|
||||
/// Manage saved peer cards and which peer keys this node accepts.
|
||||
Peers {
|
||||
#[command(subcommand)]
|
||||
command: Option<PeersCommand>,
|
||||
},
|
||||
|
||||
/// Show live/observed connections to and from this node.
|
||||
#[command(visible_alias = "list")]
|
||||
Connections,
|
||||
}
|
||||
|
||||
#[derive(Debug, Subcommand)]
|
||||
pub enum PeersCommand {
|
||||
/// Save a peer's connection card to the address book so you can reach it by
|
||||
/// name. Does NOT let them connect to you — use `accept` for that.
|
||||
Add {
|
||||
/// The peer's `goose+roam://...` card.
|
||||
card: String,
|
||||
/// Friendly nickname (defaults to a short id if omitted).
|
||||
name: Option<String>,
|
||||
},
|
||||
/// Accept inbound connections from a peer's key. The target is a saved
|
||||
/// nickname or a `goose+roam://...` card (which is also saved to the address
|
||||
/// book). An accepted peer gets goose's full ACP surface.
|
||||
Accept {
|
||||
/// A saved nickname or a `goose+roam://...` card.
|
||||
target: String,
|
||||
/// Nickname to save an inline card under (defaults to a short id).
|
||||
/// Ignored when the target is already a saved nickname.
|
||||
name: Option<String>,
|
||||
},
|
||||
/// Stop accepting a peer: a saved nickname, a card, or a raw endpoint id.
|
||||
/// A running share force-closes the peer's live connections within seconds.
|
||||
Revoke { target: String },
|
||||
/// Remove a saved peer from the address book (does not change acceptance).
|
||||
Remove { name: String },
|
||||
/// Rename a saved peer.
|
||||
Rename { from: String, to: String },
|
||||
/// List saved peers and which keys are accepted (default).
|
||||
List,
|
||||
}
|
||||
|
||||
pub async fn handle_roam_command(command: RoamCommand) -> Result<()> {
|
||||
match command {
|
||||
RoamCommand::Id { qr } => handle_id(qr).await,
|
||||
RoamCommand::Pair { name } => handle_pair(name).await,
|
||||
RoamCommand::Share { builtins, cwd, qr } => handle_share(builtins, cwd, qr).await,
|
||||
RoamCommand::Connect { target, label } => handle_connect(target, label).await,
|
||||
RoamCommand::Delegate {
|
||||
target,
|
||||
task,
|
||||
session,
|
||||
list_sessions,
|
||||
} => handle_delegate(target, task, session, list_sessions).await,
|
||||
RoamCommand::Bridge {
|
||||
target,
|
||||
listen,
|
||||
allow_remote_clients,
|
||||
label,
|
||||
} => handle_bridge(target, listen, allow_remote_clients, label).await,
|
||||
RoamCommand::Peers { command } => handle_peers(command.unwrap_or(PeersCommand::List)).await,
|
||||
RoamCommand::Connections => handle_list().await,
|
||||
}
|
||||
}
|
||||
|
||||
/// Bind a node briefly to read its live card (id + relay URLs), waiting for a
|
||||
/// relay so the card carries a reachable address.
|
||||
/// Render a card as a terminal QR code (unicode half-blocks) to stderr.
|
||||
fn print_qr(card: &str) {
|
||||
match qrcode::QrCode::new(card.as_bytes()) {
|
||||
Ok(code) => {
|
||||
let art = code
|
||||
.render::<qrcode::render::unicode::Dense1x2>()
|
||||
.quiet_zone(true)
|
||||
.build();
|
||||
eprintln!("{art}");
|
||||
eprintln!("scan with a phone camera, then paste the decoded card into your client");
|
||||
}
|
||||
Err(err) => eprintln!("could not render QR: {err}"),
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_id(qr: bool) -> Result<()> {
|
||||
let identity = load_identity()?;
|
||||
let node = RoamingNode::bind(RoamingConfig {
|
||||
identity,
|
||||
relay: resolve_relay_settings()?,
|
||||
trust: TrustBook::new(),
|
||||
trust_path: None,
|
||||
directory: Directory::new(),
|
||||
bind_addr: None,
|
||||
relay_tls: None,
|
||||
})
|
||||
.await?;
|
||||
eprintln!("contacting relay so the card carries a reachable address...");
|
||||
node.wait_online(std::time::Duration::from_secs(15)).await;
|
||||
let card = node.card();
|
||||
eprintln!("your connection card (share this with a peer):");
|
||||
println!("{}", card.encode()?);
|
||||
if qr {
|
||||
eprintln!();
|
||||
print_qr(&card.encode()?);
|
||||
}
|
||||
eprintln!();
|
||||
eprintln!(" endpoint id : {}", card.endpoint_id);
|
||||
eprintln!(" fingerprint : {}", card.fingerprint());
|
||||
eprintln!();
|
||||
eprintln!("the peer adds it with: goose roam peers add '<card>' <name>");
|
||||
eprintln!("and accepts you with: goose roam peers accept <name>");
|
||||
node.shutdown().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Interactive one-shot pairing: show this node's card + QR, read the device's
|
||||
/// card from stdin, confirm its fingerprint, then save and accept it — the
|
||||
/// same PeerBook/TrustBook writes as `peers add` + `peers accept`.
|
||||
async fn handle_pair(name: Option<String>) -> Result<()> {
|
||||
let identity = load_identity()?;
|
||||
let node = RoamingNode::bind(RoamingConfig {
|
||||
identity,
|
||||
relay: resolve_relay_settings()?,
|
||||
trust: TrustBook::new(),
|
||||
trust_path: None,
|
||||
directory: Directory::new(),
|
||||
bind_addr: None,
|
||||
relay_tls: None,
|
||||
})
|
||||
.await?;
|
||||
eprintln!("contacting relay so the card carries a reachable address...");
|
||||
node.wait_online(std::time::Duration::from_secs(15)).await;
|
||||
let card = node.card();
|
||||
let encoded = card.encode()?;
|
||||
eprintln!("your connection card:");
|
||||
println!("{encoded}");
|
||||
eprintln!();
|
||||
print_qr(&encoded);
|
||||
eprintln!();
|
||||
eprintln!(" endpoint id : {}", card.endpoint_id);
|
||||
eprintln!(" fingerprint : {}", card.fingerprint());
|
||||
node.shutdown().await?;
|
||||
eprintln!();
|
||||
eprintln!("on the new device: scan the QR with its client (or paste the card),");
|
||||
eprintln!("then copy the card from its pairing screen back here.");
|
||||
eprintln!();
|
||||
|
||||
eprint!("paste the device's card (from its pairing screen): ");
|
||||
let device_card = read_stdin_line()?;
|
||||
let device_card = device_card.trim();
|
||||
if device_card.is_empty() {
|
||||
anyhow::bail!("no card entered; pairing cancelled");
|
||||
}
|
||||
let decoded = ConnectionCard::decode(device_card)
|
||||
.context("that does not look like a goose+roam:// connection card")?;
|
||||
let name =
|
||||
name.unwrap_or_else(|| format!("device-{}", short_id(&decoded.endpoint_id.to_string())));
|
||||
|
||||
eprintln!();
|
||||
eprintln!(" endpoint id : {}", decoded.endpoint_id);
|
||||
eprintln!(" fingerprint : {}", decoded.fingerprint());
|
||||
eprintln!("verify the fingerprint matches the one shown on the device.");
|
||||
eprint!("accept this device as `{name}`? [y/N] ");
|
||||
let answer = read_stdin_line()?;
|
||||
if !matches!(answer.trim().to_lowercase().as_str(), "y" | "yes") {
|
||||
eprintln!("pairing cancelled; nothing was saved");
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
goose_roaming::PeerBook::update(peerbook_path(), |book| {
|
||||
book.save(&name, device_card, now_ms())
|
||||
})?;
|
||||
let path = trust_path();
|
||||
TrustBook::update(&path, |trust| trust.accept(&decoded.endpoint_id)).with_context(|| {
|
||||
format!(
|
||||
"trust file {} is unreadable or corrupt; refusing to modify it",
|
||||
path.display()
|
||||
)
|
||||
})?;
|
||||
|
||||
eprintln!("saved and accepted `{name}` ({})", decoded.endpoint_id);
|
||||
eprintln!("done — the device can now connect to any share/serve on this machine");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn read_stdin_line() -> Result<String> {
|
||||
let mut line = String::new();
|
||||
std::io::stdin()
|
||||
.read_line(&mut line)
|
||||
.context("failed to read from stdin")?;
|
||||
Ok(line)
|
||||
}
|
||||
|
||||
async fn handle_peers(command: PeersCommand) -> Result<()> {
|
||||
let book = goose_roaming::PeerBook::load(peerbook_path())?;
|
||||
match command {
|
||||
PeersCommand::Add { card, name } => {
|
||||
let decoded = ConnectionCard::decode(&card)?;
|
||||
let name = name.unwrap_or_else(|| short_id(&decoded.endpoint_id.to_string()));
|
||||
goose_roaming::PeerBook::update(peerbook_path(), |book| {
|
||||
book.save(&name, &card, now_ms())
|
||||
})?;
|
||||
eprintln!(
|
||||
"saved peer `{name}` -> {} (fingerprint {})",
|
||||
decoded.endpoint_id,
|
||||
decoded.fingerprint()
|
||||
);
|
||||
eprintln!("accept connections from it with: goose roam peers accept {name}");
|
||||
Ok(())
|
||||
}
|
||||
PeersCommand::Accept { target, name } => {
|
||||
// Resolve to a card: a saved name, or an inline card we also save.
|
||||
let card = match ConnectionCard::decode(&target) {
|
||||
Ok(card) => {
|
||||
let name = name.unwrap_or_else(|| short_id(&card.endpoint_id.to_string()));
|
||||
goose_roaming::PeerBook::update(peerbook_path(), |book| {
|
||||
book.save(&name, &target, now_ms())
|
||||
})?;
|
||||
card
|
||||
}
|
||||
Err(_) => {
|
||||
if name.is_some() {
|
||||
eprintln!("note: `{target}` is a saved peer; ignoring the extra name arg");
|
||||
}
|
||||
let rec = book.get(&target).ok_or_else(|| {
|
||||
anyhow::anyhow!(
|
||||
"no saved peer `{target}` and it is not a card; add it first with \
|
||||
`goose roam peers add`"
|
||||
)
|
||||
})?;
|
||||
rec.card.clone()
|
||||
}
|
||||
};
|
||||
let path = trust_path();
|
||||
TrustBook::update(&path, |trust| trust.accept(&card.endpoint_id)).with_context(
|
||||
|| {
|
||||
format!(
|
||||
"trust file {} is unreadable or corrupt; refusing to modify it",
|
||||
path.display()
|
||||
)
|
||||
},
|
||||
)?;
|
||||
eprintln!("accepting connections from {}", card.endpoint_id);
|
||||
eprintln!("verify the fingerprint out of band: {}", card.fingerprint());
|
||||
eprintln!("a running `goose roam share` picks this up on the next connection");
|
||||
Ok(())
|
||||
}
|
||||
PeersCommand::Revoke { target } => {
|
||||
let key = resolve_key(&book, &target)?;
|
||||
let path = trust_path();
|
||||
TrustBook::update(&path, |trust| trust.revoke_key(&key)).with_context(|| {
|
||||
format!(
|
||||
"trust file {} is unreadable or corrupt; refusing to modify it",
|
||||
path.display()
|
||||
)
|
||||
})?;
|
||||
eprintln!("revoked {key}; it can no longer connect");
|
||||
eprintln!("a running share also force-closes its live connections within seconds");
|
||||
Ok(())
|
||||
}
|
||||
PeersCommand::Remove { name } => {
|
||||
let existed =
|
||||
goose_roaming::PeerBook::update(peerbook_path(), |book| book.remove(&name))?;
|
||||
if existed {
|
||||
eprintln!("removed peer `{name}` from the address book");
|
||||
} else {
|
||||
eprintln!("no peer named `{name}`");
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
PeersCommand::Rename { from, to } => {
|
||||
goose_roaming::PeerBook::update(peerbook_path(), |book| book.rename(&from, &to))?;
|
||||
eprintln!("renamed `{from}` -> `{to}`");
|
||||
Ok(())
|
||||
}
|
||||
PeersCommand::List => {
|
||||
let trust =
|
||||
TrustBook::load(&trust_path()).context("trust file is unreadable or corrupt")?;
|
||||
let accepted: std::collections::HashSet<String> =
|
||||
trust.allowed_keys().into_iter().collect();
|
||||
let peers = book.list();
|
||||
if peers.is_empty() && accepted.is_empty() {
|
||||
eprintln!("no saved peers; add one with `goose roam peers add '<card>' <name>`");
|
||||
return Ok(());
|
||||
}
|
||||
println!("{:<16} {:<8} ENDPOINT ID", "NAME", "ACCEPT");
|
||||
for p in &peers {
|
||||
let accept = if accepted.contains(&p.endpoint_id) {
|
||||
"yes"
|
||||
} else {
|
||||
"no"
|
||||
};
|
||||
println!("{:<16} {accept:<8} {}", p.name, p.endpoint_id);
|
||||
}
|
||||
// Accepted keys with no saved card (accepted by raw id).
|
||||
let known: std::collections::HashSet<String> =
|
||||
peers.iter().map(|p| p.endpoint_id.clone()).collect();
|
||||
for id in &accepted {
|
||||
if !known.contains(id) {
|
||||
println!("{:<16} {:<8} {id}", "(unsaved)", "yes");
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Resolve a target (saved nickname, inline card, or raw endpoint id) to a key.
|
||||
fn resolve_key(book: &goose_roaming::PeerBook, target: &str) -> Result<EndpointId> {
|
||||
if let Ok(card) = ConnectionCard::decode(target) {
|
||||
return Ok(card.endpoint_id);
|
||||
}
|
||||
if let Some(rec) = book.get(target) {
|
||||
return Ok(rec.card.endpoint_id);
|
||||
}
|
||||
parse_endpoint_id(target)
|
||||
.map_err(|_| anyhow::anyhow!("`{target}` is not a saved peer, a card, or an endpoint id"))
|
||||
}
|
||||
|
||||
fn short_id(id: &str) -> String {
|
||||
id.chars().take(12).collect()
|
||||
}
|
||||
|
||||
fn now_ms() -> u64 {
|
||||
std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_millis() as u64)
|
||||
.unwrap_or(0)
|
||||
}
|
||||
|
||||
async fn handle_list() -> Result<()> {
|
||||
let entries = Directory::read_persisted(&directory_path());
|
||||
if entries.is_empty() {
|
||||
eprintln!("no roaming peers recorded yet");
|
||||
return Ok(());
|
||||
}
|
||||
println!("{:<10} {:<9} {:<20} ENDPOINT ID", "STATUS", "DIR", "AGENT");
|
||||
for e in entries {
|
||||
let status = if e.connected { "connected" } else { "seen" };
|
||||
let dir = match e.direction {
|
||||
goose_roaming::Direction::Inbound => "inbound",
|
||||
goose_roaming::Direction::Outbound => "outbound",
|
||||
};
|
||||
let agent = e.agent_id.unwrap_or_else(|| "-".to_string());
|
||||
let agent = if agent.chars().count() > 20 {
|
||||
let truncated: String = agent.chars().take(19).collect();
|
||||
format!("{truncated}…")
|
||||
} else {
|
||||
agent
|
||||
};
|
||||
println!("{status:<10} {dir:<9} {agent:<20} {}", e.endpoint_id);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// This node's single long-lived identity. Its public key is what peers accept
|
||||
/// and what the connection card advertises.
|
||||
pub(crate) fn load_identity() -> Result<RoamingIdentity> {
|
||||
let path = default_key_path(&Paths::config_dir());
|
||||
RoamingIdentity::load_or_create(&path).context("failed to load roaming identity")
|
||||
}
|
||||
|
||||
/// Roaming is an app-level service: every backend loads the same persisted
|
||||
/// identity, so only one process may advertise the endpoint at a time. An OS
|
||||
/// advisory lock decides ownership across all goose processes (desktop
|
||||
/// windows, CLI serves, standalone shares); it auto-releases when the owner
|
||||
/// dies — even on SIGKILL — so a standby can promote itself and paired
|
||||
/// devices keep access.
|
||||
/// `Ok(Some(file))` holds the lock; `Ok(None)` means another live process owns
|
||||
/// the endpoint (standby is reasonable); `Err` is a real failure — unwritable
|
||||
/// data dir, filesystem error — that must be surfaced, not retried silently.
|
||||
pub(crate) fn try_acquire_roam_lock_owner() -> Result<Option<std::fs::File>> {
|
||||
use fs2::FileExt as _;
|
||||
use std::io::Write as _;
|
||||
|
||||
let lock_path = Paths::data_dir().join("roam/serve.lock");
|
||||
if let Some(parent) = lock_path.parent() {
|
||||
std::fs::create_dir_all(parent)
|
||||
.with_context(|| format!("cannot create roam lock dir {}", parent.display()))?;
|
||||
}
|
||||
let mut file = std::fs::OpenOptions::new()
|
||||
.write(true)
|
||||
.create(true)
|
||||
.truncate(false)
|
||||
.open(&lock_path)
|
||||
.with_context(|| format!("cannot open roam lock file {}", lock_path.display()))?;
|
||||
if file.try_lock_exclusive().is_err() {
|
||||
return Ok(None);
|
||||
}
|
||||
file.set_len(0)?;
|
||||
writeln!(file, "{}", std::process::id())?;
|
||||
Ok(Some(file))
|
||||
}
|
||||
|
||||
pub(crate) fn try_acquire_roam_lock() -> Result<std::fs::File> {
|
||||
try_acquire_roam_lock_owner()?.ok_or_else(|| {
|
||||
anyhow::anyhow!("another goose process is already running the roaming endpoint")
|
||||
})
|
||||
}
|
||||
|
||||
async fn handle_share(
|
||||
builtins: Vec<String>,
|
||||
cwd: Option<std::path::PathBuf>,
|
||||
qr: bool,
|
||||
) -> Result<()> {
|
||||
// Same app-level lock as `serve --roam`: both bind the one persisted
|
||||
// identity, and two processes advertising the same endpoint ID would race
|
||||
// for connections while hosting different working directories.
|
||||
let _roam_lock = try_acquire_roam_lock()?;
|
||||
let identity = load_identity()?;
|
||||
|
||||
// The hosted agent runs in `--cwd` or the directory `roam share` was started
|
||||
// in; the connecting client's own path is meaningless here and is ignored.
|
||||
let session_cwd = match &cwd {
|
||||
Some(dir) => std::fs::canonicalize(dir)
|
||||
.with_context(|| format!("invalid --cwd: {}", dir.display()))?,
|
||||
None => std::env::current_dir().context("could not determine current directory")?,
|
||||
};
|
||||
// Fail here rather than advertising a share whose every session activation
|
||||
// would reject the host-imposed path.
|
||||
anyhow::ensure!(
|
||||
session_cwd.is_dir(),
|
||||
"--cwd must be a directory: {}",
|
||||
session_cwd.display()
|
||||
);
|
||||
|
||||
// Load the accepted-peer allowlist. Peers are accepted out of band with
|
||||
// `roam peers accept`; this serve loop re-reads it per connection.
|
||||
let trust = TrustBook::load(&trust_path())
|
||||
.context("trust file is unreadable or corrupt; no peer can connect until it is fixed")?;
|
||||
let accepted_count = trust.allowed_keys().len();
|
||||
if accepted_count == 0 {
|
||||
eprintln!(
|
||||
"warning: no peers are accepted yet — no one can connect.\n\
|
||||
Accept a peer's key first: goose roam peers accept <name|card>"
|
||||
);
|
||||
}
|
||||
|
||||
// The developer extension is on by default; explicitly requested builtins
|
||||
// are always loaded on top.
|
||||
let builtins = AcpBuiltinSelection {
|
||||
defaults: vec!["developer".to_string()],
|
||||
explicit: builtins,
|
||||
};
|
||||
|
||||
let node = RoamingNode::bind(RoamingConfig {
|
||||
identity,
|
||||
relay: resolve_relay_settings()?,
|
||||
trust,
|
||||
// Re-read acceptance on each connection so `peers accept`/`revoke` from
|
||||
// another process take effect against this live share without restart.
|
||||
trust_path: Some(trust_path()),
|
||||
directory: Directory::persistent_owned(directory_path()),
|
||||
bind_addr: None,
|
||||
relay_tls: None,
|
||||
})
|
||||
.await?;
|
||||
|
||||
let acp_server = Arc::new(AcpServer::new(AcpServerFactoryConfig {
|
||||
builtins,
|
||||
data_dir: Paths::data_dir(),
|
||||
config_dir: Paths::config_dir(),
|
||||
goose_platform: GoosePlatform::GooseCli,
|
||||
additional_source_roots: Vec::new(),
|
||||
session_cwd: Some(session_cwd.clone()),
|
||||
enable_scheduler: false,
|
||||
}));
|
||||
let agent_id = node.endpoint_id().to_string();
|
||||
let bridge = Arc::new(FullAcpBridge::new(
|
||||
acp_server,
|
||||
agent_id,
|
||||
session_cwd.clone(),
|
||||
));
|
||||
node.share(bridge).await?;
|
||||
|
||||
eprintln!("contacting relay...");
|
||||
if !node.wait_online(std::time::Duration::from_secs(15)).await {
|
||||
eprintln!("warning: endpoint did not come online; the card may lack a reachable address");
|
||||
}
|
||||
|
||||
eprintln!("roaming agent is live");
|
||||
eprintln!(" endpoint id : {}", node.endpoint_id());
|
||||
eprintln!(" working dir : {}", session_cwd.display());
|
||||
eprintln!(" accepted : {accepted_count} peer key(s)");
|
||||
eprintln!();
|
||||
eprintln!("your connection card (share with a peer so it can reach you):");
|
||||
println!("{}", node.card().encode()?);
|
||||
if qr {
|
||||
print_qr(&node.card().encode()?);
|
||||
}
|
||||
eprintln!();
|
||||
eprintln!("press Ctrl-C to stop sharing");
|
||||
|
||||
tokio::signal::ctrl_c().await?;
|
||||
eprintln!("\nshutting down roaming endpoint...");
|
||||
node.shutdown().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Resolve a target (saved peer nickname or inline card) to a [`ConnectionCard`].
|
||||
fn resolve_card(target: &str) -> Result<ConnectionCard> {
|
||||
if target.starts_with(CARD_SCHEME) {
|
||||
return ConnectionCard::decode(target).map_err(Into::into);
|
||||
}
|
||||
let book = goose_roaming::PeerBook::load(peerbook_path())?;
|
||||
match book.get(target) {
|
||||
Some(rec) => Ok(rec.card.clone()),
|
||||
None => anyhow::bail!(
|
||||
"no saved peer named `{target}` (and it is not a card); see `goose roam peers`"
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
/// Bind this node and dial the target's card, returning the node + authorized
|
||||
/// stream. The connection succeeds only if the remote has accepted this node's
|
||||
/// key.
|
||||
async fn dial_target(
|
||||
target: &str,
|
||||
label: Option<String>,
|
||||
) -> Result<(
|
||||
std::sync::Arc<RoamingNode>,
|
||||
goose_roaming::RoamingClientStream,
|
||||
)> {
|
||||
let card = resolve_card(target)?;
|
||||
let node = RoamingNode::bind(RoamingConfig {
|
||||
identity: load_identity()?,
|
||||
relay: resolve_relay_settings()?,
|
||||
trust: TrustBook::new(),
|
||||
trust_path: None,
|
||||
// Persist outbound observations so `roam connections` can show them.
|
||||
directory: Directory::persistent(directory_path()),
|
||||
bind_addr: None,
|
||||
relay_tls: None,
|
||||
})
|
||||
.await?;
|
||||
eprintln!("connecting to {}...", card.endpoint_id);
|
||||
let stream = node.connect(&card, label).await?;
|
||||
Ok((node, stream))
|
||||
}
|
||||
|
||||
async fn handle_connect(target: String, label: Option<String>) -> Result<()> {
|
||||
let (node, stream) = dial_target(&target, label).await?;
|
||||
let agent_label = stream.agent_id.clone();
|
||||
eprintln!("connected to `{agent_label}`");
|
||||
let result = crate::commands::roam_client::run_interactive(stream, agent_label).await;
|
||||
node.shutdown().await?;
|
||||
result
|
||||
}
|
||||
|
||||
async fn handle_delegate(
|
||||
target: String,
|
||||
task: Option<String>,
|
||||
session: Option<String>,
|
||||
list_sessions: bool,
|
||||
) -> Result<()> {
|
||||
if list_sessions {
|
||||
let (node, stream) = dial_target(&target, Some("delegate".to_string())).await?;
|
||||
eprintln!("listing sessions on `{}`...", stream.agent_id);
|
||||
let result = crate::commands::roam_client::list_sessions(stream).await;
|
||||
node.shutdown().await?;
|
||||
let sessions = result?;
|
||||
if sessions.is_empty() {
|
||||
eprintln!("no sessions on the remote agent");
|
||||
return Ok(());
|
||||
}
|
||||
println!("{:<40} {:<20} UPDATED", "SESSION ID", "TITLE");
|
||||
for s in sessions {
|
||||
let title = s.title.unwrap_or_default();
|
||||
let title = if title.chars().count() > 20 {
|
||||
format!("{}…", title.chars().take(19).collect::<String>())
|
||||
} else {
|
||||
title
|
||||
};
|
||||
println!(
|
||||
"{:<40} {title:<20} {}",
|
||||
s.session_id,
|
||||
s.updated_at.unwrap_or_default()
|
||||
);
|
||||
}
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let task = task.context("a task is required (or pass --list-sessions)")?;
|
||||
let (node, stream) = dial_target(&target, Some("delegate".to_string())).await?;
|
||||
match &session {
|
||||
Some(id) => eprintln!("delegating to `{}` session {id}...", stream.agent_id),
|
||||
None => eprintln!("delegating task to `{}`...", stream.agent_id),
|
||||
}
|
||||
let result = crate::commands::roam_client::delegate(stream, task, session).await;
|
||||
node.shutdown().await?;
|
||||
match result {
|
||||
Ok(response) => {
|
||||
println!("{response}");
|
||||
Ok(())
|
||||
}
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_bridge(
|
||||
target: String,
|
||||
listen: Option<String>,
|
||||
allow_remote_clients: bool,
|
||||
label: Option<String>,
|
||||
) -> Result<()> {
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
// Refuse a non-loopback --listen unless explicitly overridden: the TCP
|
||||
// side is an unauthenticated door to the remote agent's full ACP surface.
|
||||
if let Some(addr) = &listen {
|
||||
let parsed: std::net::SocketAddr = addr
|
||||
.parse()
|
||||
.with_context(|| format!("invalid --listen address `{addr}`"))?;
|
||||
if !parsed.ip().is_loopback() && !allow_remote_clients {
|
||||
anyhow::bail!(
|
||||
"--listen {addr} is not a loopback address; anyone who can reach it gets the \
|
||||
remote agent with no authentication. Use 127.0.0.1/[::1], or pass \
|
||||
--allow-remote-clients if you really mean to expose it."
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
let label = label.or_else(|| Some("bridge".to_string()));
|
||||
let (node, stream) = dial_target(&target, label).await?;
|
||||
let agent_id = stream.agent_id.clone();
|
||||
// The raw iroh streams carry post-handshake ACP and already implement
|
||||
// tokio's AsyncRead/AsyncWrite, so we splice them directly. `conn` must
|
||||
// outlive the splice.
|
||||
let goose_roaming::RoamingClientStream {
|
||||
conn,
|
||||
send: remote_send,
|
||||
recv: remote_recv,
|
||||
..
|
||||
} = stream;
|
||||
|
||||
let result = match listen {
|
||||
Some(addr) => {
|
||||
let listener = tokio::net::TcpListener::bind(&addr).await?;
|
||||
let local = listener.local_addr()?;
|
||||
eprintln!("bridging remote agent `{agent_id}` on tcp://{local}");
|
||||
eprintln!("point an ACP client at this address; serving one connection");
|
||||
let (socket, peer) = listener.accept().await?;
|
||||
eprintln!("ACP client connected from {peer}");
|
||||
let (lr, lw) = socket.into_split();
|
||||
crate::commands::roam_proxy::splice(lr, lw, remote_send, remote_recv).await
|
||||
}
|
||||
None => {
|
||||
// stdio is a pure ACP transport: ONLY the splice may touch stdout.
|
||||
// All status goes to stderr so an ACP client reading stdout sees a
|
||||
// clean protocol stream.
|
||||
eprintln!("bridging remote agent `{agent_id}` over stdio; speak ACP on stdin/stdout");
|
||||
let stdin = tokio::io::stdin();
|
||||
let stdout = tokio::io::stdout();
|
||||
crate::commands::roam_proxy::splice(stdin, stdout, remote_send, remote_recv).await
|
||||
}
|
||||
};
|
||||
|
||||
let _ = tokio::io::stderr().flush().await;
|
||||
drop(conn);
|
||||
node.shutdown().await?;
|
||||
result
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn no_urls_uses_default_managed_relays_not_public() {
|
||||
match build_relay_settings(vec![], None) {
|
||||
RelaySettings::Custom(entries) => {
|
||||
assert_eq!(entries.len(), DEFAULT_ROAM_RELAYS.len());
|
||||
// Managed relays register open — no auth token attached.
|
||||
assert!(entries.iter().all(|e| e.auth_token.is_none()));
|
||||
assert!(entries.iter().all(|e| e.url.contains("iroh.link")));
|
||||
}
|
||||
other => panic!("expected Custom managed relays, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn blank_urls_are_filtered_and_fall_back_to_managed() {
|
||||
match build_relay_settings(vec![" ".into(), "".into()], None) {
|
||||
RelaySettings::Custom(entries) => {
|
||||
assert_eq!(entries.len(), DEFAULT_ROAM_RELAYS.len());
|
||||
}
|
||||
other => panic!("expected Custom managed relays, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn custom_urls_without_token() {
|
||||
let settings = build_relay_settings(vec!["https://relay.example./".into()], None);
|
||||
match settings {
|
||||
RelaySettings::Custom(entries) => {
|
||||
assert_eq!(entries.len(), 1);
|
||||
assert_eq!(entries[0].url, "https://relay.example./");
|
||||
assert!(entries[0].auth_token.is_none());
|
||||
}
|
||||
other => panic!("expected Custom, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn custom_urls_apply_nonempty_token_to_each() {
|
||||
let settings = build_relay_settings(
|
||||
vec!["https://a.example./".into(), "https://b.example./".into()],
|
||||
Some("tok".into()),
|
||||
);
|
||||
match settings {
|
||||
RelaySettings::Custom(entries) => {
|
||||
assert_eq!(entries.len(), 2);
|
||||
assert!(entries
|
||||
.iter()
|
||||
.all(|e| e.auth_token.as_deref() == Some("tok")));
|
||||
}
|
||||
other => panic!("expected Custom, got {other:?}"),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn empty_token_is_ignored() {
|
||||
let settings =
|
||||
build_relay_settings(vec!["https://relay.example./".into()], Some(String::new()));
|
||||
match settings {
|
||||
RelaySettings::Custom(entries) => {
|
||||
assert!(entries[0].auth_token.is_none());
|
||||
}
|
||||
other => panic!("expected Custom, got {other:?}"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,283 @@
|
||||
//! Thin ACP client UI onto a remote roaming agent.
|
||||
//!
|
||||
//! Per design doc §9: `roam connect` is NOT a provider wrapper for a local
|
||||
//! agent loop. The **host** runs the real agent (its tools, working dir,
|
||||
//! shell); this side is just an ACP *client* that opens a session, sends
|
||||
//! prompts, and renders `session/update` notifications to the terminal.
|
||||
//!
|
||||
//! We deliberately advertise no client filesystem/terminal capabilities and do
|
||||
//! not send our local cwd — the host imposes the `share` working directory.
|
||||
|
||||
use std::io::Write;
|
||||
|
||||
use tokio::io::{AsyncBufReadExt, BufReader};
|
||||
|
||||
use agent_client_protocol::schema::v1::{
|
||||
ContentBlock, InitializeRequest, ListSessionsRequest, LoadSessionRequest, PromptRequest,
|
||||
RequestPermissionOutcome, RequestPermissionRequest, RequestPermissionResponse,
|
||||
SelectedPermissionOutcome, SessionId, SessionInfo, SessionNotification, SessionUpdate,
|
||||
};
|
||||
use agent_client_protocol::schema::ProtocolVersion;
|
||||
use agent_client_protocol::{Agent, Client, ConnectionTo};
|
||||
use anyhow::Result;
|
||||
use goose_roaming::RoamingClientStream;
|
||||
|
||||
/// Run an interactive ACP session over an authorized roaming stream, reading
|
||||
/// prompts from stdin until EOF / quit.
|
||||
pub async fn run_interactive(stream: RoamingClientStream, agent_label: String) -> Result<()> {
|
||||
let (send, recv, conn) = stream.into_futures_io();
|
||||
let transport = agent_client_protocol::ByteStreams::new(send, recv);
|
||||
|
||||
Client
|
||||
.builder()
|
||||
.name("goose-roam")
|
||||
.on_receive_notification(
|
||||
async move |notification: SessionNotification, _cx| {
|
||||
render_update(¬ification.update);
|
||||
Ok(())
|
||||
},
|
||||
agent_client_protocol::on_receive_notification!(),
|
||||
)
|
||||
.on_receive_request(
|
||||
async move |request: RequestPermissionRequest, responder, _cx| {
|
||||
// The host runs the agent, so tool-permission prompts originate
|
||||
// there. Present them to the local user and forward the choice.
|
||||
let outcome = prompt_permission(&request);
|
||||
responder.respond(RequestPermissionResponse::new(outcome))
|
||||
},
|
||||
agent_client_protocol::on_receive_request!(),
|
||||
)
|
||||
.connect_with(transport, async move |cx: ConnectionTo<Agent>| {
|
||||
let init = cx
|
||||
.send_request(InitializeRequest::new(ProtocolVersion::LATEST))
|
||||
.block_task()
|
||||
.await?;
|
||||
eprintln!(
|
||||
"connected to remote agent `{agent_label}` (protocol {:?})",
|
||||
init.protocol_version
|
||||
);
|
||||
eprintln!("type a message and press enter; Ctrl-D or /quit to end.\n");
|
||||
|
||||
// The host imposes its own working directory and ignores whatever we
|
||||
// send here (our local path is meaningless on the host machine). ACP
|
||||
// still requires a syntactically-absolute cwd, so send a placeholder.
|
||||
let cwd = std::path::PathBuf::from("/");
|
||||
let result = cx
|
||||
.build_session(cwd)
|
||||
.block_task()
|
||||
.run_until(async |mut session| {
|
||||
tracing::debug!(session_id = ?session.session_id(), "roam client: session created");
|
||||
let mut stdin = BufReader::new(tokio::io::stdin()).lines();
|
||||
loop {
|
||||
eprint!("› ");
|
||||
let _ = std::io::stderr().flush();
|
||||
let line = match stdin.next_line().await {
|
||||
Ok(Some(l)) => l,
|
||||
Ok(None) | Err(_) => break, // EOF
|
||||
};
|
||||
let line = line.trim();
|
||||
if line.is_empty() {
|
||||
continue;
|
||||
}
|
||||
if line == "/quit" || line == "/exit" {
|
||||
break;
|
||||
}
|
||||
session.send_prompt(line)?;
|
||||
// Drain updates until the turn completes; chunks are
|
||||
// rendered live via the notification handler above.
|
||||
let _ = session.read_to_string().await?;
|
||||
println!();
|
||||
}
|
||||
Ok(())
|
||||
})
|
||||
.await;
|
||||
if let Err(e) = &result {
|
||||
tracing::warn!("roam client: session ended with error: {e:?}");
|
||||
}
|
||||
result
|
||||
})
|
||||
.await?;
|
||||
|
||||
drop(conn);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// One-shot delegation: open a remote session, send a single task, return the
|
||||
/// agent's final text response. No interactive loop, no local stdin.
|
||||
///
|
||||
/// This is the reusable core a future `roam__delegate` model tool will call.
|
||||
/// Permission requests are auto-cancelled: a delegated (agent-driven) session
|
||||
/// must not block waiting for a human, and the caller isn't a person who can
|
||||
/// answer. Loop/cost safety is the caller's concern (bounded turns/deadline).
|
||||
pub async fn delegate(
|
||||
stream: RoamingClientStream,
|
||||
task: String,
|
||||
session: Option<String>,
|
||||
) -> Result<String> {
|
||||
let (send, recv, conn) = stream.into_futures_io();
|
||||
let transport = agent_client_protocol::ByteStreams::new(send, recv);
|
||||
let collected = std::sync::Arc::new(std::sync::Mutex::new(String::new()));
|
||||
let sink = collected.clone();
|
||||
let replay_sink = collected.clone();
|
||||
|
||||
Client
|
||||
.builder()
|
||||
.name("goose-roam-delegate")
|
||||
.on_receive_notification(
|
||||
async move |notification: SessionNotification, _cx| {
|
||||
if let SessionUpdate::AgentMessageChunk(chunk) = ¬ification.update {
|
||||
if let ContentBlock::Text(text) = &chunk.content {
|
||||
sink.lock().unwrap().push_str(&text.text);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
},
|
||||
agent_client_protocol::on_receive_notification!(),
|
||||
)
|
||||
.on_receive_request(
|
||||
async move |_request: RequestPermissionRequest, responder, _cx| {
|
||||
// Agent-driven session: never wait on a human. Auto-cancel.
|
||||
responder.respond(RequestPermissionResponse::new(
|
||||
RequestPermissionOutcome::Cancelled,
|
||||
))
|
||||
},
|
||||
agent_client_protocol::on_receive_request!(),
|
||||
)
|
||||
.connect_with(transport, async move |cx: ConnectionTo<Agent>| {
|
||||
cx.send_request(InitializeRequest::new(ProtocolVersion::LATEST))
|
||||
.block_task()
|
||||
.await?;
|
||||
// The host imposes its own working directory and ignores whatever we
|
||||
// send; ACP still requires a syntactically-absolute cwd.
|
||||
let cwd = std::path::PathBuf::from("/");
|
||||
match session {
|
||||
// Resume an existing remote session. The final text is
|
||||
// collected by the notification handler above, so a raw
|
||||
// session/prompt awaited to completion is all we need — no
|
||||
// ActiveSession attachment required.
|
||||
Some(id) => {
|
||||
let session_id = SessionId::from(id);
|
||||
cx.send_request(LoadSessionRequest::new(session_id.clone(), cwd))
|
||||
.block_task()
|
||||
.await?;
|
||||
// Loading replays the session's history as message chunks,
|
||||
// which the notification handler collected. Discard them so
|
||||
// the result is only the new response to this task.
|
||||
replay_sink.lock().unwrap().clear();
|
||||
cx.send_request(PromptRequest::new(session_id, vec![task.clone().into()]))
|
||||
.block_task()
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
None => {
|
||||
cx.build_session(cwd)
|
||||
.block_task()
|
||||
.run_until(async |mut session| {
|
||||
session.send_prompt(&task)?;
|
||||
let _ = session.read_to_string().await?;
|
||||
Ok(())
|
||||
})
|
||||
.await
|
||||
}
|
||||
}
|
||||
})
|
||||
.await?;
|
||||
|
||||
drop(conn);
|
||||
let result = collected.lock().unwrap().clone();
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
/// List the remote agent's sessions via ACP `session/list`. Roaming adds no
|
||||
/// session semantics — this is plain ACP over the authorized stream.
|
||||
pub async fn list_sessions(stream: RoamingClientStream) -> Result<Vec<SessionInfo>> {
|
||||
let (send, recv, conn) = stream.into_futures_io();
|
||||
let transport = agent_client_protocol::ByteStreams::new(send, recv);
|
||||
|
||||
let sessions = Client
|
||||
.builder()
|
||||
.name("goose-roam-list")
|
||||
.on_receive_notification(
|
||||
async move |_notification: SessionNotification, _cx| Ok(()),
|
||||
agent_client_protocol::on_receive_notification!(),
|
||||
)
|
||||
.connect_with(transport, async move |cx: ConnectionTo<Agent>| {
|
||||
cx.send_request(InitializeRequest::new(ProtocolVersion::LATEST))
|
||||
.block_task()
|
||||
.await?;
|
||||
let response = cx
|
||||
.send_request(ListSessionsRequest::default())
|
||||
.block_task()
|
||||
.await?;
|
||||
Ok(response.sessions)
|
||||
})
|
||||
.await?;
|
||||
|
||||
drop(conn);
|
||||
Ok(sessions)
|
||||
}
|
||||
|
||||
fn render_update(update: &SessionUpdate) {
|
||||
match update {
|
||||
SessionUpdate::AgentMessageChunk(chunk) => {
|
||||
if let ContentBlock::Text(text) = &chunk.content {
|
||||
print!("{}", text.text);
|
||||
let _ = std::io::stdout().flush();
|
||||
}
|
||||
}
|
||||
SessionUpdate::AgentThoughtChunk(chunk) => {
|
||||
if let ContentBlock::Text(text) = &chunk.content {
|
||||
eprint!("\x1b[2m{}\x1b[0m", text.text);
|
||||
}
|
||||
}
|
||||
SessionUpdate::ToolCall(tool_call) => {
|
||||
eprintln!("\n🔧 {}", tool_call.title);
|
||||
}
|
||||
SessionUpdate::ToolCallUpdate(update) => {
|
||||
if let Some(status) = &update.fields.status {
|
||||
eprintln!(" [{status:?}]");
|
||||
}
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
|
||||
fn prompt_permission(request: &RequestPermissionRequest) -> RequestPermissionOutcome {
|
||||
eprintln!("\n⚠️ the remote agent requests permission:");
|
||||
for (i, opt) in request.options.iter().enumerate() {
|
||||
eprintln!(" {}) {}", i + 1, opt.name);
|
||||
}
|
||||
eprint!("choose a number (anything else cancels): ");
|
||||
let _ = std::io::stderr().flush();
|
||||
|
||||
// Fail closed: option 1 is allow-always for goose hosts, so EOF, an empty
|
||||
// line, or a typo must cancel rather than silently granting permission.
|
||||
let Some(choice) = read_line().and_then(|l| l.trim().parse::<usize>().ok()) else {
|
||||
eprintln!(" cancelled");
|
||||
return RequestPermissionOutcome::Cancelled;
|
||||
};
|
||||
|
||||
match choice
|
||||
.checked_sub(1)
|
||||
.and_then(|idx| request.options.get(idx))
|
||||
{
|
||||
Some(opt) => RequestPermissionOutcome::Selected(SelectedPermissionOutcome::new(
|
||||
opt.option_id.clone(),
|
||||
)),
|
||||
None => {
|
||||
eprintln!(" cancelled");
|
||||
RequestPermissionOutcome::Cancelled
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn read_line() -> Option<String> {
|
||||
eprint!("› ");
|
||||
let _ = std::io::stderr().flush();
|
||||
let mut buf = String::new();
|
||||
match std::io::stdin().read_line(&mut buf) {
|
||||
Ok(0) => None, // EOF
|
||||
Ok(_) => Some(buf),
|
||||
Err(_) => None,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
//! The roaming host adapter: serve goose's **full** ACP surface to each
|
||||
//! connecting peer.
|
||||
//!
|
||||
//! Roaming is just an authenticated p2p ACP transport. This adapter is the thin
|
||||
//! seam that hands an authorized iroh stream straight to goose's real
|
||||
//! `acp::server::serve`, so a connected client gets the entire ACP surface —
|
||||
//! `session/new`, `session/list`, `session/load`, `session/prompt` — backed by
|
||||
//! the host's own `SessionManager`. Anything "session-shaped" is therefore plain
|
||||
//! ACP that happens to run over a roaming connection; roaming adds no session
|
||||
//! semantics of its own.
|
||||
//!
|
||||
//! Each accepted connection gets a **fresh** agent (never one shared across
|
||||
//! clients); every client drives its own independent sessions.
|
||||
|
||||
use std::sync::Arc;
|
||||
|
||||
use futures::future::BoxFuture;
|
||||
use futures::io::{AsyncRead, AsyncWrite};
|
||||
|
||||
use goose::acp::server::serve;
|
||||
use goose::acp::server_factory::AcpServer;
|
||||
use goose_roaming::{AcpStreamServer, EndpointId};
|
||||
|
||||
/// An [`AcpStreamServer`] that serves goose's full ACP surface, a fresh agent
|
||||
/// per connection.
|
||||
pub struct FullAcpBridge {
|
||||
server: Arc<AcpServer>,
|
||||
agent_id: String,
|
||||
/// Host-controlled working directory for sessions created over roaming.
|
||||
/// The connector's machine-local absolute path is meaningless on this
|
||||
/// host, so every roaming agent gets this instead — even when the shared
|
||||
/// `AcpServer` (e.g. `goose serve`) leaves `session_cwd` unset for its
|
||||
/// local clients.
|
||||
session_cwd: std::path::PathBuf,
|
||||
}
|
||||
|
||||
impl FullAcpBridge {
|
||||
pub fn new(
|
||||
server: Arc<AcpServer>,
|
||||
agent_id: impl Into<String>,
|
||||
session_cwd: std::path::PathBuf,
|
||||
) -> Self {
|
||||
Self {
|
||||
server,
|
||||
agent_id: agent_id.into(),
|
||||
session_cwd,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl AcpStreamServer for FullAcpBridge {
|
||||
fn serve_stream(
|
||||
&self,
|
||||
client: EndpointId,
|
||||
recv: Box<dyn AsyncRead + Send + Unpin>,
|
||||
send: Box<dyn AsyncWrite + Send + Unpin>,
|
||||
) -> BoxFuture<'static, anyhow::Result<()>> {
|
||||
let server = self.server.clone();
|
||||
let session_cwd = self.session_cwd.clone();
|
||||
Box::pin(async move {
|
||||
tracing::info!(%client, "roaming: serving full ACP surface");
|
||||
let agent = server
|
||||
.create_agent_with_session_cwd(Some(session_cwd))
|
||||
.await?;
|
||||
serve(agent, recv, send).await
|
||||
})
|
||||
}
|
||||
|
||||
fn agent_id(&self) -> String {
|
||||
self.agent_id.clone()
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
//! Transparent ACP proxy: splice a local ACP transport to a remote roaming stream.
|
||||
//!
|
||||
//! `roam connect`/`delegate` embed an ACP *client* with a built-in terminal UI.
|
||||
//! This module is the opposite composition: it exposes a remote agent as a
|
||||
//! *local ACP endpoint* so any ACP client (Zed, another editor)
|
||||
//! can drive it as if it were local.
|
||||
//!
|
||||
//! The trick is that once the roaming handshake completes, the stream carries
|
||||
//! raw ACP JSON-RPC framing — byte-for-byte what `goose acp` speaks over stdio.
|
||||
//! So bridging is a pure copy in both directions: nothing runs an agent here and
|
||||
//! nothing is deserialized. We just pump bytes local↔remote until both halves
|
||||
//! close.
|
||||
|
||||
use anyhow::Result;
|
||||
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt};
|
||||
|
||||
/// Splice a local ACP transport (`local_read`/`local_write`) to a remote
|
||||
/// roaming stream (`remote_write`/`remote_read`).
|
||||
///
|
||||
/// Runs both directions to completion: when the local client closes its input
|
||||
/// we finish the remote's send half, and when the remote agent closes we finish
|
||||
/// the local output half. This avoids truncating an in-flight ACP turn, which a
|
||||
/// "return on first close" splice would do.
|
||||
pub async fn splice<LR, LW, RW, RR>(
|
||||
mut local_read: LR,
|
||||
mut local_write: LW,
|
||||
mut remote_write: RW,
|
||||
mut remote_read: RR,
|
||||
) -> Result<()>
|
||||
where
|
||||
LR: AsyncRead + Unpin + Send,
|
||||
LW: AsyncWrite + Unpin + Send,
|
||||
RW: AsyncWrite + Unpin + Send,
|
||||
RR: AsyncRead + Unpin + Send,
|
||||
{
|
||||
let client_to_host = async {
|
||||
tokio::io::copy(&mut local_read, &mut remote_write).await?;
|
||||
remote_write.shutdown().await
|
||||
};
|
||||
let host_to_client = async {
|
||||
tokio::io::copy(&mut remote_read, &mut local_write).await?;
|
||||
local_write.shutdown().await
|
||||
};
|
||||
|
||||
tokio::try_join!(client_to_host, host_to_client)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use tokio::io::AsyncReadExt;
|
||||
|
||||
/// Bytes written by the local client reach the remote, and bytes from the
|
||||
/// remote reach the local client — both directions, to completion.
|
||||
///
|
||||
/// The test ends are used as whole `DuplexStream`s (not split): a split end
|
||||
/// only reaches EOF once *both* halves are dropped, so `splice`'s copies
|
||||
/// would never observe EOF and the bridge would hang. Dropping the whole
|
||||
/// endpoint is what signals EOF and lets `splice` finish cleanly.
|
||||
#[tokio::test]
|
||||
async fn splices_both_directions() {
|
||||
let (mut local_client, local_endpoint) = tokio::io::duplex(64);
|
||||
let (remote_endpoint, mut remote_agent) = tokio::io::duplex(64);
|
||||
let (local_read, local_write) = tokio::io::split(local_endpoint);
|
||||
let (remote_read, remote_write) = tokio::io::split(remote_endpoint);
|
||||
|
||||
let bridge = tokio::spawn(async move {
|
||||
splice(local_read, local_write, remote_write, remote_read).await
|
||||
});
|
||||
|
||||
local_client.write_all(b"initialize").await.unwrap();
|
||||
let mut buf = [0u8; 10];
|
||||
remote_agent.read_exact(&mut buf).await.unwrap();
|
||||
assert_eq!(&buf, b"initialize");
|
||||
|
||||
remote_agent.write_all(b"session-ok").await.unwrap();
|
||||
let mut buf = [0u8; 10];
|
||||
local_client.read_exact(&mut buf).await.unwrap();
|
||||
assert_eq!(&buf, b"session-ok");
|
||||
|
||||
drop(local_client);
|
||||
drop(remote_agent);
|
||||
bridge.await.unwrap().unwrap();
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user