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>
This commit is contained in:
Jimmy
2026-08-29 08:41:20 +10:00
parent f1a4eb734c
commit 4022485003
8 changed files with 288 additions and 80 deletions

View File

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

View File

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

View File

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

View File

@@ -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
));
}
}

View File

@@ -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, &[]);

View File

@@ -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<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).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<crate::mesh_llm::MeshRelayMode, String>,
closed_roster: Option<Result<Vec<String>, 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<Vec<String>>,
mode_reconcile: &mut ModeReconcileState,
) -> Result<(), String> {
let state = app.state::<AppState>();
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<String> = 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]

View File

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

View File

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