jerry/buzz
1
0
Fork 0
mirror of https://github.com/block/buzz.git synced 2026-10-30 05:51:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

revert(mesh): drop open-relay unenforced mode; mesh stays allowlist-only

Browse Source 
Remove the open-relay/open-mesh admission mode from PR #6189 at Mic's
direction: running the mesh unenforced when the Buzz relay does not
enforce NIP-43 membership is not a product mode we want, and it was the
source of all the fail-open complexity (NIP-11 mode probing, the
two-observation mode state machine, restart markers/throttles/latches).

Reverted to main's behavior: admission is always roster/allowlist
driven. On a relay with no membership snapshot the mesh simply runs
self-only — safe and boring, no mode probes, no whole-app restarts.
Removed: MeshRelayMode, relay_mesh_mode_from_nip11, the NIP-11 mode
probe, coordinator_admission.rs, mode-scoped discovery/availability, and
all their tests.

Kept from this branch (independent of the mode work):
- mesh-llm 0.76.0-rc7 upgrade and the restored hardware model ladder
- BUZZ_AGENT_LLM_TIMEOUT_SECS=660 mesh client timeout default
- desktop-standalone mesh=1 feature flag wiring
- cfg-only warning fixes in workspace.rs and restore.rs

desktop-tauri-clippy now lints BOTH cfg graphs (default and
--features mesh-llm): features are additive, so a single invocation
covers only one graph, and both ship (release-windows builds without
mesh-llm). CI keeps calling the recipe, and keeps the feature-enabled
--lib test step.

Verified on this commit: clippy -D warnings clean on both graphs;
feature-enabled lib suite 3110 passed / 0 failed / 19 ignored;
default-feature lib suite 3005 passed / 0 failed; fmt and diff checks
clean.

Signed-off-by: Michael Neale <michael.neale@gmail.com>
This commit is contained in:
Michael Neale
2026-08-29 10:59:31 +10:00
parent 502eed9028
commit d0ebf66117
10 changed files with 74 additions and 1011 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.github/workflows/ci.yml vendored
View File

@@ -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"

7
Justfile
View File

@@ -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

13
desktop/src-tauri/src/commands/agent_models.rs
View File

@@ -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()

173
desktop/src-tauri/src/commands/mesh_llm.rs
View File

@@ -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<mesh_llm::MeshRelayMode, String> {
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<mesh_llm::MeshRelayMode, String> {
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<Vec<nostr::Event>, 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<mesh_llm::MeshAvailability, String> {
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<Vec<nostr::Event>, 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<Option<Vec<String>>, 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<Vec<String>, 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<Option<Vec<String>>, 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<Vec<String>, 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<Vec<String>> {
pub(crate) async fn resolve_trusted_owner_ids_or_self_only(state: &AppState) -> Vec<String> {
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<Vec<mesh_llm::MeshServeTarget>, 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<Vec<String>>, Option<String>) {
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<String>, Option<String>) {
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,
))
}

78
desktop/src-tauri/src/commands/mesh_llm_tests.rs
View File

@@ -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![

297
desktop/src-tauri/src/mesh_llm/coordinator.rs
View File

@@ -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<Vec<String>> = 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<Option<Vec<String>>, String>,
query: Result<Vec<String>, 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<crate::mesh_llm::MeshRelayMode, String>,
closed_roster: Option<Result<Vec<String>, 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<Option<Vec<String>>, 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<Vec<String>>,
mode_reconcile: &mut ModeReconcileState,
) -> Result<(), String> {
if mode_reconcile.restart_requested() {
return Ok(());
}
let state = app.state::<AppState>();
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(&current, 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(&current, None, Ok(Some(vec!["owner-a".to_string()])));
let action = roster_reconcile_action(&current, 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(&current, None, Ok(Some(fresh)));
let action = roster_reconcile_action(&current, 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(&current, None, Ok(Some(reduced.clone())));
let action = roster_reconcile_action(&current, 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(&current, Some(&reduced), Ok(Some(reduced.clone())));
let action = roster_reconcile_action(&current, 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(
&current,
Some(&first_reduced),
Ok(Some(second_reduced.clone())),
);
let action =
roster_reconcile_action(&current, 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(&current, None, Ok(Some(Vec::new())));
let first = roster_reconcile_action(&current, None, Ok(Vec::new()));
assert_eq!(first, RosterReconcileAction::AwaitConfirm(Vec::new()));
let empty: Vec<String> = Vec::new();
let confirmed = roster_reconcile_action(&current, Some(&empty), Ok(Some(Vec::new())));
let confirmed = roster_reconcile_action(&current, 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(&current, None, Ok(Some(reduced.clone())));
let held = roster_reconcile_action(&current, None, Ok(reduced.clone()));
assert_eq!(held, RosterReconcileAction::AwaitConfirm(reduced.clone()));
let recovered =
roster_reconcile_action(&current, Some(&reduced), Ok(Some(current.clone())));
let recovered = roster_reconcile_action(&current, Some(&reduced), Ok(current.clone()));
assert_eq!(recovered, RosterReconcileAction::Keep);
}
#[test]
fn only_self_only_closed_runtimes_reprobe_mode() {
let self_only: Vec<String> = 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!(

228
desktop/src-tauri/src/mesh_llm/coordinator_admission.rs
View File

@@ -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<std::path::PathBuf, String> {
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<bool, String> {
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<ModeEvidence>,
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<bool, String> {
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<ModeEvidence, ()>,
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
);
}
}

111
desktop/src-tauri/src/mesh_llm/discovery.rs
View File

@@ -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<MeshRelayMode, String> {
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::<Result<Vec<_>, _>>()?;
if nips.contains(&43) {
Ok(MeshRelayMode::ClosedMembershipEnforced)
} else {
Ok(MeshRelayMode::OpenNoMembership)
}
}
fn dedupe_targets(targets: Vec<MeshServeTarget>) -> Vec<MeshServeTarget> {
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<nostr::Event>) -> 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<nostr::Event>,
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::<serde_json::Value>(&event.content) else {
continue;
};

6
desktop/src-tauri/src/mesh_llm/mod.rs
View File

@@ -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,

170
desktop/src-tauri/src/mesh_llm/mod_tests.rs
View File

@@ -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::<serde_json::Value>(
&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