diff --git a/CHANGELOG.md b/CHANGELOG.md index a5a611f91..7bd98e618 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,11 +5,20 @@ All notable changes to Agent Relay will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased - Patch] +## [Unreleased - Minor] + +### Added + +- `agent-relay cloud room` can invite trusted full room participants through explicit secret sinks, manage members, and establish revocable per-device Relaycast sessions without sharing the workspace key. +- `agent-relay cloud integration` exposes the existing Cloud integration catalog, connection, and disconnection lifecycle from the CLI; connected providers remain available through Relayfile's normal setup, mount, and writeback flow. +- `agent-relay agent me|presence` use scoped agent credentials for room-safe identity and presence checks. ### Fixed - Codex PTY workers now receive initial Relay tasks in one bulk write, preventing full-screen input redraws from delaying task submission for minutes. +- `agent-relay node up` now binds an OS-assigned API port atomically by default, preventing concurrent Fleet nodes from racing over a probed port; `AGENT_RELAY_BROKER_PORT` remains an explicit stable-port override. +- Newly connected Fleet brokers now advertise their spawn/release handlers immediately, so the first remote spawn is dispatched instead of remaining queued until load changes. +- `agent-relay fleet spawn --session-ref` now passes the requested session to Claude and Codex as a real resume operation, and a released agent name can be reused immediately instead of being suppressed as a duplicate spawn. ## [11.1.1] - 2026-07-23 diff --git a/crates/broker/src/cli/mod.rs b/crates/broker/src/cli/mod.rs index cc6ba4c0f..1f1f119cf 100644 --- a/crates/broker/src/cli/mod.rs +++ b/crates/broker/src/cli/mod.rs @@ -243,7 +243,7 @@ pub(crate) struct InitCommand { #[arg(long, default_value = "general")] pub(crate) channels: String, - /// Optional HTTP API port for dashboard proxy (0 = disabled) + /// Optional HTTP API port for dashboard proxy (0 = atomically OS-assigned). #[arg(long, default_value = "0")] pub(crate) api_port: u16, diff --git a/crates/broker/src/node_control.rs b/crates/broker/src/node_control.rs index ee7477139..edc50a974 100644 --- a/crates/broker/src/node_control.rs +++ b/crates/broker/src/node_control.rs @@ -1216,6 +1216,12 @@ fn handle_disconnected_command( resume_cursor, }) => { load.max_agents = manifest.max_agents.unwrap_or(load.max_agents); + // This control client is the broker provider, which owns the + // node's spawn/release capacity as soon as its socket connects. + // A fresh node has no workers yet, so no load transition would + // otherwise publish the first `handlers_live=true` snapshot and + // the engine would queue the very first spawn indefinitely. + load.handlers_live = true; *registration = Some(build_node_register( &manifest, &config.node_id, @@ -1543,6 +1549,7 @@ async fn run_connected_once( match command { Some(FleetControlCommand::RegisterNode { manifest, resume_cursor }) => { load.max_agents = manifest.max_agents.unwrap_or(load.max_agents); + load.handlers_live = true; let mut next = build_node_register(&manifest, &config.node_id, &config.node_name, &config.broker_version, resume_cursor); next.provider = Some(provider.clone()); node_register = next.clone(); @@ -2830,7 +2837,15 @@ mod tests { let register = next_node_to_server(&mut ws).await; assert!(matches!(register, BrokerToRelaycast::NodeRegister(_))); let heartbeat = next_node_to_server(&mut ws).await; - assert!(matches!(heartbeat, BrokerToRelaycast::NodeHeartbeat(_))); + match heartbeat { + BrokerToRelaycast::NodeHeartbeat(heartbeat) => { + assert!( + heartbeat.handlers_live, + "the broker provider must advertise capacity before the first spawn" + ); + } + other => panic!("expected initial node heartbeat, got {other:?}"), + } ws.send(Message::Text( serde_json::to_string(&RelaycastToBroker::Deliver(Deliver { diff --git a/crates/broker/src/runtime/fleet.rs b/crates/broker/src/runtime/fleet.rs index a4414b3b8..be2daaf1e 100644 --- a/crates/broker/src/runtime/fleet.rs +++ b/crates/broker/src/runtime/fleet.rs @@ -415,6 +415,13 @@ impl BrokerRuntime { // not correlated to the agent and a resumable `spawn:` (when // `harnessConfig.session_id` is set) silently becomes a fresh spawn. let session_ref = super::relaycast_events::relaycast_spawn_session_ref(&ws_value); + // `action.invoke` is the authoritative control request, not a + // workspace-firehose echo of a local spawn. Mark it with the same + // control key the echo guard derives so a later release + respawn of + // the same agent name is not suppressed by the five-minute + // name-scoped echo cache. + let action_control_dedup_key = + relaycast_spawn_control_dedup_key(workspace_id.as_str(), name.as_str()); super::relaycast_events::spawn_worker_from_request( name.clone(), @@ -425,7 +432,7 @@ impl BrokerRuntime { exit_after_task, &ws_value, &workspace_id, - None, + Some(&action_control_dedup_key), &workspace_state, &mut self.workers, &mut self.state, @@ -1408,6 +1415,21 @@ mod tests { .expect("valid explicit flag")); } + #[test] + fn action_invoke_spawn_control_key_allows_immediate_name_reuse() { + let local_key = relaycast_spawn_control_dedup_key("ws_1", "worker-a"); + + // Each node action is already correlated by its invocation id. Passing + // the matching control key tells the legacy firehose echo guard not to + // consume or reject the reusable worker name. + for _ in 0..2 { + assert!(!relaycast_ws_should_apply_local_spawn_echo_dedup( + Some(local_key.as_str()), + &local_key, + )); + } + } + #[test] fn fleet_initial_session_ref_prefers_explicit_spec_session() { let spec = test_agent_spec(Some("session-spec"), Some("session-harness")); @@ -1425,17 +1447,16 @@ mod tests { #[tokio::test] async fn action_invoke_spawn_seeds_authoritative_cursor_before_resumed_delivery() { - // An `action.invoke` spawn carrying `harnessConfig.session_id` must - // forward a non-None session_ref (and the invocation id) into the node - // `agent.register` it emits, so the spawn resumes the session and the - // invocation is correlated to the agent (Bug 2). Previously both were - // hardcoded to None on this path. + // The Fleet CLI sends `session_ref` at the top level. It must be + // forwarded with the invocation id into the node `agent.register`, so + // the spawn resumes the session and the invocation is correlated to + // the agent. let ws_value = json!({ + "session_ref": "sess-resume-7", "agent": { "harnessConfig": { "runtime": "pty", "command": "codex", - "sessionId": "sess-resume-7", } } }); @@ -1443,7 +1464,7 @@ mod tests { assert_eq!( session_ref.as_deref(), Some("sess-resume-7"), - "session ref must be derived from harnessConfig.session_id" + "session ref must be derived from the action input" ); // Drive the exact registration step the spawn path uses and capture the @@ -1659,6 +1680,84 @@ mod tests { None ); } + + #[test] + fn relaycast_spawn_session_ref_supports_action_and_harness_shapes() { + let explicit = json!({ + "session_ref": " session-explicit ", + "agent": { + "harnessConfig": { + "runtime": "pty", + "command": "codex", + "sessionId": "session-harness", + } + } + }); + assert_eq!( + super::super::relaycast_events::relaycast_spawn_session_ref(&explicit).as_deref(), + Some("session-explicit"), + "the Fleet action field must take precedence over its compatibility fallback" + ); + + let nested_camel = json!({"agent": {"sessionRef": "session-nested"}}); + assert_eq!( + super::super::relaycast_events::relaycast_spawn_session_ref(&nested_camel).as_deref(), + Some("session-nested") + ); + + let harness_only = json!({ + "agent": { + "harnessConfig": { + "runtime": "pty", + "command": "codex", + "sessionId": "session-harness", + } + } + }); + assert_eq!( + super::super::relaycast_events::relaycast_spawn_session_ref(&harness_only).as_deref(), + Some("session-harness") + ); + } + + #[test] + fn relaycast_spawn_spec_session_id_prefers_requested_resume() { + assert_eq!( + super::super::relaycast_events::relaycast_spawn_spec_session_id( + "codex", + Some(" requested-session "), + Some("harness-session"), + ) + .as_deref(), + Some("requested-session") + ); + assert_eq!( + super::super::relaycast_events::relaycast_spawn_spec_session_id( + "claude", + None, + Some(" harness-session "), + ) + .as_deref(), + Some("harness-session") + ); + assert_eq!( + super::super::relaycast_events::relaycast_spawn_spec_session_id( + "codex", + Some(" "), + None, + ), + None + ); + assert_eq!( + super::super::relaycast_events::relaycast_spawn_spec_session_id( + "pool", + Some("metadata-only-session"), + None, + ), + None, + "custom capacity harnesses retain session_ref metadata without receiving Codex/Claude argv" + ); + } #[tokio::test] async fn prune_fleet_inventory_entry_publishes_without_removed_agent() { let (tx, mut rx) = mpsc::channel(4); diff --git a/crates/broker/src/runtime/relaycast_events.rs b/crates/broker/src/runtime/relaycast_events.rs index d1660d048..294bd61b8 100644 --- a/crates/broker/src/runtime/relaycast_events.rs +++ b/crates/broker/src/runtime/relaycast_events.rs @@ -25,13 +25,37 @@ impl BrokerRuntime { } } -/// Derive the initial session ref for a spawn request from its `ws_value`, -/// mirroring `spawn_worker_from_request`'s own `session_id` derivation (the -/// harness config's `session_id`). Returns `None` when the harness config is -/// absent or invalid, or carries no session id. Used by the node `action.invoke` -/// spawn path to forward a resumable session ref into `agent.register`, matching -/// the sidecar's `fleet_initial_session_ref(&spec)`. +/// Derive the initial session ref for a spawn request from its `ws_value`. +/// +/// Fleet CLI/API callers send `session_ref` as a top-level action input, while +/// older firehose-style payloads may carry it under `agent` or in +/// `harnessConfig.session_id`. Prefer the explicit action field and retain the +/// harness fallback so both shapes resume the worker and register the same +/// session with the node control plane. pub(super) fn relaycast_spawn_session_ref(ws_value: &Value) -> Option { + let explicit = ["session_ref", "sessionRef"] + .iter() + .find_map(|key| { + ws_value + .get(*key) + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + }) + .or_else(|| { + let agent = ws_value.get("agent")?; + ["session_ref", "sessionRef"].iter().find_map(|key| { + agent + .get(*key) + .and_then(Value::as_str) + .map(str::trim) + .filter(|value| !value.is_empty()) + }) + }); + if let Some(session_ref) = explicit { + return Some(session_ref.to_string()); + } + relaycast_harness_config(ws_value) .ok() .flatten() @@ -40,6 +64,30 @@ pub(super) fn relaycast_spawn_session_ref(ws_value: &Value) -> Option { .map(ToOwned::to_owned) } +pub(super) fn relaycast_spawn_spec_session_id( + cli: &str, + session_ref: Option<&str>, + harness_session_id: Option<&str>, +) -> Option { + let normalized_cli = crate::cli::command_parse::normalize_cli_name(cli); + let supports_resume = normalized_cli == "codex" + || normalized_cli == "claude" + || normalized_cli.starts_with("claude:"); + supports_resume + .then_some(session_ref) + .flatten() + .and_then(|value| { + let value = value.trim(); + (!value.is_empty()).then(|| value.to_string()) + }) + .or_else(|| { + harness_session_id.and_then(|value| { + let value = value.trim(); + (!value.is_empty()).then(|| value.to_string()) + }) + }) +} + fn relaycast_harness_config(value: &Value) -> Result, String> { let agent = value.get("agent"); let harness_id = agent @@ -355,10 +403,15 @@ pub(super) async fn spawn_worker_from_request( .as_ref() .map(ResolvedHarnessConfig::runtime) .unwrap_or(AgentRuntime::Pty); - let session_id = harness_config + let harness_session_id = harness_config .as_ref() .and_then(ResolvedHarnessConfig::session_id) .map(ToOwned::to_owned); + let session_id = relaycast_spawn_spec_session_id( + &cli, + session_ref.as_deref(), + harness_session_id.as_deref(), + ); tracing::info!(name = %name, cli = %cli, task = ?task, channel = ?channel, "handling spawn request from relaycast WS"); let channels = channel diff --git a/crates/broker/src/worker.rs b/crates/broker/src/worker.rs index a77febbe2..8b6786824 100644 --- a/crates/broker/src/worker.rs +++ b/crates/broker/src/worker.rs @@ -350,16 +350,25 @@ impl WorkerRegistry { spec.model = Some(model); } let mut harness_session_args = Vec::new(); - if spec.session_id.is_none() { + if let Some(session_id) = spec.session_id.as_deref() { + apply_requested_session_reference( + &cli_lower, + session_id, + &mut effective_args, + &mut harness_session_args, + )?; + } else { if is_claude { spec.session_id = prepare_claude_session_args(&mut effective_args); } else if is_codex { match codex_session_reference(&effective_args) { - CodexSessionReference::Known(thread_id) => { + CodexSessionReference::Resume(thread_id) => { spec.session_id = Some(thread_id); } - CodexSessionReference::Unknown => {} - CodexSessionReference::None => { + CodexSessionReference::Fork(_) + | CodexSessionReference::AmbiguousVariadicImage + | CodexSessionReference::Unknown => {} + CodexSessionReference::None | CodexSessionReference::VariadicImage => { if codex_has_positional_arg(&effective_args) { tracing::debug!( worker = %spec.name, @@ -454,27 +463,16 @@ impl WorkerRegistry { spec.model = Some(model.clone()); } - let has_extra = bypass_flag.is_some() - || model_flag.is_some() - || !effective_args.is_empty() - || !mcp_args.is_empty() - || !harness_session_args.is_empty(); - if has_extra { + let pty_cli_args = ordered_pty_cli_args( + bypass_flag, + model_flag.as_deref(), + &mcp_args, + &effective_args, + &harness_session_args, + ); + if !pty_cli_args.is_empty() { command.arg("--"); - if let Some(flag) = bypass_flag { - command.arg(flag); - } - if let Some(ref model) = model_flag { - command.arg("--model"); - command.arg(model); - } - for arg in &mcp_args { - command.arg(arg); - } - for arg in &effective_args { - command.arg(arg); - } - for arg in &harness_session_args { + for arg in &pty_cli_args { command.arg(arg); } } @@ -582,16 +580,26 @@ impl WorkerRegistry { spec.model = Some(model); } let mut harness_session_args = Vec::new(); - if spec.session_id.is_none() { + if let Some(session_id) = spec.session_id.as_deref() { + apply_requested_session_reference( + &cli_lower, + session_id, + &mut effective_args, + &mut harness_session_args, + )?; + } else { if is_claude { spec.session_id = prepare_claude_session_args(&mut effective_args); } else if is_codex { match codex_session_reference(&effective_args) { - CodexSessionReference::Known(thread_id) => { + CodexSessionReference::Resume(thread_id) => { spec.session_id = Some(thread_id); } - CodexSessionReference::Unknown => {} - CodexSessionReference::None => { + CodexSessionReference::Fork(_) + | CodexSessionReference::AmbiguousVariadicImage + | CodexSessionReference::Unknown => {} + CodexSessionReference::None + | CodexSessionReference::VariadicImage => { if codex_has_positional_arg(&effective_args) { tracing::debug!( worker = %spec.name, @@ -688,27 +696,16 @@ impl WorkerRegistry { spec.model = Some(model.clone()); } - let has_extra = bypass_flag.is_some() - || model_flag.is_some() - || !effective_args.is_empty() - || !mcp_args.is_empty() - || !harness_session_args.is_empty(); - if has_extra { + let pty_cli_args = ordered_pty_cli_args( + bypass_flag, + model_flag.as_deref(), + &mcp_args, + &effective_args, + &harness_session_args, + ); + if !pty_cli_args.is_empty() { command.arg("--"); - if let Some(flag) = bypass_flag { - command.arg(flag); - } - if let Some(ref model) = model_flag { - command.arg("--model"); - command.arg(model); - } - for arg in &mcp_args { - command.arg(arg); - } - for arg in &effective_args { - command.arg(arg); - } - for arg in &harness_session_args { + for arg in &pty_cli_args { command.arg(arg); } } @@ -1242,7 +1239,10 @@ fn is_loopback_endpoint_host(endpoint: &reqwest::Url) -> bool { #[derive(Debug, Clone, PartialEq, Eq)] enum CodexSessionReference { - Known(String), + Resume(String), + Fork(String), + VariadicImage, + AmbiguousVariadicImage, Unknown, None, } @@ -1269,6 +1269,93 @@ fn prepare_claude_session_args(args: &mut Vec) -> Option { Some(session_id) } +fn ordered_pty_cli_args( + bypass_flag: Option<&str>, + model: Option<&str>, + mcp_args: &[String], + effective_args: &[String], + harness_session_args: &[String], +) -> Vec { + let mut args = Vec::new(); + if let Some(flag) = bypass_flag { + args.push(flag.to_string()); + } + if let Some(model) = model { + args.push("--model".to_string()); + args.push(model.to_string()); + } + args.extend_from_slice(mcp_args); + // Codex options such as --image are variadic and can consume an appended + // `resume `. Put the broker-owned subcommand before user options; + // Codex accepts its resume options after the session positional. + args.extend_from_slice(harness_session_args); + args.extend_from_slice(effective_args); + args +} + +fn apply_requested_session_reference( + cli_lower: &str, + session_id: &str, + args: &mut Vec, + harness_session_args: &mut Vec, +) -> Result<()> { + let session_id = session_id.trim(); + if session_id.is_empty() { + anyhow::bail!("session_ref must not be empty"); + } + + if cli_lower == "claude" || cli_lower.starts_with("claude:") { + if let Some(existing) = + cli_flag_value(args, "--resume").or_else(|| cli_flag_value(args, "-r")) + { + if existing != session_id { + anyhow::bail!( + "session_ref conflicts with the Claude session argument already configured" + ); + } + return Ok(()); + } + if cli_flag_present( + args, + &["--session-id", "--resume", "-r", "--continue", "-c"], + ) { + anyhow::bail!("session_ref requires an explicit Claude session id"); + } + args.push("--resume".to_string()); + args.push(session_id.to_string()); + return Ok(()); + } + + if cli_lower == "codex" { + match codex_session_reference(args) { + CodexSessionReference::Resume(existing) if existing == session_id => return Ok(()), + CodexSessionReference::Resume(_) => { + anyhow::bail!( + "session_ref conflicts with the Codex session argument already configured" + ); + } + CodexSessionReference::Fork(_) => { + anyhow::bail!("session_ref cannot be combined with a Codex fork"); + } + CodexSessionReference::AmbiguousVariadicImage => { + anyhow::bail!( + "session_ref cannot safely disambiguate Codex resume/fork values after --image" + ); + } + CodexSessionReference::Unknown => { + anyhow::bail!("session_ref requires an explicit Codex session id"); + } + CodexSessionReference::None | CodexSessionReference::VariadicImage => { + harness_session_args.push("resume".to_string()); + harness_session_args.push(session_id.to_string()); + return Ok(()); + } + } + } + + anyhow::bail!("session_ref resume is supported only for Claude and Codex PTY harnesses"); +} + fn codex_session_reference(args: &[String]) -> CodexSessionReference { let mut index = 0; let mut skip_next = false; @@ -1282,6 +1369,16 @@ fn codex_session_reference(args: &[String]) -> CodexSessionReference { if arg == "--" { return CodexSessionReference::None; } + if codex_is_variadic_image_arg(arg) { + return if args[index + 1..] + .iter() + .any(|value| value == "resume" || value == "fork") + { + CodexSessionReference::AmbiguousVariadicImage + } else { + CodexSessionReference::VariadicImage + }; + } if codex_flag_consumes_next_arg(arg) { if args.get(index + 1).is_none() { return CodexSessionReference::Unknown; @@ -1290,6 +1387,15 @@ fn codex_session_reference(args: &[String]) -> CodexSessionReference { index += 1; continue; } + if arg.starts_with('-') { + if arg.contains('=') || codex_flag_without_value(arg) { + index += 1; + continue; + } + // An unknown option may consume the following token. Fail closed + // instead of mistaking that value for a resume/fork subcommand. + return CodexSessionReference::Unknown; + } if arg == "resume" || arg == "fork" { let Some(next) = args.get(index + 1).map(String::as_str) else { return CodexSessionReference::Unknown; @@ -1297,7 +1403,11 @@ fn codex_session_reference(args: &[String]) -> CodexSessionReference { if next == "--last" || next.starts_with('-') { return CodexSessionReference::Unknown; } - return CodexSessionReference::Known(next.to_string()); + return if arg == "resume" { + CodexSessionReference::Resume(next.to_string()) + } else { + CodexSessionReference::Fork(next.to_string()) + }; } index += 1; } @@ -1314,6 +1424,12 @@ fn codex_has_positional_arg(args: &[String]) -> bool { if arg == "--" { return true; } + if codex_is_variadic_image_arg(arg) { + // At the root command, --image consumes subsequent positional + // values. With a broker-owned resume prefix those same options are + // safely interpreted by the resume subcommand. + return false; + } if codex_flag_consumes_next_arg(arg) { skip_next = true; continue; @@ -1335,14 +1451,44 @@ fn codex_flag_consumes_next_arg(arg: &str) -> bool { "--model" | "-m" | "--profile" + | "-p" | "--config" | "-c" + | "--enable" + | "--disable" + | "--remote" + | "--remote-auth-token-env" | "--sandbox" | "-s" + | "--local-provider" | "--ask-for-approval" + | "-a" | "--approval-policy" | "--cd" + | "-C" | "--cwd" + | "--add-dir" + ) +} + +fn codex_is_variadic_image_arg(arg: &str) -> bool { + arg == "--image" || arg == "-i" || arg.starts_with("--image=") || arg.starts_with("-i=") +} + +fn codex_flag_without_value(arg: &str) -> bool { + matches!( + arg, + "--strict-config" + | "--oss" + | "--dangerously-bypass-approvals-and-sandbox" + | "--dangerously-bypass-hook-trust" + | "--full-auto" + | "--search" + | "--no-alt-screen" + | "--help" + | "-h" + | "--version" + | "-V" ) } @@ -2023,6 +2169,215 @@ mod tests { assert_eq!(args, vec!["--resume=session-2".to_string()]); } + #[test] + fn requested_session_reference_adds_claude_resume_args() { + let mut args = vec!["--model".to_string(), "claude-opus-4-1".to_string()]; + let mut harness_session_args = Vec::new(); + + apply_requested_session_reference( + "claude", + "session-claude-1", + &mut args, + &mut harness_session_args, + ) + .expect("Claude session resume"); + + assert_eq!( + args, + vec![ + "--model".to_string(), + "claude-opus-4-1".to_string(), + "--resume".to_string(), + "session-claude-1".to_string(), + ] + ); + assert!(harness_session_args.is_empty()); + } + + #[test] + fn requested_session_reference_adds_codex_resume_args() { + let mut args = vec!["--profile".to_string(), "work".to_string()]; + let mut harness_session_args = Vec::new(); + + apply_requested_session_reference( + "codex", + "thread-codex-1", + &mut args, + &mut harness_session_args, + ) + .expect("Codex session resume"); + + assert_eq!(args, vec!["--profile".to_string(), "work".to_string()]); + assert_eq!( + harness_session_args, + vec!["resume".to_string(), "thread-codex-1".to_string()] + ); + } + + #[test] + fn requested_session_reference_does_not_treat_flag_value_as_codex_resume() { + let mut args = vec!["--enable".to_string(), "resume".to_string()]; + let mut harness_session_args = Vec::new(); + + apply_requested_session_reference( + "codex", + "thread-codex-1", + &mut args, + &mut harness_session_args, + ) + .expect("Codex session resume"); + + assert_eq!(args, vec!["--enable".to_string(), "resume".to_string()]); + assert_eq!( + harness_session_args, + vec!["resume".to_string(), "thread-codex-1".to_string()] + ); + } + + #[test] + fn requested_session_reference_precedes_variadic_codex_image_args() { + let mut args = vec!["--image".to_string(), "/tmp/review.png".to_string()]; + let mut harness_session_args = Vec::new(); + + apply_requested_session_reference( + "codex", + "thread-codex-1", + &mut args, + &mut harness_session_args, + ) + .expect("Codex session resume"); + + let ordered = ordered_pty_cli_args( + Some("--dangerously-bypass-approvals-and-sandbox"), + Some("gpt-5.4"), + &[ + "-c".to_string(), + "mcp_servers.agent-relay.enabled=true".to_string(), + ], + &args, + &harness_session_args, + ); + assert_eq!( + ordered, + vec![ + "--dangerously-bypass-approvals-and-sandbox", + "--model", + "gpt-5.4", + "-c", + "mcp_servers.agent-relay.enabled=true", + "resume", + "thread-codex-1", + "--image", + "/tmp/review.png", + ] + ); + } + + #[test] + fn requested_session_reference_accepts_matching_resume_before_codex_images() { + let mut args = vec![ + "resume".to_string(), + "thread-codex-1".to_string(), + "--image".to_string(), + "/tmp/review.png".to_string(), + ]; + let mut harness_session_args = Vec::new(); + + apply_requested_session_reference( + "codex", + "thread-codex-1", + &mut args, + &mut harness_session_args, + ) + .expect("matching explicit Codex resume"); + + assert!(harness_session_args.is_empty()); + } + + #[test] + fn requested_session_reference_rejects_ambiguous_variadic_codex_image_values() { + let mut args = vec![ + "--image".to_string(), + "/tmp/review.png".to_string(), + "resume".to_string(), + ]; + let mut harness_session_args = Vec::new(); + + let error = apply_requested_session_reference( + "codex", + "thread-codex-1", + &mut args, + &mut harness_session_args, + ) + .expect_err("ambiguous variadic values must fail closed"); + + assert!(error.to_string().contains("after --image")); + assert!(harness_session_args.is_empty()); + } + + #[test] + fn requested_session_reference_rejects_ambiguous_codex_option() { + let mut args = vec!["--future-option".to_string(), "resume".to_string()]; + let mut harness_session_args = Vec::new(); + + let error = apply_requested_session_reference( + "codex", + "thread-codex-1", + &mut args, + &mut harness_session_args, + ) + .expect_err("unknown Codex option arity must fail closed"); + + assert!(error + .to_string() + .contains("requires an explicit Codex session id")); + assert!(harness_session_args.is_empty()); + } + + #[test] + fn requested_session_reference_rejects_conflicting_cli_session() { + let mut args = vec!["resume".to_string(), "thread-other".to_string()]; + let mut harness_session_args = Vec::new(); + + let error = apply_requested_session_reference( + "codex", + "thread-requested", + &mut args, + &mut harness_session_args, + ) + .expect_err("conflicting resume must fail closed"); + + assert!(error.to_string().contains("conflicts")); + assert!(harness_session_args.is_empty()); + } + + #[test] + fn requested_session_reference_rejects_new_or_forked_sessions() { + let mut claude_args = vec!["--session-id".to_string(), "session-requested".to_string()]; + let mut claude_harness_args = Vec::new(); + let claude_error = apply_requested_session_reference( + "claude", + "session-requested", + &mut claude_args, + &mut claude_harness_args, + ) + .expect_err("a requested session must resume instead of starting"); + assert!(claude_error + .to_string() + .contains("explicit Claude session id")); + + let mut codex_args = vec!["fork".to_string(), "thread-requested".to_string()]; + let mut codex_harness_args = Vec::new(); + let codex_error = apply_requested_session_reference( + "codex", + "thread-requested", + &mut codex_args, + &mut codex_harness_args, + ) + .expect_err("a requested session must resume instead of forking"); + assert!(codex_error.to_string().contains("Codex fork")); + } + #[test] fn codex_session_reference_detects_resume_and_fork_ids() { assert_eq!( @@ -2032,11 +2387,11 @@ mod tests { "resume".into(), "thread-1".into() ]), - CodexSessionReference::Known("thread-1".to_string()) + CodexSessionReference::Resume("thread-1".to_string()) ); assert_eq!( codex_session_reference(&["fork".into(), "thread-2".into()]), - CodexSessionReference::Known("thread-2".to_string()) + CodexSessionReference::Fork("thread-2".to_string()) ); assert_eq!( codex_session_reference(&["resume".into(), "--last".into()]), @@ -2048,6 +2403,28 @@ mod tests { codex_session_reference(&["--profile".into()]), CodexSessionReference::Unknown ); + assert_eq!( + codex_session_reference(&[ + "--image".into(), + "/tmp/review.png".into(), + "resume".into(), + "thread-3".into(), + ]), + CodexSessionReference::AmbiguousVariadicImage + ); + assert_eq!( + codex_session_reference(&["--image".into(), "/tmp/review.png".into()]), + CodexSessionReference::VariadicImage + ); + assert_eq!( + codex_session_reference(&[ + "resume".into(), + "thread-4".into(), + "--image".into(), + "/tmp/review.png".into(), + ]), + CodexSessionReference::Resume("thread-4".to_string()) + ); } #[test] @@ -2064,6 +2441,10 @@ mod tests { "Fix the bug".into(), ])); assert!(codex_has_positional_arg(&["exec".into()])); + assert!(!codex_has_positional_arg(&[ + "--image".into(), + "/tmp/review.png".into(), + ])); } #[test] diff --git a/packages/cli/README.md b/packages/cli/README.md index 08d643015..4d12f1550 100644 --- a/packages/cli/README.md +++ b/packages/cli/README.md @@ -62,6 +62,13 @@ agent-relay fleet spawn codex \ --task "Use https://agentrelay.com/skill, ACK over Relay, then wait for details." \ --node sf-mini +# Resume a known Claude/Codex CLI session on its origin node. +agent-relay fleet spawn codex \ + --name api-worker \ + --task "Resume over Relay and continue the prior task." \ + --node sf-mini \ + --session-ref + # Omit --node for automatic eligible-node placement. agent-relay fleet spawn codex --name api-worker --task "Review the current diff." @@ -78,6 +85,11 @@ set `RELAY_AGENT_TOKEN` to the token returned by `agent-relay agent register `. Automatic placement and release need only the workspace key. +`--session-ref` is a real CLI resume, not a logical collaboration label. Pass +the actual Claude session ID or Codex thread ID and target its origin node. +Omit it to start a new CLI session. The project’s Agent Relay workspace remains +pinned independently until you explicitly create or select another workspace. + To run as a Cloud-managed node, first redeem a one-time enrollment token, then start the node: ```bash @@ -85,6 +97,74 @@ agent-relay cloud enroll --token ocl_node_enr_... agent-relay node up ``` +## Cloud multiplayer rooms + +Cloud room membership is scoped to one Relay workspace. Every v1 invite creates +a trusted full room participant: they receive their own revocable Relaycast +human credential and may use all ordinary agent-level collaboration actions. +The workspace key itself is never shared, so owner-key administration and Agent +Relay Cloud organization administration remain owner-only. + +```bash +# Owner: invite and manage people in this workspace. +agent-relay cloud room invite \ + --workspace rw_7ccfea89 \ + --email teammate@example.com \ + --token-file ./teammate.room-invite +agent-relay cloud room invites --workspace rw_7ccfea89 +agent-relay cloud room members --workspace rw_7ccfea89 + +# Share the owner-only token file over a secure channel. The invitee keeps the +# token out of shell history and process arguments. +# Tokens use the consumer-neutral relay_room_inv_ prefix followed by exactly +# 43 URL-safe characters. +read -rs ROOM_INVITATION_TOKEN +printf '%s' "$ROOM_INVITATION_TOKEN" | + agent-relay cloud room accept --token-stdin +unset ROOM_INVITATION_TOKEN + +# Trusted clients establish one stable session per device. +# --json intentionally includes the participant credential; capture it in +# memory and do not log or persist it. +agent-relay cloud room session \ + --workspace rw_7ccfea89 \ + --device-id client-macbook \ + --json + +# Explicitly ending or replacing the device session revokes the old scoped token. +agent-relay cloud room revoke-session \ + --workspace rw_7ccfea89 \ + --device-id client-macbook + +# Participants use their scoped token for agent-level Relaycast operations; an +# ambient owner workspace key is never consulted when --token is present. +agent-relay agent presence \ + --token at_live_... \ + --base-url https://cast.agentrelay.com + +# Owner: revoke access and active room sessions. +agent-relay cloud room remove-member --workspace rw_7ccfea89 +``` + +There is no room-specific integration grant or credential service. Connect the +workspace provider through the existing Cloud integration API, then use the +normal Relayfile workflow for setup, mounts, reads, and writebacks: + +```bash +# Owner: discover or connect a provider through Cloud. +agent-relay cloud integration catalog +agent-relay cloud integration connect linear --workspace rw_7ccfea89 +agent-relay cloud integration connections --workspace rw_7ccfea89 + +# Member clients use Relayfile directly, including its OAuth/backend selection +# and durable writeback queue. +relayfile integration available +relayfile integration connect linear +RELAYFILE_LOCAL_DIR="$PWD/.integrations" relayfile setup +RELAYFILE_LOCAL_DIR="$PWD/.integrations" relayfile status +RELAYFILE_LOCAL_DIR="$PWD/.integrations" relayfile writeback status +``` + `local` remains as a deprecated hidden alias of `node` (it prints a one-time warning). Node workflow runs use Relayflows for YAML, TypeScript, and Python workflow files. diff --git a/packages/cli/src/cli/bootstrap.test.ts b/packages/cli/src/cli/bootstrap.test.ts index a3289c46a..e28920d6c 100644 --- a/packages/cli/src/cli/bootstrap.test.ts +++ b/packages/cli/src/cli/bootstrap.test.ts @@ -62,6 +62,18 @@ const expectedLeafCommands = [ 'cloud worker start', 'cloud worker status', 'cloud worker logs', + 'cloud room invite', + 'cloud room invites', + 'cloud room revoke-invite', + 'cloud room members', + 'cloud room remove-member', + 'cloud room session', + 'cloud room revoke-session', + 'cloud room accept', + 'cloud integration catalog', + 'cloud integration connections', + 'cloud integration connect', + 'cloud integration disconnect', // workspace 'workspace create', 'workspace active', @@ -74,6 +86,8 @@ const expectedLeafCommands = [ 'agent list', 'agent add', 'agent remove', + 'agent me', + 'agent presence', // channel 'channel create', 'channel list', diff --git a/packages/cli/src/cli/commands/agent.test.ts b/packages/cli/src/cli/commands/agent.test.ts new file mode 100644 index 000000000..fdffd946a --- /dev/null +++ b/packages/cli/src/cli/commands/agent.test.ts @@ -0,0 +1,57 @@ +import { Command } from 'commander'; +import { describe, expect, it, vi } from 'vitest'; + +import { registerAgentCommands } from './agent.js'; + +function createHarness() { + const agentRelay = { + agents: { + me: vi.fn(async () => ({ id: 'agent_1', name: 'room-human' })), + presence: vi.fn(async () => [{ agent: 'room-human', status: 'online' }]), + }, + }; + const createAgentRelay = vi.fn(() => agentRelay); + const createWorkspaceRelay = vi.fn(); + const program = new Command(); + program.exitOverride(); + registerAgentCommands(program, { + createAgentRelay: createAgentRelay as never, + createWorkspaceRelay: createWorkspaceRelay as never, + log: vi.fn(), + error: vi.fn(), + exit: ((code: number) => { + throw new Error(`exit:${code}`); + }) as never, + }); + return { program, agentRelay, createAgentRelay, createWorkspaceRelay }; +} + +describe('agent-scoped identity commands', () => { + it.each([ + ['me', 'me'], + ['presence', 'presence'], + ] as const)('uses the agent credential for agent %s', async (command, method) => { + const { program, agentRelay, createAgentRelay, createWorkspaceRelay } = createHarness(); + + await program.parseAsync([ + 'node', + 'agent-relay', + 'agent', + command, + '--token', + 'at_live_room_human', + '--workspace-key', + 'rk_live_owner_must_not_win', + '--base-url', + 'https://cast.agentrelay.test', + ]); + + expect(createAgentRelay).toHaveBeenCalledWith({ + token: 'at_live_room_human', + workspaceKey: 'rk_live_owner_must_not_win', + baseUrl: 'https://cast.agentrelay.test', + }); + expect(createWorkspaceRelay).not.toHaveBeenCalled(); + expect(agentRelay.agents[method]).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/cli/src/cli/commands/agent.ts b/packages/cli/src/cli/commands/agent.ts index a7d906a49..324691ed1 100644 --- a/packages/cli/src/cli/commands/agent.ts +++ b/packages/cli/src/cli/commands/agent.ts @@ -53,6 +53,24 @@ export function registerAgentCommands( }); }); + addSdkOptions(group.command('me').description('Show the current agent identity')).action( + async (opts: Record) => { + await runSdk(deps, async () => { + const relay = deps.createAgentRelay(sdkOptionsFromOpts(opts)); + printJson(deps, await relay.agents.me()); + }); + } + ); + + addSdkOptions(group.command('presence').description('List visible agent presence')).action( + async (opts: Record) => { + await runSdk(deps, async () => { + const relay = deps.createAgentRelay(sdkOptionsFromOpts(opts)); + printJson(deps, await relay.agents.presence()); + }); + } + ); + addSdkOptions( group .command('add') diff --git a/packages/cli/src/cli/commands/cloud-integration.test.ts b/packages/cli/src/cli/commands/cloud-integration.test.ts new file mode 100644 index 000000000..da896f1cf --- /dev/null +++ b/packages/cli/src/cli/commands/cloud-integration.test.ts @@ -0,0 +1,290 @@ +import { Command } from 'commander'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { registerCloudIntegrationCommands } from './cloud-integration.js'; +import type { CloudDependencies } from './cloud.js'; + +vi.mock('@agent-relay/cloud', () => ({ + defaultApiUrl: () => 'https://cloud.test', +})); + +type Deps = Pick; + +const auth = { + apiUrl: 'https://cloud.test', + accessToken: 'access-secret', + refreshToken: 'refresh-secret', + accessTokenExpiresAt: '2999-01-01T00:00:00.000Z', +}; + +function response(value: unknown, status = 200): Response { + return new Response(value === null ? null : JSON.stringify(value), { + status, + headers: value === null ? undefined : { 'content-type': 'application/json' }, + }); +} + +function harness() { + const exit = vi.fn((code: number) => { + throw new Error(`exit:${code}`); + }) as unknown as Deps['exit']; + const deps: Deps = { + log: vi.fn(), + error: vi.fn(), + exit, + ensureCloudSession: vi.fn(async () => ({ auth, client: {} as never })) as Deps['ensureCloudSession'], + authorizedApiFetch: vi.fn(async () => ({ + response: response({}), + auth, + })) as Deps['authorizedApiFetch'], + }; + const program = new Command(); + program.exitOverride(); + const cloud = program.command('cloud'); + registerCloudIntegrationCommands(cloud, deps); + return { program, deps, integration: cloud.commands[0] }; +} + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe('registerCloudIntegrationCommands', () => { + it('exposes only the existing Cloud connection lifecycle', () => { + const { integration } = harness(); + expect(integration.commands.map((command) => command.name())).toEqual([ + 'catalog', + 'connections', + 'connect', + 'disconnect', + ]); + }); + + it('lists dynamic providers without requiring room-specific capabilities', async () => { + const { program, deps } = harness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: response({ + providers: [ + { + id: 'linear', + displayName: 'Linear', + backends: ['nango'], + apiKey: 'must-not-print', + }, + ], + version: 'abcdef123456', + }), + auth, + }); + + await program.parseAsync(['node', 'agent-relay', 'cloud', 'integration', 'catalog', '--json']); + + expect(deps.authorizedApiFetch).toHaveBeenCalledWith( + auth, + '/api/v1/integrations/catalog?dynamic=true', + { method: 'GET' }, + { interactive: false } + ); + const output = vi.mocked(deps.log).mock.calls.flat().join('\n'); + expect(output).toContain('"id": "linear"'); + expect(output).not.toContain('must-not-print'); + }); + + it('filters the catalog locally by backend and search text', async () => { + const { program, deps } = harness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: response({ + providers: [ + { id: 'linear', displayName: 'Linear', backends: ['nango'] }, + { id: 'github', displayName: 'GitHub', backends: ['composio'] }, + ], + version: '1', + }), + auth, + }); + + await program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'integration', + 'catalog', + '--search', + 'git', + '--backend', + 'composio', + '--json', + ]); + + const output = vi.mocked(deps.log).mock.calls.flat().join('\n'); + expect(output).toContain('"id": "github"'); + expect(output).not.toContain('"id": "linear"'); + }); + + it('lists workspace connections using the existing endpoint', async () => { + const { program, deps } = harness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: response([{ id: 'linear', status: 'connected', token: 'hidden' }]), + auth, + }); + + await program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'integration', + 'connections', + '--workspace', + 'rw_7ccfea89', + '--json', + ]); + + expect(deps.authorizedApiFetch).toHaveBeenCalledWith( + auth, + '/api/v1/workspaces/rw_7ccfea89/integrations', + { method: 'GET' }, + { interactive: false } + ); + expect(vi.mocked(deps.log).mock.calls.flat().join('\n')).not.toContain('hidden'); + }); + + it('renders connected providers for humans by default', async () => { + const { program, deps } = harness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: response([{ provider: 'linear', status: 'connected' }]), + auth, + }); + + await program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'integration', + 'connections', + '--workspace', + 'rw_7ccfea89', + ]); + + expect(deps.log).toHaveBeenCalledWith('linear connected'); + }); + + it('creates a provider connection session', async () => { + const { program, deps } = harness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: response({ + connectLink: 'https://cloud.test/connect/opaque', + workspaceId: '00000000-0000-4000-8000-000000000020', + relayWorkspaceId: 'rw_7ccfea89', + backend: 'nango', + providers: [{ id: 'linear' }], + expiresAt: '2026-07-30T00:00:00.000Z', + }), + auth, + }); + + await program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'integration', + 'connect', + 'linear', + '--workspace', + 'rw_7ccfea89', + '--backend', + 'nango', + ]); + + expect(deps.authorizedApiFetch).toHaveBeenCalledWith( + auth, + '/api/v1/workspaces/rw_7ccfea89/integrations/connect-session', + { + method: 'POST', + body: JSON.stringify({ + allowedIntegrations: ['linear'], + requestedBackend: 'nango', + }), + }, + { interactive: false } + ); + expect(deps.log).toHaveBeenCalledWith('https://cloud.test/connect/opaque'); + }); + + it('disconnects a provider through the existing status endpoint', async () => { + const { program, deps } = harness(); + + await program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'integration', + 'disconnect', + 'linear', + '--workspace', + 'rw_7ccfea89', + ]); + + expect(deps.authorizedApiFetch).toHaveBeenCalledWith( + auth, + '/api/v1/workspaces/rw_7ccfea89/integrations/linear/status', + { method: 'DELETE' }, + { interactive: false } + ); + expect(deps.log).toHaveBeenCalledWith('Disconnected linear.'); + }); + + it('rejects invalid workspace and provider IDs before authenticating', async () => { + const { program, deps } = harness(); + + await expect( + program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'integration', + 'connect', + '../linear', + '--workspace', + 'not-a-workspace', + ]) + ).rejects.toThrow('exit:1'); + + expect(deps.ensureCloudSession).not.toHaveBeenCalled(); + }); + + it('does not reuse a login bound to a different explicit API host', async () => { + const { program, deps } = harness(); + + await expect( + program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'integration', + 'catalog', + '--api-url', + 'https://other.test', + ]) + ).rejects.toThrow('exit:1'); + + expect(deps.authorizedApiFetch).not.toHaveBeenCalled(); + expect(deps.error).toHaveBeenCalledWith(expect.stringContaining('Cloud login is bound to')); + }); + + it('maps authorization failures to a stable error without reflecting response bodies', async () => { + const { program, deps } = harness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: response({ error: 'private server detail' }, 403), + auth, + }); + + await expect( + program.parseAsync(['node', 'agent-relay', 'cloud', 'integration', 'catalog']) + ).rejects.toThrow('exit:1'); + + expect(deps.error).toHaveBeenCalledWith( + 'You do not have permission to perform that integration operation.' + ); + expect(vi.mocked(deps.error).mock.calls.flat().join('\n')).not.toContain('private server detail'); + }); +}); diff --git a/packages/cli/src/cli/commands/cloud-integration.ts b/packages/cli/src/cli/commands/cloud-integration.ts new file mode 100644 index 000000000..ea5d3e80c --- /dev/null +++ b/packages/cli/src/cli/commands/cloud-integration.ts @@ -0,0 +1,387 @@ +import { Command, InvalidArgumentError } from 'commander'; + +import { defaultApiUrl } from '@agent-relay/cloud'; +import { stripAnsiFast } from '@agent-relay/utils'; + +import type { CloudDependencies } from './cloud.js'; + +type Dependencies = Pick< + CloudDependencies, + 'log' | 'error' | 'exit' | 'ensureCloudSession' | 'authorizedApiFetch' +>; +type CloudAuth = Awaited>['auth']; + +const WORKSPACE_UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +const RELAY_WORKSPACE = /^rw_[a-z0-9]{8}$/; +const PROVIDER_ID = /^[a-z0-9][a-z0-9_-]{0,127}$/; + +function isObject(value: unknown): value is Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +function object(value: unknown, label: string): Record { + if (!isObject(value)) throw new Error(`Cloud returned an invalid ${label} response.`); + return value; +} + +function string(value: unknown, label: string): string { + if (typeof value !== 'string' || !value.trim()) { + throw new Error(`Cloud returned an invalid ${label} response.`); + } + return value.trim(); +} + +function connectionProviderIds(value: unknown): string[] { + if (!Array.isArray(value)) { + throw new Error('Cloud returned an invalid integration connection response.'); + } + return value.map((entry) => { + if (typeof entry === 'string') return string(entry, 'integration connection'); + return string(object(entry, 'integration connection provider').id, 'integration connection'); + }); +} + +function workspaceId(value: string): string { + const normalized = value.trim(); + if (!WORKSPACE_UUID.test(normalized) && !RELAY_WORKSPACE.test(normalized)) { + throw new Error( + 'Unsupported Cloud workspace identifier. Use a Cloud workspace UUID or unified rw_ workspace ID.' + ); + } + return normalized; +} + +function providerId(value: string): string { + const normalized = value.trim().toLowerCase(); + if (!PROVIDER_ID.test(normalized)) throw new Error('Invalid integration provider ID.'); + return normalized; +} + +function backend(value: string): 'nango' | 'composio' { + if (value === 'nango' || value === 'composio') return value; + throw new InvalidArgumentError('Expected backend to be one of: nango, composio'); +} + +function canonicalApiUrl(value: string): string { + let url: URL; + try { + url = new URL(value); + } catch { + throw new Error('Invalid Cloud API URL.'); + } + const loopback = url.hostname === 'localhost' || url.hostname === '127.0.0.1' || url.hostname === '[::1]'; + if ( + (url.protocol !== 'https:' && !(url.protocol === 'http:' && loopback)) || + url.username || + url.password || + url.search || + url.hash + ) { + throw new Error('Invalid Cloud API URL.'); + } + return url.toString().replace(/\/+$/, ''); +} + +function cloudError(response: Response): Error { + if (response.status === 401) { + return new Error('Cloud login required. Run `agent-relay cloud login` and retry.'); + } + if (response.status === 403) { + return new Error('You do not have permission to perform that integration operation.'); + } + if (response.status === 404) { + return new Error('The integration resource was not found or is no longer available.'); + } + if (response.status === 409) { + return new Error('The integration operation conflicts with its current lifecycle state.'); + } + if (response.status === 429) { + return new Error('Cloud integration rate limit exceeded. Wait and retry.'); + } + return new Error(`Cloud integration request failed (${response.status}).`); +} + +async function requestWithAuth( + deps: Dependencies, + path: string, + init: RequestInit, + apiUrl?: string, + priorAuth?: CloudAuth +): Promise<{ payload: unknown; auth: CloudAuth }> { + const requested = apiUrl ?? defaultApiUrl(); + const auth = + priorAuth ?? + ( + await deps.ensureCloudSession({ + apiUrl: requested, + interactive: false, + }) + ).auth; + if (apiUrl && canonicalApiUrl(auth.apiUrl) !== canonicalApiUrl(requested)) { + throw new Error( + `Cloud login is bound to ${canonicalApiUrl( + auth.apiUrl + )}. Run \`agent-relay cloud login --api-url ${canonicalApiUrl( + requested + )} --force\` before using this host.` + ); + } + const result = await deps.authorizedApiFetch(auth, path, init, { + interactive: false, + }); + const payload = (await result.response.json().catch(() => null)) as unknown; + if (!result.response.ok) throw cloudError(result.response); + return { payload, auth: result.auth }; +} + +async function request( + deps: Dependencies, + path: string, + init: RequestInit, + apiUrl?: string +): Promise { + return (await requestWithAuth(deps, path, init, apiUrl)).payload; +} + +async function action(deps: Dependencies, fn: () => Promise): Promise { + try { + await fn(); + } catch (error) { + deps.error(error instanceof Error ? error.message : String(error)); + deps.exit(1); + } +} + +function terminal(value: string): string { + return ( + stripAnsiFast(value) + // eslint-disable-next-line no-control-regex + .replace(/[\u0000-\u001f\u007f-\u009f\u202a-\u202e\u2066-\u2069]/g, '�') + .trim() + ); +} + +function secretField(key: string): boolean { + return /(?:token|secret|password|authorization|credential|api[_-]?key)/i.test(key); +} + +function sanitize(value: unknown): unknown { + if (Array.isArray(value)) return value.map(sanitize); + if (!isObject(value)) return typeof value === 'string' ? terminal(value) : value; + return Object.fromEntries( + Object.entries(value) + .filter(([key]) => !secretField(key)) + .map(([key, nested]) => [key, sanitize(nested)]) + ); +} + +function json(deps: Dependencies, payload: unknown): void { + deps.log(JSON.stringify(payload, null, 2)); +} + +function normalizeCatalog(payload: unknown): { + providers: Array>; + version: string; +} { + const response = object(payload, 'integration catalog'); + if (!Array.isArray(response.providers)) { + throw new Error('Cloud returned an invalid integration catalog response.'); + } + return { + providers: response.providers.map((entry) => { + const provider = object(entry, 'integration provider'); + string(provider.id, 'integration provider'); + return sanitize(provider) as Record; + }), + version: string(response.version, 'integration catalog'), + }; +} + +function renderProviders(catalog: ReturnType, deps: Dependencies): void { + for (const provider of catalog.providers) { + const backends = Array.isArray(provider.backends) + ? provider.backends.map(String) + : [provider.backend].filter(Boolean).map(String); + deps.log( + [ + terminal(String(provider.id ?? 'unknown')), + backends.length > 0 ? backends.join(',') : 'backend-unspecified', + ].join(' ') + ); + } +} + +function renderConnections(payload: unknown, deps: Dependencies): void { + if (!Array.isArray(payload)) { + throw new Error('Cloud returned an invalid integration connection list.'); + } + if (payload.length === 0) { + deps.log('No connected workspace integrations.'); + return; + } + for (const entry of payload) { + const connection = object(entry, 'integration connection'); + const id = [connection.id, connection.provider, connection.providerId].find( + (value) => typeof value === 'string' && value.trim() + ); + if (typeof id !== 'string') { + throw new Error('Cloud returned an invalid integration connection list.'); + } + const status = + typeof connection.status === 'string' && connection.status.trim() + ? terminal(connection.status) + : undefined; + deps.log([terminal(id), status].filter(Boolean).join(' ')); + } +} + +export function registerCloudIntegrationCommands(cloudCommand: Command, deps: Dependencies): void { + const integration = cloudCommand + .command('integration') + .description('Manage Agent Relay Cloud integration connections'); + + integration + .command('catalog') + .description('Discover static and dynamic Cloud integrations') + .option('--api-url ', 'Cloud API base URL') + .option('--static', 'Exclude dynamic Nango and Composio catalog entries') + .option('--search ', 'Filter providers by ID or display name') + .option('--backend ', 'Filter providers by nango or composio', backend) + .option('--json', 'Output the integration catalog as JSON') + .action( + async (options: { + apiUrl?: string; + static?: boolean; + search?: string; + backend?: 'nango' | 'composio'; + json?: boolean; + }) => { + await action(deps, async () => { + const catalog = normalizeCatalog( + await request( + deps, + `/api/v1/integrations/catalog?dynamic=${options.static ? 'false' : 'true'}`, + { method: 'GET' }, + options.apiUrl + ) + ); + const query = options.search?.trim().toLowerCase(); + const payload = { + ...catalog, + providers: catalog.providers.filter((provider) => { + const haystack = `${String(provider.id ?? '')} ${String( + provider.displayName ?? '' + )}`.toLowerCase(); + const backends = Array.isArray(provider.backends) + ? provider.backends + : [provider.backend].filter(Boolean); + return ( + (!query || haystack.includes(query)) && + (!options.backend || backends.includes(options.backend)) + ); + }), + }; + if (options.json) json(deps, payload); + else renderProviders(payload, deps); + }); + } + ); + + integration + .command('connections') + .description('List connected workspace integrations') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output connections as JSON') + .action(async (options: { workspace: string; apiUrl?: string; json?: boolean }) => { + await action(deps, async () => { + const id = workspaceId(options.workspace); + const payload = sanitize( + await request( + deps, + `/api/v1/workspaces/${encodeURIComponent(id)}/integrations`, + { method: 'GET' }, + options.apiUrl + ) + ); + if (options.json) json(deps, payload); + else renderConnections(payload, deps); + }); + }); + + integration + .command('connect') + .description('Create a Cloud connection session for a provider') + .argument('', 'Provider ID from the catalog') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .option('--backend ', 'Connection backend: nango or composio', backend) + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output safe connection-session details as JSON') + .action( + async ( + providerInput: string, + options: { + workspace: string; + backend?: 'nango' | 'composio'; + apiUrl?: string; + json?: boolean; + } + ) => { + await action(deps, async () => { + const id = workspaceId(options.workspace); + const provider = providerId(providerInput); + const response = object( + await request( + deps, + `/api/v1/workspaces/${encodeURIComponent(id)}/integrations/connect-session`, + { + method: 'POST', + body: JSON.stringify({ + allowedIntegrations: [provider], + ...(options.backend ? { requestedBackend: options.backend } : {}), + }), + }, + options.apiUrl + ), + 'integration connection' + ); + const payload = { + connectLink: string(response.connectLink, 'integration connection'), + workspaceId: string(response.workspaceId, 'integration connection'), + relayWorkspaceId: string(response.relayWorkspaceId, 'integration connection'), + backend: string(response.backend, 'integration connection'), + providers: connectionProviderIds(response.providers), + ...(typeof response.expiresAt === 'string' ? { expiresAt: response.expiresAt } : {}), + }; + if (options.json) json(deps, payload); + else deps.log(payload.connectLink); + }); + } + ); + + integration + .command('disconnect') + .description('Disconnect a provider from the workspace') + .argument('', 'Provider ID') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output the disconnection result as JSON') + .action( + async (providerInput: string, options: { workspace: string; apiUrl?: string; json?: boolean }) => { + await action(deps, async () => { + const id = workspaceId(options.workspace); + const provider = providerId(providerInput); + await request( + deps, + `/api/v1/workspaces/${encodeURIComponent(id)}/integrations/${encodeURIComponent( + provider + )}/status`, + { method: 'DELETE' }, + options.apiUrl + ); + if (options.json) json(deps, { success: true }); + else deps.log(`Disconnected ${terminal(provider)}.`); + }); + } + ); +} diff --git a/packages/cli/src/cli/commands/cloud-room.test.ts b/packages/cli/src/cli/commands/cloud-room.test.ts new file mode 100644 index 000000000..bd9ab3a3b --- /dev/null +++ b/packages/cli/src/cli/commands/cloud-room.test.ts @@ -0,0 +1,424 @@ +import fs from 'node:fs'; +import os from 'node:os'; +import path from 'node:path'; +import { Command } from 'commander'; +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +import { registerCloudRoomCommands } from './cloud-room.js'; +import type { CloudDependencies } from './cloud.js'; + +vi.mock('@agent-relay/cloud', () => ({ + defaultApiUrl: () => 'https://cloud.test', +})); + +type RoomDeps = Pick< + CloudDependencies, + 'log' | 'error' | 'exit' | 'ensureCloudSession' | 'authorizedApiFetch' +>; + +const auth = { + apiUrl: 'https://cloud.test', + accessToken: 'access-secret', + refreshToken: 'refresh-secret', + accessTokenExpiresAt: '2999-01-01T00:00:00.000Z', +}; + +function jsonResponse(body: unknown, status = 200, headers?: HeadersInit): Response { + return new Response(JSON.stringify(body), { + status, + headers: { 'content-type': 'application/json', ...headers }, + }); +} + +function roomInvitationToken(character = 'A') { + return `relay_room_inv_${character.repeat(43)}`; +} + +function invite(token = roomInvitationToken()) { + return { + invite: { + id: 'invite_1', + email: 'person@example.com', + role: 'participant', + token, + expiresAt: '2026-07-30T00:00:00.000Z', + createdAt: '2026-07-23T00:00:00.000Z', + }, + }; +} + +function createHarness(roomIo?: Parameters[2]) { + const exit = vi.fn((code: number) => { + throw new Error(`exit:${code}`); + }) as unknown as RoomDeps['exit']; + const deps: RoomDeps = { + log: vi.fn(), + error: vi.fn(), + exit, + ensureCloudSession: vi.fn(async () => ({ auth, client: {} as never })) as RoomDeps['ensureCloudSession'], + authorizedApiFetch: vi.fn(async () => ({ + response: jsonResponse({}), + auth, + })) as RoomDeps['authorizedApiFetch'], + }; + const program = new Command(); + program.exitOverride(); + const cloud = program.command('cloud'); + registerCloudRoomCommands(cloud, deps, roomIo); + return { program, deps, room: cloud.commands[0] }; +} + +beforeEach(() => { + vi.clearAllMocks(); +}); + +describe('registerCloudRoomCommands', () => { + it('registers the complete trusted-participant lifecycle', () => { + const { room } = createHarness(); + expect(room.commands.map((command) => command.name())).toEqual([ + 'invite', + 'invites', + 'revoke-invite', + 'members', + 'remove-member', + 'accept', + 'revoke-session', + 'session', + ]); + }); + + it('creates only participant invitations', async () => { + const { program, deps } = createHarness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: jsonResponse(invite()), + auth, + }); + + await program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'room', + 'invite', + '--workspace', + 'rw_7ccfea89', + '--email', + 'Person@Example.com', + '--expires-in', + '600', + '--token-stdout', + ]); + + expect(deps.authorizedApiFetch).toHaveBeenCalledWith( + auth, + '/api/v1/workspaces/rw_7ccfea89/room/invites', + { + method: 'POST', + body: JSON.stringify({ + email: 'person@example.com', + role: 'participant', + expiresInSeconds: 600, + }), + }, + { interactive: false } + ); + expect(deps.log).toHaveBeenCalledWith(roomInvitationToken()); + }); + + it('does not expose viewer or email-delivery invite options', () => { + const { room } = createHarness(); + const inviteCommand = room.commands.find((command) => command.name() === 'invite'); + expect(inviteCommand?.options.map((option) => option.long)).not.toContain('--role'); + expect(inviteCommand?.options.map((option) => option.long)).not.toContain('--email-delivery'); + }); + + it('requires exactly one invitation token sink before authenticating', async () => { + const { program, deps } = createHarness(); + + await expect( + program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'room', + 'invite', + '--workspace', + 'rw_7ccfea89', + '--email', + 'person@example.com', + ]) + ).rejects.toThrow('exit:1'); + + expect(deps.ensureCloudSession).not.toHaveBeenCalled(); + expect(deps.error).toHaveBeenCalledWith( + 'Use exactly one invitation token sink: --token-stdout, --token-file, or --json.' + ); + }); + + it('writes a token only to a new owner-only file', async () => { + const directory = fs.mkdtempSync(path.join(os.tmpdir(), 'relay-room-invite-')); + const tokenFile = path.join(directory, 'token'); + const { program, deps } = createHarness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: jsonResponse(invite(roomInvitationToken('B'))), + auth, + }); + + try { + await program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'room', + 'invite', + '--workspace', + 'rw_7ccfea89', + '--email', + 'person@example.com', + '--token-file', + tokenFile, + ]); + expect(fs.readFileSync(tokenFile, 'utf8')).toBe(`${roomInvitationToken('B')}\n`); + if (process.platform !== 'win32') { + expect(fs.statSync(tokenFile).mode & 0o777).toBe(0o600); + } + } finally { + fs.rmSync(directory, { recursive: true, force: true }); + } + }); + + it('revokes a newly-created invite when its token cannot be written', async () => { + const { program, deps } = createHarness({ + writeSecretFile: vi.fn(async () => { + throw new Error('disk full'); + }), + }); + vi.mocked(deps.authorizedApiFetch) + .mockResolvedValueOnce({ response: jsonResponse(invite()), auth }) + .mockResolvedValueOnce({ response: jsonResponse({ ok: true }), auth }); + + await expect( + program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'room', + 'invite', + '--workspace', + 'rw_7ccfea89', + '--email', + 'person@example.com', + '--token-file', + '/unused', + ]) + ).rejects.toThrow('exit:1'); + + expect(deps.authorizedApiFetch).toHaveBeenNthCalledWith( + 2, + auth, + '/api/v1/workspaces/rw_7ccfea89/room/invites/invite_1', + { method: 'DELETE' }, + { interactive: false } + ); + }); + + it('accepts an invitation only from an explicit secret source', async () => { + const token = roomInvitationToken('C'); + const { program, deps } = createHarness({ + readStdin: vi.fn(async () => `${token}\n`), + }); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: jsonResponse({ + membership: { + id: 'member_1', + workspaceId: 'rw_7ccfea89', + role: 'participant', + }, + }), + auth, + }); + + await program.parseAsync(['node', 'agent-relay', 'cloud', 'room', 'accept', '--token-stdin', '--json']); + + expect(deps.authorizedApiFetch).toHaveBeenCalledWith( + auth, + '/api/v1/room/invites/accept', + { + method: 'POST', + body: JSON.stringify({ token }), + }, + { interactive: false } + ); + }); + + it('rejects invitation tokens outside the Relay Room wire contract', async () => { + for (const token of [ + `product_inv_${'A'.repeat(43)}`, + `relay_room_inv_${'A'.repeat(42)}`, + `relay_room_inv_${'A'.repeat(44)}`, + `relay_room_inv_${'A'.repeat(42)}!`, + ]) { + const { program, deps } = createHarness({ + readStdin: vi.fn(async () => `${token}\n`), + }); + + await expect( + program.parseAsync(['node', 'agent-relay', 'cloud', 'room', 'accept', '--token-stdin']) + ).rejects.toThrow('exit:1'); + expect(deps.error).toHaveBeenCalledWith('Invalid room invitation token.'); + expect(deps.ensureCloudSession).not.toHaveBeenCalled(); + expect(deps.authorizedApiFetch).not.toHaveBeenCalled(); + } + }); + + it('rejects non-participant member responses', async () => { + const { program, deps } = createHarness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: jsonResponse({ + members: [ + { + id: 'member_1', + userId: 'user_1', + email: 'person@example.com', + name: null, + role: 'viewer', + status: 'active', + joinedAt: '2026-07-23T00:00:00.000Z', + }, + ], + }), + auth, + }); + + await expect( + program.parseAsync(['node', 'agent-relay', 'cloud', 'room', 'members', '--workspace', 'rw_7ccfea89']) + ).rejects.toThrow('exit:1'); + expect(deps.error).toHaveBeenCalledWith('Cloud room returned an invalid member list response.'); + }); + + it('creates a human Relaycast session for a participant device', async () => { + const { program, deps } = createHarness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: jsonResponse({ + role: 'participant', + relaycastBaseUrl: 'https://relay.example.com', + agentName: 'human-person-device', + agentToken: 'at_live_device_secret', + }), + auth, + }); + + await program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'room', + 'session', + '--workspace', + 'rw_7ccfea89', + '--device-id', + 'herdr-device', + '--json', + ]); + + expect(deps.authorizedApiFetch).toHaveBeenCalledWith( + auth, + '/api/v1/workspaces/rw_7ccfea89/room/session', + { + method: 'POST', + body: JSON.stringify({ deviceId: 'herdr-device' }), + }, + { interactive: false } + ); + const output = vi.mocked(deps.log).mock.calls.flat().join('\n'); + expect(output).toContain('"role": "participant"'); + expect(output).toContain('at_live_device_secret'); + }); + + it('rejects observer sessions from Cloud', async () => { + const { program, deps } = createHarness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: jsonResponse({ + role: 'viewer', + relaycastBaseUrl: 'https://relay.example.com', + observerToken: 'ot_live_observer', + }), + auth, + }); + + await expect( + program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'room', + 'session', + '--workspace', + 'rw_7ccfea89', + '--device-id', + 'herdr-device', + ]) + ).rejects.toThrow('exit:1'); + }); + + it('revokes the current device session', async () => { + const { program, deps } = createHarness(); + + await program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'room', + 'revoke-session', + '--workspace', + 'rw_7ccfea89', + '--device-id', + 'herdr-device', + ]); + + expect(deps.authorizedApiFetch).toHaveBeenCalledWith( + auth, + '/api/v1/workspaces/rw_7ccfea89/room/session', + { + method: 'DELETE', + body: JSON.stringify({ deviceId: 'herdr-device' }), + }, + { interactive: false } + ); + }); + + it('does not reuse a login bound to another explicit API host', async () => { + const { program, deps } = createHarness(); + + await expect( + program.parseAsync([ + 'node', + 'agent-relay', + 'cloud', + 'room', + 'members', + '--workspace', + 'rw_7ccfea89', + '--api-url', + 'https://other.test', + ]) + ).rejects.toThrow('exit:1'); + + expect(deps.authorizedApiFetch).not.toHaveBeenCalled(); + }); + + it('maps rate limits without reflecting response bodies', async () => { + const { program, deps } = createHarness(); + vi.mocked(deps.authorizedApiFetch).mockResolvedValueOnce({ + response: jsonResponse({ error: 'private detail' }, 429, { 'retry-after': '5' }), + auth, + }); + + await expect( + program.parseAsync(['node', 'agent-relay', 'cloud', 'room', 'members', '--workspace', 'rw_7ccfea89']) + ).rejects.toThrow('exit:1'); + + expect(deps.error).toHaveBeenCalledWith('Cloud room rate limit exceeded. Retry-After: 5 seconds.'); + expect(vi.mocked(deps.error).mock.calls.flat().join('\n')).not.toContain('private detail'); + }); +}); diff --git a/packages/cli/src/cli/commands/cloud-room.ts b/packages/cli/src/cli/commands/cloud-room.ts new file mode 100644 index 000000000..677e3f715 --- /dev/null +++ b/packages/cli/src/cli/commands/cloud-room.ts @@ -0,0 +1,794 @@ +import fs from 'node:fs/promises'; +import { constants as fsConstants } from 'node:fs'; +import process from 'node:process'; +import { Command, InvalidArgumentError } from 'commander'; + +import { defaultApiUrl } from '@agent-relay/cloud'; +import { stripAnsiFast } from '@agent-relay/utils'; + +import type { CloudDependencies } from './cloud.js'; + +type CloudRoomDependencies = Pick< + CloudDependencies, + 'log' | 'error' | 'exit' | 'ensureCloudSession' | 'authorizedApiFetch' +>; +type CloudAuth = Awaited>['auth']; + +type RoomRole = 'participant'; + +const CLOUD_WORKSPACE_UUID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; +const UNIFIED_WORKSPACE_ID_PATTERN = /^rw_[a-z0-9]{8}$/; +const ROOM_RESOURCE_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_-]{0,127}$/; +const DEFAULT_INVITATION_LIFETIME_SECONDS = 7 * 24 * 60 * 60; +const MIN_INVITATION_LIFETIME_SECONDS = 60; +const MAX_INVITATION_LIFETIME_SECONDS = 30 * 24 * 60 * 60; +const MAX_ROOM_SECRET_LENGTH = 2_048; +const ROOM_INVITATION_TOKEN_PATTERN = /^relay_room_inv_[A-Za-z0-9_-]{43}$/; + +interface CloudRoomIo { + readStdin: () => Promise; + readSecretFile: (filePath: string) => Promise; + writeSecretFile: (filePath: string, value: string) => Promise; +} + +function isObject(value: unknown): value is Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +function requireObject(value: unknown, label: string): Record { + if (!isObject(value)) { + throw new Error(`Cloud room returned an invalid ${label} response.`); + } + return value; +} + +function requireStringField(record: Record, key: string, label: string): string { + const value = record[key]; + if (typeof value !== 'string' || !value.trim()) { + throw new Error(`Cloud room returned an invalid ${label} response.`); + } + return value.trim(); +} + +function requireNullableStringField( + record: Record, + key: string, + label: string +): string | null { + const value = record[key]; + if (value === null) return null; + if (typeof value !== 'string') { + throw new Error(`Cloud room returned an invalid ${label} response.`); + } + return value; +} + +function requireIsoDateField(record: Record, key: string, label: string): string { + const value = requireStringField(record, key, label); + if (!Number.isFinite(Date.parse(value))) { + throw new Error(`Cloud room returned an invalid ${label} response.`); + } + return value; +} + +function requireResponseRole(record: Record, label: string): RoomRole { + const role = record.role; + if (role !== 'participant') { + throw new Error(`Cloud room returned an invalid ${label} response.`); + } + return role; +} + +function containsForbiddenCredentialField(value: unknown): boolean { + if (Array.isArray(value)) return value.some(containsForbiddenCredentialField); + if (!isObject(value)) return false; + return Object.entries(value).some( + ([key, nested]) => + /^(?:workspaceKey|relaycastApiKey|relayfileToken|relayfileCredentials|accessToken|refreshToken|authorization|apiKey)$/i.test( + key + ) || containsForbiddenCredentialField(nested) + ); +} + +type RoomInvite = { + id: string; + email: string; + role: RoomRole; + expiresAt: string; + createdAt: string; + token?: string; +}; + +function normalizeInvite(value: unknown, requireToken: boolean): RoomInvite { + const invite = requireObject(value, 'invitation'); + const normalized: RoomInvite = { + id: requireStringField(invite, 'id', 'invitation'), + email: requireStringField(invite, 'email', 'invitation'), + role: requireResponseRole(invite, 'invitation'), + expiresAt: requireIsoDateField(invite, 'expiresAt', 'invitation'), + createdAt: requireIsoDateField(invite, 'createdAt', 'invitation'), + }; + if (requireToken) { + normalized.token = requireInvitationToken(requireStringField(invite, 'token', 'invitation')); + } + return normalized; +} + +function normalizeInviteCreate(payload: unknown): { invite: RoomInvite & { token: string } } { + const response = requireObject(payload, 'invitation'); + return { + invite: normalizeInvite(response.invite, true) as RoomInvite & { token: string }, + }; +} + +function normalizeInviteList(payload: unknown): { invites: RoomInvite[] } { + const response = requireObject(payload, 'invitation list'); + if (!Array.isArray(response.invites)) { + throw new Error('Cloud room returned an invalid invitation list response.'); + } + return { invites: response.invites.map((invite) => normalizeInvite(invite, false)) }; +} + +type RoomMember = { + id: string; + userId: string; + email: string | null; + name: string | null; + role: RoomRole; + status: 'active' | 'revoking'; + joinedAt: string; +}; + +function normalizeMemberList(payload: unknown): { members: RoomMember[] } { + const response = requireObject(payload, 'member list'); + if (!Array.isArray(response.members)) { + throw new Error('Cloud room returned an invalid member list response.'); + } + return { + members: response.members.map((value) => { + const member = requireObject(value, 'member list'); + const status = requireStringField(member, 'status', 'member list'); + if (status !== 'active' && status !== 'revoking') { + throw new Error('Cloud room returned an invalid member list response.'); + } + return { + id: requireStringField(member, 'id', 'member list'), + userId: requireStringField(member, 'userId', 'member list'), + email: requireNullableStringField(member, 'email', 'member list'), + name: requireNullableStringField(member, 'name', 'member list'), + role: requireResponseRole(member, 'member list'), + status, + joinedAt: requireIsoDateField(member, 'joinedAt', 'member list'), + }; + }), + }; +} + +function normalizeMembership(payload: unknown): { + membership: { id: string; workspaceId: string; role: RoomRole }; +} { + const response = requireObject(payload, 'membership'); + const membership = requireObject(response.membership, 'membership'); + return { + membership: { + id: requireStringField(membership, 'id', 'membership'), + workspaceId: requireStringField(membership, 'workspaceId', 'membership'), + role: requireResponseRole(membership, 'membership'), + }, + }; +} + +type RoomSession = { + role: 'participant'; + relaycastBaseUrl: string; + agentName: string; + agentToken: string; +}; + +function normalizeRoomSession(payload: unknown): RoomSession { + const response = requireObject(payload, 'session'); + const role = requireResponseRole(response, 'session'); + const relaycastBaseUrl = requireRelaycastBaseUrl( + requireStringField(response, 'relaycastBaseUrl', 'session') + ); + if ( + typeof response.agentToken !== 'string' || + !response.agentToken.trim().startsWith('at_live_') || + response.observerToken !== undefined + ) { + throw new Error('Cloud room returned an invalid participant session response.'); + } + return { + role, + relaycastBaseUrl, + agentName: requireStringField(response, 'agentName', 'participant session'), + agentToken: response.agentToken.trim(), + }; +} + +function parsePositiveInteger(value: string): number { + if (!/^[1-9][0-9]*$/.test(value)) { + throw new InvalidArgumentError('Expected a positive whole number.'); + } + const parsed = Number(value); + if ( + !Number.isSafeInteger(parsed) || + parsed < MIN_INVITATION_LIFETIME_SECONDS || + parsed > MAX_INVITATION_LIFETIME_SECONDS + ) { + throw new InvalidArgumentError( + `Expected between ${MIN_INVITATION_LIFETIME_SECONDS} and ${MAX_INVITATION_LIFETIME_SECONDS} seconds.` + ); + } + return parsed; +} + +function requireWorkspaceId(value: string): string { + const workspaceId = value.trim(); + if (!CLOUD_WORKSPACE_UUID_PATTERN.test(workspaceId) && !UNIFIED_WORKSPACE_ID_PATTERN.test(workspaceId)) { + throw new Error( + 'Unsupported Cloud workspace identifier. Use a Cloud workspace UUID or unified rw_ workspace ID.' + ); + } + return workspaceId; +} + +function requireResourceId(value: string, label: string): string { + const resourceId = value.trim(); + // Cloud owns these opaque identifiers. Only reject values that cannot be + // safely carried in a URL path or terminal; encodeURIComponent handles the + // remaining printable characters without coupling the CLI to an ID format. + if ( + !resourceId || + resourceId.length > 512 || + // eslint-disable-next-line no-control-regex + /[\u0000-\u001f\u007f-\u009f\u202a-\u202e\u2066-\u2069]/.test(resourceId) + ) { + throw new Error(`Invalid ${label}.`); + } + return resourceId; +} + +function requireEmail(value: string): string { + const email = value.trim().toLowerCase(); + if (email.length > 320 || /[\r\n]/.test(email) || !/^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(email)) { + throw new Error('A valid email address is required.'); + } + return email; +} + +function requireDeviceId(value: string): string { + const deviceId = value.trim(); + if (!ROOM_RESOURCE_ID_PATTERN.test(deviceId)) { + throw new Error('Invalid device ID. Use 1-128 letters, numbers, underscores, or hyphens.'); + } + return deviceId; +} + +/** Keep Cloud/user-provided text from controlling or escaping the terminal. */ +function sanitizeTerminalCell(value: string): string { + // eslint-disable-next-line no-control-regex + return stripAnsiFast(value).replace(/[\u0000-\u001f\u007f-\u009f\u202a-\u202e\u2066-\u2069]/g, '�'); +} + +async function defaultReadStdin(): Promise { + let input = ''; + for await (const chunk of process.stdin) { + input += String(chunk); + if (input.length > MAX_ROOM_SECRET_LENGTH + 1) { + throw new Error('Invalid room invitation token.'); + } + } + return input; +} + +async function defaultReadSecretFile(filePath: string): Promise { + const noFollow = process.platform === 'win32' ? 0 : fsConstants.O_NOFOLLOW; + const handle = await fs.open(filePath, fsConstants.O_RDONLY | noFollow); + try { + const metadata = await handle.stat(); + if (!metadata.isFile()) { + throw new Error('Room invitation token file must be a regular file.'); + } + if (process.platform !== 'win32' && (metadata.mode & 0o077) !== 0) { + throw new Error('Room invitation token file must have owner-only permissions (0600).'); + } + if (metadata.size > MAX_ROOM_SECRET_LENGTH + 1) { + throw new Error('Invalid room invitation token.'); + } + return handle.readFile('utf8'); + } finally { + await handle.close(); + } +} + +async function defaultWriteSecretFile(filePath: string, value: string): Promise { + const noFollow = process.platform === 'win32' ? 0 : fsConstants.O_NOFOLLOW; + const handle = await fs.open( + filePath, + fsConstants.O_WRONLY | fsConstants.O_CREAT | fsConstants.O_EXCL | noFollow, + 0o600 + ); + try { + await handle.writeFile(`${value}\n`, 'utf8'); + await handle.sync(); + } finally { + await handle.close(); + } +} + +function requireInvitationToken(value: string): string { + const token = value.trim(); + if (!ROOM_INVITATION_TOKEN_PATTERN.test(token)) { + throw new Error('Invalid room invitation token.'); + } + return token; +} + +function requireRelaycastBaseUrl(value: string): string { + let url: URL; + try { + url = new URL(value); + } catch { + throw new Error('Cloud room returned an invalid session response.'); + } + const loopback = url.hostname === 'localhost' || url.hostname === '127.0.0.1' || url.hostname === '[::1]'; + if ( + (url.protocol !== 'https:' && !(url.protocol === 'http:' && loopback)) || + url.username || + url.password || + url.search || + url.hash + ) { + throw new Error('Cloud room returned an invalid session response.'); + } + return url.toString().replace(/\/+$/, ''); +} + +function canonicalApiBaseUrl(value: string): string { + let url: URL; + try { + url = new URL(value); + } catch { + throw new Error('Invalid Cloud API URL.'); + } + const loopback = url.hostname === 'localhost' || url.hostname === '127.0.0.1' || url.hostname === '[::1]'; + if ( + (url.protocol !== 'https:' && !(url.protocol === 'http:' && loopback)) || + url.username || + url.password || + url.search || + url.hash + ) { + throw new Error('Invalid Cloud API URL.'); + } + return url.toString().replace(/\/+$/, ''); +} + +function cloudRoomError(response: Response): Error { + if (response.status === 401) { + return new Error('Cloud login required. Run `agent-relay cloud login` and retry.'); + } + if (response.status === 403) { + return new Error('You do not have permission to perform that room operation.'); + } + if (response.status === 404) { + return new Error('The room resource was not found or is no longer available.'); + } + if (response.status === 409) { + return new Error('The room operation conflicts with the current membership state.'); + } + if (response.status === 410) { + return new Error('The room invitation has expired or was already used.'); + } + if (response.status === 429) { + const retryAfter = response.headers.get('retry-after')?.trim(); + return new Error( + `Cloud room rate limit exceeded.${ + retryAfter ? ` Retry-After: ${retryAfter} seconds.` : ' Wait and retry.' + }` + ); + } + if (response.status >= 400 && response.status < 500) { + return new Error(`Cloud rejected the room request (${response.status}).`); + } + return new Error(`Cloud room request failed (${response.status}).`); +} + +async function requestRoomWithAuth( + deps: CloudRoomDependencies, + path: string, + init: RequestInit, + apiUrl?: string, + priorAuth?: CloudAuth +): Promise<{ payload: unknown; auth: CloudAuth }> { + const requestedApiUrl = apiUrl ?? defaultApiUrl(); + const auth = + priorAuth ?? + ( + await deps.ensureCloudSession({ + apiUrl: requestedApiUrl, + interactive: false, + }) + ).auth; + if (apiUrl && canonicalApiBaseUrl(auth.apiUrl) !== canonicalApiBaseUrl(requestedApiUrl)) { + throw new Error( + `Cloud login is bound to ${canonicalApiBaseUrl( + auth.apiUrl + )}. Run \`agent-relay cloud login --api-url ${canonicalApiBaseUrl( + requestedApiUrl + )} --force\` before using this host.` + ); + } + const result = await deps.authorizedApiFetch(auth, path, init, { + interactive: false, + }); + const payload = (await result.response.json().catch(() => null)) as unknown; + if (!result.response.ok) { + throw cloudRoomError(result.response); + } + if (containsForbiddenCredentialField(payload)) { + throw new Error('Cloud room returned a forbidden workspace or integration credential.'); + } + return { payload, auth: result.auth }; +} + +async function requestRoom( + deps: CloudRoomDependencies, + path: string, + init: RequestInit, + apiUrl?: string +): Promise { + return (await requestRoomWithAuth(deps, path, init, apiUrl)).payload; +} + +async function runRoomAction(deps: CloudRoomDependencies, action: () => Promise): Promise { + try { + await action(); + } catch (error) { + deps.error(error instanceof Error ? error.message : String(error)); + deps.exit(1); + } +} + +function logJson(deps: CloudRoomDependencies, payload: unknown): void { + deps.log(JSON.stringify(payload, null, 2)); +} + +function textField(record: object, key: string): string | undefined { + const value = (record as Record)[key]; + return typeof value === 'string' && value.trim() ? sanitizeTerminalCell(value.trim()) : undefined; +} + +function renderInvites(payload: { invites: RoomInvite[] }, deps: CloudRoomDependencies): void { + const { invites } = payload; + if (invites.length === 0) { + deps.log('No active room invitations.'); + return; + } + for (const invite of invites) { + const id = textField(invite, 'id') ?? 'unknown'; + const email = textField(invite, 'email') ?? 'unknown'; + const role = textField(invite, 'role') ?? 'unknown'; + const expiresAt = textField(invite, 'expiresAt'); + deps.log([id, email, role, expiresAt ? `expires ${expiresAt}` : undefined].filter(Boolean).join(' ')); + } +} + +function renderMembers(payload: { members: RoomMember[] }, deps: CloudRoomDependencies): void { + const { members } = payload; + if (members.length === 0) { + deps.log('No room members.'); + return; + } + for (const member of members) { + const id = textField(member, 'id') ?? 'unknown'; + const email = textField(member, 'email') ?? textField(member, 'name') ?? 'unknown'; + const role = textField(member, 'role') ?? 'unknown'; + const status = textField(member, 'status'); + deps.log([id, email, role, status].filter(Boolean).join(' ')); + } +} + +export function registerCloudRoomCommands( + cloudCommand: Command, + deps: CloudRoomDependencies, + ioOverrides: Partial = {} +): void { + const io: CloudRoomIo = { + readStdin: defaultReadStdin, + readSecretFile: defaultReadSecretFile, + writeSecretFile: defaultWriteSecretFile, + ...ioOverrides, + }; + const room = cloudCommand.command('room').description('Manage workspace-scoped multiplayer rooms'); + + room + .command('invite') + .description('Invite a full participant to a workspace room') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .requiredOption('--email ', 'Email address bound to the invitation') + .option('--api-url ', 'Cloud API base URL') + .option( + '--expires-in ', + 'Invitation lifetime in seconds', + parsePositiveInteger, + DEFAULT_INVITATION_LIFETIME_SECONDS + ) + .option('--token-stdout', 'Print only the one-time invitation token') + .option('--token-file ', 'Write the token to a new owner-only 0600 file') + .option('--json', 'Output the invitation and its one-time token as JSON') + .action( + async (options: { + workspace: string; + email: string; + expiresIn: number; + apiUrl?: string; + tokenStdout?: boolean; + tokenFile?: string; + json?: boolean; + }) => { + await runRoomAction(deps, async () => { + const manualSinkCount = [options.tokenStdout, Boolean(options.tokenFile), options.json].filter( + Boolean + ).length; + if (manualSinkCount !== 1) { + throw new Error( + 'Use exactly one invitation token sink: --token-stdout, --token-file, or --json.' + ); + } + const workspaceId = requireWorkspaceId(options.workspace); + const email = requireEmail(options.email); + const created = await requestRoomWithAuth( + deps, + `/api/v1/workspaces/${encodeURIComponent(workspaceId)}/room/invites`, + { + method: 'POST', + body: JSON.stringify({ + email, + role: 'participant', + expiresInSeconds: options.expiresIn, + }), + }, + options.apiUrl + ); + const payload = normalizeInviteCreate(created.payload); + if (options.json) { + logJson(deps, payload); + return; + } + if (options.tokenStdout) { + deps.log(payload.invite.token); + return; + } + try { + await io.writeSecretFile(options.tokenFile ?? '', payload.invite.token); + } catch (writeError) { + try { + await requestRoomWithAuth( + deps, + `/api/v1/workspaces/${encodeURIComponent( + workspaceId + )}/room/invites/${encodeURIComponent(payload.invite.id)}`, + { method: 'DELETE' }, + options.apiUrl, + created.auth + ); + } catch (cleanupError) { + throw new AggregateError( + [writeError, cleanupError], + `Could not write the invitation token. Revocation could not be confirmed; revoke invitation ${sanitizeTerminalCell( + payload.invite.id + )} before retrying.`, + { cause: writeError } + ); + } + throw writeError; + } + deps.log(`Created full room-participant invitation for ${email}.`); + deps.log( + `Wrote the one-time invitation token to ${sanitizeTerminalCell(options.tokenFile ?? '')}.` + ); + }); + } + ); + + room + .command('invites') + .description('List workspace room invitations') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output invitations as JSON') + .action(async (options: { workspace: string; apiUrl?: string; json?: boolean }) => { + await runRoomAction(deps, async () => { + const workspaceId = requireWorkspaceId(options.workspace); + const payload = normalizeInviteList( + await requestRoom( + deps, + `/api/v1/workspaces/${encodeURIComponent(workspaceId)}/room/invites`, + { method: 'GET' }, + options.apiUrl + ) + ); + if (options.json) { + logJson(deps, payload); + return; + } + renderInvites(payload, deps); + }); + }); + + room + .command('revoke-invite') + .description('Revoke an unused workspace room invitation') + .argument('', 'Invitation ID') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output the revocation response as JSON') + .action( + async (inviteIdInput: string, options: { workspace: string; apiUrl?: string; json?: boolean }) => { + await runRoomAction(deps, async () => { + const workspaceId = requireWorkspaceId(options.workspace); + const inviteId = requireResourceId(inviteIdInput, 'invitation ID'); + await requestRoom( + deps, + `/api/v1/workspaces/${encodeURIComponent(workspaceId)}/room/invites/${encodeURIComponent(inviteId)}`, + { method: 'DELETE' }, + options.apiUrl + ); + if (options.json) { + logJson(deps, { ok: true }); + return; + } + deps.log(`Revoked room invitation ${inviteId}.`); + }); + } + ); + + room + .command('members') + .description('List workspace room members') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output members as JSON') + .action(async (options: { workspace: string; apiUrl?: string; json?: boolean }) => { + await runRoomAction(deps, async () => { + const workspaceId = requireWorkspaceId(options.workspace); + const payload = normalizeMemberList( + await requestRoom( + deps, + `/api/v1/workspaces/${encodeURIComponent(workspaceId)}/room/members`, + { method: 'GET' }, + options.apiUrl + ) + ); + if (options.json) { + logJson(deps, payload); + return; + } + renderMembers(payload, deps); + }); + }); + + room + .command('remove-member') + .description('Remove a member and revoke their live room access') + .argument('', 'Workspace membership ID') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output the removal response as JSON') + .action( + async (memberIdInput: string, options: { workspace: string; apiUrl?: string; json?: boolean }) => { + await runRoomAction(deps, async () => { + const workspaceId = requireWorkspaceId(options.workspace); + const memberId = requireResourceId(memberIdInput, 'membership ID'); + await requestRoom( + deps, + `/api/v1/workspaces/${encodeURIComponent(workspaceId)}/room/members/${encodeURIComponent(memberId)}`, + { method: 'DELETE' }, + options.apiUrl + ); + if (options.json) { + logJson(deps, { ok: true }); + return; + } + deps.log(`Removed room member ${memberId} and revoked their room sessions.`); + }); + } + ); + + room + .command('accept') + .description('Accept an email-bound room invitation') + .option('--token-stdin', 'Read the single-use invitation token from stdin') + .option('--token-file ', 'Read the invitation token from an owner-only 0600 file') + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output the accepted membership as JSON') + .action( + async (options: { tokenStdin?: boolean; tokenFile?: string; apiUrl?: string; json?: boolean }) => { + await runRoomAction(deps, async () => { + if (Boolean(options.tokenStdin) === Boolean(options.tokenFile)) { + throw new Error('Use exactly one of --token-stdin or --token-file.'); + } + const token = requireInvitationToken( + options.tokenStdin ? await io.readStdin() : await io.readSecretFile(options.tokenFile ?? '') + ); + const payload = normalizeMembership( + await requestRoom( + deps, + '/api/v1/room/invites/accept', + { + method: 'POST', + body: JSON.stringify({ token }), + }, + options.apiUrl + ) + ); + if (options.json) { + logJson(deps, payload); + return; + } + deps.log('Room invitation accepted.'); + }); + } + ); + + room + .command('revoke-session') + .description('Revoke this member’s scoped session for one device') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .requiredOption('--device-id ', 'Stable non-secret identifier for this client') + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output the revocation response as JSON') + .action(async (options: { workspace: string; deviceId: string; apiUrl?: string; json?: boolean }) => { + await runRoomAction(deps, async () => { + const workspaceId = requireWorkspaceId(options.workspace); + const deviceId = requireDeviceId(options.deviceId); + await requestRoom( + deps, + `/api/v1/workspaces/${encodeURIComponent(workspaceId)}/room/session`, + { + method: 'DELETE', + body: JSON.stringify({ deviceId }), + }, + options.apiUrl + ); + if (options.json) { + logJson(deps, { ok: true }); + return; + } + deps.log(`Revoked room session for device ${sanitizeTerminalCell(deviceId)}.`); + }); + }); + + room + .command('session') + .description('Create or resume this device’s full room-participant session') + .requiredOption('--workspace ', 'Cloud UUID or unified rw_ workspace ID') + .requiredOption('--device-id ', 'Stable non-secret identifier for this client') + .option('--api-url ', 'Cloud API base URL') + .option('--json', 'Output the session, including its participant credential') + .action(async (options: { workspace: string; deviceId: string; apiUrl?: string; json?: boolean }) => { + await runRoomAction(deps, async () => { + const workspaceId = requireWorkspaceId(options.workspace); + const deviceId = requireDeviceId(options.deviceId); + const payload = normalizeRoomSession( + await requestRoom( + deps, + `/api/v1/workspaces/${encodeURIComponent(workspaceId)}/room/session`, + { + method: 'POST', + body: JSON.stringify({ deviceId }), + }, + options.apiUrl + ) + ); + if (options.json) { + logJson(deps, payload); + return; + } + deps.log('Full-participant room session ready.'); + deps.log('Scoped credentials are hidden. Trusted clients may request them explicitly with --json.'); + }); + }); +} diff --git a/packages/cli/src/cli/commands/cloud.test.ts b/packages/cli/src/cli/commands/cloud.test.ts index b12c45e35..d121343ae 100644 --- a/packages/cli/src/cli/commands/cloud.test.ts +++ b/packages/cli/src/cli/commands/cloud.test.ts @@ -115,6 +115,8 @@ describe('registerCloudCommands', () => { expect(cloud).toBeDefined(); expect(cloud?.commands.map((command) => command.name())).toEqual([ 'worker', + 'room', + 'integration', 'login', 'logout', 'session', diff --git a/packages/cli/src/cli/commands/cloud.ts b/packages/cli/src/cli/commands/cloud.ts index 191f64552..191b38852 100644 --- a/packages/cli/src/cli/commands/cloud.ts +++ b/packages/cli/src/cli/commands/cloud.ts @@ -32,6 +32,8 @@ import { import { defaultExit } from '../lib/exit.js'; import { errorClassName } from '../lib/telemetry-helpers.js'; import { track } from '../telemetry/index.js'; +import { registerCloudRoomCommands } from './cloud-room.js'; +import { registerCloudIntegrationCommands } from './cloud-integration.js'; import { registerCloudWorkerCommands } from './cloud-worker.js'; const CLOUD_SYNC_PATCH_EXCLUDES = [ @@ -416,6 +418,8 @@ export function registerCloudCommands(program: Command, overrides: Partial { expect(relay.getStatus).toHaveBeenCalledTimes(1); }); + it('up lets the broker atomically bind an OS-assigned API port when configured with port zero', async () => { + const relay = createRelayMock({ apiPort: 43123 }); + const { program, deps } = createHarness({ + relay, + env: { AGENT_RELAY_BROKER_PORT: '0' }, + }); + + const exitCode = await runCommand(program, ['up']); + + expect(exitCode).toBeUndefined(); + expect(deps.isPortInUse).not.toHaveBeenCalled(); + expect(deps.createRelay).toHaveBeenCalledWith('/tmp/project', 0, undefined, undefined); + expect(deps.log).toHaveBeenCalledWith('Relay API: http://localhost:43123'); + }); + + it('up shuts down a port-zero broker that does not report its assigned API port', async () => { + const relay = createRelayMock({ apiPort: undefined }); + const { program, deps } = createHarness({ + relay, + env: { AGENT_RELAY_BROKER_PORT: '0' }, + }); + + const exitCode = await runCommand(program, ['up']); + + expect(exitCode).toBe(1); + expect(relay.shutdown).toHaveBeenCalledTimes(1); + expect(deps.error).toHaveBeenCalledWith( + 'Failed to start broker: Broker started without reporting its OS-assigned API port.' + ); + }); + + it('up shuts down a port-zero broker when startup status validation rejects', async () => { + const relay = createRelayMock({ + apiPort: 43123, + getStatus: vi.fn(async () => { + throw new Error('startup status unavailable'); + }), + }); + const { program, deps } = createHarness({ + relay, + env: { AGENT_RELAY_BROKER_PORT: '0' }, + }); + + const exitCode = await runCommand(program, ['up']); + + expect(exitCode).toBe(1); + expect(relay.shutdown).toHaveBeenCalledTimes(1); + expect(deps.error).toHaveBeenCalledWith('Failed to start broker: startup status unavailable'); + }); + + it('up shuts down a fixed-port broker when startup status validation rejects', async () => { + const relay = createRelayMock({ + getStatus: vi.fn(async () => { + throw new Error('startup status unavailable'); + }), + }); + const { program, deps } = createHarness({ relay }); + + const exitCode = await runCommand(program, ['up']); + + expect(exitCode).toBe(1); + expect(relay.shutdown).toHaveBeenCalledTimes(1); + expect(deps.error).toHaveBeenCalledWith('Failed to start broker: startup status unavailable'); + }); + it('up enables the local broker API', async () => { const relay = createRelayMock(); const { program, deps } = createHarness({ relay }); diff --git a/packages/cli/src/cli/commands/core.ts b/packages/cli/src/cli/commands/core.ts index 19aab00f3..5b5008160 100644 --- a/packages/cli/src/cli/commands/core.ts +++ b/packages/cli/src/cli/commands/core.ts @@ -61,6 +61,8 @@ export interface CoreRelay { workspaceKey?: string; /** PID of the underlying broker process, when available. */ brokerPid?: number; + /** Actual HTTP API port bound by the broker, including OS-assigned ports. */ + apiPort?: number; } export interface CoreFileSystem { @@ -148,11 +150,13 @@ async function createDefaultRelay( brokerName?: string, verbose = false ): Promise { - const binaryArgs: BrokerInitArgs = {}; - if (apiPort > 0) { - binaryArgs.persist = true; - binaryArgs.apiPort = apiPort; - } + // This is the `up` command's broker factory. `up` is persistent even when + // port 0 delegates atomic port selection to the OS; the connection file is + // how later `status`, `down`, and enrolled-node recovery find that broker. + const binaryArgs: BrokerInitArgs = { + persist: true, + apiPort, + }; const stateDir = process.env.AGENT_RELAY_STATE_DIR; if (stateDir) { binaryArgs.stateDir = stateDir; @@ -186,6 +190,10 @@ async function createDefaultRelay( get brokerPid() { return client.brokerPid; }, + get apiPort() { + const port = Number.parseInt(new URL(client.baseUrl).port, 10); + return Number.isInteger(port) && port > 0 ? port : undefined; + }, }; return relay; } diff --git a/packages/cli/src/cli/commands/node.test.ts b/packages/cli/src/cli/commands/node.test.ts index 298a0c2fd..572481e2d 100644 --- a/packages/cli/src/cli/commands/node.test.ts +++ b/packages/cli/src/cli/commands/node.test.ts @@ -105,6 +105,26 @@ describe('registerNodeCommands', () => { expect(up.options.map((option) => option.long)).toContain('--config'); }); + it('defaults node startup to an atomically OS-assigned broker API port', async () => { + const { program, env } = createNodeHarness(); + + await program.parseAsync(['node', 'up'], { from: 'user' }); + + expect(env.AGENT_RELAY_BROKER_PORT).toBe('0'); + expect(brokerMocks.runUpCommand).toHaveBeenCalledTimes(1); + }); + + it('preserves an explicit broker base port for node startup', async () => { + const { program, env } = createNodeHarness({ + env: { AGENT_RELAY_BROKER_PORT: '4100' }, + }); + + await program.parseAsync(['node', 'up'], { from: 'user' }); + + expect(env.AGENT_RELAY_BROKER_PORT).toBe('4100'); + expect(brokerMocks.runUpCommand).toHaveBeenCalledTimes(1); + }); + it('picks up a persisted enrollment and wires its creds into the env', async () => { const resolveEnrollment = vi.fn( () => enrollmentRecord diff --git a/packages/cli/src/cli/commands/node.ts b/packages/cli/src/cli/commands/node.ts index a8bf1bc13..36711aa53 100644 --- a/packages/cli/src/cli/commands/node.ts +++ b/packages/cli/src/cli/commands/node.ts @@ -159,6 +159,10 @@ function applyResolvedNodeSession( */ async function runNodeUp(options: UpCommandOptions, deps: NodeCommandDependencies): Promise { const env = deps.core.env; + // Fleet nodes may be started concurrently on one machine. Let the broker + // bind an ephemeral API port atomically unless the operator explicitly + // selected a stable broker base port. + env.AGENT_RELAY_BROKER_PORT ??= '0'; // An explicit workspace key (flag or env) is a direct workspace choice; the // enrollment store records workspace ids, not keys, so a stored enrollment // cannot be matched against it — skip pickup entirely rather than risk diff --git a/packages/cli/src/cli/lib/broker-lifecycle.ts b/packages/cli/src/cli/lib/broker-lifecycle.ts index d05266c39..f07163594 100644 --- a/packages/cli/src/cli/lib/broker-lifecycle.ts +++ b/packages/cli/src/cli/lib/broker-lifecycle.ts @@ -317,12 +317,13 @@ async function resolveApiPortWithFallback( /** * The broker base port. `AGENT_RELAY_BROKER_PORT` overrides the default so - * multiple brokers can run side by side (e.g. in tests); the broker HTTP API - * binds near `basePort + 1` with fallback scanning. + * multiple brokers can run side by side. A value of `0` asks the OS to assign + * the API port atomically during broker bind, which avoids probe-then-bind + * races in concurrent test stacks. */ export function resolveBrokerBasePort(deps: Pick): number { const raw = Number.parseInt(deps.env.AGENT_RELAY_BROKER_PORT ?? '', 10); - return Number.isFinite(raw) && raw > 0 ? raw : DEFAULT_BROKER_BASE_PORT; + return Number.isFinite(raw) && raw >= 0 ? raw : DEFAULT_BROKER_BASE_PORT; } export async function startBrokerWithPortFallback( @@ -332,6 +333,30 @@ export async function startBrokerWithPortFallback( brokerName?: string, verbose?: boolean ): Promise<{ relay: CoreRelay; apiPort: number }> { + if (basePort === 0) { + vlog(deps, verbose, 'Asking the OS to assign the broker API port...'); + const candidate = await deps.createRelay(paths.projectRoot, 0, brokerName, verbose); + try { + await candidate.getStatus(); + if (!candidate.apiPort) { + throw new Error('Broker started without reporting its OS-assigned API port.'); + } + } catch (startupError) { + try { + await candidate.shutdown(); + } catch (cleanupError) { + throw new AggregateError( + [startupError, cleanupError], + 'Broker startup validation failed and cleanup also failed.', + { cause: startupError } + ); + } + throw startupError; + } + vlog(deps, verbose, `API port assigned: ${candidate.apiPort}`); + return { relay: candidate, apiPort: candidate.apiPort }; + } + // Resolve a free API port BEFORE spawning the broker. This avoids // spawning (and flocking) multiple --persist brokers during retry, // which caused stale-flock "already running" errors. @@ -344,7 +369,20 @@ export async function startBrokerWithPortFallback( const candidate = await deps.createRelay(paths.projectRoot, apiPort, brokerName, verbose); vlog(deps, verbose, 'Broker client created. Checking broker status...'); - await candidate.getStatus(); + try { + await candidate.getStatus(); + } catch (startupError) { + try { + await candidate.shutdown(); + } catch (cleanupError) { + throw new AggregateError( + [startupError, cleanupError], + 'Broker startup validation failed and cleanup also failed.', + { cause: startupError } + ); + } + throw startupError; + } vlog(deps, verbose, 'Broker status check passed.'); return { relay: candidate, apiPort }; } diff --git a/packages/cli/src/cli/lib/sdk-client.test.ts b/packages/cli/src/cli/lib/sdk-client.test.ts index 439bcd301..65e947453 100644 --- a/packages/cli/src/cli/lib/sdk-client.test.ts +++ b/packages/cli/src/cli/lib/sdk-client.test.ts @@ -5,6 +5,7 @@ import path from 'node:path'; import { afterEach, beforeEach, describe, expect, it } from 'vitest'; import { + createAgentRelay, resolveAgentToken, resolveBaseUrl, resolveWorkspaceKey, @@ -105,4 +106,21 @@ describe('sdk client option resolution', () => { expect(resolveAgentToken({ token: ' at_123 ' })).toBe('at_123'); expect(resolveAgentToken({ token: ' ', env: { RELAY_AGENT_TOKEN: ' at_env ' } })).toBe('at_env'); }); + + it('uses an agent token as the transport credential instead of an ambient owner workspace key', () => { + setWorkspaceKey('ops', 'rk_live_owner_secret'); + writeProjectWorkspaceKey(projectDataDir(), 'rk_live_project_owner_secret'); + + const relay = createAgentRelay({ + env: { + AGENT_RELAY_HOME: dir, + RELAY_AGENT_TOKEN: 'at_live_participant_scoped', + }, + }) as { workspaceKey?: string; toJSON(): unknown }; + + expect(relay.workspaceKey).toBeUndefined(); + expect(JSON.stringify(relay)).not.toContain('rk_live_owner_secret'); + expect(JSON.stringify(relay)).not.toContain('rk_live_project_owner_secret'); + expect(JSON.stringify(relay)).not.toContain('at_live_participant_scoped'); + }); }); diff --git a/packages/cli/src/cli/lib/sdk-client.ts b/packages/cli/src/cli/lib/sdk-client.ts index 3082928d5..ae1e2b216 100644 --- a/packages/cli/src/cli/lib/sdk-client.ts +++ b/packages/cli/src/cli/lib/sdk-client.ts @@ -69,9 +69,18 @@ export function createWorkspaceRelay(options: SdkClientOptions = {}): AgentRelay */ export function createAgentRelay(options: SdkClientOptions = {}): AgentRelayAgent { const token = resolveAgentToken(options); + // Agent tokens are valid Relaycast transport credentials and already bind + // the caller to exactly one workspace. Prefer the scoped token itself over + // every ambient workspace-key source so invited humans cannot accidentally + // inherit the local owner's rk_live credential from this project or machine. + if (token) { + return new AgentRelay({ + agentToken: token, + baseUrl: resolveBaseUrl(options), + }); + } return new AgentRelay({ workspaceKey: resolveWorkspaceKey(options), baseUrl: resolveBaseUrl(options), - ...(token ? { agentToken: token } : {}), }); } diff --git a/packages/harness-driver/src/spawn-config.ts b/packages/harness-driver/src/spawn-config.ts index aa65d76ff..f3c95b235 100644 --- a/packages/harness-driver/src/spawn-config.ts +++ b/packages/harness-driver/src/spawn-config.ts @@ -4,7 +4,7 @@ import type { EventBus } from './event-bus.js'; import type { HarnessDriverEvents } from './lifecycle-hooks.js'; export interface BrokerInitArgs { - /** Optional HTTP API port for the broker (0 = disabled). */ + /** Optional HTTP API port for the broker (0 = atomically OS-assigned). */ apiPort?: number; /** Bind address for the HTTP API. Defaults to 127.0.0.1 in the broker. */ apiBind?: string; diff --git a/packages/sdk/src/__tests__/agent-relay.test.ts b/packages/sdk/src/__tests__/agent-relay.test.ts index 6c48e35cf..be3c883cd 100644 --- a/packages/sdk/src/__tests__/agent-relay.test.ts +++ b/packages/sdk/src/__tests__/agent-relay.test.ts @@ -1,3 +1,4 @@ +import { inspect } from 'node:util'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; const relaycastMocks = vi.hoisted(() => { @@ -98,6 +99,44 @@ describe('AgentRelay workspace setup', () => { }); }); + it('uses an agent token as the only Relaycast transport credential', () => { + const relay = new AgentRelay({ + agentToken: 'at_live_participant_scoped', + baseUrl: 'https://api.example.test', + }); + + expect(relay.workspaceKey).toBeUndefined(); + expect(relaycastMocks.relayCast).toHaveBeenCalledWith({ + apiKey: 'at_live_participant_scoped', + baseUrl: 'https://api.example.test', + }); + }); + + it('redacts credentials from JSON, object spread, and Node inspection', () => { + const relay = new AgentRelay({ + workspaceKey: 'rk_live_owner_marker', + agentToken: 'at_live_agent_marker', + observerToken: 'ot_live_observer_marker', + baseUrl: 'https://user:password@example.test', + }); + + const rendered = [ + JSON.stringify(relay), + JSON.stringify({ ...relay }), + inspect(relay), + inspect({ ...relay }), + ].join('\n'); + + expect(rendered).not.toContain('rk_live_owner_marker'); + expect(rendered).not.toContain('at_live_agent_marker'); + expect(rendered).not.toContain('ot_live_observer_marker'); + expect(rendered).not.toContain('password'); + expect(JSON.parse(JSON.stringify(relay))).toEqual({ + type: 'AgentRelay', + authenticated: true, + }); + }); + it('passes explicit Relaycast telemetry through existing workspace clients', () => { const relay = new AgentRelay({ workspaceKey: 'rk_live_existing', diff --git a/packages/sdk/src/agent-relay.ts b/packages/sdk/src/agent-relay.ts index d7d38ed4d..b8967b49a 100644 --- a/packages/sdk/src/agent-relay.ts +++ b/packages/sdk/src/agent-relay.ts @@ -190,6 +190,26 @@ export class AgentRelay implements AgentRelayAgent { if (onError) { this.errorHooks.add(onError); } + // Credentials can live in several implementation objects (the workspace + // key, observer token, messaging options, and agent-client map). Keep every + // own field non-enumerable so object spread and generic serializers cannot + // accidentally copy those values into logs. + for (const property of Object.keys(this)) { + Object.defineProperty(this, property, { enumerable: false }); + } + } + + /** Safe JSON/log representation. Deliberately excludes URLs and credentials. */ + toJSON(): { type: 'AgentRelay'; authenticated: boolean } { + return { + type: 'AgentRelay', + authenticated: Boolean(this.workspaceKey || this.observerToken || this.messagingOptions.agentToken), + }; + } + + /** Node's util.inspect hook follows the same credential-free contract. */ + [Symbol.for('nodejs.util.inspect.custom')](): ReturnType { + return this.toJSON(); } static async createWorkspace(input: string | AgentRelayCreateWorkspaceInput): Promise { diff --git a/packages/sdk/src/messaging/relaycast-client.ts b/packages/sdk/src/messaging/relaycast-client.ts index 80a99198a..8782d98ac 100644 --- a/packages/sdk/src/messaging/relaycast-client.ts +++ b/packages/sdk/src/messaging/relaycast-client.ts @@ -218,14 +218,16 @@ export interface RelaycastMessagingOptions extends RelaycastTelemetryOptions { export function createRelaycastClient(options: RelaycastMessagingOptions): RelaycastWorkspaceLike { if (options.relaycast) return options.relaycast; - const workspaceKey = options.workspaceKey ?? options.apiKey; - if (!workspaceKey) { - throw new Error('RelaycastMessagingClient requires workspaceKey when relaycast is not provided.'); + const credential = options.workspaceKey ?? options.apiKey ?? options.agentToken; + if (!credential) { + throw new Error( + 'RelaycastMessagingClient requires workspaceKey or agentToken when relaycast is not provided.' + ); } return new RelayCast( definedOptions({ - apiKey: workspaceKey, + apiKey: credential, baseUrl: options.baseUrl, retryPolicy: options.retryPolicy, ...relaycastTelemetryOptions({ diff --git a/tests/e2e/fleet/fleet-e2e.test.ts b/tests/e2e/fleet/fleet-e2e.test.ts index f295a9709..2a2700abc 100644 --- a/tests/e2e/fleet/fleet-e2e.test.ts +++ b/tests/e2e/fleet/fleet-e2e.test.ts @@ -79,7 +79,6 @@ describe.skipIf(!pre.ok)('Cloud-enrolled node startup', () => { engineBaseUrl: engine.baseUrl, brokerBinary: pre.brokerBinary!, tmpRoot, - brokerPort: await getFreePort(), capacityHarnesses: 'claude', usePersistedEnrollment: true, }); @@ -102,9 +101,17 @@ describe.skipIf(!pre.ok)('Cloud-enrolled node startup', () => { async () => { const nodes = await getNodes(engine, workspaceKey, { name: 'cloud-enrolled' }); const match = nodes.find((node) => node.id === 'node_cloud_enrolled'); - return match?.live && match.handlers_live ? match : null; + // The broker provider is independently ready for spawn/release before + // the config-backed action provider finishes registering. + return match?.live && + match.handlers_live && + match.capabilities.some((capability) => capability.name === 'cloud:ping') && + match.tags?.includes('cloud-enrolled') && + match.tags.includes('e2e') + ? match + : null; }, - { timeoutMs: 30_000, label: 'Cloud-enrolled node online with live handlers' } + { timeoutMs: 30_000, label: 'Cloud-enrolled node online with broker and action handlers' } ); expect(enrolled.name).toBe('cloud-enrolled'); @@ -176,7 +183,6 @@ describe.skipIf(!pre.ok)('two-node fleet scenario matrix', () => { engineBaseUrl: engine.baseUrl, brokerBinary: pre.brokerBinary!, tmpRoot, - brokerPort: await getFreePort(), // Pin capacity so the node advertises a distinct harness (`claude`) plus the // shared `pool`. A `spawn:` shadow delegates to the broker's native // capacity for that harness, so every shadow the node defines (spawn:claude, @@ -193,7 +199,6 @@ describe.skipIf(!pre.ok)('two-node fleet scenario matrix', () => { engineBaseUrl: engine.baseUrl, brokerBinary: pre.brokerBinary!, tmpRoot, - brokerPort: await getFreePort(), // Distinct `codex` plus the shared `pool` (see node-a's note). capacityHarnesses: 'codex,pool', }); @@ -207,9 +212,22 @@ describe.skipIf(!pre.ok)('two-node fleet scenario matrix', () => { const nodes = await getNodes(engine, workspaceKey); const a = node(nodes, 'node-a'); const b = node(nodes, 'node-b'); - return a?.live && a.handlers_live && b?.live && b.handlers_live ? nodes : null; + const aCapabilities = new Set(a?.capabilities.map((capability) => capability.name)); + const bCapabilities = new Set(b?.capabilities.map((capability) => capability.name)); + // handlers_live covers the broker provider too, so wait for the + // separately connected action providers before asserting their union. + return a?.live && + a.handlers_live && + aCapabilities.has('echo') && + aCapabilities.has('work') && + b?.live && + b.handlers_live && + bCapabilities.has('ping') && + bCapabilities.has('work') + ? nodes + : null; }, - { timeoutMs: 45_000, label: 'both nodes online+handlers_live' } + { timeoutMs: 45_000, label: 'both nodes online with broker and action handlers' } ); }, 60_000); @@ -269,7 +287,6 @@ describe.skipIf(!pre.ok)('two-node fleet scenario matrix', () => { engineBaseUrl: engine.baseUrl, brokerBinary: pre.brokerBinary!, tmpRoot, - brokerPort: await getFreePort(), }); badNode.start(); try { diff --git a/tests/e2e/fleet/harness.ts b/tests/e2e/fleet/harness.ts index 52b8f0888..cae3909e1 100644 --- a/tests/e2e/fleet/harness.ts +++ b/tests/e2e/fleet/harness.ts @@ -371,7 +371,9 @@ export class FleetNode { engineBaseUrl: string; brokerBinary: string; tmpRoot: string; - brokerPort: number; + /** Optional explicit broker base port; omission exercises the production + * `node up` default of an atomically OS-assigned API port. */ + brokerPort?: number; /** Pins the broker's `spawn:` capacity set (AGENT_RELAY_NODE_HARNESSES) * so two nodes on one host advertise distinct capabilities. */ capacityHarnesses?: string; @@ -498,7 +500,7 @@ export class FleetNode { }), AGENT_RELAY_PROJECT: this.projectDir, AGENT_RELAY_STATE_DIR: stateDir, - AGENT_RELAY_BROKER_PORT: String(o.brokerPort), + ...(o.brokerPort === undefined ? {} : { AGENT_RELAY_BROKER_PORT: String(o.brokerPort) }), ...(o.capacityHarnesses ? { AGENT_RELAY_NODE_HARNESSES: o.capacityHarnesses } : {}), }), stdio: ['ignore', 'pipe', 'pipe'],