diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 00474ce0ff..3eec4dc2e8 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -214,7 +214,7 @@ jobs: run: just desktop-tauri-test env: CMAKE_POLICY_VERSION_MINIMUM: "3.5" - - name: Desktop Mesh admission tests + - name: Desktop Mesh feature tests run: cargo test --manifest-path desktop/src-tauri/Cargo.toml --features mesh-llm --lib env: CMAKE_POLICY_VERSION_MINIMUM: "3.5" diff --git a/Justfile b/Justfile index 038f575bad..4df797071a 100644 --- a/Justfile +++ b/Justfile @@ -210,9 +210,12 @@ _ensure-migrations: _ensure-services ./scripts/seed-local-community.sh # Run clippy on the desktop Tauri Rust crate -# Mesh support is part of the shipped desktop contract, so local `just check` -# and CI both lint the same feature-enabled graph. +# Features are additive, so a single invocation lints only one cfg graph. +# Both graphs ship (release-windows builds without mesh-llm), so lint both: +# the default graph covers the `#[cfg(not(feature = "mesh-llm"))]` arms and +# the feature-enabled graph covers the mesh code. desktop-tauri-clippy: _ensure-sidecar-stubs + cargo clippy --manifest-path {{desktop_tauri_manifest}} --workspace --all-targets -- -D warnings cargo clippy --manifest-path {{desktop_tauri_manifest}} --workspace --all-targets --features mesh-llm -- -D warnings # Check the desktop Tauri Rust crate compiles diff --git a/desktop/src-tauri/src/commands/agent_models.rs b/desktop/src-tauri/src/commands/agent_models.rs index 7f5139a08b..05b1abad90 100644 --- a/desktop/src-tauri/src/commands/agent_models.rs +++ b/desktop/src-tauri/src/commands/agent_models.rs @@ -245,9 +245,16 @@ pub async fn discover_agent_models( if input.provider.as_deref().map(str::trim) == Some(crate::managed_agents::RELAY_MESH_PROVIDER_ID) { - let availability = crate::commands::mesh_llm::resolve_mesh_availability(&state) - .await - .map_err(|error| format!("Buzz shared compute model discovery failed: {error}"))?; + let events = crate::relay::query_relay( + &state, + &[ + crate::mesh_llm::mesh_status_filter(), + crate::mesh_llm::relay_membership_filter(), + ], + ) + .await + .map_err(|error| format!("Buzz shared compute model discovery failed: {error}"))?; + let availability = crate::mesh_llm::availability_from_events(events); if availability.models.is_empty() { return Err(availability.reason.unwrap_or_else(|| { "No live Buzz shared compute models are available".to_string() diff --git a/desktop/src-tauri/src/commands/mesh_llm.rs b/desktop/src-tauri/src/commands/mesh_llm.rs index e5ff5e39d5..7356cd7fc0 100644 --- a/desktop/src-tauri/src/commands/mesh_llm.rs +++ b/desktop/src-tauri/src/commands/mesh_llm.rs @@ -176,55 +176,9 @@ fn advance_mesh_status_cursor( Ok(cursor) } -/// Resolve whether a relay enforces NIP-43 relay membership, from its NIP-11 -/// document. -/// -/// The relay's advertisement is authoritative only to the extent that the -/// HTTP transport is authenticated: `wss://` relay URLs produce HTTPS NIP-11 -/// reads, while a `ws://` override is vulnerable to on-path modification. -/// -/// The relay advertises NIP-43 only when membership is actually enforced -/// (`nip11_facts` in `buzz-relay`), so this is the one signal that separates -/// "no roster because nobody is a direct member here" from "no roster because -/// this read was incomplete". A probe failure is reported as `Err` and treated -/// as enforcing by callers: an unreachable relay must never be able to relax -/// admission. -async fn probe_relay_mesh_mode(relay_url: &str) -> Result { - let http_url = relay::relay_http_base_url(relay_url); - let client = reqwest::Client::builder() - .timeout(std::time::Duration::from_secs(5)) - .build() - .map_err(|error| format!("failed to build NIP-11 client: {error}"))?; - let document: serde_json::Value = client - .get(&http_url) - .header("Accept", "application/nostr+json") - .send() - .await - .map_err(|error| format!("NIP-11 probe failed: {error}"))? - .error_for_status() - .map_err(|error| format!("NIP-11 probe returned an unsuccessful HTTP status: {error}"))? - .json() - .await - .map_err(|error| format!("NIP-11 document was not JSON: {error}"))?; - mesh_llm::relay_mesh_mode_from_nip11(&document) -} - -/// Probe the relay's current membership mode. -/// -/// Only a successful, strictly valid NIP-11 document proves a mode. Transport, -/// HTTP, and document errors remain errors so callers can preserve a running -/// admission policy instead of turning an observation failure into a mode -/// transition. -pub(crate) async fn resolved_relay_mesh_mode( - relay_url: &str, -) -> Result { - probe_relay_mesh_mode(relay_url).await -} - -async fn query_mesh_discovery_events_for_mode( +async fn query_mesh_discovery_events_at( state: &AppState, relay_url: &str, - mode: mesh_llm::MeshRelayMode, ) -> Result, String> { let api_base_url = relay::relay_http_base_url(relay_url); let mut events = @@ -239,27 +193,13 @@ async fn query_mesh_discovery_events_for_mode( // means the query is incomplete: surface it as an error so the // reconcile loop keeps the current allowlist instead of flapping the // node down to self-only on a successful-but-empty response. - // - // A relay that does not enforce membership never populates - // `relay_members` from sign-in, so there is no roster to wait for and - // no incomplete read to guard against. Fall through and read status - // notes unfiltered so the caller can run the mesh unenforced. - if !mode.is_open() { - if !mesh_llm::has_membership_snapshot(&events) { - return Err("relay returned no membership snapshot".to_string()); - } - return Ok(events); + if !mesh_llm::has_membership_snapshot(&events) { + return Err("relay returned no membership snapshot".to_string()); } + return Ok(events); } let mut status_filter = mesh_llm::mesh_status_filter(); - // On an enforcing relay, scope status reads to current members. On an open - // relay the snapshot is not a trust boundary even when one exists (added - // out of band) — admission runs unenforced there, so scoping reads to a - // stray roster would hide peers the mesh admits. Read every reporter's - // note and let freshness + endpoint-binding checks gate routing. - if !mode.is_open() && !member_pubkeys.is_empty() { - status_filter["authors"] = serde_json::json!(member_pubkeys); - } + status_filter["authors"] = serde_json::json!(member_pubkeys); let mut previous_cursor: Option<(u64, String)> = None; loop { @@ -279,17 +219,8 @@ async fn query_mesh_discovery_events_for_mode( } } -/// Resolve model availability using the same relay-mode probe and discovery -/// query as mesh startup and routing. Agent model pickers must go through this -/// path too: the legacy closed-mode helper rejects every open relay because an -/// open relay correctly publishes no membership snapshot. -pub(crate) async fn resolve_mesh_availability( - state: &AppState, -) -> Result { - let relay_url = relay::relay_ws_url_with_override(state); - let mode = resolved_relay_mesh_mode(&relay_url).await?; - let events = query_mesh_discovery_events_for_mode(state, &relay_url, mode).await?; - Ok(mesh_llm::availability_from_events_for_mode(events, mode)) +async fn query_mesh_discovery_events(state: &AppState) -> Result, String> { + query_mesh_discovery_events_at(state, &relay::relay_ws_url_with_override(state)).await } /// Resolve the admission roster by intersecting member-signed mesh status @@ -300,42 +231,16 @@ pub(crate) async fn resolve_mesh_availability( /// never be collapsed into "self-only", or a transient relay blip de-admits /// every other member. `reconcile_roster` relies on this to keep the current /// allowlist on error instead of restarting the node down to self-only. -/// `None` means the relay does not enforce membership: there is no roster to -/// intersect and the node must run unenforced (`TrustPolicy::Off`), exactly as -/// stock MeshLLM does. `Some` carries an enforcing relay's roster. -pub(crate) async fn resolve_trusted_owner_ids( - state: &AppState, -) -> Result>, String> { - resolve_trusted_owner_ids_at(state, &relay::relay_ws_url_with_override(state)).await +pub(crate) async fn resolve_trusted_owner_ids(state: &AppState) -> Result, String> { + let events = query_mesh_discovery_events(state).await?; + Ok(mesh_llm::owner_ids_from_events(&events)) } pub(crate) async fn resolve_trusted_owner_ids_at( state: &AppState, relay_url: &str, -) -> Result>, String> { - let mode = resolved_relay_mesh_mode(relay_url).await?; - let events = query_mesh_discovery_events_for_mode(state, relay_url, mode).await?; - if mode.is_open() { - return Ok(None); - } - Ok(Some(mesh_llm::owner_ids_from_events(&events))) -} - -/// Refresh the roster of a runtime that is already membership-enforcing. -/// -/// Its admission mode is established, so reconciliation reads only the -/// authoritative membership/status snapshot. NIP-11 timeouts or changes cannot -/// alter this closed-runtime path. -pub(crate) async fn resolve_closed_trusted_owner_ids_at( - state: &AppState, - relay_url: &str, ) -> Result, String> { - let events = query_mesh_discovery_events_for_mode( - state, - relay_url, - mesh_llm::MeshRelayMode::ClosedMembershipEnforced, - ) - .await?; + let events = query_mesh_discovery_events_at(state, relay_url).await?; Ok(mesh_llm::owner_ids_from_events(&events)) } @@ -343,14 +248,12 @@ pub(crate) async fn resolve_closed_trusted_owner_ids_at( /// (an empty roster) when the relay query fails. This is safe only at start: /// there is no established allowlist to preserve yet. The periodic /// `reconcile_roster` path must NOT use this — it has a live roster to keep. -pub(crate) async fn resolve_trusted_owner_ids_or_self_only( - state: &AppState, -) -> Option> { +pub(crate) async fn resolve_trusted_owner_ids_or_self_only(state: &AppState) -> Vec { match resolve_trusted_owner_ids(state).await { Ok(owners) => owners, Err(error) => { eprintln!("buzz-mesh: roster query failed; allowing only this node: {error}"); - Some(Vec::new()) + Vec::new() } } } @@ -388,13 +291,12 @@ pub(crate) async fn resolve_buzz_mesh_join_targets_at( state: &AppState, relay_url: &str, ) -> Result, String> { - let mode = resolved_relay_mesh_mode(relay_url).await?; - let events = query_mesh_discovery_events_for_mode(state, relay_url, mode).await?; + let events = query_mesh_discovery_events_at(state, relay_url).await?; let self_owner_id = mesh_llm::ensure_owner_identity() .map_err(|error| format!("failed to load mesh owner identity: {error}"))? .owner_id; Ok(buzz_mesh_join_targets( - mesh_llm::availability_from_events_for_mode(events, mode).serve_targets, + mesh_llm::availability_from_events(events).serve_targets, &self_owner_id, )) } @@ -406,29 +308,15 @@ pub(crate) async fn resolve_buzz_mesh_join_targets_at( async fn resolve_buzz_mesh_startup_at( state: &AppState, relay_url: &str, -) -> (Option>, Option) { - let mode = match resolved_relay_mesh_mode(relay_url).await { - Ok(mode) => mode, - Err(error) => { - // Initial startup has no established policy to preserve, so - // ambiguous mode evidence starts isolated and retries later. - eprintln!( - "buzz-mesh: startup NIP-11 probe failed; allowing only this node and starting isolated for now: {error}" - ); - return (Some(Vec::new()), None); - } - }; - match query_mesh_discovery_events_for_mode(state, relay_url, mode).await { +) -> (Vec, Option) { + match query_mesh_discovery_events_at(state, relay_url).await { Ok(events) => { - // An open relay has no roster to enforce: `None` runs the node - // unenforced instead of collapsing it to a self-only allowlist. - let trusted_owner_ids = - (!mode.is_open()).then(|| mesh_llm::owner_ids_from_events(&events)); + let trusted_owner_ids = mesh_llm::owner_ids_from_events(&events); let join_token = mesh_llm::ensure_owner_identity() .ok() .and_then(|identity| { buzz_mesh_join_targets( - mesh_llm::availability_from_events_for_mode(events, mode).serve_targets, + mesh_llm::availability_from_events(events).serve_targets, &identity.owner_id, ) .into_iter() @@ -444,7 +332,7 @@ async fn resolve_buzz_mesh_startup_at( eprintln!( "buzz-mesh: startup discovery failed; allowing only this node and starting isolated for now: {error}" ); - (Some(Vec::new()), None) + (Vec::new(), None) } } } @@ -485,7 +373,7 @@ pub(crate) async fn restore_mesh_sharing(app: &AppHandle, state: &AppState) -> C join_token, mesh_name: Some(buzz_mesh_name_for_relay(&relay_url)), relay_url: Some(relay_url), - trusted_owner_ids, + trusted_owner_ids: Some(trusted_owner_ids), }; let started = mesh_llm::DesktopMeshRuntime::start(request) .await @@ -558,17 +446,10 @@ pub async fn mesh_start_node( // Frontend requests never carry a roster. Resolve it and the bootstrap // endpoint from one snapshot so UI startup does not repeat relay probes. - let roster_unresolved = request.trusted_owner_ids.is_none(); - if roster_unresolved || request.join_token.is_none() { + if request.trusted_owner_ids.is_none() || request.join_token.is_none() { let (trusted_owner_ids, join_token) = resolve_buzz_mesh_startup_at(&state, &relay_url).await; - // Adopt the resolver's decision only when the caller supplied no - // roster. `None` from the resolver is meaningful (open relay: run - // unenforced), so this cannot use `get_or_insert`, which would treat - // it as "nothing resolved" and leave the field untouched. - if roster_unresolved { - request.trusted_owner_ids = trusted_owner_ids; - } + request.trusted_owner_ids.get_or_insert(trusted_owner_ids); if request.join_token.is_none() { request.join_token = join_token; } @@ -709,7 +590,7 @@ pub(crate) async fn ensure_client_node_for_model( join_token: Some(join_token.clone()), mesh_name: Some(buzz_mesh_name(state)), relay_url: Some(relay::relay_ws_url_with_override(state)), - trusted_owner_ids: resolve_trusted_owner_ids_or_self_only(state).await, + trusted_owner_ids: Some(resolve_trusted_owner_ids_or_self_only(state).await), }; let mut runtime = state.mesh_llm_runtime.lock().await; if let Some(existing) = runtime.as_ref() { @@ -757,11 +638,9 @@ pub(crate) async fn resolve_mesh_bootstrap_target( if model_id.is_empty() { return Ok(None); } - let relay_url = relay::relay_ws_url_with_override(state); - let mode = resolved_relay_mesh_mode(&relay_url).await?; - let events = query_mesh_discovery_events_for_mode(state, &relay_url, mode).await?; + let events = query_mesh_discovery_events(state).await?; Ok(pick_serve_target_for_model( - mesh_llm::availability_from_events_for_mode(events, mode).serve_targets, + mesh_llm::availability_from_events(events).serve_targets, model_id, )) } diff --git a/desktop/src-tauri/src/commands/mesh_llm_tests.rs b/desktop/src-tauri/src/commands/mesh_llm_tests.rs index d89c1b0b73..c4e1ae2e42 100644 --- a/desktop/src-tauri/src/commands/mesh_llm_tests.rs +++ b/desktop/src-tauri/src/commands/mesh_llm_tests.rs @@ -1,6 +1,5 @@ use super::*; use crate::app_state::build_app_state; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; fn target(model_id: &str, endpoint_addr: &str) -> mesh_llm::MeshServeTarget { mesh_llm::MeshServeTarget { @@ -28,83 +27,6 @@ fn reported_target( target } -#[tokio::test] -async fn relay_mode_probe_rechecks_the_same_url_without_caching() { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") - .await - .expect("bind test NIP-11 server"); - let addr = listener.local_addr().expect("test server address"); - let server = tokio::spawn(async move { - for body in [ - r#"{"supported_nips":[1,11,42]}"#, - r#"{"supported_nips":[1,11,42,43]}"#, - r#"{"supported_nips":[1,"43"]}"#, - ] { - let (mut stream, _) = listener.accept().await.expect("accept NIP-11 request"); - let mut request = vec![0; 2048]; - let bytes_read = stream.read(&mut request).await.expect("read request"); - let request = String::from_utf8_lossy(&request[..bytes_read]); - assert!(request.starts_with("GET / HTTP/1.1")); - assert!(request - .to_ascii_lowercase() - .contains("accept: application/nostr+json")); - - let response = format!( - "HTTP/1.1 200 OK\r\nContent-Type: application/nostr+json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", - body.len() - ); - stream - .write_all(response.as_bytes()) - .await - .expect("write response"); - } - }); - let relay_url = format!("ws://{addr}"); - - assert_eq!( - probe_relay_mesh_mode(&relay_url).await, - Ok(mesh_llm::MeshRelayMode::OpenNoMembership) - ); - assert_eq!( - probe_relay_mesh_mode(&relay_url).await, - Ok(mesh_llm::MeshRelayMode::ClosedMembershipEnforced) - ); - assert!(probe_relay_mesh_mode(&relay_url).await.is_err()); - server.await.expect("NIP-11 server task"); -} - -#[tokio::test] -async fn relay_mode_probe_rejects_non_success_with_valid_open_body() { - let listener = tokio::net::TcpListener::bind("127.0.0.1:0") - .await - .expect("bind test NIP-11 server"); - let addr = listener.local_addr().expect("test server address"); - let server = tokio::spawn(async move { - let (mut stream, _) = listener.accept().await.expect("accept NIP-11 request"); - let mut request = vec![0; 2048]; - let bytes_read = stream.read(&mut request).await.expect("read request"); - let request = String::from_utf8_lossy(&request[..bytes_read]); - assert!(request.starts_with("GET / HTTP/1.1")); - - let body = r#"{"supported_nips":[1,11,42]}"#; - let response = format!( - "HTTP/1.1 503 Service Unavailable\r\nContent-Type: application/nostr+json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", - body.len() - ); - stream - .write_all(response.as_bytes()) - .await - .expect("write response"); - }); - let relay_url = format!("ws://{addr}"); - - let error = probe_relay_mesh_mode(&relay_url) - .await - .expect_err("non-success HTTP status must not prove an open relay"); - assert!(error.contains("503 Service Unavailable"), "{error}"); - server.await.expect("NIP-11 server task"); -} - #[test] fn buzz_mesh_join_uses_the_same_live_member_from_every_other_node() { let targets = vec![ diff --git a/desktop/src-tauri/src/mesh_llm/coordinator.rs b/desktop/src-tauri/src/mesh_llm/coordinator.rs index 5de0cd2c86..066fa46373 100644 --- a/desktop/src-tauri/src/mesh_llm/coordinator.rs +++ b/desktop/src-tauri/src/mesh_llm/coordinator.rs @@ -11,10 +11,6 @@ use std::time::Duration; use nostr::Tag; use tauri::{AppHandle, Manager}; -use super::coordinator_admission::{ - clear_mode_restart, mode_reconcile_action, request_mode_restart, ModeEvidence, - ModeReconcileAction, ModeReconcileState, -}; use crate::app_state::AppState; /// Client-owned parameterized-replaceable discovery note. We use the standard @@ -36,6 +32,7 @@ const INGRESS_WATCHDOG_MAX: Duration = Duration::from_secs(120); /// mesh instead of remaining independent islands. const MESH_JOIN_POLL_INTERVAL: Duration = Duration::from_secs(15); const MESH_JOIN_RETRY_MAX: Duration = Duration::from_secs(120); + pub struct MeshCoordinator { _status_publisher: tokio::task::JoinHandle<()>, _roster_watcher: tokio::task::JoinHandle<()>, @@ -66,15 +63,12 @@ pub async fn start_coordinator(app: AppHandle) { }); let roster_app = app.clone(); let roster_watcher = tokio::spawn(async move { - // Carries pending observations across polls. Roster reductions and - // admission-mode changes must both be seen twice before a restart. + // Carries a shrink awaiting confirmation across polls (hysteresis): + // a reduced roster must be seen twice in a row before we tear down. let mut pending_shrink: Option> = None; - let mut mode_reconcile = ModeReconcileState::default(); loop { tokio::time::sleep(ROSTER_POLL_INTERVAL).await; - if let Err(error) = - reconcile_roster(&roster_app, &mut pending_shrink, &mut mode_reconcile).await - { + if let Err(error) = reconcile_roster(&roster_app, &mut pending_shrink).await { eprintln!("buzz-mesh: roster reconcile failed: {error}"); } } @@ -240,28 +234,14 @@ fn roster_shrinks(current: &[String], fresh: &[String]) -> bool { /// /// Rules: /// - query failed (`Err`) → `Keep` (never de-admit on a relay blip) -/// - relay enforces no membership (`Ok(None)`), roster beyond self → `Keep` (see below) -/// - relay enforces no membership (`Ok(None)`), self-only roster → `RestartProcess` (see below) /// - resolved roster == current → `Keep` (no-op) /// - grows (only additions) → `RestartProcess` immediately (fast admission) /// - shrinks/empties, first observation → `AwaitConfirm` (hold, re-check next poll) /// - shrinks/empties, confirmed → `RestartProcess` (same reduced roster twice) -/// -/// `Ok(None)` means the relay does not enforce NIP-43 membership, so there is -/// no roster to reconcile against. A node enforcing an allowlist that admits -/// *other* owners keeps it: relaxing a live multi-party allowlist on the -/// strength of a NIP-11 read would let a relay downgrade (or a spoofed -/// document) de-restrict a running mesh. A **self-only** allowlist is -/// different: it admits nobody else, so it protects nothing — and it is -/// exactly the fallback a transient NIP-11 probe failure at startup produces -/// on an open relay. Without recovery here that node stays at "1 NODE" until -/// an app restart (the original bug, reintroduced by a single HTTP blip). -/// Restarting resolves the roster afresh, which on an open relay yields -/// `None` → `TrustPolicy::Off`. fn roster_reconcile_action( current_owners: &[String], pending_shrink: Option<&[String]>, - query: Result>, String>, + query: Result, String>, ) -> RosterReconcileAction { let fresh = match query { Err(error) => { @@ -270,20 +250,7 @@ fn roster_reconcile_action( ); return RosterReconcileAction::Keep; } - Ok(None) if current_owners.is_empty() => { - eprintln!( - "buzz-mesh: relay does not enforce membership and this node is isolated by the \ - startup fallback; restarting to run the mesh unenforced" - ); - return RosterReconcileAction::RestartProcess; - } - Ok(None) => { - eprintln!( - "buzz-mesh: relay does not enforce membership; keeping this node's current allowlist" - ); - return RosterReconcileAction::Keep; - } - Ok(Some(fresh)) => fresh, + Ok(fresh) => fresh, }; if fresh == current_owners { @@ -303,81 +270,10 @@ fn roster_reconcile_action( } } -fn runtime_needs_mode_probe(current_owners: &[String]) -> bool { - current_owners.is_empty() -} - -#[derive(Debug, PartialEq, Eq)] -enum OpenRuntimeObservation { - Open, - ClosedWithRoster, - ClosedWithoutRoster(String), - Unknown(String), -} - -fn classify_open_runtime_observation( - mode: Result, - closed_roster: Option, String>>, -) -> OpenRuntimeObservation { - match mode { - Ok(mode) if mode.is_open() => OpenRuntimeObservation::Open, - Ok(_) => match closed_roster { - Some(Ok(_)) => OpenRuntimeObservation::ClosedWithRoster, - Some(Err(error)) => OpenRuntimeObservation::ClosedWithoutRoster(error), - None => OpenRuntimeObservation::ClosedWithoutRoster( - "closed relay roster was not queried".to_string(), - ), - }, - Err(error) => OpenRuntimeObservation::Unknown(error), - } -} - -async fn observe_open_runtime_relay(state: &AppState, relay_url: &str) -> OpenRuntimeObservation { - let mode = crate::commands::mesh_llm::resolved_relay_mesh_mode(relay_url).await; - let closed_roster = if matches!(&mode, Ok(mode) if !mode.is_open()) { - Some(crate::commands::mesh_llm::resolve_closed_trusted_owner_ids_at(state, relay_url).await) - } else { - None - }; - classify_open_runtime_observation(mode, closed_roster) -} - -#[derive(Debug, PartialEq, Eq)] -enum SelfOnlyModeAction { - Keep, - AwaitConfirm, - RestartProcess, -} - -fn reconcile_self_only_mode( - state: &mut ModeReconcileState, - query: &Result>, String>, -) -> SelfOnlyModeAction { - match query { - Ok(None) => match mode_reconcile_action(state, Ok(ModeEvidence::Open), false) { - ModeReconcileAction::Keep => SelfOnlyModeAction::Keep, - ModeReconcileAction::AwaitConfirm => SelfOnlyModeAction::AwaitConfirm, - ModeReconcileAction::RestartProcess => SelfOnlyModeAction::RestartProcess, - }, - Err(_) => { - let _ = mode_reconcile_action(state, Err(()), false); - SelfOnlyModeAction::Keep - } - Ok(Some(_)) => { - let _ = mode_reconcile_action(state, Ok(ModeEvidence::Closed), false); - SelfOnlyModeAction::Keep - } - } -} - async fn reconcile_roster( app: &AppHandle, pending_shrink: &mut Option>, - mode_reconcile: &mut ModeReconcileState, ) -> Result<(), String> { - if mode_reconcile.restart_requested() { - return Ok(()); - } let state = app.state::(); let current_request = { let runtime = state.mesh_llm_runtime.lock().await; @@ -385,68 +281,11 @@ async fn reconcile_roster( Some(runtime) => runtime.start_request().clone(), None => { *pending_shrink = None; - *mode_reconcile = ModeReconcileState::default(); return Ok(()); } } }; - let current_owners = current_request.trusted_owner_ids.as_ref(); - // Closed runtimes keep the existing roster-only reconciliation path. Open - // runtimes re-probe NIP-11 so a later open -> closed relay restart is not - // hidden by a permanent mode cache. - let relay_url = current_request - .relay_url - .as_deref() - .map(str::to_owned) - .unwrap_or_else(|| crate::relay::relay_ws_url_with_override(&state)); - let Some(current_owners) = current_owners else { - let observation = observe_open_runtime_relay(&state, &relay_url).await; - let evidence = match &observation { - OpenRuntimeObservation::Open => Ok(ModeEvidence::Open), - OpenRuntimeObservation::ClosedWithRoster - | OpenRuntimeObservation::ClosedWithoutRoster(_) => Ok(ModeEvidence::Closed), - OpenRuntimeObservation::Unknown(_) => Err(()), - }; - match mode_reconcile_action(mode_reconcile, evidence, true) { - ModeReconcileAction::Keep => { - if let OpenRuntimeObservation::Unknown(error) = observation { - eprintln!( - "buzz-mesh: open-relay mode recheck failed once; keeping the running policy pending one retry: {error}" - ); - } - } - ModeReconcileAction::AwaitConfirm => eprintln!( - "buzz-mesh: relay membership mode changed; awaiting confirmation before restarting Buzz" - ), - ModeReconcileAction::RestartProcess => { - let reason = match &observation { - OpenRuntimeObservation::ClosedWithRoster => "open-to-closed", - OpenRuntimeObservation::ClosedWithoutRoster(_) => { - "open-to-closed-without-roster" - } - OpenRuntimeObservation::Unknown(_) => "open-mode-unknown", - OpenRuntimeObservation::Open => return Ok(()), - }; - if !request_mode_restart(app, mode_reconcile, reason)? { - eprintln!( - "buzz-mesh: mode-policy restart is already pending or throttled after a recent attempt; keeping the running policy" - ); - } else { - match &observation { - OpenRuntimeObservation::ClosedWithRoster => eprintln!( - "buzz-mesh: relay repeatedly reports membership enforcement; restarting Buzz to apply its allowlist" - ), - OpenRuntimeObservation::ClosedWithoutRoster(error) => eprintln!( - "buzz-mesh: relay repeatedly reports membership enforcement but its roster is unavailable; restarting Buzz fail-closed: {error}" - ), - OpenRuntimeObservation::Unknown(error) => eprintln!( - "buzz-mesh: open-relay mode remained unknown across two polls; restarting Buzz fail-closed: {error}" - ), - OpenRuntimeObservation::Open => {} - } - } - } - } + let Some(current_owners) = current_request.trusted_owner_ids.as_ref() else { *pending_shrink = None; return Ok(()); }; @@ -455,42 +294,12 @@ async fn reconcile_roster( // other member on a transient relay blip (the flapping restart loop). Keep // the current allowlist and try again on the next poll. A shrink is held // for one extra poll (hysteresis) so a single short-read never tears down. - let query = if runtime_needs_mode_probe(current_owners) { - // Self-only is the fail-safe startup policy when the initial NIP-11 - // probe is inconclusive. Re-probe this isolated state so a later valid - // open result can recover it; errors still keep it self-only. - crate::commands::mesh_llm::resolve_trusted_owner_ids_at(&state, &relay_url).await - } else { - crate::commands::mesh_llm::resolve_closed_trusted_owner_ids_at(&state, &relay_url) - .await - .map(Some) - }; - if matches!(query, Ok(Some(_))) { - // A successful closed-roster read is decisive non-restarting mode - // evidence, so a prior cross-process mode-restart throttle may clear. - clear_mode_restart(app); - } - if current_owners.is_empty() { - match reconcile_self_only_mode(mode_reconcile, &query) { - SelfOnlyModeAction::Keep => {} - SelfOnlyModeAction::AwaitConfirm => { - eprintln!( - "buzz-mesh: isolated startup fallback observed an open relay; awaiting confirmation before restarting Buzz" - ); - return Ok(()); - } - SelfOnlyModeAction::RestartProcess => { - if !request_mode_restart(app, mode_reconcile, "self-only-to-open")? { - eprintln!( - "buzz-mesh: open-mode recovery restart is already pending or throttled after a recent attempt; keeping the self-only fallback" - ); - } - return Ok(()); - } - } - } else { - *mode_reconcile = ModeReconcileState::default(); - } + let relay_url = current_request + .relay_url + .as_deref() + .map(str::to_owned) + .unwrap_or_else(|| crate::relay::relay_ws_url_with_override(&state)); + let query = crate::commands::mesh_llm::resolve_trusted_owner_ids_at(&state, &relay_url).await; match roster_reconcile_action(current_owners, pending_shrink.as_deref(), query) { RosterReconcileAction::Keep => { *pending_shrink = None; @@ -749,44 +558,10 @@ mod tests { ); } - #[test] - fn self_only_wiring_never_restarts_after_one_open_observation() { - let mut state = ModeReconcileState::default(); - let query = Ok(None); - assert_eq!( - reconcile_self_only_mode(&mut state, &query), - SelfOnlyModeAction::AwaitConfirm - ); - } - - #[test] - fn self_only_wiring_restarts_after_two_open_observations() { - let mut state = ModeReconcileState::default(); - let query = Ok(None); - assert_eq!( - reconcile_self_only_mode(&mut state, &query), - SelfOnlyModeAction::AwaitConfirm - ); - assert_eq!( - reconcile_self_only_mode(&mut state, &query), - SelfOnlyModeAction::RestartProcess - ); - } - - // The other side of that rule: an allowlist that admits OTHER owners is - // never relaxed on the strength of a NIP-11 read. A relay downgrade (or a - // spoofed document) must not de-restrict a running mesh. - #[test] - fn open_relay_report_keeps_a_multi_party_allowlist() { - let current = vec!["owner-a".to_string(), "owner-b".to_string()]; - let action = roster_reconcile_action(¤t, None, Ok(None)); - assert_eq!(action, RosterReconcileAction::Keep); - } - #[test] fn unchanged_roster_is_a_noop() { let current = vec!["owner-a".to_string()]; - let action = roster_reconcile_action(¤t, None, Ok(Some(vec!["owner-a".to_string()]))); + let action = roster_reconcile_action(¤t, None, Ok(vec!["owner-a".to_string()])); assert_eq!(action, RosterReconcileAction::Keep); } @@ -795,7 +570,7 @@ mod tests { fn roster_growth_requests_process_restart_immediately() { let current = vec!["owner-a".to_string()]; let fresh = vec!["owner-a".to_string(), "owner-c".to_string()]; - let action = roster_reconcile_action(¤t, None, Ok(Some(fresh))); + let action = roster_reconcile_action(¤t, None, Ok(fresh)); assert_eq!(action, RosterReconcileAction::RestartProcess); } @@ -804,7 +579,7 @@ mod tests { fn roster_shrink_awaits_confirmation_first() { let current = vec!["owner-a".to_string(), "owner-b".to_string()]; let reduced = vec!["owner-a".to_string()]; - let action = roster_reconcile_action(¤t, None, Ok(Some(reduced.clone()))); + let action = roster_reconcile_action(¤t, None, Ok(reduced.clone())); assert_eq!(action, RosterReconcileAction::AwaitConfirm(reduced)); } @@ -813,7 +588,7 @@ mod tests { fn roster_shrink_requests_process_restart_once_confirmed() { let current = vec!["owner-a".to_string(), "owner-b".to_string()]; let reduced = vec!["owner-a".to_string()]; - let action = roster_reconcile_action(¤t, Some(&reduced), Ok(Some(reduced.clone()))); + let action = roster_reconcile_action(¤t, Some(&reduced), Ok(reduced.clone())); assert_eq!(action, RosterReconcileAction::RestartProcess); } @@ -824,11 +599,8 @@ mod tests { let current = vec!["a".to_string(), "b".to_string(), "c".to_string()]; let first_reduced = vec!["a".to_string(), "b".to_string()]; let second_reduced = vec!["a".to_string()]; - let action = roster_reconcile_action( - ¤t, - Some(&first_reduced), - Ok(Some(second_reduced.clone())), - ); + let action = + roster_reconcile_action(¤t, Some(&first_reduced), Ok(second_reduced.clone())); assert_eq!(action, RosterReconcileAction::AwaitConfirm(second_reduced)); } @@ -837,10 +609,10 @@ mod tests { #[test] fn genuinely_empty_roster_awaits_then_restarts_to_self_only() { let current = vec!["owner-a".to_string()]; - let first = roster_reconcile_action(¤t, None, Ok(Some(Vec::new()))); + let first = roster_reconcile_action(¤t, None, Ok(Vec::new())); assert_eq!(first, RosterReconcileAction::AwaitConfirm(Vec::new())); let empty: Vec = Vec::new(); - let confirmed = roster_reconcile_action(¤t, Some(&empty), Ok(Some(Vec::new()))); + let confirmed = roster_reconcile_action(¤t, Some(&empty), Ok(Vec::new())); assert_eq!(confirmed, RosterReconcileAction::RestartProcess); } @@ -849,35 +621,12 @@ mod tests { fn roster_shrink_then_recovery_keeps_allowlist() { let current = vec!["owner-a".to_string(), "owner-b".to_string()]; let reduced = vec!["owner-a".to_string()]; - let held = roster_reconcile_action(¤t, None, Ok(Some(reduced.clone()))); + let held = roster_reconcile_action(¤t, None, Ok(reduced.clone())); assert_eq!(held, RosterReconcileAction::AwaitConfirm(reduced.clone())); - let recovered = - roster_reconcile_action(¤t, Some(&reduced), Ok(Some(current.clone()))); + let recovered = roster_reconcile_action(¤t, Some(&reduced), Ok(current.clone())); assert_eq!(recovered, RosterReconcileAction::Keep); } - #[test] - fn only_self_only_closed_runtimes_reprobe_mode() { - let self_only: Vec = Vec::new(); - let closed = vec!["owner-a".to_string()]; - assert!(runtime_needs_mode_probe(&self_only)); - assert!(!runtime_needs_mode_probe(&closed)); - } - - #[test] - fn open_to_closed_with_unavailable_roster_fails_closed() { - let observation = classify_open_runtime_observation( - Ok(crate::mesh_llm::MeshRelayMode::ClosedMembershipEnforced), - Some(Err("relay returned no membership snapshot".to_string())), - ); - assert_eq!( - observation, - OpenRuntimeObservation::ClosedWithoutRoster( - "relay returned no membership snapshot".to_string() - ) - ); - } - #[test] fn member_heartbeat_leaves_room_before_admission_status_expires() { assert!( diff --git a/desktop/src-tauri/src/mesh_llm/coordinator_admission.rs b/desktop/src-tauri/src/mesh_llm/coordinator_admission.rs deleted file mode 100644 index 54d8ccc77b..0000000000 --- a/desktop/src-tauri/src/mesh_llm/coordinator_admission.rs +++ /dev/null @@ -1,228 +0,0 @@ -use std::time::Duration; - -use tauri::{AppHandle, Manager}; - -/// Prevent a flaky mode probe from creating an unbounded whole-app restart -/// loop while changing admission policy. -const MODE_RESTART_COOLDOWN: Duration = Duration::from_secs(10 * 60); -const MODE_RESTART_MARKER: &str = "mesh-mode-recovery-restart"; - -fn mode_restart_marker(app: &AppHandle) -> Result { - Ok(app - .path() - .app_data_dir() - .map_err(|error| format!("failed to resolve app data directory: {error}"))? - .join(MODE_RESTART_MARKER)) -} - -fn restart_marker_is_throttled( - modified: std::time::SystemTime, - now: std::time::SystemTime, -) -> bool { - // A clock rollback makes the marker appear to be in the future. Fail safe: - // retain the throttle rather than disabling the only cross-process bound. - now.duration_since(modified) - .map_or(true, |elapsed| elapsed < MODE_RESTART_COOLDOWN) -} - -fn mode_restart_is_throttled(app: &AppHandle) -> Result { - let marker = mode_restart_marker(app)?; - let modified = match std::fs::metadata(marker).and_then(|metadata| metadata.modified()) { - Ok(modified) => modified, - Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), - Err(error) => return Err(format!("failed to read mesh restart marker: {error}")), - }; - Ok(restart_marker_is_throttled( - modified, - std::time::SystemTime::now(), - )) -} - -fn mark_mode_restart(app: &AppHandle, reason: &str) -> Result<(), String> { - let marker = mode_restart_marker(app)?; - if let Some(parent) = marker.parent() { - std::fs::create_dir_all(parent) - .map_err(|error| format!("failed to create app data directory: {error}"))?; - } - std::fs::write(marker, format!("{reason}\n")) - .map_err(|error| format!("failed to write mesh restart marker: {error}")) -} - -pub(super) fn clear_mode_restart(app: &AppHandle) { - if let Ok(marker) = mode_restart_marker(app) { - match std::fs::remove_file(marker) { - Ok(()) => {} - Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} - Err(error) => eprintln!("buzz-mesh: failed to clear mesh restart marker: {error}"), - } - } -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(super) enum ModeEvidence { - Open, - Closed, -} - -#[derive(Debug, Default, PartialEq, Eq)] -pub(super) struct ModeReconcileState { - pending: Option, - consecutive_unknown: u8, - restart_requested: bool, -} - -impl ModeReconcileState { - pub(super) fn restart_requested(&self) -> bool { - self.restart_requested - } - - pub(super) fn latch_restart(&mut self) -> bool { - if self.restart_requested { - false - } else { - self.restart_requested = true; - true - } - } -} - -#[derive(Debug, PartialEq, Eq)] -pub(super) enum ModeReconcileAction { - Keep, - AwaitConfirm, - RestartProcess, -} - -pub(super) fn request_mode_restart( - app: &AppHandle, - state: &mut ModeReconcileState, - reason: &str, -) -> Result { - if state.restart_requested || mode_restart_is_throttled(app)? { - return Ok(false); - } - mark_mode_restart(app, reason)?; - if !state.latch_restart() { - return Ok(false); - } - app.request_restart(); - Ok(true) -} - -/// Require two consecutive mode observations before restarting the whole app. -/// Unknown observations preserve a running policy once, but a second -/// consecutive unknown while admission is unenforced restarts fail-closed. -pub(super) fn mode_reconcile_action( - state: &mut ModeReconcileState, - evidence: Result, - running_open: bool, -) -> ModeReconcileAction { - match evidence { - Err(()) => { - state.pending = None; - state.consecutive_unknown = state.consecutive_unknown.saturating_add(1); - if running_open && state.consecutive_unknown >= 2 { - state.consecutive_unknown = 0; - ModeReconcileAction::RestartProcess - } else { - ModeReconcileAction::Keep - } - } - Ok(observed) => { - state.consecutive_unknown = 0; - let transition = (running_open && observed == ModeEvidence::Closed) - || (!running_open && observed == ModeEvidence::Open); - if !transition { - state.pending = None; - return ModeReconcileAction::Keep; - } - if state.pending == Some(observed) { - state.pending = None; - ModeReconcileAction::RestartProcess - } else { - state.pending = Some(observed); - ModeReconcileAction::AwaitConfirm - } - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn restart_latch_only_arms_once() { - let mut state = ModeReconcileState::default(); - assert!(state.latch_restart()); - assert!(state.restart_requested()); - assert!(!state.latch_restart()); - } - - #[test] - fn clock_rollback_keeps_restart_throttled() { - let now = std::time::SystemTime::UNIX_EPOCH + Duration::from_secs(100); - let future_marker = now + Duration::from_secs(1); - assert!(restart_marker_is_throttled(future_marker, now)); - } - - #[test] - fn old_restart_marker_is_not_throttled() { - let marker = std::time::SystemTime::UNIX_EPOCH; - let now = marker + MODE_RESTART_COOLDOWN + Duration::from_secs(1); - assert!(!restart_marker_is_throttled(marker, now)); - } - - #[test] - fn transition_requires_two_matching_observations() { - let mut state = ModeReconcileState::default(); - assert_eq!( - mode_reconcile_action(&mut state, Ok(ModeEvidence::Closed), true), - ModeReconcileAction::AwaitConfirm - ); - assert_eq!( - mode_reconcile_action(&mut state, Ok(ModeEvidence::Closed), true), - ModeReconcileAction::RestartProcess - ); - } - - #[test] - fn recovery_cancels_pending_restart() { - let mut state = ModeReconcileState::default(); - assert_eq!( - mode_reconcile_action(&mut state, Ok(ModeEvidence::Closed), true), - ModeReconcileAction::AwaitConfirm - ); - assert_eq!( - mode_reconcile_action(&mut state, Ok(ModeEvidence::Open), true), - ModeReconcileAction::Keep - ); - assert_eq!(state, ModeReconcileState::default()); - } - - #[test] - fn persistently_unknown_open_runtime_restarts_fail_closed() { - let mut state = ModeReconcileState::default(); - assert_eq!( - mode_reconcile_action(&mut state, Err(()), true), - ModeReconcileAction::Keep - ); - assert_eq!( - mode_reconcile_action(&mut state, Err(()), true), - ModeReconcileAction::RestartProcess - ); - } - - #[test] - fn isolated_open_recovery_requires_confirmation() { - let mut state = ModeReconcileState::default(); - assert_eq!( - mode_reconcile_action(&mut state, Ok(ModeEvidence::Open), false), - ModeReconcileAction::AwaitConfirm - ); - assert_eq!( - mode_reconcile_action(&mut state, Ok(ModeEvidence::Open), false), - ModeReconcileAction::RestartProcess - ); - } -} diff --git a/desktop/src-tauri/src/mesh_llm/discovery.rs b/desktop/src-tauri/src/mesh_llm/discovery.rs index d347a16d80..2df4a7f8c5 100644 --- a/desktop/src-tauri/src/mesh_llm/discovery.rs +++ b/desktop/src-tauri/src/mesh_llm/discovery.rs @@ -20,66 +20,6 @@ fn status_is_fresh(event: &nostr::Event, now: u64) -> bool { >= now } -/// Whether the relay enforces NIP-43 relay membership. -/// -/// This is a property of the *relay deployment*, not of any particular query -/// result, and it is resolved from the relay's NIP-11 document (NIP-43 is -/// advertised only when membership is actually enforced). It must never be -/// inferred from an absent membership snapshot: on a closed relay a snapshot -/// can also be missing because of a transient replication gap, and treating -/// that as "open" would silently drop admission enforcement exactly when the -/// relay is least healthy. -#[derive(Clone, Copy, Debug, Eq, PartialEq)] -pub enum MeshRelayMode { - /// Membership is enforced (`BUZZ_REQUIRE_RELAY_MEMBERSHIP=true`). Every - /// participant is a NIP-43 direct member, so the membership snapshot is the - /// trust boundary for both admission and routing. - ClosedMembershipEnforced, - /// Membership is not enforced. Signing in never writes a `relay_members` - /// row, so the snapshot lists at most whoever was added out of band and is - /// structurally useless as a trust boundary. Mesh runs unenforced (stock - /// MeshLLM behaviour) and is scoped by the relay-derived mesh topic. - OpenNoMembership, -} - -impl MeshRelayMode { - pub(crate) fn is_open(self) -> bool { - matches!(self, Self::OpenNoMembership) - } -} - -/// Resolve [`MeshRelayMode`] from a relay's NIP-11 document. -/// -/// `buzz-relay` lists NIP-43 in `supported_nips` only when it has a stable -/// signing key *and* `BUZZ_REQUIRE_RELAY_MEMBERSHIP=true`, so the advertisement -/// is a faithful proxy for "membership is enforced here" when the field is -/// well formed. A missing or malformed `supported_nips` field is rejected so -/// callers keep the fail-closed fallback rather than treating an ambiguous -/// response as evidence that membership is disabled. -pub fn relay_mesh_mode_from_nip11(document: &serde_json::Value) -> Result { - let nips = document - .get("supported_nips") - .and_then(serde_json::Value::as_array) - .ok_or_else(|| "NIP-11 document has no supported_nips array".to_string())?; - let nips = nips - .iter() - .enumerate() - .map(|(index, nip)| { - nip.as_u64() - .filter(|value| *value <= u32::MAX as u64) - .map(|value| value as u32) - .ok_or_else(|| { - format!("NIP-11 supported_nips[{index}] is not a valid u32 NIP number") - }) - }) - .collect::, _>>()?; - if nips.contains(&43) { - Ok(MeshRelayMode::ClosedMembershipEnforced) - } else { - Ok(MeshRelayMode::OpenNoMembership) - } -} - fn dedupe_targets(targets: Vec) -> Vec { let mut by_endpoint_and_model = BTreeMap::<(String, String), MeshServeTarget>::new(); for target in targets { @@ -232,47 +172,14 @@ fn endpoint_binding_is_valid(event: &nostr::Event, content: &serde_json::Value) .is_ok() } -#[cfg(test)] pub fn availability_from_events(events: Vec) -> MeshAvailability { - availability_from_events_for_mode(events, MeshRelayMode::ClosedMembershipEnforced) -} - -/// Resolve availability from relay events, scoped to the relay's membership mode. -/// -/// On a membership-enforcing relay the NIP-43 snapshot gates routing: a serve -/// target is selectable only when its status note is signed by a current -/// member. On a relay that does not enforce membership there is no such -/// roster to intersect with — requiring one would make every serving peer -/// permanently undiscoverable — so status notes are accepted on their own -/// merits (fresh, well-formed, and carrying a valid owner-signed endpoint -/// binding). The mesh topic is derived from the relay origin, so this widens -/// discovery to that relay's participants, not to the public internet. -pub fn availability_from_events_for_mode( - events: Vec, - mode: MeshRelayMode, -) -> MeshAvailability { if events.is_empty() { return MeshAvailability::unavailable("Buzz shared compute status is not published yet"); } - // On an open relay the snapshot is not a trust boundary even when one - // happens to exist (added out of band): admission runs `TrustPolicy::Off` - // there, so intersecting routing with a stray roster would hide peers the - // mesh happily admits. Ignore membership entirely and accept status notes - // on their own merits. - let members = if mode.is_open() { - None - } else { - match latest_membership_list(&events) { - Some(members) => Some(members), - // An enforcing relay always publishes a snapshot (even a - // zero-member one), so its absence means an incomplete read: - // stay closed. - None => { - return MeshAvailability::unavailable( - "Buzz shared compute is waiting for the current member roster", - ); - } - } + let Some(members) = latest_membership_list(&events) else { + return MeshAvailability::unavailable( + "Buzz shared compute is waiting for the current member roster", + ); }; // Status is replaceable per member pubkey, so a query returns multiple @@ -284,14 +191,12 @@ pub fn availability_from_events_for_mode( let now = nostr::Timestamp::now().as_secs(); for event in events { - if event.kind.as_u16() as u64 != MESH_STATUS_KIND || !status_is_fresh(&event, now) { + if event.kind.as_u16() as u64 != MESH_STATUS_KIND + || !status_is_fresh(&event, now) + || !members.contains(&event.pubkey.to_hex().to_ascii_lowercase()) + { continue; } - if let Some(members) = members.as_ref() { - if !members.contains(&event.pubkey.to_hex().to_ascii_lowercase()) { - continue; - } - } let Ok(content) = serde_json::from_str::(&event.content) else { continue; }; diff --git a/desktop/src-tauri/src/mesh_llm/mod.rs b/desktop/src-tauri/src/mesh_llm/mod.rs index 1aa7ead858..e206c53886 100644 --- a/desktop/src-tauri/src/mesh_llm/mod.rs +++ b/desktop/src-tauri/src/mesh_llm/mod.rs @@ -1,16 +1,12 @@ use std::collections::BTreeMap; mod coordinator; -mod coordinator_admission; pub(crate) use coordinator::{publish_current_status_once, publish_stopped_status_once_at}; pub use coordinator::{start_coordinator, MeshCoordinator, KIND_BUZZ_MESH_MEMBER_STATUS}; mod discovery; -#[cfg(test)] -pub use discovery::availability_from_events; pub use discovery::{ - availability_from_events_for_mode, mesh_status_filter, owner_ids_from_events, - relay_membership_filter, relay_mesh_mode_from_nip11, MeshRelayMode, + availability_from_events, mesh_status_filter, owner_ids_from_events, relay_membership_filter, }; pub(crate) use discovery::{ current_member_pubkeys, has_membership_snapshot, MESH_STATUS_PAGE_SIZE, diff --git a/desktop/src-tauri/src/mesh_llm/mod_tests.rs b/desktop/src-tauri/src/mesh_llm/mod_tests.rs index a1db42225f..557cd040fa 100644 --- a/desktop/src-tauri/src/mesh_llm/mod_tests.rs +++ b/desktop/src-tauri/src/mesh_llm/mod_tests.rs @@ -665,176 +665,6 @@ fn owner_roster_without_membership_list_fails_closed() { assert!(super::owner_ids_from_events(&events).is_empty()); } -#[test] -fn relay_mesh_mode_reads_nip43_advertisement() { - use super::MeshRelayMode; - - // `buzz-relay` advertises NIP-43 only when membership is enforced, so the - // advertisement is the deployment-mode signal. - assert_eq!( - super::relay_mesh_mode_from_nip11(&json!({"supported_nips": [1, 11, 42, 43]})), - Ok(MeshRelayMode::ClosedMembershipEnforced) - ); - // The live open relay's actual shape (no 43). - assert_eq!( - super::relay_mesh_mode_from_nip11( - &json!({"supported_nips": [1, 2, 10, 11, 16, 17, 23, 25, 29, 33, 38, 42, 50, 56]}) - ), - Ok(MeshRelayMode::OpenNoMembership) - ); - // Missing, malformed, or partially malformed mode evidence must fail at - // the caller. An ambiguous NIP-11 response is not proof of either mode. - assert!(super::relay_mesh_mode_from_nip11(&json!({})).is_err()); - assert!(super::relay_mesh_mode_from_nip11(&json!({"supported_nips": "not-an-array"})).is_err()); - for malformed in [ - json!({"supported_nips": [1, "43"]}), - json!({"supported_nips": [1, {"nip": 43}]}), - json!({"supported_nips": [1, -1]}), - json!({"supported_nips": [1, 43.5]}), - json!({"supported_nips": [1, 4_294_967_296_u64]}), - ] { - assert!( - super::relay_mesh_mode_from_nip11(&malformed).is_err(), - "partially malformed supported_nips must not prove a relay mode: {malformed}" - ); - } -} - -#[test] -fn open_relay_routes_without_a_membership_snapshot() { - use super::MeshRelayMode; - - // The bug this fixes: on a relay that does not enforce membership, no - // kind:13534 snapshot ever lists ordinary users, so requiring one made - // every serving peer permanently undiscoverable — each node sat at - // "1 NODE" with an empty peer list forever. - let endpoint = test_endpoint_token(); - let events = vec![signed_reporter_target( - &"a".repeat(64), - "model-open", - &endpoint, - )]; - - let closed = super::availability_from_events_for_mode( - events.clone(), - MeshRelayMode::ClosedMembershipEnforced, - ); - assert!( - closed.serve_targets.is_empty(), - "an enforcing relay must still wait for its roster" - ); - - let open = super::availability_from_events_for_mode(events, MeshRelayMode::OpenNoMembership); - assert_eq!(open.serve_targets.len(), 1); - assert_eq!(open.serve_targets[0].endpoint_addr, endpoint); -} - -#[test] -fn open_relay_ignores_a_stray_membership_snapshot() { - use super::MeshRelayMode; - - // On an open relay a kind:13534 snapshot can still exist (members added - // out of band), but it is not a trust boundary there: admission runs - // `TrustPolicy::Off` regardless. Intersecting routing with that stray - // roster would hide peers the mesh happily admits — a hybrid state where - // admission and discovery disagree. Routing must ignore it entirely. - let reporter_secret = "a".repeat(64); - let reporter_pubkey = nostr::Keys::parse(&reporter_secret) - .unwrap() - .public_key() - .to_hex(); - let endpoint = test_endpoint_token(); - let events = vec![ - // Snapshot deliberately lists someone else, NOT the reporter. - signed_membership_event(&["f".repeat(64)]), - signed_reporter_target(&reporter_secret, "model-open", &endpoint), - ]; - - let closed = super::availability_from_events_for_mode( - events.clone(), - MeshRelayMode::ClosedMembershipEnforced, - ); - assert!( - closed.serve_targets.is_empty(), - "an enforcing relay still intersects routing with the roster" - ); - - let open = super::availability_from_events_for_mode(events, MeshRelayMode::OpenNoMembership); - assert_eq!( - open.serve_targets.len(), - 1, - "an open relay accepts a non-roster reporter's status: {reporter_pubkey} was hidden" - ); - assert_eq!(open.serve_targets[0].endpoint_addr, endpoint); -} - -#[test] -fn closed_relay_without_snapshot_never_silently_opens_up() { - use super::MeshRelayMode; - - // Regression guard for the risk this change introduces. On an enforcing - // relay a missing snapshot is a transient read gap, NOT permission to - // admit everyone: if that ever resolved to the open path, one replication - // blip would drop admission enforcement on a closed relay. - let events = vec![ - signed_reporter_target(&"b".repeat(64), "model-a", &test_endpoint_token()), - signed_reporter_target(&"c".repeat(64), "model-b", &test_endpoint_token()), - ]; - - let availability = - super::availability_from_events_for_mode(events, MeshRelayMode::ClosedMembershipEnforced); - assert!(availability.serve_targets.is_empty()); - assert_eq!( - availability.reason.as_deref(), - Some("Buzz shared compute is waiting for the current member roster") - ); -} - -#[test] -fn open_relay_still_enforces_freshness_and_endpoint_binding() { - use super::MeshRelayMode; - - // Dropping the membership intersection must not drop the other - // protections: a stale note is still unroutable, and an unsigned endpoint - // substitution is still rejected. - let stale = signed_reporter_target_at( - &"d".repeat(64), - "stale-model", - &test_endpoint_token(), - 1_000, - ); - let stale_availability = - super::availability_from_events_for_mode(vec![stale], MeshRelayMode::OpenNoMembership); - assert!( - stale_availability.serve_targets.is_empty(), - "freshness still gates routing on an open relay" - ); - - let signed_endpoint = test_endpoint_token(); - let mut forged = serde_json::from_str::( - &signed_reporter_target(&"e".repeat(64), "model-x", &signed_endpoint).content, - ) - .expect("status content is json"); - // Swap in an endpoint the owner never signed for. Assert the substitution - // is real so this test can never pass vacuously by reusing the same token. - let substituted_endpoint = test_endpoint_token(); - assert_ne!(signed_endpoint, substituted_endpoint); - forged["serveTargets"][0]["endpointAddr"] = json!(substituted_endpoint); - let keys = nostr::Keys::parse(&"e".repeat(64)).unwrap(); - let substituted = super::coordinator::build_status_report_event(forged) - .expect("status builder") - .sign_with_keys(&keys) - .expect("signs"); - let forged_availability = super::availability_from_events_for_mode( - vec![substituted], - MeshRelayMode::OpenNoMembership, - ); - assert!( - forged_availability.serve_targets.is_empty(), - "endpoint binding is still verified on an open relay" - ); -} - #[test] fn serving_usage_extracts_local_and_remote_attempts() { // Captured shape from a live serving node's status payload. The local vs