diff --git a/Cargo.lock b/Cargo.lock index d9a47e17c..0159860fc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3756,6 +3756,7 @@ dependencies = [ "infer", "nostr", "reqwest 0.13.3", + "rustls", "serde", "serde_json", "sha2 0.11.0", diff --git a/crates/sprout-acp/Cargo.toml b/crates/sprout-acp/Cargo.toml index a4f24860c..e3f2505c2 100644 --- a/crates/sprout-acp/Cargo.toml +++ b/crates/sprout-acp/Cargo.toml @@ -76,3 +76,9 @@ evalexpr = { workspace = true } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } +nostr = { workspace = true } +tokio-tungstenite = { workspace = true } +futures-util = { workspace = true } +serde_json = { workspace = true } +uuid = { workspace = true } +rustls = { version = "0.23", default-features = false, features = ["ring", "std"] } diff --git a/crates/sprout-acp/src/config.rs b/crates/sprout-acp/src/config.rs index 71b6c71b6..47a4b3d84 100644 --- a/crates/sprout-acp/src/config.rs +++ b/crates/sprout-acp/src/config.rs @@ -810,10 +810,19 @@ impl Config { )); } + // Serverless if explicitly flagged OR the relay URL is a comma-separated + // LIST. A multi-relay URL is only ever produced by a serverless + // workspace, so this is a robust fallback when SPROUT_SERVERLESS wasn't + // set true at launch (e.g. the agent was restored before the desktop + // applied the workspace's serverless flag). Without this the agent runs + // in server mode against public relays → "No auth challenge received" + // and an HTTP-bridge channel-discovery crash. + let serverless = args.serverless || args.relay_url.contains(','); + let config = Config { keys, relay_url: args.relay_url, - serverless: args.serverless, + serverless, agent_command, agent_args, mcp_command: args.mcp_command, @@ -860,8 +869,9 @@ impl Config { other => format!("respond_to={other}"), }; format!( - "relay={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}", + "relay={} serverless={} pubkey={} agent_cmd={} {} mcp_cmd={} idle_timeout={}s max_turn={}s agents={} heartbeat={}s subscribe={:?} dedup={:?} meh={:?} ignore_self={} context_limit={} max_turns_per_session={} presence={} typing={} memory={} model={} permission_mode={} {}", self.relay_url, + self.serverless, self.keys.public_key().to_hex(), self.agent_command, self.agent_args.join(" "), diff --git a/crates/sprout-acp/src/relay.rs b/crates/sprout-acp/src/relay.rs index bd67670fb..dd156cfa8 100644 --- a/crates/sprout-acp/src/relay.rs +++ b/crates/sprout-acp/src/relay.rs @@ -107,8 +107,13 @@ pub struct RestClient { /// When true, `base_url` is ignored and `ws_url` is used. See /// docs/SPROUT_LITE_MODE.md. pub serverless: bool, - /// WebSocket URL of the relay (used only in serverless mode). - pub ws_url: String, + /// WebSocket URLs of the relays (used only in serverless mode). In + /// serverless we fan out across ALL relays: queries merge+dedup results + /// and publishes succeed if ANY relay accepts. Relays don't gossip, so a + /// read must union every relay and a write must tolerate one relay being + /// down or rate-limiting — the standard Nostr client model (damus + /// `RelayPool`, nostr-tools `SimplePool`). + pub ws_urls: Vec, } /// Whether an HTTP status code is retriable (transient server/rate-limit errors). @@ -307,10 +312,45 @@ impl RestClient { /// NIP-42 AUTH challenge if the relay sends one. Returns a JSON array of /// event objects (same shape as the HTTP `/query` bridge). async fn query_ws(&self, filters: &[nostr::Filter]) -> Result { + // Fan out across all relays, merge results, dedup by event id. Relays + // don't gossip — a complete read must union every relay. A single + // relay failing (503, connect error) does NOT fail the whole query as + // long as at least one relay answered. + let mut by_id: std::collections::HashMap = std::collections::HashMap::new(); + let mut last_err: Option = None; + let mut any_ok = false; + for relay in &self.ws_urls { + match self.query_ws_one(relay, filters).await { + Ok(events) => { + any_ok = true; + for ev in events { + if let Some(id) = ev.get("id").and_then(|v| v.as_str()) { + by_id.entry(id.to_string()).or_insert(ev); + } + } + } + Err(e) => { + tracing::warn!("query relay failed (continuing): {e}"); + last_err = Some(e); + } + } + } + if !any_ok { + return Err(last_err.unwrap_or(RelayError::ConnectionClosed)); + } + Ok(Value::Array(by_id.into_values().collect())) + } + + /// Query a single relay over plain WS (serverless mode): REQ → collect + /// EVENTs until EOSE → CLOSE. Answers a NIP-42 AUTH challenge if sent. + async fn query_ws_one( + &self, + relay: &str, + filters: &[nostr::Filter], + ) -> Result, RelayError> { use futures_util::{SinkExt, StreamExt}; - let parsed = self - .ws_url + let parsed = relay .parse::() .map_err(|e| RelayError::Http(format!("invalid relay URL: {e}")))?; let (ws, _) = timeout(CONNECT_TIMEOUT, connect_async(parsed.as_str())) @@ -356,7 +396,7 @@ impl RestClient { "CLOSED" if sub_matches => return Ok(()), "AUTH" => { if let Some(challenge) = arr.get(1).and_then(|v| v.as_str()) { - if let Ok(json) = self.ws_auth_message(challenge) { + if let Ok(json) = self.ws_auth_message(relay, challenge) { let _ = write.send(Message::Text(json.into())).await; } } @@ -379,12 +419,12 @@ impl RestClient { // Timeout or early close → return whatever arrived (public relays are // often slow to EOSE; partial results beat a hard failure). match collect { - Ok(Ok(())) | Err(_) => Ok(Value::Array(events)), + Ok(Ok(())) | Err(_) => Ok(events), Ok(Err(e)) => { if events.is_empty() { Err(e) } else { - Ok(Value::Array(events)) + Ok(events) } } } @@ -393,10 +433,42 @@ impl RestClient { /// Publish a signed event over a plain WebSocket (serverless mode) and wait /// for the relay's `OK`. Answers a NIP-42 AUTH challenge if sent. async fn submit_event_ws(&self, event: &Event) -> Result { + // Fan out to every relay; succeed if ANY accepts. This makes the + // agent's replies resilient to a single relay rate-limiting ("noting + // too much") or being down — exactly what the desktop relay pool does. + let event_id = event.id.to_hex(); + let mut last_err: Option = None; + for relay in &self.ws_urls { + match self.submit_event_ws_one(relay, event).await { + Ok(v) => { + let accepted = v.get("accepted").and_then(|b| b.as_bool()).unwrap_or(false); + if accepted { + return Ok(v); + } + let msg = v + .get("message") + .and_then(|m| m.as_str()) + .unwrap_or("") + .to_string(); + tracing::warn!("relay {relay} rejected event (trying next): {msg}"); + last_err = Some(RelayError::Http(format!("relay rejected: {msg}"))); + } + Err(e) => { + tracing::warn!("relay {relay} publish failed (trying next): {e}"); + last_err = Some(e); + } + } + } + Err(last_err + .unwrap_or_else(|| RelayError::Http(format!("no relay accepted event {event_id}")))) + } + + /// Publish a signed event to a single relay over plain WS (serverless mode) + /// and wait for the relay's `OK`. Answers a NIP-42 AUTH challenge if sent. + async fn submit_event_ws_one(&self, relay: &str, event: &Event) -> Result { use futures_util::{SinkExt, StreamExt}; - let parsed = self - .ws_url + let parsed = relay .parse::() .map_err(|e| RelayError::Http(format!("invalid relay URL: {e}")))?; let (ws, _) = timeout(CONNECT_TIMEOUT, connect_async(parsed.as_str())) @@ -444,7 +516,7 @@ impl RestClient { } "AUTH" => { if let Some(challenge) = arr.get(1).and_then(|v| v.as_str()) { - if let Ok(json) = self.ws_auth_message(challenge) { + if let Ok(json) = self.ws_auth_message(relay, challenge) { let _ = write.send(Message::Text(json.into())).await; let _ = write.send(Message::Text(event_msg.clone().into())).await; } @@ -475,8 +547,8 @@ impl RestClient { } /// Build a NIP-42 `["AUTH", ]` message string for serverless writes. - fn ws_auth_message(&self, challenge: &str) -> Result { - let url = nostr::RelayUrl::parse(&self.ws_url) + fn ws_auth_message(&self, relay: &str, challenge: &str) -> Result { + let url = nostr::RelayUrl::parse(relay) .map_err(|e| RelayError::Http(format!("invalid relay URL: {e}")))?; let event = EventBuilder::auth(challenge.to_string(), url) .sign_with_keys(&self.keys) @@ -622,8 +694,12 @@ pub struct HarnessRelay { cmd_tx: mpsc::Sender, /// HTTP client for HTTP bridge calls. http: reqwest::Client, - /// WebSocket URL of the relay. + /// Primary relay URL (first in the list). Used for HTTP-bridge (server) + /// mode and as a display label. relay_url: String, + /// Full relay list (serverless mode). Reads/writes fan out across all of + /// these — see `RestClient` for the merge/first-accepts semantics. + relay_urls: Vec, /// Keys used for NIP-42 signing and NIP-98 HTTP auth. keys: Keys, /// Optional NIP-OA auth tag for relay membership delegation. @@ -774,6 +850,7 @@ impl HarnessRelay { .build() .map_err(|e| RelayError::Http(format!("failed to build HTTP client: {e}")))?, relay_url: primary, + relay_urls, keys: keys.clone(), auth_tag, serverless, @@ -906,7 +983,11 @@ impl HarnessRelay { .as_ref() .and_then(|t| serde_json::to_string(t.as_slice()).ok()), serverless: self.serverless, - ws_url: self.relay_url.clone(), + ws_urls: if self.relay_urls.is_empty() { + vec![self.relay_url.clone()] + } else { + self.relay_urls.clone() + }, } } @@ -2081,7 +2162,31 @@ async fn handle_ws_message( ); if is_auth_error { - // Auth errors require a full reconnect (re-handshake). + // Serverless: a generic/paid public relay (e.g. nostr.land, + // nostr.wine) may permanently require auth or payment to + // read. Re-handshaking won't help — reconnecting just hits + // the same auth-required CLOSED forever (a reconnect storm + // that churns ALL relays and starves message processing). + // Drop this subscription on THIS relay and keep the + // connection alive; the OTHER relays still serve the agent. + if state.serverless { + warn!( + "serverless: relay {relay_url} rejected subscription {subscription_id} ({message}) — \ + dropping it on this relay (other relays still serve the agent), not reconnecting" + ); + // Return true (keep the connection alive, NO reconnect) + // and stop tracking this sub on this relay so we don't + // resubscribe it into an infinite auth-required storm. + if let Some(channel_id) = channel_id_from_sub_id(&subscription_id) { + state.active_subscriptions.remove(&channel_id); + } else if subscription_id == OBSERVER_CONTROL_SUB_ID { + state.observer_control_sub_active = false; + } else if subscription_id == MEMBERSHIP_NOTIF_SUB_ID { + state.membership_sub_active = false; + } + return true; + } + // Server mode: auth errors require a full reconnect (re-handshake). return false; } diff --git a/crates/sprout-acp/tests/e2e_agent_responds.rs b/crates/sprout-acp/tests/e2e_agent_responds.rs new file mode 100644 index 000000000..dc6451114 --- /dev/null +++ b/crates/sprout-acp/tests/e2e_agent_responds.rs @@ -0,0 +1,268 @@ +//! End-to-end test: a serverless agent actually RESPONDS in a channel. +//! +//! This is the test that proves the whole chain works, against real public +//! relays, with no LLM: +//! +//! 1. create a channel (kind:39000) + add the agent as a member (kind:39002) +//! 2. spawn the real `sprout-acp` binary with a STUB agent (a shell script +//! that speaks minimal ACP and posts a reply via the `sprout` CLI) +//! 3. publish a message mentioning the agent +//! 4. assert the agent's reply lands on the relay +//! +//! It exercises: serverless detection (comma relay list), multi-relay connect, +//! channel discovery (39002 over WS), subscription, the respond gate, the ACP +//! prompt turn, and the reply publish via the `sprout` CLI. +//! +//! Run with: +//! cargo test -p sprout-acp --test e2e_agent_responds -- --ignored --nocapture + +use std::process::Stdio; +use std::time::Duration; + +use futures_util::{SinkExt, StreamExt}; +use nostr::{EventBuilder, Keys, Kind, ToBech32}; +use tokio::process::Command; +use tokio_tungstenite::{connect_async, tungstenite::Message}; + +const RELAYS_DEFAULT: &str = "wss://relay.damus.io,wss://nos.lol"; + +fn relays() -> String { + std::env::var("RELAY_URL").unwrap_or_else(|_| RELAYS_DEFAULT.to_string()) +} + +/// Publish a signed event to one relay over plain WS and wait briefly for OK. +/// Tolerant: a relay hiccup (503, connect error, rejected) does not panic — +/// we publish to multiple relays and only need one to accept. +async fn publish(relay: &str, event: &nostr::Event) -> bool { + let ws = match connect_async(relay).await { + Ok((ws, _)) => ws, + Err(e) => { + eprintln!(" (publish connect to {relay} failed: {e} — skipping)"); + return false; + } + }; + let (mut write, mut read) = ws.split(); + let msg = serde_json::json!(["EVENT", event]).to_string(); + if write.send(Message::Text(msg.into())).await.is_err() { + return false; + } + // Drain briefly for the OK; report whether it was accepted. + let accepted = tokio::time::timeout(Duration::from_secs(3), async { + while let Some(Ok(m)) = read.next().await { + if let Message::Text(t) = m { + if let Ok(v) = serde_json::from_str::(&t) { + let arr = v.as_array().cloned().unwrap_or_default(); + if arr.first().and_then(|x| x.as_str()) == Some("OK") { + let ok = arr.get(2).and_then(|x| x.as_bool()).unwrap_or(false); + if !ok { + eprintln!(" (publish to {relay} rejected: {t})"); + } + return ok; + } + } + } + } + false + }) + .await + .unwrap_or(false); + let _ = write.close().await; + accepted +} + +/// Publish to every relay in a comma list; returns true if any accepted. +async fn publish_all(relay_list: &str, event: &nostr::Event) -> bool { + let mut any = false; + for r in relay_list.split(',') { + if publish(r.trim(), event).await { + any = true; + } + } + any +} + +/// Query one relay for events matching a filter; collect until EOSE/timeout. +/// Tolerant: a relay hiccup (503, connect error) returns empty rather than +/// panicking — the caller retries across attempts/relays. +async fn query(relay: &str, filter: serde_json::Value) -> Vec { + let ws = match connect_async(relay).await { + Ok((ws, _)) => ws, + Err(e) => { + eprintln!(" (query connect to {relay} failed: {e} — treating as empty)"); + return Vec::new(); + } + }; + let (mut write, mut read) = ws.split(); + let sub = "q1"; + let req = serde_json::json!(["REQ", sub, filter]).to_string(); + write.send(Message::Text(req.into())).await.expect("req"); + let mut out = Vec::new(); + let _ = tokio::time::timeout(Duration::from_secs(8), async { + while let Some(Ok(m)) = read.next().await { + if let Message::Text(t) = m { + let v: serde_json::Value = match serde_json::from_str(&t) { + Ok(v) => v, + Err(_) => continue, + }; + let arr = v.as_array().cloned().unwrap_or_default(); + match arr.first().and_then(|x| x.as_str()) { + Some("EVENT") if arr.get(1).and_then(|x| x.as_str()) == Some(sub) => { + if let Some(ev) = arr.get(2) { + if let Ok(e) = serde_json::from_value::(ev.clone()) { + out.push(e); + } + } + } + Some("EOSE") => break, + _ => {} + } + } + } + }) + .await; + let _ = write.close().await; + out +} + +#[tokio::test] +#[ignore = "network: hits live public relays; spawns sprout-acp + sprout binaries"] +async fn agent_responds_in_channel_e2e() { + let _ = rustls::crypto::ring::default_provider().install_default(); + + let relay_list = relays(); + + // Locate the built binaries (same target dir as this test binary). + let acp_bin = env!("CARGO_BIN_EXE_sprout-acp"); + // sprout CLI lives next to it in the target dir. + let target_dir = std::path::Path::new(acp_bin) + .parent() + .unwrap() + .to_path_buf(); + let sprout_bin = target_dir.join("sprout"); + assert!( + sprout_bin.exists(), + "sprout CLI not built at {sprout_bin:?} — run `cargo build -p sprout-cli` first" + ); + let stub = concat!(env!("CARGO_MANIFEST_DIR"), "/tests/stub_agent.sh"); + + // Identities: the human (creator) and the agent. + let human = Keys::generate(); + let agent = Keys::generate(); + let channel = uuid::Uuid::new_v4().to_string(); + let agent_pk = agent.public_key().to_hex(); + let human_pk = human.public_key().to_hex(); + let reply_marker = format!("stub-reply-{}", &channel[..8]); + + eprintln!("channel={channel}\nhuman={human_pk}\nagent={agent_pk}\nrelays={relay_list}"); + + // 1. Channel metadata (39000) + members (39002 with BOTH human and agent). + let meta = EventBuilder::new(Kind::Custom(39000), "") + .tags(vec![ + nostr::Tag::parse(["d", &channel]).unwrap(), + nostr::Tag::parse(["name", "e2e-agent-test"]).unwrap(), + nostr::Tag::parse(["t", "stream"]).unwrap(), + nostr::Tag::parse(["public"]).unwrap(), + ]) + .sign_with_keys(&human) + .unwrap(); + let members = EventBuilder::new(Kind::Custom(39002), "") + .tags(vec![ + nostr::Tag::parse(["d", &channel]).unwrap(), + nostr::Tag::parse(["p", &human_pk, "", "owner"]).unwrap(), + nostr::Tag::parse(["p", &agent_pk, "", "member"]).unwrap(), + ]) + .sign_with_keys(&human) + .unwrap(); + // Publish membership to all relays so discovery finds it. + let meta_ok = publish_all(&relay_list, &meta).await; + let members_ok = publish_all(&relay_list, &members).await; + assert!( + meta_ok && members_ok, + "no relay accepted channel metadata/membership (meta_ok={meta_ok}, members_ok={members_ok}) — relays may be down/rate-limiting" + ); + eprintln!("published channel metadata + membership"); + + // 2. Spawn the real sprout-acp harness with the stub agent. + let log_path = std::env::temp_dir().join(format!("acp-e2e-{}.log", &channel[..8])); + let harness_log_path = + std::env::temp_dir().join(format!("acp-e2e-harness-{}.log", &channel[..8])); + let harness_log = std::fs::File::create(&harness_log_path).expect("create harness log"); + // tracing logs go to stdout via `fmt()`; capture both stdout+stderr so the + // diagnostic dump shows discovery/subscribe/dispatch. + let harness_log_out = harness_log.try_clone().expect("clone harness log"); + let mut child = Command::new(acp_bin) + .env("SPROUT_RELAY_URL", &relay_list) + .env( + "SPROUT_PRIVATE_KEY", + agent.secret_key().to_bech32().unwrap(), + ) + .env("SPROUT_ACP_AGENT_COMMAND", "bash") + .env("SPROUT_ACP_AGENT_ARGS", stub) + .env("SPROUT_ACP_RESPOND_TO", "anyone") + .env("SPROUT_ACP_SUBSCRIBE", "all") + .env("SPROUT_ACP_NO_MENTION_FILTER", "true") + .env("SPROUT_ACP_AGENTS", "1") + .env("STUB_AGENT_CHANNEL", &channel) + .env("STUB_AGENT_REPLY", &reply_marker) + .env("STUB_AGENT_SPROUT_BIN", &sprout_bin) + .env("STUB_AGENT_LOG", &log_path) + .env("RUST_LOG", "sprout_acp=debug") + .stdout(Stdio::from(harness_log_out)) + .stderr(Stdio::from(harness_log)) + .kill_on_drop(true) + .spawn() + .expect("spawn sprout-acp"); + + // Give the harness time to connect to all relays + discover the channel. + tokio::time::sleep(Duration::from_secs(8)).await; + + // 3. Publish a message into the channel (the human talking to the agent). + let msg = EventBuilder::new(Kind::Custom(9), "@agent hello, please reply") + .tags(vec![ + nostr::Tag::parse(["h", &channel]).unwrap(), + nostr::Tag::parse(["p", &agent_pk]).unwrap(), + ]) + .sign_with_keys(&human) + .unwrap(); + let msg_ok = publish_all(&relay_list, &msg).await; + assert!(msg_ok, "no relay accepted the @mention message"); + eprintln!("published @mention message; waiting for agent reply…"); + + // 4. Poll the relays for the agent's reply (kind 9 from agent, content + // marker). Query EVERY relay and merge — the reply may land on only one + // relay (the agent publishes to whichever accepts first), and a single + // relay may be 503-ing, so polling just one relay can miss it entirely. + let mut found = false; + for attempt in 0..20 { + tokio::time::sleep(Duration::from_secs(3)).await; + let mut events: Vec = Vec::new(); + for r in relay_list.split(',') { + let mut got = query( + r.trim(), + serde_json::json!({"kinds":[9],"#h":[channel],"limit":50}), + ) + .await; + events.append(&mut got); + } + if events + .iter() + .any(|e| e.content.contains(&reply_marker) && e.pubkey == agent.public_key()) + { + found = true; + eprintln!("✅ agent reply found on relay after {}s", (attempt + 1) * 3); + break; + } + eprintln!(" …attempt {attempt}: {} msgs, no reply yet", events.len()); + } + + let _ = child.kill().await; + if !found { + if let Ok(h) = std::fs::read_to_string(&harness_log_path) { + eprintln!("--- harness (sprout-acp) log ---\n{h}\n-----------------------------------"); + } + if let Ok(log) = std::fs::read_to_string(&log_path) { + eprintln!("--- stub agent log ---\n{log}\n----------------------"); + } + panic!("agent never posted a reply to the channel (see logs above)"); + } +} diff --git a/crates/sprout-acp/tests/stub_agent.sh b/crates/sprout-acp/tests/stub_agent.sh new file mode 100755 index 000000000..2075223b3 --- /dev/null +++ b/crates/sprout-acp/tests/stub_agent.sh @@ -0,0 +1,58 @@ +#!/usr/bin/env bash +# Minimal ACP agent stub for end-to-end testing. +# +# Speaks just enough of the Agent Client Protocol (JSON-RPC over stdio) to let +# the sprout-acp harness drive a full turn: +# - initialize → returns capabilities +# - session/new → returns a sessionId +# - session/prompt → posts a reply to the channel via the `sprout` CLI, +# then returns stopReason=end_turn +# +# It needs NO LLM. On a prompt it replies with a fixed marker so the test can +# assert the reply landed on the relay. The channel id is passed via the +# STUB_AGENT_CHANNEL env var; the reply text via STUB_AGENT_REPLY; the sprout +# binary via STUB_AGENT_SPROUT_BIN. Auth (SPROUT_RELAY_URL / SPROUT_PRIVATE_KEY) +# is inherited from the harness, exactly as the real agent receives it. + +set -euo pipefail + +reply_sent=0 + +send_response() { + # $1 = id, $2 = result json + printf '{"jsonrpc":"2.0","id":%s,"result":%s}\n' "$1" "$2" +} + +while IFS= read -r line; do + [ -z "$line" ] && continue + method=$(printf '%s' "$line" | sed -n 's/.*"method":"\([^"]*\)".*/\1/p') + id=$(printf '%s' "$line" | sed -n 's/.*"id":\([0-9]*\).*/\1/p') + + case "$method" in + initialize) + send_response "$id" '{"protocolVersion":1,"agentCapabilities":{"promptCapabilities":{"embeddedContext":true}}}' + ;; + session/new) + send_response "$id" '{"sessionId":"stub-session-1"}' + ;; + session/prompt) + if [ "$reply_sent" -eq 0 ]; then + reply_sent=1 + # Post the reply to the channel using the sprout CLI — the real reply + # mechanism per base_prompt.md. Auth env is inherited from the harness. + "${STUB_AGENT_SPROUT_BIN}" messages send \ + --channel "${STUB_AGENT_CHANNEL}" \ + --content "${STUB_AGENT_REPLY}" >/dev/null 2>>"${STUB_AGENT_LOG:-/dev/stderr}" || \ + echo "stub: sprout messages send failed" >>"${STUB_AGENT_LOG:-/dev/stderr}" + fi + send_response "$id" '{"stopReason":"end_turn"}' + ;; + "") + # Response or notification without a method — ignore. + ;; + *) + # Unknown request with an id — ack with empty result to avoid hangs. + [ -n "$id" ] && send_response "$id" '{}' + ;; + esac +done diff --git a/crates/sprout-cli/Cargo.toml b/crates/sprout-cli/Cargo.toml index 1836b00c0..33f5b4b99 100644 --- a/crates/sprout-cli/Cargo.toml +++ b/crates/sprout-cli/Cargo.toml @@ -28,6 +28,10 @@ tokio = { workspace = true, features = ["macros", "rt-multi-thread", "time"] } # WebSocket transport for serverless mode (generic relays, no HTTP bridge) tokio-tungstenite = { workspace = true } futures-util = { workspace = true } +# rustls provider install for the serverless WS transport. The CLI links +# aws-lc-rs (via reqwest's rustls); install that provider at startup so +# tokio-tungstenite's direct TLS config has a process-level provider. +rustls = { version = "0.23", default-features = false, features = ["aws_lc_rs", "std"] } # Serialization — JSON body building and response passthrough serde = { workspace = true } diff --git a/crates/sprout-cli/src/client.rs b/crates/sprout-cli/src/client.rs index e3d2eb8db..90f13e8f8 100644 --- a/crates/sprout-cli/src/client.rs +++ b/crates/sprout-cli/src/client.rs @@ -123,8 +123,11 @@ fn sign_nip98( pub struct SproutClient { http: reqwest::Client, relay_url: String, // base URL, no trailing slash, e.g. "https://relay.sprout.place" - /// WebSocket URL (ws/wss). Used only in serverless mode. - ws_url: String, + /// WebSocket URLs (ws/wss). Used only in serverless mode. In serverless we + /// fan out across all relays — publish succeeds if ANY relay accepts, and + /// queries merge+dedup results — matching the desktop relay pool and the + /// standard Nostr client model (relays don't gossip; clients talk to many). + ws_urls: Vec, /// Serverless mode: talk to a generic relay over plain WebSocket instead /// of the Sprout HTTP bridge. See docs/SPROUT_LITE_MODE.md. serverless: bool, @@ -144,6 +147,17 @@ impl SproutClient { auth_tag: Option, auth_tag_json: Option, ) -> Result { + // `ws_url` may be a comma-separated list of relays in serverless mode. + let ws_urls: Vec = ws_url + .split(',') + .map(|s| s.trim().to_string()) + .filter(|s| !s.is_empty()) + .collect(); + let ws_urls = if ws_urls.is_empty() { + vec![ws_url] + } else { + ws_urls + }; let http = reqwest::Client::builder() .timeout(Duration::from_secs(10)) .connect_timeout(Duration::from_secs(5)) @@ -152,7 +166,7 @@ impl SproutClient { Ok(Self { http, relay_url, - ws_url, + ws_urls, serverless, keys, auth_tag, @@ -312,10 +326,47 @@ impl SproutClient { /// JSON-array string of event objects (same shape as the HTTP `/query` /// bridge response, so downstream parsing is unchanged). async fn query_ws(&self, filters: &[serde_json::Value]) -> Result { + // Fan out across all relays, merge results, dedup by event id. Relays + // don't gossip — a given event may live on only one relay, so a + // complete read must union every relay (the standard Nostr client + // model; see damus RelayPool / nostr-tools SimplePool). + let mut by_id: std::collections::HashMap = + std::collections::HashMap::new(); + let mut last_err: Option = None; + let mut any_ok = false; + for relay in &self.ws_urls { + match self.query_ws_one(relay, filters).await { + Ok(events) => { + any_ok = true; + for ev in events { + if let Some(id) = ev.get("id").and_then(|v| v.as_str()) { + by_id.entry(id.to_string()).or_insert(ev); + } + } + } + Err(e) => last_err = Some(e), + } + } + if !any_ok { + if let Some(e) = last_err { + return Err(e); + } + } + let merged: Vec = by_id.into_values().collect(); + Ok(serde_json::Value::Array(merged).to_string()) + } + + /// Query a single relay over plain WS: REQ → collect EVENTs until EOSE → + /// CLOSE. Answers a NIP-42 AUTH challenge if the relay sends one. + async fn query_ws_one( + &self, + relay: &str, + filters: &[serde_json::Value], + ) -> Result, CliError> { use futures_util::{SinkExt, StreamExt}; use tokio_tungstenite::{connect_async, tungstenite::Message}; - let (ws, _) = connect_async(self.ws_url.as_str()) + let (ws, _) = connect_async(relay) .await .map_err(|e| CliError::NetworkMsg(format!("relay connect failed: {e}")))?; let (mut write, mut read) = ws.split(); @@ -360,7 +411,7 @@ impl SproutClient { "CLOSED" if sub_matches => return Ok(()), "AUTH" => { if let Some(challenge) = arr.get(1).and_then(|v| v.as_str()) { - if let Ok(json) = self.ws_auth_message(challenge) { + if let Ok(json) = self.ws_auth_message(relay, challenge) { let _ = write.send(Message::Text(json.into())).await; } } @@ -379,12 +430,12 @@ impl SproutClient { let _ = write.close().await; match collect { - Ok(Ok(())) | Err(_) => Ok(serde_json::Value::Array(events).to_string()), + Ok(Ok(())) | Err(_) => Ok(events), Ok(Err(e)) => { if events.is_empty() { Err(CliError::NetworkMsg(e)) } else { - Ok(serde_json::Value::Array(events).to_string()) + Ok(events) } } } @@ -394,10 +445,32 @@ impl SproutClient { /// a NIP-42 AUTH challenge if sent. Returns the relay response as a JSON /// string (`{event_id, accepted, message}`) matching the HTTP bridge shape. async fn submit_event_ws(&self, event: &nostr::Event) -> Result { + // Fan out to every relay; succeed if ANY accepts. This is what makes + // serverless writes resilient to a single relay rate-limiting + // ("noting too much") or being down — the desktop relay pool does the + // same. We try relays in order and return on the first acceptance. + let event_id = event.id.to_hex(); + let mut last_err: Option = None; + for relay in &self.ws_urls { + match self.submit_event_ws_one(relay, event).await { + Ok(msg) => return Ok(msg), + Err(e) => last_err = Some(e), + } + } + Err(last_err + .unwrap_or_else(|| CliError::Other(format!("no relay accepted event {event_id}")))) + } + + /// Publish a signed event to a single relay over plain WS and wait for OK. + async fn submit_event_ws_one( + &self, + relay: &str, + event: &nostr::Event, + ) -> Result { use futures_util::{SinkExt, StreamExt}; use tokio_tungstenite::{connect_async, tungstenite::Message}; - let (ws, _) = connect_async(self.ws_url.as_str()) + let (ws, _) = connect_async(relay) .await .map_err(|e| CliError::NetworkMsg(format!("relay connect failed: {e}")))?; let (mut write, mut read) = ws.split(); @@ -436,7 +509,7 @@ impl SproutClient { } "AUTH" => { if let Some(challenge) = arr.get(1).and_then(|v| v.as_str()) { - if let Ok(json) = self.ws_auth_message(challenge) { + if let Ok(json) = self.ws_auth_message(relay, challenge) { let _ = write.send(Message::Text(json.into())).await; let _ = write.send(Message::Text(event_msg.clone().into())).await; } @@ -470,8 +543,8 @@ impl SproutClient { } /// Build a NIP-42 `["AUTH", ]` message string for serverless writes. - fn ws_auth_message(&self, challenge: &str) -> Result { - let url = nostr::RelayUrl::parse(&self.ws_url) + fn ws_auth_message(&self, relay: &str, challenge: &str) -> Result { + let url = nostr::RelayUrl::parse(relay) .map_err(|e| CliError::Other(format!("invalid relay URL: {e}")))?; let event = EventBuilder::auth(challenge.to_string(), url) .sign_with_keys(&self.keys) diff --git a/crates/sprout-cli/src/lib.rs b/crates/sprout-cli/src/lib.rs index a40afd55c..3de7c77f6 100644 --- a/crates/sprout-cli/src/lib.rs +++ b/crates/sprout-cli/src/lib.rs @@ -1069,9 +1069,42 @@ pub enum PackCmd { // --------------------------------------------------------------------------- async fn run(cli: Cli) -> Result<(), CliError> { - let relay_url = client::normalize_relay_url(&cli.relay); - // In serverless mode we need the WebSocket URL (ws/wss), not the HTTP form. - let ws_url = client::to_ws_url(&cli.relay); + // Install the process-level rustls CryptoProvider. The serverless WS + // transport (tokio-tungstenite → tokio-rustls) constructs a TLS config + // directly and, unlike reqwest, does NOT auto-install a provider — without + // this, the first `wss://` connection panics ("Could not automatically + // determine the process-level CryptoProvider"). The CLI links aws-lc-rs + // (via reqwest's rustls), so install that. Idempotent; ignore the result + // if something already installed one. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + + // Serverless workspaces inject SPROUT_RELAY_URL as a comma-separated LIST + // (`wss://a,wss://b,...`). Detect serverless from that (a multi-relay URL is + // only ever produced by a serverless workspace) so the agent's reply via + // `sprout messages send` works even when SPROUT_SERVERLESS wasn't set. + let serverless = cli.serverless || cli.relay.contains(','); + let first_relay = cli + .relay + .split(',') + .next() + .map(str::trim) + .unwrap_or(cli.relay.as_str()); + let relay_url = client::normalize_relay_url(first_relay); + // In serverless mode pass the FULL relay list as a comma-separated set of + // ws/wss URLs — `SproutClient` fans out reads/writes across all of them + // (relays don't gossip; a write must tolerate one relay rate-limiting and a + // read must union every relay). In server mode only `relay_url` is used. + let ws_url: String = if serverless { + cli.relay + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .map(client::to_ws_url) + .collect::>() + .join(",") + } else { + client::to_ws_url(first_relay) + }; // Pack commands are local-only — no relay connection needed. if let Cmd::Pack(ref sub) = cli.command { @@ -1105,14 +1138,7 @@ async fn run(cli: Cli) -> Result<(), CliError> { _ => (None, None), }; - let client = SproutClient::new( - relay_url, - ws_url, - cli.serverless, - keys, - auth_tag, - auth_tag_json, - )?; + let client = SproutClient::new(relay_url, ws_url, serverless, keys, auth_tag, auth_tag_json)?; match cli.command { Cmd::Messages(sub) => commands::messages::dispatch(sub, &client, &cli.format).await, @@ -1232,6 +1258,7 @@ mod tests { "members", "purpose", "remove-member", + "search", "topic", "unarchive", "update" @@ -1270,7 +1297,7 @@ mod tests { fn subcommand_counts_are_stable() { let expected: Vec<(&str, usize)> = vec![ ("canvas", 2), - ("channels", 14), + ("channels", 15), ("dms", 3), ("feed", 1), ("messages", 8), diff --git a/desktop/src/features/workspaces/defaultRelays.ts b/desktop/src/features/workspaces/defaultRelays.ts index 73c5f8c10..f73ab5f3d 100644 --- a/desktop/src/features/workspaces/defaultRelays.ts +++ b/desktop/src/features/workspaces/defaultRelays.ts @@ -2,9 +2,12 @@ * Default public Nostr relays offered as quick-picks when creating a * serverless workspace. * - * Sourced from the deez mesh client's `DEFAULT_RELAYS` - * (`deez/crates/mesh-client/src/network/nostr.rs`) so Sprout's serverless mode - * and the mesh ecosystem converge on the same well-known relays. + * Originally sourced from the deez mesh client's `DEFAULT_RELAYS` + * (`deez/crates/mesh-client/src/network/nostr.rs`), but trimmed to the FREE, + * open relays. `nostr.land` and `nostr.wine` are PAID relays: they answer + * `auth-required` / `restricted: Pay for access` on reads, which is useless for + * an open serverless workspace and previously caused an auth-required reconnect + * storm in the agent. Keep only relays that allow anonymous read/write. * * These are only suggestions — users can type any relay URL. They apply only * to serverless workspaces; Sprout-server workspaces use their own relay. @@ -13,8 +16,6 @@ export const DEFAULT_PUBLIC_RELAYS: readonly string[] = [ "wss://relay.damus.io", "wss://nos.lol", "wss://relay.nostr.band", - "wss://nostr.land", - "wss://nostr.wine", ] as const; /** The relays pre-filled (comma-joined) when a user first enables serverless mode. */