diff --git a/.env.example b/.env.example index db6d8c83c..07f59f34f 100644 --- a/.env.example +++ b/.env.example @@ -210,3 +210,10 @@ RUST_LOG=buzz_relay=debug,buzz_db=debug,buzz_auth=debug,buzz_pubsub=debug,tower_ # These are accepted for backward compatibility but the canonical names above # are preferred: # BUZZ_ACP_PRIVATE_KEY → BUZZ_PRIVATE_KEY + +# Optional relay join policy. Markdown is served by the relay so every join +# surface can present the same documents. Each document and the independent age +# attestation are optional; configuring any one enables policy acceptance. +# BUZZ_TERMS_OF_SERVICE_MARKDOWN="# Terms of Service\n\nFull terms here." +# BUZZ_PRIVACY_POLICY_MARKDOWN="# Privacy Policy\n\nFull policy here." +# BUZZ_AGE_ATTESTATION_REQUIRED=true diff --git a/CHANGELOG.md b/CHANGELOG.md index eabe1baa0..bedbcd706 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,15 @@ # Changelog +## v0.4.9 + +- Restyle onboarding: branded landing screen, yellow/gradient backgrounds, new starter avatars ([#1982](https://github.com/block/buzz/pull/1982)) ([`831c80c1a`](https://github.com/block/buzz/commit/831c80c1a5d0811c7ac6ae832a95b2fd4ddc4e89)) +- Guide CLI installation and subscription sign-in ([#1980](https://github.com/block/buzz/pull/1980)) ([`8d3666c5f`](https://github.com/block/buzz/commit/8d3666c5f8ea07abe65355a1afe7de95ea247441)) +- unify channel add + search into one entry point ([#1964](https://github.com/block/buzz/pull/1964)) ([`3dd236eb6`](https://github.com/block/buzz/commit/3dd236eb6fd7c0dd86a9174d2852bc9ae9861912)) +- Apply optional relay join policy across join flows ([#1894](https://github.com/block/buzz/pull/1894)) ([`6c2d66757`](https://github.com/block/buzz/commit/6c2d667575cbc372ba42d26134448660fb1d2ee9)) +- fix(desktop): preserve relaunch through mesh shutdown ([#1966](https://github.com/block/buzz/pull/1966)) ([`a1626f96c`](https://github.com/block/buzz/commit/a1626f96cea57e415dad72558df04494ab4c2596)) +- Persist agent audiences with native inline mentions ([#1949](https://github.com/block/buzz/pull/1949)) ([`19dc33bda`](https://github.com/block/buzz/commit/19dc33bda6e9e7f09703c4efcd721235de644ce9)) + + ## v0.4.8 - fix(desktop): restore default community join option ([#1969](https://github.com/block/buzz/pull/1969)) ([`a84fe13a8`](https://github.com/block/buzz/commit/a84fe13a8c419f3d2d6ff3b577b431629b8e3e07)) diff --git a/Cargo.lock b/Cargo.lock index 7fcd5cd8a..fc08b299c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -774,6 +774,7 @@ dependencies = [ "evalexpr", "futures-util", "hex", + "httparse", "nix 0.31.3", "nostr", "reqwest 0.13.4", @@ -1151,6 +1152,7 @@ dependencies = [ "opentelemetry-otlp 0.32.0", "opentelemetry_sdk 0.32.1", "postcard", + "pulldown-cmark", "rand 0.10.1", "redis", "reqwest 0.13.4", @@ -1525,7 +1527,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2163,7 +2165,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccc2776f0c61eca1ca32528f85548abd1a4be8fb53d1b21c013e4f18da1e7090" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.117", ] [[package]] @@ -2374,7 +2376,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2612,7 +2614,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -2995,8 +2997,8 @@ dependencies = [ "libc", "log", "rustversion", - "windows-link 0.1.3", - "windows-result 0.3.4", + "windows-link 0.2.1", + "windows-result 0.4.1", ] [[package]] @@ -5554,7 +5556,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -6738,6 +6740,24 @@ version = "3.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "95067976aca6421a523e491fce939a3e65249bac4b977adee0ee9771568e8aa3" +[[package]] +name = "pulldown-cmark" +version = "0.13.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f068eba8e7071c5f9511831b44f32c740d5adf574e990f946ddb53db2f314e" +dependencies = [ + "bitflags 2.13.0", + "memchr", + "pulldown-cmark-escape", + "unicase", +] + +[[package]] +name = "pulldown-cmark-escape" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "007d8adb5ddab6f8e3f491ac63566a7d5002cc7ed73901f72057943fa71ae1ae" + [[package]] name = "pxfm" version = "0.1.29" @@ -6843,7 +6863,7 @@ dependencies = [ "once_cell", "socket2", "tracing", - "windows-sys 0.59.0", + "windows-sys 0.60.2", ] [[package]] @@ -7490,7 +7510,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7549,7 +7569,7 @@ dependencies = [ "security-framework 3.7.0", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -7812,7 +7832,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521" dependencies = [ "libc", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -8709,7 +8729,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] @@ -10008,7 +10028,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.59.0", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/buzz-acp/Cargo.toml b/crates/buzz-acp/Cargo.toml index d3afb5062..d04784980 100644 --- a/crates/buzz-acp/Cargo.toml +++ b/crates/buzz-acp/Cargo.toml @@ -78,3 +78,4 @@ nix = { version = "0.31", default-features = false, features = ["signal"] } [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } +httparse = "1" diff --git a/crates/buzz-acp/src/acp.rs b/crates/buzz-acp/src/acp.rs index ebfd8fb4a..4b8d043ab 100644 --- a/crates/buzz-acp/src/acp.rs +++ b/crates/buzz-acp/src/acp.rs @@ -121,6 +121,17 @@ fn agent_error_from_json(error: &serde_json::Value) -> AcpError { AcpError::AgentError { code, message } } +fn build_initialize_params() -> serde_json::Value { + serde_json::json!({ + "protocolVersion": 2, + "clientCapabilities": build_client_capabilities(), + "clientInfo": { + "name": "buzz-acp", + "version": env!("CARGO_PKG_VERSION") + }, + }) +} + /// ACP client that owns an agent subprocess and communicates over its stdio. /// /// One `AcpClient` per agent process. Multiple sessions can be created on the @@ -333,6 +344,29 @@ pub(crate) fn build_codex_config_env( Ok(Some(serde_json::Value::Object(base).to_string())) } +fn build_client_capabilities() -> serde_json::Value { + serde_json::json!({ + // Signal to ACP adapters that Buzz can hand users to terminal-native + // auth flows. Adapters decide which auth methods to expose; Buzz does + // not hardcode vendor login commands from this capability. + "auth": { + "terminal": true + }, + // Signal to goose that we handle `_goose/unstable/session/update` + // notifications. Without this the custom notification is suppressed + // on goose's side and usage data is never emitted. + "_meta": { + "goose": { + "customNotifications": true + }, + // Non-standard extension used by claude-agent-acp to advertise the + // exact terminal login argv for subscription auth. Unknown `_meta` + // keys are ignored by other adapters. + "terminal-auth": true + } + }) +} + impl AcpClient { /// Kill the agent subprocess and wait for it to exit (no zombies). /// @@ -501,28 +535,20 @@ impl AcpClient { pub async fn initialize(&mut self) -> Result { // Requesting version 2 is an intentional temporary pin — we are squatting // on ACP v2 ahead of the upstream ACP RFD. Revisit when that RFD merges. - let params = serde_json::json!({ - "protocolVersion": 2, - "clientCapabilities": { - // Signal to goose that we handle `_goose/unstable/session/update` - // notifications. Without this the custom notification is suppressed - // on goose's side and usage data is never emitted. - "_meta": { - "goose": { - "customNotifications": true - } - } - }, - "clientInfo": { - "name": "buzz-acp", - "version": env!("CARGO_PKG_VERSION") - } - }); + let params = build_initialize_params(); let result = self.send_request("initialize", params).await?; tracing::debug!(target: "acp::init", "initialize response: {result}"); Ok(result) } + /// Send the ACP `authenticate` request for an adapter-advertised method. + pub async fn authenticate(&mut self, method_id: &str) -> Result { + let params = serde_json::json!({ + "methodId": method_id, + }); + self.send_request("authenticate", params).await + } + /// Send `session/new` and return the full response alongside the session ID. /// /// `cwd` must be an absolute path. `mcp_servers` may be empty. @@ -2067,13 +2093,7 @@ mod tests { "method": "initialize", "params": { "protocolVersion": 2, - "clientCapabilities": { - "_meta": { - "goose": { - "customNotifications": true - } - } - }, + "clientCapabilities": build_client_capabilities(), "clientInfo": { "name": "buzz-acp", "version": "0.1.0" @@ -2086,6 +2106,11 @@ mod tests { Some("buzz-acp") ); assert!(msg["params"]["clientCapabilities"].is_object()); + assert_eq!( + msg["params"]["clientCapabilities"]["auth"]["terminal"].as_bool(), + Some(true), + "terminal auth capability must be advertised so adapters can expose terminal login methods" + ); assert_eq!( msg["params"]["clientCapabilities"]["_meta"]["goose"]["customNotifications"].as_bool(), Some(true), diff --git a/crates/buzz-acp/src/config.rs b/crates/buzz-acp/src/config.rs index e8e8fdd51..befb7aa6a 100644 --- a/crates/buzz-acp/src/config.rs +++ b/crates/buzz-acp/src/config.rs @@ -175,6 +175,18 @@ impl std::fmt::Display for PermissionMode { about = "Query available models from the configured agent" )] pub struct ModelsArgs { + /// Agent binary to spawn (e.g. "goose", "claude-agent-acp", "codex-acp"). + #[command(flatten)] + pub agent: AuthAgentArgs, + + /// Output structured JSON instead of human-readable text. + #[arg(long)] + pub json: bool, +} + +/// Shared agent-spawn flags for lightweight local ACP helper subcommands. +#[derive(Debug, Parser)] +pub struct AuthAgentArgs { /// Agent binary to spawn (e.g. "goose", "claude-agent-acp", "codex-acp"). #[arg(long, env = "BUZZ_ACP_AGENT_COMMAND", default_value = "goose")] pub agent_command: String, @@ -187,12 +199,38 @@ pub struct ModelsArgs { value_delimiter = ',' )] pub agent_args: Vec, +} + +/// CLI args for `buzz-acp auth-methods` — query adapter-advertised login methods. +#[derive(Debug, Parser)] +#[command( + name = "buzz-acp auth-methods", + about = "Query adapter-advertised ACP authentication methods" +)] +pub struct AuthMethodsArgs { + #[command(flatten)] + pub agent: AuthAgentArgs, /// Output structured JSON instead of human-readable text. #[arg(long)] pub json: bool, } +/// CLI args for `buzz-acp authenticate` — start an adapter-owned login flow. +#[derive(Debug, Parser)] +#[command( + name = "buzz-acp authenticate", + about = "Start an adapter-owned ACP authentication flow" +)] +pub struct AuthenticateArgs { + #[command(flatten)] + pub agent: AuthAgentArgs, + + /// Adapter-advertised auth method id to invoke. + #[arg(long)] + pub method_id: String, +} + #[derive(Debug, Parser)] #[command( name = "buzz-acp", diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 77fb4a613..2b6e8d45d 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -28,7 +28,10 @@ use buzz_core::observer::{ OBSERVER_MAX_PLAINTEXT_LEN, }; use clap::Parser; -use config::{Config, DedupMode, ModelsArgs, MultipleEventHandling, RespondTo, SubscribeMode}; +use config::{ + AuthAgentArgs, AuthMethodsArgs, AuthenticateArgs, Config, DedupMode, ModelsArgs, + MultipleEventHandling, RespondTo, SubscribeMode, +}; use filter::SubscriptionRule; use futures_util::FutureExt; use nostr::{PublicKey, ToBech32}; @@ -46,7 +49,7 @@ use uuid::Uuid; /// /// This avoids clap rejecting harness flags (like `--private-key`) that aren't /// declared on the subcommand's `Parser`. The `models` path has its own -/// `ModelsArgs` parser; the default path uses the existing `CliArgs`. +/// dedicated parser; the default path uses the existing `CliArgs`. /// /// **Constraint**: subcommand must be argv[1] — flags before the subcommand /// name (e.g., `buzz-acp --verbose models`) are not supported. @@ -54,9 +57,13 @@ fn is_subcommand(name: &str) -> bool { std::env::args().nth(1).map(|a| a == name).unwrap_or(false) } -/// Timeout for the `buzz-acp models` subcommand (spawn + init + session/new). +/// Timeout for lightweight helper subcommands (spawn + initialize + model/method probes). const MODELS_TIMEOUT: Duration = Duration::from_secs(10); +/// Timeout for `buzz-acp authenticate`. Browser-based vendor auth can require +/// human interaction, so it must not share the short probe timeout. +const AUTHENTICATE_TIMEOUT: Duration = Duration::from_secs(10 * 60); + /// Publish a kind:20001 presence update event via the WebSocket connection. /// /// Ephemeral kinds (20000-29999) are rejected by the HTTP bridge, so presence @@ -1076,8 +1083,8 @@ async fn tokio_main() -> Result<()> { .install_default() .expect("failed to install rustls crypto provider"); if is_subcommand("models") { - // Strip the "models" token so clap doesn't reject it as a positional. - // Keeps argv[0] (binary name) and passes everything after "models". + // Strip the subcommand token so clap doesn't reject it as a positional. + // Keeps argv[0] (binary name) and passes everything after the subcommand. let filtered: Vec = std::env::args() .enumerate() .filter(|(i, _)| *i != 1) @@ -1087,6 +1094,26 @@ async fn tokio_main() -> Result<()> { return run_models(args).await; } + if is_subcommand("auth-methods") { + let filtered: Vec = std::env::args() + .enumerate() + .filter(|(i, _)| *i != 1) + .map(|(_, a)| a) + .collect(); + let args = AuthMethodsArgs::parse_from(&filtered); + return run_auth_methods(args).await; + } + + if is_subcommand("authenticate") { + let filtered: Vec = std::env::args() + .enumerate() + .filter(|(i, _)| *i != 1) + .map(|(_, a)| a) + .collect(); + let args = AuthenticateArgs::parse_from(&filtered); + return run_authenticate(args).await; + } + tracing_subscriber::fmt() .with_env_filter( EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("buzz_acp=info")), @@ -3325,14 +3352,130 @@ async fn spawn_and_init( } } -/// `buzz-acp models` — spawn an agent, query its available models, exit. -/// +async fn spawn_auth_client(agent: &AuthAgentArgs) -> Result { + let agent_args = config::normalize_agent_args(&agent.agent_command, agent.agent_args.clone()); + AcpClient::spawn(&agent.agent_command, &agent_args, &[], false).await +} + +fn extract_auth_methods(init_result: &serde_json::Value) -> Vec { + init_result + .get("authMethods") + .and_then(|methods| methods.as_array()) + .cloned() + .unwrap_or_default() +} + +/// `buzz-acp auth-methods` — spawn an adapter, initialize it, print authMethods. +async fn run_auth_methods(args: AuthMethodsArgs) -> Result<()> { + let mut client = match spawn_auth_client(&args.agent).await { + Ok(c) => c, + Err(e) => { + eprintln!("error: failed to spawn agent: {e}"); + std::process::exit(1); + } + }; + + let init_result = match tokio::time::timeout(MODELS_TIMEOUT, client.initialize()).await { + Ok(Ok(result)) => result, + Ok(Err(e)) => { + client.shutdown().await; + eprintln!("error: agent initialize failed: {e}"); + std::process::exit(1); + } + Err(_) => { + client.shutdown().await; + eprintln!("error: agent timed out ({MODELS_TIMEOUT:?})"); + std::process::exit(1); + } + }; + + let methods = extract_auth_methods(&init_result); + client.shutdown().await; + + if args.json { + let output = serde_json::json!({ "methods": methods }); + println!("{}", serde_json::to_string_pretty(&output)?); + } else if methods.is_empty() { + println!("No auth methods advertised."); + } else { + for method in methods { + let id = method + .get("id") + .and_then(|value| value.as_str()) + .unwrap_or("unknown"); + let name = method + .get("name") + .and_then(|value| value.as_str()) + .unwrap_or(id); + println!("{id}\t{name}"); + } + } + Ok(()) +} + +/// `buzz-acp authenticate` — invoke one adapter-owned auth method. +async fn run_authenticate(args: AuthenticateArgs) -> Result<()> { + let mut client = match spawn_auth_client(&args.agent).await { + Ok(c) => c, + Err(e) => { + eprintln!("error: failed to spawn agent: {e}"); + std::process::exit(1); + } + }; + + let init_result = match tokio::time::timeout(MODELS_TIMEOUT, client.initialize()).await { + Ok(Ok(result)) => result, + Ok(Err(e)) => { + client.shutdown().await; + eprintln!("error: agent initialize failed: {e}"); + std::process::exit(1); + } + Err(_) => { + client.shutdown().await; + eprintln!("error: agent initialize timed out ({MODELS_TIMEOUT:?})"); + std::process::exit(1); + } + }; + + let supports_method = extract_auth_methods(&init_result) + .iter() + .any(|method| method.get("id").and_then(|id| id.as_str()) == Some(args.method_id.as_str())); + if !supports_method { + client.shutdown().await; + eprintln!( + "error: auth method '{}' is not advertised by this adapter", + args.method_id + ); + std::process::exit(1); + } + + let result = + tokio::time::timeout(AUTHENTICATE_TIMEOUT, client.authenticate(&args.method_id)).await; + + match result { + Ok(Ok(_)) => { + client.shutdown().await; + Ok(()) + } + Ok(Err(e)) => { + client.shutdown().await; + eprintln!("error: authenticate failed: {e}"); + std::process::exit(1); + } + Err(_) => { + client.shutdown().await; + eprintln!("error: authenticate timed out ({AUTHENTICATE_TIMEOUT:?})"); + std::process::exit(1); + } + } +} + /// Flow: spawn → initialize → session/new → print models → shutdown. /// No relay connection, no MCP servers, no subscriptions. ~2-5s total. async fn run_models(args: ModelsArgs) -> Result<()> { use acp::{extract_model_config_options, extract_model_state}; - let agent_args = config::normalize_agent_args(&args.agent_command, args.agent_args); + let agent_args = config::normalize_agent_args(&args.agent.agent_command, args.agent.agent_args); let cwd = std::env::current_dir() .unwrap_or_else(|_| std::path::PathBuf::from("/")) .to_string_lossy() @@ -3340,13 +3483,14 @@ async fn run_models(args: ModelsArgs) -> Result<()> { // Spawn outside the timeout so we always own the child for cleanup. // `models` subcommand doesn't use persona packs — no extra env, no codex config. - let mut client = match AcpClient::spawn(&args.agent_command, &agent_args, &[], false).await { - Ok(c) => c, - Err(e) => { - eprintln!("error: failed to spawn agent: {e}"); - std::process::exit(1); - } - }; + let mut client = + match AcpClient::spawn(&args.agent.agent_command, &agent_args, &[], false).await { + Ok(c) => c, + Err(e) => { + eprintln!("error: failed to spawn agent: {e}"); + std::process::exit(1); + } + }; // Initialize + session/new under a timeout. Client is owned above, // so shutdown() runs on all paths (success, error, timeout). diff --git a/crates/buzz-acp/src/relay.rs b/crates/buzz-acp/src/relay.rs index 0f26fc5d4..67d01fd38 100644 --- a/crates/buzz-acp/src/relay.rs +++ b/crates/buzz-acp/src/relay.rs @@ -7,6 +7,10 @@ //! //! ## Architecture //! +//! `HarnessRelay::connect()` retries a transient initial connect/auth failure +//! (e.g. a dropped handshake on a spotty link) with bounded jittered backoff +//! before giving up; a terminal configuration/auth error fails immediately. +//! //! A background tokio task owns the WebSocket stream. It: //! - Responds to Ping frames with Pong (preventing relay disconnect on long turns) //! - Forwards `BuzzEvent`s through an `mpsc` channel @@ -56,6 +60,24 @@ const SINCE_SKEW_SECS: u64 = 5; const AUTH_TIMEOUT: Duration = Duration::from_secs(5); /// Timeout for the TCP + WebSocket handshake in `do_connect`. const CONNECT_TIMEOUT: Duration = Duration::from_secs(10); +/// Backoff delay values shared by the initial-connect retry in +/// `HarnessRelay::connect()` and `try_autonomous_reconnect`'s post-start +/// reconnect loop — a spotty link should get consistent retry pacing whether +/// the failure happens at agent startup or later. Bounded so a dead relay +/// can't hang either path forever. +/// +/// The two callers consume this differently: `retry_initial_connect` sleeps +/// before every entry (1 immediate attempt + up to 5 delayed retries, all 5 +/// values used), while `try_autonomous_reconnect` skips the sleep after its +/// final attempt (5 attempts total, only the first 4 values used) — so +/// "shared values," not "identical schedule." +const STARTUP_CONNECT_BACKOFFS: [Duration; 5] = [ + Duration::from_secs(1), + Duration::from_secs(2), + Duration::from_secs(4), + Duration::from_secs(8), + Duration::from_secs(16), +]; use std::time::Instant; @@ -517,10 +539,15 @@ impl HarnessRelay { agent_pubkey_hex: &str, auth_tag: Option, ) -> Result { - // Perform the initial connection and auth handshake. + // Perform the initial connection and auth handshake, retrying + // transient failures (dropped handshake, timeout) with bounded + // jittered backoff. A terminal error (bad URL, bad auth tag, + // rejected/invalid signing key) fails immediately — see + // `is_terminal_connect_error`. // Finding #8: capture the handshake buffer and pass it to the background // task so buffered messages aren't silently discarded. - let (ws, handshake_buffer) = do_connect(relay_url, keys, auth_tag.as_ref()).await?; + let (ws, handshake_buffer) = + retry_initial_connect(|| do_connect(relay_url, keys, auth_tag.as_ref())).await?; let (event_tx, event_rx) = mpsc::channel::>(event_channel_capacity()); let (observer_control_tx, observer_control_rx) = @@ -2123,14 +2150,11 @@ async fn try_autonomous_reconnect( observer_control_tx: &mpsc::Sender, auth_tag: Option<&nostr::Tag>, ) -> ReconnectOutcome { - // Finding #42: 5 attempts, up to 16s base backoff. - let backoffs = [ - Duration::from_secs(1), - Duration::from_secs(2), - Duration::from_secs(4), - Duration::from_secs(8), - Duration::from_secs(16), - ]; + // Finding #42: 5 attempts, up to 16s base backoff. Shares delay values + // with the initial-connect retry in `HarnessRelay::connect()` + // (STARTUP_CONNECT_BACKOFFS) — see its doc comment for how the two + // loops consume the array differently. + let backoffs = STARTUP_CONNECT_BACKOFFS; for (attempt, delay) in backoffs.iter().enumerate() { info!( @@ -2710,6 +2734,204 @@ pub(crate) fn parse_relay_message(text: &str) -> Result bool { + match err { + RelayError::Http(_) | RelayError::Json(_) | RelayError::UnexpectedMessage(_) => true, + RelayError::WebSocket(e) => is_terminal_ws_error(e.as_ref()), + RelayError::AuthFailed(message) => is_terminal_auth_failure(message), + RelayError::NoAuthChallenge | RelayError::ConnectionClosed | RelayError::Timeout => false, + } +} + +/// Exhaustive classification of `tungstenite::Error` inner variants for +/// startup connect retry. No wildcard — a tungstenite upgrade forces +/// reclassification at compile time. +fn is_terminal_ws_error(err: &tokio_tungstenite::tungstenite::Error) -> bool { + use tokio_tungstenite::tungstenite::error::ProtocolError; + use tokio_tungstenite::tungstenite::Error as WsError; + + match err { + // Deterministic pre-connect / handshake-shape failures. + WsError::Url(_) + | WsError::Capacity(_) + | WsError::Utf8(_) + | WsError::HttpFormat(_) + | WsError::AttackAttempt => true, + + // Non-101 HTTP: terminal unless 408/429/5xx. + WsError::Http(resp) => { + let status = resp.status().as_u16(); + !(status == 408 || status == 429 || (500..600).contains(&status)) + } + + // Protocol errors: most are deterministic upgrade mismatches. + WsError::Protocol(p) => !matches!( + p, + ProtocolError::HandshakeIncomplete | ProtocolError::ResetWithoutClosingHandshake + ), + + // Io: split by error source and rustls variant. tokio-rustls wraps + // rustls errors as io::Error(InvalidData, rustls_err). Deterministic + // cert/config/incompatibility failures (allowlist) are terminal; + // ambiguous protocol, decrypt, and server-alert shapes stay transient + // under the bounded retry budget. Plain transport Io (reset, EOF, + // timeout, refused) also stays transient. + // Relies on a single rustls version in the dep tree (0.23.40); + // a version split would break the downcast. + WsError::Io(e) => is_terminal_rustls_io_error(e), + + WsError::ConnectionClosed => false, + + // Deterministic TLS config failures. On our rustls build the only + // connect-time Tls variant is InvalidDnsName; certificate validation + // failures arrive wrapped inside Io (terminal via source-chain + // downcast above). + WsError::Tls(_) => true, + + // Unreachable during connect_async; kept fail-safe transient. + WsError::AlreadyClosed | WsError::WriteBufferFull(_) => false, + } +} + +/// Walks an `io::Error` for a `rustls::Error` and inspects its variant. +/// Returns `true` (terminal) only for deterministic cert/config/incompatibility +/// failures that retry cannot fix. Ambiguous protocol, decrypt, and server-alert +/// shapes return `false` (transient) — retries are bounded and the feature's +/// purpose is resilience. +/// +/// Relies on a single rustls version in the dep tree (0.23.40); a version split +/// would break the downcast. +fn is_terminal_rustls_io_error(err: &std::io::Error) -> bool { + use std::error::Error as _; + + fn find_rustls_error(err: &std::io::Error) -> Option<&rustls::Error> { + // First check the direct inner payload (io::Error stores it via + // get_ref — source() skips to *its* source). + if let Some(inner) = err.get_ref() { + if let Some(re) = inner.downcast_ref::() { + return Some(re); + } + } + // Walk the source chain for deeper wrapping. + let mut source = err.source(); + while let Some(e) = source { + if let Some(re) = e.downcast_ref::() { + return Some(re); + } + source = e.source(); + } + None + } + + let Some(rustls_err) = find_rustls_error(err) else { + return false; + }; + + matches!( + rustls_err, + rustls::Error::InvalidCertificate(_) + | rustls::Error::InvalidCertRevocationList(_) + | rustls::Error::NoCertificatesPresented + | rustls::Error::UnsupportedNameType + | rustls::Error::PeerIncompatible(_) + ) +} + +/// Whether a relay's `OK false ` denial during NIP-42 auth is +/// terminal, per the NIP-01 machine-readable prefixes the relay actually +/// sends (`crates/buzz-relay/src/handlers/auth.rs`). +/// +/// `error:` marks the relay's own dependency failures (e.g. a ban-state DB +/// lookup that couldn't run) — the relay is failing closed on itself, not +/// rejecting the caller, and a later attempt can succeed once the +/// dependency recovers. `invalid:`, `auth-required:`, `restricted:`, and +/// `blocked:` are explicit rejections of this identity/config (bad +/// signature, ban, non-member, allowlist denial) that retrying without +/// changing anything cannot fix. An unrecognized prefix is treated as +/// terminal — failing fast on an unknown denial is safer than retrying one +/// that might be a real rejection. +fn is_terminal_auth_failure(message: &str) -> bool { + !message.trim_start().starts_with("error:") +} + +/// Retry `op` with bounded jittered backoff, stopping immediately on a +/// terminal error (see [`is_terminal_connect_error`]). Used by +/// `HarnessRelay::connect()` so a transient failure during the initial +/// WebSocket/NIP-42 handshake — e.g. a dropped connection on a spotty link — +/// doesn't fail agent startup outright. +/// +/// Generic over the success type so the backoff/classification logic can be +/// exercised in tests without a real socket. Returns the last transient +/// error if all attempts are exhausted. +async fn retry_initial_connect(mut op: F) -> Result +where + F: FnMut() -> Fut, + Fut: std::future::Future>, +{ + let mut last_err = None; + + for (attempt, delay) in std::iter::once(None) + .chain(STARTUP_CONNECT_BACKOFFS.iter().map(|d| Some(*d))) + .enumerate() + { + if let Some(base) = delay { + let jittered = jittered_duration(base); + info!( + "retrying initial relay connect (attempt {attempt}) in {:.1}s", + jittered.as_secs_f64() + ); + tokio::time::sleep(jittered).await; + } + + match op().await { + Ok(v) => return Ok(v), + Err(e) if is_terminal_connect_error(&e) => { + warn!("initial relay connect failed with terminal error: {e}"); + return Err(e); + } + Err(e) => { + warn!("initial relay connect attempt {attempt} failed: {e}"); + last_err = Some(e); + } + } + } + + Err(last_err.unwrap_or(RelayError::ConnectionClosed)) +} + /// Perform a single WebSocket connect + NIP-42 auth handshake. /// /// Returns `(ws, buffer)` on success. @@ -3712,4 +3934,639 @@ mod tests { "the dropped channel must not be resubscribed — the loop cannot re-form" ); } + + // ── startup connect retry ──────────────────────────────────────────── + + /// Table-driven coverage of every `RelayError` variant and every + /// `tungstenite::Error` inner variant. Exhaustive — adding a new + /// tungstenite variant without updating this table is a compile error + /// in `is_terminal_ws_error` (no wildcard), and a missing row here + /// is a code-review gap, not a silent misclassification. + #[test] + fn connect_error_classification_matches_every_relay_error_variant() { + use tokio_tungstenite::tungstenite::error::{ + CapacityError, Error as WsError, ProtocolError, SubProtocolError, TlsError, UrlError, + }; + use tokio_tungstenite::tungstenite::http; + + fn ws(e: WsError) -> RelayError { + RelayError::WebSocket(Box::new(e)) + } + + let cases: Vec<(&str, RelayError, bool)> = vec![ + // ── outer RelayError variants ── + ("Http: bad URL", RelayError::Http("bad url".into()), true), + ( + "Json: malformed relay frame", + RelayError::Json(serde_json::from_str::<()>("not json").unwrap_err()), + true, + ), + ( + "UnexpectedMessage: unknown frame type", + RelayError::UnexpectedMessage("unknown message type: WAT".into()), + true, + ), + ( + "AuthFailed: relay dependency fault (NIP-01 `error:` prefix)", + RelayError::AuthFailed("error: internal error checking restriction state".into()), + false, + ), + ( + "AuthFailed: bad signature (`invalid:` prefix)", + RelayError::AuthFailed("invalid: bad signature".into()), + true, + ), + ( + "AuthFailed: banned (`blocked:` prefix)", + RelayError::AuthFailed("blocked: you are banned from this community".into()), + true, + ), + ( + "AuthFailed: not a member (`restricted:` prefix)", + RelayError::AuthFailed("restricted: not a relay member".into()), + true, + ), + ( + "AuthFailed: allowlist denial (`auth-required:` prefix)", + RelayError::AuthFailed("auth-required: verification failed".into()), + true, + ), + ( + "AuthFailed: unrecognized prefix fails safe as terminal", + RelayError::AuthFailed("some new denial reason".into()), + true, + ), + ( + "NoAuthChallenge: relay silence is link/relay-timing noise", + RelayError::NoAuthChallenge, + false, + ), + ("ConnectionClosed", RelayError::ConnectionClosed, false), + ("Timeout", RelayError::Timeout, false), + // ── WebSocket inner: terminal ── + ( + "WebSocket(Url): unsupported scheme", + ws(WsError::Url(UrlError::UnsupportedUrlScheme)), + true, + ), + ( + "WebSocket(Url): missing host", + ws(WsError::Url(UrlError::NoHostName)), + true, + ), + ( + "WebSocket(Url): empty host", + ws(WsError::Url(UrlError::EmptyHostName)), + true, + ), + ( + "WebSocket(Url): TLS feature not enabled", + ws(WsError::Url(UrlError::TlsFeatureNotEnabled)), + true, + ), + ( + "WebSocket(Url): unable to connect", + ws(WsError::Url(UrlError::UnableToConnect("addr".into()))), + true, + ), + ( + "WebSocket(Url): no path or query", + ws(WsError::Url(UrlError::NoPathOrQuery)), + true, + ), + ( + "WebSocket(Capacity): message too long", + ws(WsError::Capacity(CapacityError::MessageTooLong { + size: 100, + max_size: 50, + })), + true, + ), + ( + "WebSocket(Capacity): too many headers", + ws(WsError::Capacity(CapacityError::TooManyHeaders)), + true, + ), + ( + "WebSocket(Utf8): encoding error", + ws(WsError::Utf8("invalid utf-8".into())), + true, + ), + ( + "WebSocket(HttpFormat): malformed HTTP", + ws(WsError::HttpFormat( + http::Response::builder().status(9999).body(()).unwrap_err(), + )), + true, + ), + ("WebSocket(AttackAttempt)", ws(WsError::AttackAttempt), true), + // ── WebSocket inner: Http status split ── + ( + "WebSocket(Http): 200 = plain HTTPS endpoint → terminal", + ws(WsError::Http(Box::new( + http::Response::builder().status(200).body(None).unwrap(), + ))), + true, + ), + ( + "WebSocket(Http): 301 redirect → terminal", + ws(WsError::Http(Box::new( + http::Response::builder().status(301).body(None).unwrap(), + ))), + true, + ), + ( + "WebSocket(Http): 404 not found → terminal", + ws(WsError::Http(Box::new( + http::Response::builder().status(404).body(None).unwrap(), + ))), + true, + ), + ( + "WebSocket(Http): 403 forbidden → terminal", + ws(WsError::Http(Box::new( + http::Response::builder().status(403).body(None).unwrap(), + ))), + true, + ), + ( + "WebSocket(Http): 408 request timeout → transient", + ws(WsError::Http(Box::new( + http::Response::builder().status(408).body(None).unwrap(), + ))), + false, + ), + ( + "WebSocket(Http): 429 too many requests → transient", + ws(WsError::Http(Box::new( + http::Response::builder().status(429).body(None).unwrap(), + ))), + false, + ), + ( + "WebSocket(Http): 500 internal server error → transient", + ws(WsError::Http(Box::new( + http::Response::builder().status(500).body(None).unwrap(), + ))), + false, + ), + ( + "WebSocket(Http): 502 bad gateway → transient", + ws(WsError::Http(Box::new( + http::Response::builder().status(502).body(None).unwrap(), + ))), + false, + ), + ( + "WebSocket(Http): 503 service unavailable → transient", + ws(WsError::Http(Box::new( + http::Response::builder().status(503).body(None).unwrap(), + ))), + false, + ), + // ── WebSocket inner: Protocol variants ── + ( + "Protocol(WrongHttpMethod): deterministic upgrade mismatch", + ws(WsError::Protocol(ProtocolError::WrongHttpMethod)), + true, + ), + ( + "Protocol(WrongHttpVersion): deterministic upgrade mismatch", + ws(WsError::Protocol(ProtocolError::WrongHttpVersion)), + true, + ), + ( + "Protocol(MissingConnectionUpgradeHeader)", + ws(WsError::Protocol( + ProtocolError::MissingConnectionUpgradeHeader, + )), + true, + ), + ( + "Protocol(MissingUpgradeWebSocketHeader)", + ws(WsError::Protocol( + ProtocolError::MissingUpgradeWebSocketHeader, + )), + true, + ), + ( + "Protocol(MissingSecWebSocketVersionHeader)", + ws(WsError::Protocol( + ProtocolError::MissingSecWebSocketVersionHeader, + )), + true, + ), + ( + "Protocol(MissingSecWebSocketKey)", + ws(WsError::Protocol(ProtocolError::MissingSecWebSocketKey)), + true, + ), + ( + "Protocol(SecWebSocketAcceptKeyMismatch)", + ws(WsError::Protocol( + ProtocolError::SecWebSocketAcceptKeyMismatch, + )), + true, + ), + ( + "Protocol(SecWebSocketSubProtocolError)", + ws(WsError::Protocol( + ProtocolError::SecWebSocketSubProtocolError( + SubProtocolError::ServerSentSubProtocolNoneRequested, + ), + )), + true, + ), + ( + "Protocol(JunkAfterRequest)", + ws(WsError::Protocol(ProtocolError::JunkAfterRequest)), + true, + ), + ( + "Protocol(CustomResponseSuccessful)", + ws(WsError::Protocol(ProtocolError::CustomResponseSuccessful)), + true, + ), + ( + "Protocol(InvalidHeader)", + ws(WsError::Protocol(ProtocolError::InvalidHeader(Box::new( + http::header::UPGRADE, + )))), + true, + ), + ( + "Protocol(HttparseError)", + ws(WsError::Protocol(ProtocolError::HttparseError( + httparse::Error::TooManyHeaders, + ))), + true, + ), + ( + "Protocol(SendAfterClosing)", + ws(WsError::Protocol(ProtocolError::SendAfterClosing)), + true, + ), + ( + "Protocol(ReceivedAfterClosing)", + ws(WsError::Protocol(ProtocolError::ReceivedAfterClosing)), + true, + ), + ( + "Protocol(NonZeroReservedBits)", + ws(WsError::Protocol(ProtocolError::NonZeroReservedBits)), + true, + ), + ( + "Protocol(UnmaskedFrameFromClient)", + ws(WsError::Protocol(ProtocolError::UnmaskedFrameFromClient)), + true, + ), + ( + "Protocol(MaskedFrameFromServer)", + ws(WsError::Protocol(ProtocolError::MaskedFrameFromServer)), + true, + ), + ( + "Protocol(FragmentedControlFrame)", + ws(WsError::Protocol(ProtocolError::FragmentedControlFrame)), + true, + ), + ( + "Protocol(ControlFrameTooBig)", + ws(WsError::Protocol(ProtocolError::ControlFrameTooBig)), + true, + ), + ( + "Protocol(UnknownControlFrameType)", + ws(WsError::Protocol(ProtocolError::UnknownControlFrameType( + 0xF, + ))), + true, + ), + ( + "Protocol(UnknownDataFrameType)", + ws(WsError::Protocol(ProtocolError::UnknownDataFrameType(0xF))), + true, + ), + ( + "Protocol(UnexpectedContinueFrame)", + ws(WsError::Protocol(ProtocolError::UnexpectedContinueFrame)), + true, + ), + ( + "Protocol(ExpectedFragment)", + ws(WsError::Protocol(ProtocolError::ExpectedFragment( + tokio_tungstenite::tungstenite::protocol::frame::coding::Data::Text, + ))), + true, + ), + ( + "Protocol(InvalidOpcode)", + ws(WsError::Protocol(ProtocolError::InvalidOpcode(0xF))), + true, + ), + ( + "Protocol(InvalidCloseSequence)", + ws(WsError::Protocol(ProtocolError::InvalidCloseSequence)), + true, + ), + // ── Protocol: transient exceptions ── + ( + "Protocol(HandshakeIncomplete): connection dropped mid-handshake", + ws(WsError::Protocol(ProtocolError::HandshakeIncomplete)), + false, + ), + ( + "Protocol(ResetWithoutClosingHandshake): abrupt reset", + ws(WsError::Protocol( + ProtocolError::ResetWithoutClosingHandshake, + )), + false, + ), + // ── WebSocket(Io): transport (transient) ── + ( + "Io(other): plain transport failure is transient", + ws(WsError::Io(std::io::Error::other("reset"))), + false, + ), + ( + "Io(ConnectionReset): transport reset is transient", + ws(WsError::Io(std::io::ErrorKind::ConnectionReset.into())), + false, + ), + ( + "Io(UnexpectedEof): transport EOF is transient", + ws(WsError::Io(std::io::ErrorKind::UnexpectedEof.into())), + false, + ), + ( + "Io(TimedOut): transport timeout is transient", + ws(WsError::Io(std::io::ErrorKind::TimedOut.into())), + false, + ), + // ── WebSocket(Io): rustls-sourced, variant-inspected ── + // Production shape: tokio-rustls wraps rustls errors as + // io::Error(InvalidData, rustls::Error). Only deterministic + // cert/config/incompatibility variants are terminal. + ( + "Io(rustls InvalidCertificate(Expired)): production-shaped expired cert is terminal", + ws(WsError::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + rustls::Error::InvalidCertificate(rustls::CertificateError::Expired), + ))), + true, + ), + ( + "Io(rustls InvalidCertificate(NotValidForName)): hostname mismatch is terminal", + ws(WsError::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + rustls::Error::InvalidCertificate( + rustls::CertificateError::NotValidForName, + ), + ))), + true, + ), + ( + "Io(rustls NoCertificatesPresented): missing cert is terminal", + ws(WsError::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + rustls::Error::NoCertificatesPresented, + ))), + true, + ), + // ── WebSocket(Io): rustls-sourced, ambiguous (transient) ── + // Protocol, decrypt, alert, and general errors may be caused by + // network conditions or transient server failures — retryable + // under the bounded budget. + ( + "Io(rustls General): ambiguous general error is transient", + ws(WsError::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + rustls::Error::General("protocol error".into()), + ))), + false, + ), + ( + "Io(rustls AlertReceived(InternalError)): server alert is transient", + ws(WsError::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + rustls::Error::AlertReceived(rustls::AlertDescription::InternalError), + ))), + false, + ), + ( + "Io(rustls DecryptError): corrupted record is transient", + ws(WsError::Io(std::io::Error::new( + std::io::ErrorKind::InvalidData, + rustls::Error::DecryptError, + ))), + false, + ), + ( + "WebSocket(ConnectionClosed): link-level closure", + ws(WsError::ConnectionClosed), + false, + ), + // ── WebSocket(Tls): deterministic config (terminal, pins the arm) ── + // These shapes are constructible but not reachable through our + // rustls production connector — cert failures arrive as Io above. + // Kept to pin the Tls(_) => true arm. + ( + "Tls(Rustls(General)): pins Tls arm terminal", + ws(WsError::Tls( + rustls::Error::General("tls handshake failed".into()).into(), + )), + true, + ), + ( + "Tls(InvalidDnsName): only reachable connect-time Tls variant", + ws(WsError::Tls(TlsError::InvalidDnsName)), + true, + ), + ( + "Tls(Rustls(InvalidCertificate(Expired))): pins Tls arm terminal", + ws(WsError::Tls( + rustls::Error::InvalidCertificate(rustls::CertificateError::Expired).into(), + )), + true, + ), + ( + "WebSocket(AlreadyClosed): unreachable at connect, fail-safe transient", + ws(WsError::AlreadyClosed), + false, + ), + ( + "WebSocket(WriteBufferFull): unreachable at connect, fail-safe transient", + ws(WsError::WriteBufferFull(Box::new( + tokio_tungstenite::tungstenite::Message::Text("x".into()), + ))), + false, + ), + ]; + + for (label, err, want_terminal) in cases { + assert_eq!( + is_terminal_connect_error(&err), + want_terminal, + "{label}: expected terminal={want_terminal}" + ); + } + } + + /// A literal `https://…` URL through production `do_connect()` must fail + /// fast as terminal — the relay endpoint is a plain HTTPS server, not a + /// WebSocket endpoint, and tungstenite returns `Error::Http` (non-101 + /// response) or `Error::Url(UnsupportedUrlScheme)` depending on how far + /// the handshake gets. Either way it must not be retried. + #[tokio::test] + async fn do_connect_wrong_scheme_is_terminal() { + let keys = nostr::Keys::generate(); + let err = do_connect("https://example.com", &keys, None) + .await + .unwrap_err(); + assert!( + is_terminal_connect_error(&err), + "wrong-scheme URL should be terminal, got: {err}" + ); + } + + /// A transient failure (e.g. connection dropped mid-handshake on a spotty + /// link) must be retried and can still succeed once the link recovers. + #[tokio::test(start_paused = true)] + async fn retry_initial_connect_retries_transient_failure_then_succeeds() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let attempts = AtomicUsize::new(0); + let result: Result<&'static str, RelayError> = retry_initial_connect(|| { + let n = attempts.fetch_add(1, Ordering::SeqCst); + async move { + if n < 2 { + Err(RelayError::ConnectionClosed) + } else { + Ok("connected") + } + } + }) + .await; + + assert_eq!(result.unwrap(), "connected"); + assert_eq!( + attempts.load(Ordering::SeqCst), + 3, + "should succeed on the 3rd attempt (2 transient failures + 1 success)" + ); + } + + /// A terminal error (bad auth, bad config) must not be retried — the + /// same call would fail identically every time, so retrying just delays + /// surfacing a real problem to the caller. + #[tokio::test(start_paused = true)] + async fn retry_initial_connect_does_not_retry_terminal_error() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let attempts = AtomicUsize::new(0); + let result: Result<(), RelayError> = retry_initial_connect(|| { + attempts.fetch_add(1, Ordering::SeqCst); + async { Err(RelayError::AuthFailed("invalid: bad signature".into())) } + }) + .await; + + assert!(matches!(result, Err(RelayError::AuthFailed(_)))); + assert_eq!( + attempts.load(Ordering::SeqCst), + 1, + "a terminal error must fail on the first attempt with no retries" + ); + } + + /// A relay-side dependency fault (NIP-01 `error:` prefix) is transient — + /// the relay is failing closed on itself, not rejecting this identity — + /// so it must be retried rather than surfaced immediately like a real + /// auth rejection. + #[tokio::test(start_paused = true)] + async fn retry_initial_connect_retries_relay_dependency_fault() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let attempts = AtomicUsize::new(0); + let result: Result<&'static str, RelayError> = retry_initial_connect(|| { + let n = attempts.fetch_add(1, Ordering::SeqCst); + async move { + if n < 1 { + Err(RelayError::AuthFailed( + "error: internal error checking restriction state".into(), + )) + } else { + Ok("connected") + } + } + }) + .await; + + assert_eq!(result.unwrap(), "connected"); + assert_eq!( + attempts.load(Ordering::SeqCst), + 2, + "a relay dependency fault must be retried, not surfaced immediately" + ); + } + + /// Once every attempt (1 initial + N backoff retries) is exhausted, the + /// last transient error is returned rather than retrying forever — a + /// dead relay must not hang agent startup indefinitely. + #[tokio::test(start_paused = true)] + async fn retry_initial_connect_exhausts_and_returns_last_error() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let attempts = AtomicUsize::new(0); + let result: Result<(), RelayError> = retry_initial_connect(|| { + attempts.fetch_add(1, Ordering::SeqCst); + async { Err(RelayError::Timeout) } + }) + .await; + + assert!( + matches!(result, Err(RelayError::Timeout)), + "must surface the last attempt's error, not a generic one" + ); + assert_eq!( + attempts.load(Ordering::SeqCst), + STARTUP_CONNECT_BACKOFFS.len() + 1, + "must attempt exactly once plus one retry per backoff entry" + ); + } + + /// Backoff sleeps must actually elapse (not be skipped) — this pins the + /// bounded-but-real-delay contract using `tokio::time::pause` so the + /// test itself stays fast (virtual time, not wall-clock sleeps). + #[tokio::test(start_paused = true)] + async fn retry_initial_connect_sleeps_between_attempts() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let attempts = AtomicUsize::new(0); + let call = retry_initial_connect(|| { + let n = attempts.fetch_add(1, Ordering::SeqCst); + async move { + if n < 1 { + Err(RelayError::ConnectionClosed) + } else { + Ok(()) + } + } + }); + tokio::pin!(call); + + // Before the first backoff elapses, the retry must still be pending + // (i.e. it actually slept rather than immediately retrying). + tokio::select! { + biased; + _ = tokio::time::sleep(Duration::from_millis(1)) => {} + _ = &mut call => panic!("must not resolve before the backoff sleep elapses"), + } + assert_eq!(attempts.load(Ordering::SeqCst), 1); + + // Advancing past the (jittered, ≤1.2x) first backoff lets it proceed. + let result = call.await; + assert!(result.is_ok()); + assert_eq!(attempts.load(Ordering::SeqCst), 2); + } } diff --git a/crates/buzz-db/src/lib.rs b/crates/buzz-db/src/lib.rs index 07c67a824..5e20c267e 100644 --- a/crates/buzz-db/src/lib.rs +++ b/crates/buzz-db/src/lib.rs @@ -2689,6 +2689,30 @@ impl Db { relay_members::add_relay_member(&self.pool, community, pubkey, role, added_by).await } + /// Claims relay membership via an invite and atomically persists the + /// accepted policy version when a policy is configured. + pub async fn claim_relay_membership( + &self, + community: CommunityId, + pubkey: &str, + role: &str, + policy_version: Option<&str>, + ) -> Result { + relay_members::claim_relay_membership(&self.pool, community, pubkey, role, policy_version) + .await + } + + /// Returns whether a member has persisted acceptance evidence for a policy version. + pub async fn has_join_policy_acceptance( + &self, + community: CommunityId, + pubkey: &str, + policy_version: &str, + ) -> Result { + relay_members::has_join_policy_acceptance(&self.pool, community, pubkey, policy_version) + .await + } + /// Removes a relay member from `community` atomically, refusing to delete the owner. pub async fn remove_relay_member( &self, diff --git a/crates/buzz-db/src/migration.rs b/crates/buzz-db/src/migration.rs index 1d8c961a2..256498846 100644 --- a/crates/buzz-db/src/migration.rs +++ b/crates/buzz-db/src/migration.rs @@ -549,7 +549,7 @@ mod tests { let mut migrations: Vec<_> = MIGRATOR.iter().collect(); migrations.sort_by_key(|migration| migration.version); - assert_eq!(migrations.len(), 19); + assert_eq!(migrations.len(), 20); assert_eq!(migrations[0].version, 1); assert_eq!(&*migrations[0].description, "initial schema"); assert!(migrations[0] @@ -812,6 +812,18 @@ mod tests { .sql .as_str() .contains("purge_soft_deleted_buzz_mesh_status")); + + // Join policy acceptances landed concurrently with mesh status retention; + // keep both additive migrations in a single, unambiguous sequence. + assert_eq!(migrations[19].version, 20); + assert!(migrations[19] + .sql + .as_str() + .contains("CREATE TABLE join_policy_acceptances")); + assert!(!migrations[0] + .sql + .as_str() + .contains("join_policy_acceptances")); } #[test] @@ -1054,7 +1066,7 @@ mod tests { run_migrations(&pool) .await .expect("retry succeeds after operator repair"); - assert_eq!(applied_versions(&pool).await.last().copied(), Some(19)); + assert_eq!(applied_versions(&pool).await.last().copied(), Some(20)); } #[tokio::test] diff --git a/crates/buzz-db/src/relay_members.rs b/crates/buzz-db/src/relay_members.rs index d354b057b..3bd114fda 100644 --- a/crates/buzz-db/src/relay_members.rs +++ b/crates/buzz-db/src/relay_members.rs @@ -114,6 +114,67 @@ pub async fn add_relay_member( Ok(result.rows_affected() > 0) } +/// Claims relay membership via an invite and atomically persists policy evidence. +/// +/// Returns `true` when membership was inserted, or `false` when the pubkey was +/// already a member. A configured `policy_version` is recorded in the same +/// transaction, so membership cannot be granted without its acceptance record. +pub async fn claim_relay_membership( + pool: &PgPool, + community: CommunityId, + pubkey: &str, + role: &str, + policy_version: Option<&str>, +) -> Result { + let mut tx = pool.begin().await?; + let inserted = sqlx::query( + "INSERT INTO relay_members (community_id, pubkey, role, added_by) \ + VALUES ($1, $2, $3, 'invite') \ + ON CONFLICT (community_id, pubkey) DO NOTHING", + ) + .bind(community.as_uuid()) + .bind(pubkey) + .bind(role) + .execute(&mut *tx) + .await? + .rows_affected() + > 0; + + if let Some(version) = policy_version { + sqlx::query( + "INSERT INTO join_policy_acceptances (community_id, pubkey, policy_version) \ + VALUES ($1, $2, $3) ON CONFLICT DO NOTHING", + ) + .bind(community.as_uuid()) + .bind(pubkey) + .bind(version) + .execute(&mut *tx) + .await?; + } + + tx.commit().await?; + Ok(inserted) +} + +/// Returns whether a member has persisted acceptance evidence for a policy version. +pub async fn has_join_policy_acceptance( + pool: &PgPool, + community: CommunityId, + pubkey: &str, + policy_version: &str, +) -> Result { + let row = sqlx::query( + "SELECT 1 FROM join_policy_acceptances \ + WHERE community_id = $1 AND pubkey = $2 AND policy_version = $3", + ) + .bind(community.as_uuid()) + .bind(pubkey) + .bind(policy_version) + .fetch_optional(pool) + .await?; + Ok(row.is_some()) +} + /// The result of a relay member removal attempt. #[derive(Debug, PartialEq)] pub enum RemoveResult { @@ -544,6 +605,38 @@ mod tests { (community, owner) } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn invite_claim_persists_policy_version_and_legacy_claim_does_not() { + let pool = setup_pool().await; + let community = make_test_community(&pool).await; + let policy_member = test_pubkey(); + let legacy_member = test_pubkey(); + let version = "a".repeat(64); + + assert!( + claim_relay_membership(&pool, community, &policy_member, "member", Some(&version),) + .await + .expect("claim membership with policy") + ); + assert!( + has_join_policy_acceptance(&pool, community, &policy_member, &version) + .await + .expect("policy acceptance lookup") + ); + + assert!( + claim_relay_membership(&pool, community, &legacy_member, "member", None) + .await + .expect("legacy claim membership") + ); + assert!( + !has_join_policy_acceptance(&pool, community, &legacy_member, &version) + .await + .expect("legacy acceptance lookup") + ); + } + /// NIP-43 admission confinement: a pubkey admitted to community A is *not* /// admitted to community B. This is the exact mutation #1285 targets — a /// `WHERE pubkey = $1` membership check (no community predicate) would let an diff --git a/crates/buzz-relay/Cargo.toml b/crates/buzz-relay/Cargo.toml index 36cc71dcb..31eac113a 100644 --- a/crates/buzz-relay/Cargo.toml +++ b/crates/buzz-relay/Cargo.toml @@ -77,6 +77,7 @@ moka = { workspace = true } metrics = { workspace = true } metrics-exporter-prometheus = { workspace = true } metrics-util = { workspace = true } +pulldown-cmark = { version = "0.13.4", default-features = false, features = ["html"] } [features] dev = ["buzz-auth/dev"] diff --git a/crates/buzz-relay/src/api/invites.rs b/crates/buzz-relay/src/api/invites.rs index 3b09e2a38..cf7ee3a6e 100644 --- a/crates/buzz-relay/src/api/invites.rs +++ b/crates/buzz-relay/src/api/invites.rs @@ -19,7 +19,7 @@ use std::time::Duration; use axum::{ extract::State, http::{HeaderMap, StatusCode}, - response::Json, + response::{Html, Json}, }; use serde::Deserialize; use serde_json::Value; @@ -54,6 +54,138 @@ pub struct MintInviteRequest { pub struct ClaimInviteRequest { /// The invite code to redeem. pub code: String, + /// Relay-issued proof of accepting the configured terms, when required. + #[serde(default)] + pub policy_receipt: Option, +} + +/// Body for `POST /api/invites/accept-policy`. +#[derive(Debug, Deserialize)] +pub struct AcceptPolicyRequest { + /// Invite code the acceptance receipt will be bound to. + pub code: String, + /// Policy revision displayed by the client. + pub policy_version: String, + /// Minimum-age assertion, required only when configured by the operator. + #[serde(default)] + pub age_confirmed: bool, +} + +/// Public join policy shared by every client-side join surface. +pub async fn join_policy(State(state): State>) -> Json { + match &state.config.join_policy { + Some(policy) => Json(serde_json::json!({ + "policy": { + "terms_markdown": policy.terms_markdown, + "privacy_markdown": policy.privacy_markdown, + "age_attestation_required": policy.age_attestation_required, + "version": policy.version + } + })), + None => Json(serde_json::json!({})), + } +} + +/// `GET /api/join-policy/terms` — Terms of Service as a standalone HTML page. +/// +/// Serves the operator-configured Markdown as a real browser page so desktop +/// clients can hand the link to the system browser instead of rendering the +/// document inside the webview (which requires app chrome the onboarding +/// surfaces don't have). 404 when no terms document is configured. +pub async fn join_policy_terms( + State(state): State>, +) -> Result, (StatusCode, Json)> { + policy_document_page(&state, "Terms of Service", |policy| { + policy.terms_markdown.as_deref() + }) +} + +/// `GET /api/join-policy/privacy` — Privacy Policy as a standalone HTML page. +pub async fn join_policy_privacy( + State(state): State>, +) -> Result, (StatusCode, Json)> { + policy_document_page(&state, "Privacy Policy", |policy| { + policy.privacy_markdown.as_deref() + }) +} + +fn policy_document_page( + state: &AppState, + title: &str, + select: impl Fn(&crate::config::JoinPolicyConfig) -> Option<&str>, +) -> Result, (StatusCode, Json)> { + let markdown = state + .config + .join_policy + .as_ref() + .and_then(select) + .ok_or_else(|| api_error(StatusCode::NOT_FOUND, "join_policy_not_configured"))?; + Ok(Html(render_policy_document(title, markdown))) +} + +/// Render operator Markdown into a minimal self-contained HTML page. +/// +/// Raw HTML embedded in the Markdown is escaped and rendered as text — the +/// operator authors a policy document, not a web page, and this keeps the +/// endpoint from serving arbitrary operator-controlled markup. +fn render_policy_document(title: &str, markdown: &str) -> String { + use pulldown_cmark::{html, Event, Parser}; + + let mut body = String::new(); + html::push_html( + &mut body, + Parser::new(markdown).map(|event| match event { + Event::Html(raw) => Event::Text(raw.into_string().into()), + Event::InlineHtml(raw) => Event::Text(raw.into_string().into()), + other => other, + }), + ); + + // Titles are fixed literals today; escape anyway so a future caller + // can't accidentally inject markup through this seam. + let escaped_title = title + .replace('&', "&") + .replace('<', "<") + .replace('>', ">"); + + format!( + "\n\n\n\n\ + \n\ + {escaped_title}\n\ + \n\ + \n\n{body}\n\n" + ) +} + +/// Exchange explicit policy acceptance for a short-lived, invite-bound receipt. +pub async fn accept_policy( + State(state): State>, + body: axum::body::Bytes, +) -> Result, (StatusCode, Json)> { + let Some(policy) = &state.config.join_policy else { + return Err(api_error( + StatusCode::NOT_FOUND, + "join_policy_not_configured", + )); + }; + let request: AcceptPolicyRequest = serde_json::from_slice(&body).map_err(|e| { + api_error( + StatusCode::BAD_REQUEST, + &format!("invalid policy acceptance JSON: {e}"), + ) + })?; + if request.policy_version != policy.version + || (policy.age_attestation_required && !request.age_confirmed) + { + return Err(api_error( + StatusCode::BAD_REQUEST, + "join_policy_not_accepted", + )); + } + let key = invite_token::derive_invite_key(&state.relay_keypair); + let receipt = invite_token::mint_policy_acceptance(&key, &request.code, &policy.version); + Ok(Json(serde_json::json!({ "receipt": receipt }))) } /// Shared prelude: bind the tenant from the Host header and verify the NIP-98 @@ -186,9 +318,27 @@ pub async fn claim_invite( )?; let claimer_hex = pubkey.to_hex(); + if let Some(policy) = &state.config.join_policy { + let receipt = request + .policy_receipt + .as_deref() + .ok_or_else(|| api_error(StatusCode::FORBIDDEN, "join_policy_required"))?; + invite_token::verify_policy_acceptance(&key, receipt, &request.code, &policy.version) + .map_err(|_| api_error(StatusCode::FORBIDDEN, "join_policy_required"))?; + } + let was_inserted = state .db - .add_relay_member(tenant.community(), &claimer_hex, &payload.r, Some("invite")) + .claim_relay_membership( + tenant.community(), + &claimer_hex, + &payload.r, + state + .config + .join_policy + .as_ref() + .map(|policy| policy.version.as_str()), + ) .await .map_err(|e| internal_error(&format!("invite claim insert: {e}")))?; @@ -361,14 +511,17 @@ mod tests { /// a fresh community on `host`; returns `None` when Postgres is unavailable. async fn invite_test_state(host: &str) -> Option> { let mut config = crate::config::Config::from_env().ok()?; - config.database_url = TEST_DB_URL.to_string(); + let database_url = std::env::var("BUZZ_TEST_DATABASE_URL") + .or_else(|_| std::env::var("DATABASE_URL")) + .unwrap_or_else(|_| TEST_DB_URL.to_string()); + config.database_url = database_url.clone(); config.redis_url = "redis://127.0.0.1:1".to_string(); config.relay_url = format!("wss://{host}"); // The claim route must work on relays where membership is enforced — // that is the entire point of an invite. config.require_relay_membership = true; - let pool = sqlx::PgPool::connect(TEST_DB_URL).await.ok()?; + let pool = sqlx::PgPool::connect(&database_url).await.ok()?; let db = buzz_db::Db::from_pool(pool.clone()); db.ensure_configured_community(host).await.ok()?; @@ -505,6 +658,215 @@ mod tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn join_policy_gate_end_to_end() { + let host = format!("invites-policy-{}.example", Uuid::new_v4().simple()); + let owner = Keys::generate(); + let joiner = Keys::generate(); + let Some(state) = invite_test_state(&host).await else { + return; + }; + // Force the join policy on regardless of env. + let mut state_inner = (*state).clone(); + let mut config = state_inner.config.as_ref().clone(); + config.join_policy = Some(crate::config::JoinPolicyConfig { + terms_markdown: Some("# Terms".to_string()), + privacy_markdown: Some("# Privacy".to_string()), + age_attestation_required: true, + version: "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa".to_string(), + }); + state_inner.config = Arc::new(config); + let state = Arc::new(state_inner); + + let community = state + .db + .lookup_community_by_host(&host) + .await + .expect("lookup") + .expect("community exists"); + state + .db + .add_relay_member(community.id, &owner.public_key().to_hex(), "owner", None) + .await + .expect("seed owner"); + + // Mint an invite. + let response = post_json( + state.clone(), + &host, + "/api/invites", + &owner, + "{}".to_string(), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let json = read_json(response).await; + let code = json + .get("code") + .and_then(Value::as_str) + .expect("code") + .to_string(); + + // 1. Claim WITHOUT receipt -> 403 (checkbox bypass). + let response = post_json( + state.clone(), + &host, + "/api/invites/claim", + &joiner, + serde_json::json!({ "code": code }).to_string(), + ) + .await; + assert_eq!( + response.status(), + StatusCode::FORBIDDEN, + "no-receipt claim must fail" + ); + + // 2. Forged receipt (wrong key) -> 403. + let forged = crate::invite_token::mint_policy_acceptance( + &[9u8; 32], + &code, + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ); + let response = post_json( + state.clone(), + &host, + "/api/invites/claim", + &joiner, + serde_json::json!({ "code": code, "policy_receipt": forged }).to_string(), + ) + .await; + assert_eq!( + response.status(), + StatusCode::FORBIDDEN, + "forged receipt must fail" + ); + + // 3. Receipt bound to a DIFFERENT invite code -> 403. + let key = crate::invite_token::derive_invite_key(&state.relay_keypair); + let other = crate::invite_token::mint_policy_acceptance( + &key, + "some-other-code", + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ); + let response = post_json( + state.clone(), + &host, + "/api/invites/claim", + &joiner, + serde_json::json!({ "code": code, "policy_receipt": other }).to_string(), + ) + .await; + assert_eq!( + response.status(), + StatusCode::FORBIDDEN, + "cross-invite receipt must fail" + ); + + // 4. Receipt for a STALE policy version -> 403. + let stale = crate::invite_token::mint_policy_acceptance( + &key, + &code, + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", + ); + let response = post_json( + state.clone(), + &host, + "/api/invites/claim", + &joiner, + serde_json::json!({ "code": code, "policy_receipt": stale }).to_string(), + ) + .await; + assert_eq!( + response.status(), + StatusCode::FORBIDDEN, + "stale-version receipt must fail" + ); + + // 5. accept-policy without age confirmation -> 400. + let response = post_json( + state.clone(), + &host, + "/api/invites/accept-policy", + &joiner, + serde_json::json!({ "code": code, "policy_version": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "age_confirmed": false }) + .to_string(), + ) + .await; + assert_eq!( + response.status(), + StatusCode::BAD_REQUEST, + "age not confirmed must be rejected when required" + ); + + // 5b. accept-policy with stale version -> 400. + let response = post_json( + state.clone(), + &host, + "/api/invites/accept-policy", + &joiner, + serde_json::json!({ "code": code, "policy_version": "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb", "age_confirmed": true }) + .to_string(), + ) + .await; + assert_eq!(response.status(), StatusCode::BAD_REQUEST); + + // 6. Legit flow: accept-policy -> receipt -> claim OK. + let response = post_json( + state.clone(), + &host, + "/api/invites/accept-policy", + &joiner, + serde_json::json!({ "code": code, "policy_version": "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", "age_confirmed": true }) + .to_string(), + ) + .await; + assert_eq!(response.status(), StatusCode::OK); + let receipt = read_json(response) + .await + .get("receipt") + .and_then(Value::as_str) + .expect("receipt") + .to_string(); + + let response = post_json( + state.clone(), + &host, + "/api/invites/claim", + &joiner, + serde_json::json!({ "code": code, "policy_receipt": receipt }).to_string(), + ) + .await; + assert_eq!( + response.status(), + StatusCode::OK, + "legit receipt claim must succeed" + ); + let json = read_json(response).await; + assert_eq!(json.get("status").and_then(Value::as_str), Some("joined")); + + let member = state + .db + .get_relay_member(community.id, &joiner.public_key().to_hex()) + .await + .expect("member lookup") + .expect("joiner is now a member"); + assert_eq!(member.role, "member"); + assert!( + state + .db + .has_join_policy_acceptance( + community.id, + &joiner.public_key().to_hex(), + "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + ) + .await + .expect("policy acceptance lookup"), + "accepted policy version must be persisted", + ); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn non_admin_cannot_mint() { @@ -610,4 +972,87 @@ mod tests { let response = post_json(state, &host_b, "/api/invites/claim", &joiner, body).await; assert_eq!(response.status(), StatusCode::FORBIDDEN); } + + #[test] + fn policy_document_renders_markdown_and_escapes_raw_html() { + let page = super::render_policy_document( + "Terms of Service", + "# Terms\n\nBe kind & honest.\n\n", + ); + assert!(page.contains("Terms of Service"), "{page}"); + assert!(page.contains("

Terms

"), "{page}"); + // `&` inside prose must be entity-encoded by the HTML writer. + assert!(page.contains("Be kind & honest."), "{page}"); + // Raw HTML in operator Markdown renders as escaped text, never markup. + assert!(!page.contains("