From 40224850032f5300fb8e13c49fe5e90b8da3e80d Mon Sep 17 00:00:00 2001 From: Jimmy <1fe240cd1a8cf775f6f3060f115e5a303181f3abf28ad4cb0c2515f4a02b36a8@meshllm.communities.buzz.xyz> Date: Sat, 29 Aug 2026 08:41:20 +1000 Subject: [PATCH] fix(mesh): bound admission mode transitions Require two consecutive observations before whole-app restarts, fail closed after repeated unknown mode reads, and throttle self-only-to-open recovery restarts across process launches. Add feature-enabled mesh clippy to CI and resolve the warnings it exposes. Clarify transport trust and catalog contracts. Co-authored-by: Jimmy <1fe240cd1a8cf775f6f3060f115e5a303181f3abf28ad4cb0c2515f4a02b36a8@meshllm.communities.buzz.xyz> Signed-off-by: Jimmy <1fe240cd1a8cf775f6f3060f115e5a303181f3abf28ad4cb0c2515f4a02b36a8@meshllm.communities.buzz.xyz> --- .github/workflows/ci.yml | 4 + desktop/src-tauri/src/commands/mesh_llm.rs | 4 + desktop/src-tauri/src/commands/workspace.rs | 10 +- .../src-tauri/src/managed_agents/restore.rs | 32 ++-- desktop/src-tauri/src/mesh_llm/catalog.rs | 14 +- desktop/src-tauri/src/mesh_llm/coordinator.rs | 140 +++++++++------ .../src/mesh_llm/coordinator_admission.rs | 163 ++++++++++++++++++ desktop/src-tauri/src/mesh_llm/mod.rs | 1 + 8 files changed, 288 insertions(+), 80 deletions(-) create mode 100644 desktop/src-tauri/src/mesh_llm/coordinator_admission.rs diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 00474ce0ff..87c45a8b4a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -214,6 +214,10 @@ jobs: run: just desktop-tauri-test env: CMAKE_POLICY_VERSION_MINIMUM: "3.5" + - name: Desktop Mesh clippy + run: cargo clippy --manifest-path desktop/src-tauri/Cargo.toml --workspace --all-targets --features mesh-llm -- -D warnings + env: + CMAKE_POLICY_VERSION_MINIMUM: "3.5" - name: Desktop Mesh admission tests run: cargo test --manifest-path desktop/src-tauri/Cargo.toml --features mesh-llm --lib env: diff --git a/desktop/src-tauri/src/commands/mesh_llm.rs b/desktop/src-tauri/src/commands/mesh_llm.rs index 1b7620dcf9..e5ff5e39d5 100644 --- a/desktop/src-tauri/src/commands/mesh_llm.rs +++ b/desktop/src-tauri/src/commands/mesh_llm.rs @@ -179,6 +179,10 @@ fn advance_mesh_status_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 diff --git a/desktop/src-tauri/src/commands/workspace.rs b/desktop/src-tauri/src/commands/workspace.rs index 77d519b94b..128a54753d 100644 --- a/desktop/src-tauri/src/commands/workspace.rs +++ b/desktop/src-tauri/src/commands/workspace.rs @@ -345,7 +345,7 @@ pub async fn apply_workspace( } } }); - return Ok(()); + Ok(()) } #[cfg(not(feature = "mesh-llm"))] @@ -364,9 +364,11 @@ pub async fn apply_workspace( return Ok(()); } - assert_current_apply_generation(&state.workspace_apply_generation, apply_generation)?; - - Ok(()) + #[cfg(not(feature = "mesh-llm"))] + { + assert_current_apply_generation(&state.workspace_apply_generation, apply_generation)?; + Ok(()) + } } #[cfg(test)] diff --git a/desktop/src-tauri/src/managed_agents/restore.rs b/desktop/src-tauri/src/managed_agents/restore.rs index a225f492d3..1c64cdfb91 100644 --- a/desktop/src-tauri/src/managed_agents/restore.rs +++ b/desktop/src-tauri/src/managed_agents/restore.rs @@ -534,22 +534,6 @@ pub(crate) fn spawn_pending_profile_reconciliations(app: &tauri::AppHandle, work } } -#[cfg(test)] -mod profile_reconcile_tests { - use super::profile_reconcile_completed; - use crate::commands::ProfileReconcileOutcome; - - #[test] - fn skipped_reconciliation_never_retires_pending_work() { - assert!(profile_reconcile_completed( - ProfileReconcileOutcome::Reconciled - )); - assert!(!profile_reconcile_completed( - ProfileReconcileOutcome::SkippedDisabled - )); - } -} - #[cfg(feature = "mesh-llm")] fn persist_restore_error( app: &tauri::AppHandle, @@ -567,3 +551,19 @@ fn persist_restore_error( record.last_error = Some(error); save_managed_agents(app, &records) } + +#[cfg(test)] +mod profile_reconcile_tests { + use super::profile_reconcile_completed; + use crate::commands::ProfileReconcileOutcome; + + #[test] + fn skipped_reconciliation_never_retires_pending_work() { + assert!(profile_reconcile_completed( + ProfileReconcileOutcome::Reconciled + )); + assert!(!profile_reconcile_completed( + ProfileReconcileOutcome::SkippedDisabled + )); + } +} diff --git a/desktop/src-tauri/src/mesh_llm/catalog.rs b/desktop/src-tauri/src/mesh_llm/catalog.rs index 9990b16427..00cd8ea8fa 100644 --- a/desktop/src-tauri/src/mesh_llm/catalog.rs +++ b/desktop/src-tauri/src/mesh_llm/catalog.rs @@ -17,8 +17,8 @@ use mesh_llm_system::vram::{format_rated_capacity, rated_capacity_gb}; /// /// The recommended ladder follows rated unified memory: /// - below 32 GB: Gemma 4 E4B; -/// - 32 through 64 GB: Qwen3.5 9B; -/// - above 64 GB: Qwen3.8 27B. +/// - 32 through 79 GB: Qwen3.5 9B; +/// - 80 GB and above: Qwen3.8 27B. /// /// The Qwen entries are canonicalized from mesh-llm's compiled /// `MODEL_CATALOG` rather than synthesized. @@ -28,13 +28,15 @@ const CURATED_MEDIUM: &str = "unsloth/Qwen3.5-9B-GGUF:Q4_K_M"; const CURATED_MEDIUM_ALIAS: &str = "Qwen3.5-9B-Vision-Q4_K_M"; const CURATED_SMALL: &str = "unsloth/gemma-4-E4B-it-GGUF:Q4_K_M"; const CURATED_SMALL_ALIAS: &str = "Gemma-4-E4B-it-Q4_K_M"; -/// Superseded large pick, still mapped so machines that already selected it -/// keep resolving to a model id Mesh's OpenAI ingress accepts. +/// Superseded large alias retained only to preserve the historical string +/// canonicalization contract. Availability in a particular Mesh runtime is +/// determined by that runtime's compiled catalog. const LEGACY_LARGE: &str = "unsloth/gemma-4-26B-A4B-it-GGUF:UD-Q4_K_M"; const LEGACY_LARGE_ALIAS: &str = "gemma-4-26B-A4B-it-UD-Q4_K_M"; /// Rated-capacity boundary for the balanced Qwen3.5 9B tier. const CURATED_MEDIUM_MIN_RATED_GB: u64 = 32; -/// Qwen3.8 27B is reserved for rated capacity classes above 64 GB. +/// Qwen3.8 27B is reserved for 80 GB-and-larger rated capacity classes, +/// leaving headroom beyond the model weights for KV cache and runtime use. const CURATED_LARGE_MIN_RATED_GB: u64 = 80; /// The Buzz-curated recommendation for a machine's rated memory capacity. @@ -266,7 +268,7 @@ mod tests { assert_eq!(CURATED_MEDIUM, "unsloth/Qwen3.5-9B-GGUF:Q4_K_M"); assert_eq!(CURATED_LARGE, "unsloth/Qwen3.8-27B-GGUF:Q4_K_M"); - // Qwen3.8 is reserved for rated capacity classes above 64 GB. + // Qwen3.8 is reserved for 80 GB-and-larger rated capacity classes. let large = build_catalog(None, 80_000_000_000, 80.0, &[]); assert_eq!(large.recommended.as_deref(), Some(CURATED_LARGE)); let big = build_catalog(None, 128_000_000_000, 128.0, &[]); diff --git a/desktop/src-tauri/src/mesh_llm/coordinator.rs b/desktop/src-tauri/src/mesh_llm/coordinator.rs index 92693974cf..75ecc4c10b 100644 --- a/desktop/src-tauri/src/mesh_llm/coordinator.rs +++ b/desktop/src-tauri/src/mesh_llm/coordinator.rs @@ -11,6 +11,10 @@ use std::time::Duration; use nostr::Tag; use tauri::{AppHandle, Manager}; +use super::coordinator_admission::{ + clear_mode_recovery_restart, mark_mode_recovery_restart, mode_reconcile_action, + mode_recovery_restart_is_throttled, ModeEvidence, ModeReconcileAction, ModeReconcileState, +}; use crate::app_state::AppState; /// Client-owned parameterized-replaceable discovery note. We use the standard @@ -32,7 +36,6 @@ 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<()>, @@ -63,12 +66,15 @@ pub async fn start_coordinator(app: AppHandle) { }); let roster_app = app.clone(); let roster_watcher = tokio::spawn(async move { - // Carries a shrink awaiting confirmation across polls (hysteresis): - // a reduced roster must be seen twice in a row before we tear down. + // Carries pending observations across polls. Roster reductions and + // admission-mode changes must both be seen twice before a restart. 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).await { + if let Err(error) = + reconcile_roster(&roster_app, &mut pending_shrink, &mut mode_reconcile).await + { eprintln!("buzz-mesh: roster reconcile failed: {error}"); } } @@ -297,8 +303,8 @@ fn roster_reconcile_action( } } -fn runtime_needs_mode_probe(current_owners: Option<&[String]>) -> bool { - current_owners.is_none_or(<[String]>::is_empty) +fn runtime_needs_mode_probe(current_owners: &[String]) -> bool { + current_owners.is_empty() } #[derive(Debug, PartialEq, Eq)] @@ -309,13 +315,6 @@ enum OpenRuntimeObservation { Unknown(String), } -fn open_runtime_should_restart(observation: &OpenRuntimeObservation) -> bool { - matches!( - observation, - OpenRuntimeObservation::ClosedWithRoster | OpenRuntimeObservation::ClosedWithoutRoster(_) - ) -} - fn classify_open_runtime_observation( mode: Result, closed_roster: Option, String>>, @@ -346,6 +345,7 @@ async fn observe_open_runtime_relay(state: &AppState, relay_url: &str) -> OpenRu async fn reconcile_roster( app: &AppHandle, pending_shrink: &mut Option>, + mode_reconcile: &mut ModeReconcileState, ) -> Result<(), String> { let state = app.state::(); let current_request = { @@ -354,6 +354,7 @@ async fn reconcile_roster( Some(runtime) => runtime.start_request().clone(), None => { *pending_shrink = None; + *mode_reconcile = ModeReconcileState::default(); return Ok(()); } } @@ -368,27 +369,44 @@ async fn reconcile_roster( .map(str::to_owned) .unwrap_or_else(|| crate::relay::relay_ws_url_with_override(&state)); if current_owners.is_none() { - let observation = observe_open_runtime_relay(&state, &relay_url).await; - if open_runtime_should_restart(&observation) { - match &observation { - OpenRuntimeObservation::ClosedWithRoster => eprintln!( - "buzz-mesh: relay now enforces membership; restarting Buzz to apply its allowlist" - ), - OpenRuntimeObservation::ClosedWithoutRoster(error) => eprintln!( - "buzz-mesh: relay now enforces membership but its roster is unavailable; restarting Buzz fail-closed: {error}" - ), - OpenRuntimeObservation::Open | OpenRuntimeObservation::Unknown(_) => {} - } - app.request_restart(); - } else if let OpenRuntimeObservation::Unknown(error) = observation { - eprintln!( - "buzz-mesh: open-relay mode recheck failed; keeping the running policy: {error}" - ); - } - *pending_shrink = None; - return Ok(()); + clear_mode_recovery_restart(app); } 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 => { + 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 => {} + } + app.request_restart(); + } + } + *pending_shrink = None; return Ok(()); }; // A failed roster query must NOT be treated as "the roster became empty": @@ -396,7 +414,7 @@ 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(Some(current_owners)) { + 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. @@ -406,6 +424,38 @@ async fn reconcile_roster( .await .map(Some) }; + if current_owners.is_empty() { + match &query { + Ok(None) => { + match mode_reconcile_action(mode_reconcile, Ok(ModeEvidence::Open), false) { + ModeReconcileAction::Keep => return Ok(()), + ModeReconcileAction::AwaitConfirm => { + eprintln!( + "buzz-mesh: isolated startup fallback observed an open relay; awaiting confirmation before restarting Buzz" + ); + return Ok(()); + } + ModeReconcileAction::RestartProcess => { + if mode_recovery_restart_is_throttled(app)? { + eprintln!( + "buzz-mesh: open-mode recovery restart is throttled after a recent attempt; keeping the self-only fallback" + ); + return Ok(()); + } + mark_mode_recovery_restart(app)?; + } + } + } + Err(_) => { + let _ = mode_reconcile_action(mode_reconcile, Err(()), false); + } + Ok(Some(_)) => { + let _ = mode_reconcile_action(mode_reconcile, Ok(ModeEvidence::Closed), false); + } + } + } else { + *mode_reconcile = ModeReconcileState::default(); + } match roster_reconcile_action(current_owners, pending_shrink.as_deref(), query) { RosterReconcileAction::Keep => { *pending_shrink = None; @@ -766,12 +816,11 @@ mod tests { } #[test] - fn only_open_or_self_only_runtimes_reprobe_mode() { + 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(None)); - assert!(runtime_needs_mode_probe(Some(&self_only))); - assert!(!runtime_needs_mode_probe(Some(&closed))); + assert!(runtime_needs_mode_probe(&self_only)); + assert!(!runtime_needs_mode_probe(&closed)); } #[test] @@ -786,23 +835,6 @@ mod tests { "relay returned no membership snapshot".to_string() ) ); - assert!(open_runtime_should_restart(&observation)); - } - - #[test] - fn open_runtime_restarts_for_any_proven_closed_mode() { - assert!(!open_runtime_should_restart(&OpenRuntimeObservation::Open)); - assert!(open_runtime_should_restart( - &OpenRuntimeObservation::ClosedWithRoster - )); - assert!(open_runtime_should_restart( - &OpenRuntimeObservation::ClosedWithoutRoster( - "relay returned no membership snapshot".to_string() - ) - )); - assert!(!open_runtime_should_restart( - &OpenRuntimeObservation::Unknown("NIP-11 timed out".to_string()) - )); } #[test] diff --git a/desktop/src-tauri/src/mesh_llm/coordinator_admission.rs b/desktop/src-tauri/src/mesh_llm/coordinator_admission.rs new file mode 100644 index 0000000000..19806d770a --- /dev/null +++ b/desktop/src-tauri/src/mesh_llm/coordinator_admission.rs @@ -0,0 +1,163 @@ +use std::time::Duration; + +use tauri::{AppHandle, Manager}; + +/// Prevent a flaky startup mode probe from creating an unbounded whole-app +/// restart loop while recovering a self-only fallback into open mode. +const MODE_RECOVERY_RESTART_COOLDOWN: Duration = Duration::from_secs(10 * 60); +const MODE_RECOVERY_RESTART_MARKER: &str = "mesh-mode-recovery-restart"; + +fn mode_recovery_restart_marker(app: &AppHandle) -> Result { + Ok(app + .path() + .app_data_dir() + .map_err(|error| format!("failed to resolve app data directory: {error}"))? + .join(MODE_RECOVERY_RESTART_MARKER)) +} + +pub(super) fn mode_recovery_restart_is_throttled(app: &AppHandle) -> Result { + let marker = mode_recovery_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(modified + .elapsed() + .is_ok_and(|elapsed| elapsed < MODE_RECOVERY_RESTART_COOLDOWN)) +} + +pub(super) fn mark_mode_recovery_restart(app: &AppHandle) -> Result<(), String> { + let marker = mode_recovery_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, b"self-only-to-open\n") + .map_err(|error| format!("failed to write mesh restart marker: {error}")) +} + +pub(super) fn clear_mode_recovery_restart(app: &AppHandle) { + if let Ok(marker) = mode_recovery_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, +} + +#[derive(Debug, PartialEq, Eq)] +pub(super) enum ModeReconcileAction { + Keep, + AwaitConfirm, + RestartProcess, +} + +/// 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 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/mod.rs b/desktop/src-tauri/src/mesh_llm/mod.rs index d6a2b925db..1aa7ead858 100644 --- a/desktop/src-tauri/src/mesh_llm/mod.rs +++ b/desktop/src-tauri/src/mesh_llm/mod.rs @@ -1,6 +1,7 @@ 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};