diff --git a/.env.example b/.env.example index fb1a3c2aa0..9e35203524 100644 --- a/.env.example +++ b/.env.example @@ -159,6 +159,11 @@ RUST_LOG=buzz_relay=debug,buzz_db=debug,buzz_auth=debug,buzz_pubsub=debug,tower_ # Path to a file containing the heartbeat prompt. # BUZZ_ACP_HEARTBEAT_PROMPT_FILE= +# ── Desktop development ────────────────────────────────────────────────────── +# DEV-only: replay first-run onboarding and the Welcome Team kickoff on each +# app launch while keeping the current identity and relay data. +# VITE_BUZZ_FORCE_FRESH_ONBOARDING=true + # ── Subscription & filtering ───────────────────────────────────────────────── # Subscribe mode: "mentions" (default), "all", or "config" (rule-based). # BUZZ_ACP_SUBSCRIBE=mentions diff --git a/.intersect/sadscan.yaml b/.intersect/sadscan.yaml new file mode 100644 index 0000000000..a321714bcb --- /dev/null +++ b/.intersect/sadscan.yaml @@ -0,0 +1,4 @@ +# Cargo registry checksums are integrity hashes generated by Cargo, not payment data. +exclude_rules_for_files: + sq.pii.cc.visa: + - Cargo.lock diff --git a/CHANGELOG.md b/CHANGELOG.md index 1b924d7135..ae835ea9dd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,25 @@ # Changelog +## v0.4.13 + +- fix(desktop): compare relay media origins case-insensitively so uploads render when the saved community URL has uppercase characters +- fix(desktop): invalidate stale media lookups so an in-flight relay-origin fetch cannot repopulate caches after a community reset + +## v0.4.12 + +- fix(desktop): proxy authenticated relay media after cold-start community initialization +- fix(desktop): keep shared-compute consumers admitted (WIP: multi-workspace publishing) ([#2000](https://github.com/block/buzz/pull/2000)) ([`5d77fa57`](https://github.com/block/buzz/commit/5d77fa57)) +- feat(desktop): reskin provider, config, community, profile, and team onboarding pages ([#2003](https://github.com/block/buzz/pull/2003)) ([`f054df73`](https://github.com/block/buzz/commit/f054df73)) + + +## v0.4.11 + +- Fix policy receipts for cold-launch invite links ([#2005](https://github.com/block/buzz/pull/2005)) ([`1d8d006a`](https://github.com/block/buzz/commit/1d8d006aaf078ffd5b92eae06334b7bb908b680e)) +- Preserve persistent agent audience order ([#1989](https://github.com/block/buzz/pull/1989)) ([`fb8e6827`](https://github.com/block/buzz/commit/fb8e6827c758c73c9c6e16965dcdcecd00a5827c)) +- feat(desktop): create channels in sections ([#1996](https://github.com/block/buzz/pull/1996)) ([`9b649740`](https://github.com/block/buzz/commit/9b6497402f343e380f6c79b97a01c2d52bd12f17)) +- feat(desktop): graduate community rail ([#1995](https://github.com/block/buzz/pull/1995)) ([`f06b59ff`](https://github.com/block/buzz/commit/f06b59ff13bef0c031c8e4c3bdf979e120ea76d5)) + + ## v0.4.10 - feat(desktop): onboarding step dots, identity/key page restyle, overflow scroll ([#1993](https://github.com/block/buzz/pull/1993)) ([`f2522995c`](https://github.com/block/buzz/commit/f2522995cbea67c96b79d5de6c19c665e465027d)) diff --git a/crates/buzz-acp/src/lib.rs b/crates/buzz-acp/src/lib.rs index 2b6e8d45d1..3d86125298 100644 --- a/crates/buzz-acp/src/lib.rs +++ b/crates/buzz-acp/src/lib.rs @@ -1286,12 +1286,6 @@ async fn tokio_main() -> Result<()> { let presence_publisher = relay.event_publisher(); let presence_keys = config.keys.clone(); - if config.presence_enabled { - match publish_presence(&presence_publisher, &presence_keys, "online").await { - Ok(_) => tracing::info!("presence set to online"), - Err(e) => tracing::warn!("failed to set initial presence: {e}"), - } - } // Priority: BUZZ_AUTH_TAG (NIP-OA attestation) → --agent-owner flag. let startup_owner: Option = resolve_agent_owner(&config); @@ -1413,6 +1407,16 @@ async fn tokio_main() -> Result<()> { } } + // Online means the harness can receive work, not merely that its socket is + // connected. Publishing after channel subscriptions gives desktop callers + // a durable readiness boundary before they send a startup mention. + if config.presence_enabled { + match publish_presence(&presence_publisher, &presence_keys, "online").await { + Ok(_) => tracing::info!("presence set to online"), + Err(e) => tracing::warn!("failed to set initial presence: {e}"), + } + } + let dedup_mode = config.dedup_mode; let mut queue = EventQueue::new(dedup_mode).with_in_flight_deadline(config.max_turn_duration_secs); diff --git a/desktop/package.json b/desktop/package.json index 0d5dca3fbe..f07b282d4f 100644 --- a/desktop/package.json +++ b/desktop/package.json @@ -1,7 +1,7 @@ { "name": "buzz", "private": true, - "version": "0.4.10", + "version": "0.4.13", "type": "module", "scripts": { "dev": "vite", diff --git a/desktop/playwright.config.ts b/desktop/playwright.config.ts index fa6f9ae19f..b6296f0f3a 100644 --- a/desktop/playwright.config.ts +++ b/desktop/playwright.config.ts @@ -20,6 +20,7 @@ export default defineConfig({ name: "smoke", testMatch: [ "**/smoke.spec.ts", + "**/onboarding-docked-cta-screenshots.spec.ts", "**/navigation.spec.ts", "**/channels.spec.ts", "**/channel-shared-header-backdrop.spec.ts", diff --git a/desktop/scripts/check-file-sizes.mjs b/desktop/scripts/check-file-sizes.mjs index 15674a14ec..bf7177e6c2 100644 --- a/desktop/scripts/check-file-sizes.mjs +++ b/desktop/scripts/check-file-sizes.mjs @@ -202,11 +202,6 @@ const overrides = new Map([ // mapper. This is the existing API boundary; split remains queued. // team-instructions-first-class: createManagedAgent Tauri bridge threads the // new teamId input through to the backend (+1 line). - // #1970 merge-skew: add-community deep-link wiring (+4) was based before - // #1923 (+3) landed, so PR CI was green while the merged total crossed the - // cap (996 -> 1003). Pre-existing red on main, not one PR's feature debt. - // Queued to split with the rest of this list. - ["src/app/AppShell.tsx", 1003], ["src/shared/api/tauri.ts", 1305], // doctor-npm-eacces-preflight: hint field added to InstallStepResult (+1 line). // codex-acp-package-swap: "adapter_outdated" variant added to AcpAvailabilityStatus (+1 line). @@ -376,7 +371,10 @@ const overrides = new Map([ // here beside `build_app_state` and its sibling client; its doc comment // carries the load-bearing SSRF rationale. Extraction would only relocate, // not reduce, the security-critical code. - ["src-tauri/src/app_state.rs", 1068], + // +5 (1068 -> 1073): merge with main, which independently added the + // managed_agent_profile_reconcile_enabled flag (field + doc + init) under + // its own 1042-line override. Union of two separately approved additions. + ["src-tauri/src/app_state.rs", 1073], // multi-slot splitting + no-op suppression (#1309): the ReadStateManager // class grew from ~700 lines to ~1019 with the addition of // splitContextsIntoBudgetedSlots (pure fn + 5 tests), publishSplitSlots, diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index c59ba28343..2db15a9b54 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -972,7 +972,7 @@ dependencies = [ [[package]] name = "buzz-desktop" -version = "0.4.10" +version = "0.4.13" dependencies = [ "anyhow", "arboard", diff --git a/desktop/src-tauri/Cargo.toml b/desktop/src-tauri/Cargo.toml index cf8bb280ec..68005f41d3 100644 --- a/desktop/src-tauri/Cargo.toml +++ b/desktop/src-tauri/Cargo.toml @@ -2,7 +2,7 @@ [package] name = "buzz-desktop" -version = "0.4.10" +version = "0.4.13" description = "Buzz desktop app" authors = ["you"] edition = "2021" diff --git a/desktop/src-tauri/src/app_state.rs b/desktop/src-tauri/src/app_state.rs index 6d509318ec..3d910c3133 100644 --- a/desktop/src-tauri/src/app_state.rs +++ b/desktop/src-tauri/src/app_state.rs @@ -34,6 +34,10 @@ pub struct AppState { /// restore. `apply_workspace` consumes it after installing the workspace /// relay and identity, so agents never start against the fallback relay. pub managed_agent_restore_pending: AtomicBool, + /// Whether desktop may repair managed-agent kind:0 profiles from its local + /// records. Disabled by the agent-managed profiles experiment so an agent's + /// own profile updates are not overwritten on start or restore. + pub managed_agent_profile_reconcile_enabled: AtomicBool, /// Shared shutdown signal checked by launch-time agent restoration. pub shutdown_started: AtomicBool, /// Serializes the restore spawn/register transition with shutdown cleanup, @@ -196,6 +200,7 @@ pub fn build_app_state() -> AppState { ), relay_url_override: Mutex::new(None), managed_agent_restore_pending: AtomicBool::new(false), + managed_agent_profile_reconcile_enabled: AtomicBool::new(true), shutdown_started: AtomicBool::new(false), managed_agent_restore_transition: Mutex::new(()), identity_mutation: Mutex::new(()), diff --git a/desktop/src-tauri/src/commands/agent_settings.rs b/desktop/src-tauri/src/commands/agent_settings.rs index 54b52bd197..b654158401 100644 --- a/desktop/src-tauri/src/commands/agent_settings.rs +++ b/desktop/src-tauri/src/commands/agent_settings.rs @@ -1,4 +1,5 @@ -use tauri::{AppHandle, Manager}; +use std::sync::atomic::Ordering; +use tauri::{AppHandle, Manager, State}; use crate::{ app_state::AppState, @@ -10,6 +11,13 @@ use crate::{ util::now_iso, }; +#[tauri::command] +pub fn set_agent_managed_profiles(enabled: bool, state: State<'_, AppState>) { + state + .managed_agent_profile_reconcile_enabled + .store(!enabled, Ordering::Release); +} + #[tauri::command] pub async fn set_managed_agent_start_on_app_launch( pubkey: String, diff --git a/desktop/src-tauri/src/commands/agents.rs b/desktop/src-tauri/src/commands/agents.rs index 1718d54de2..6ebe6b7c4a 100644 --- a/desktop/src-tauri/src/commands/agents.rs +++ b/desktop/src-tauri/src/commands/agents.rs @@ -991,7 +991,11 @@ pub async fn start_managed_agent( // profile is published on the relay. This self-heals cases where the initial // profile sync at creation time failed silently. For legacy records (pre-PR-921) // with no persisted avatar, this also backfills the avatar from the relay. - if result.is_ok() { + if result.is_ok() + && state + .managed_agent_profile_reconcile_enabled + .load(std::sync::atomic::Ordering::Acquire) + { let reconcile_pubkey = pubkey.clone(); let reconcile_app = app.clone(); tauri::async_runtime::spawn(async move { diff --git a/desktop/src-tauri/src/commands/agents_profile.rs b/desktop/src-tauri/src/commands/agents_profile.rs index 236fd9f4cc..0675d4c48f 100644 --- a/desktop/src-tauri/src/commands/agents_profile.rs +++ b/desktop/src-tauri/src/commands/agents_profile.rs @@ -81,6 +81,13 @@ pub(crate) async fn reconcile_agent_profile( &relay_ws_url_with_override(state), ); + if !state + .managed_agent_profile_reconcile_enabled + .load(std::sync::atomic::Ordering::Acquire) + { + return Ok(()); + } + // Query the relay for the agent's existing kind:0 profile. let existing = query_agent_profile(state, &relay_url, agent_pubkey).await?; @@ -136,6 +143,13 @@ pub(crate) async fn reconcile_agent_profile( let agent_keys = Keys::parse(&data.private_key_nsec) .map_err(|e| format!("failed to parse agent keys: {e}"))?; + if !state + .managed_agent_profile_reconcile_enabled + .load(std::sync::atomic::Ordering::Acquire) + { + return Ok(()); + } + sync_managed_agent_profile( state, &relay_url, diff --git a/desktop/src-tauri/src/commands/canvas.rs b/desktop/src-tauri/src/commands/canvas.rs index c40e5d4e9a..191fafe60d 100644 --- a/desktop/src-tauri/src/commands/canvas.rs +++ b/desktop/src-tauri/src/commands/canvas.rs @@ -23,14 +23,22 @@ pub async fn get_canvas( .await?; let Some(event) = events.first() else { - return Ok(serde_json::json!({ "content": "" })); + // Explicit nulls: the TS caller distinguishes "no canvas yet" from + // "canvas exists" via `updated_at`/`author`, so these keys must be + // present (absent keys deserialize as `undefined`, not `null`). + return Ok(serde_json::json!({ + "content": "", + "event_id": null, + "updated_at": null, + "author": null, + })); }; Ok(serde_json::json!({ "content": event.content, "event_id": event.id.to_hex(), - "created_at": event.created_at.as_secs(), - "pubkey": event.pubkey.to_hex(), + "updated_at": event.created_at.as_secs(), + "author": event.pubkey.to_hex(), })) } diff --git a/desktop/src-tauri/src/commands/mesh_llm.rs b/desktop/src-tauri/src/commands/mesh_llm.rs index 36cb458db1..ad55d95060 100644 --- a/desktop/src-tauri/src/commands/mesh_llm.rs +++ b/desktop/src-tauri/src/commands/mesh_llm.rs @@ -47,17 +47,77 @@ const RELAY_MESH_RUNTIME_NO_TARGET: &str = pub type CmdResult = Result; +fn advance_mesh_status_cursor( + filter: &mut serde_json::Value, + page: &[nostr::Event], +) -> Result<(u64, String), String> { + let last = page + .last() + .ok_or_else(|| "cannot advance an empty mesh status page".to_string())?; + let cursor = (last.created_at.as_secs(), last.id.to_hex()); + filter["until"] = serde_json::json!(cursor.0); + filter["before_id"] = serde_json::json!(cursor.1); + Ok(cursor) +} + +async fn query_mesh_discovery_events(state: &AppState) -> Result, String> { + let mut events = relay::query_relay(state, &[mesh_llm::relay_membership_filter()]).await?; + let member_pubkeys = mesh_llm::current_member_pubkeys(&events); + if member_pubkeys.is_empty() { + // Distinguish "relay returned a membership snapshot listing zero + // members" (authoritative empty — allowed to shrink the roster to + // self-only) from "no membership snapshot came back at all" (a + // transient gap / replication lag). The relay publishes an explicit + // kind:13534 event even for a zero-member community, so its absence + // 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. + 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(); + status_filter["authors"] = serde_json::json!(member_pubkeys); + let mut previous_cursor: Option<(u64, String)> = None; + + loop { + let page = relay::query_relay(state, &[status_filter.clone()]).await?; + let done = page.len() < mesh_llm::MESH_STATUS_PAGE_SIZE; + if !done { + let cursor = advance_mesh_status_cursor(&mut status_filter, &page)?; + if previous_cursor.as_ref() == Some(&cursor) { + return Err("mesh status pagination did not advance".to_string()); + } + previous_cursor = Some(cursor); + } + events.extend(page); + if done { + return Ok(events); + } + } +} + /// Resolve the admission roster by intersecting member-signed mesh status -/// reporters with the current NIP-43 direct-member list. Missing membership or -/// a failed query returns an empty roster, which the runtime normalizes to -/// self-only admission. -pub(crate) async fn resolve_trusted_owner_ids(state: &AppState) -> Vec { - let filters = [ - mesh_llm::mesh_status_filter(), - mesh_llm::relay_membership_filter(), - ]; - match relay::query_relay(state, &filters).await { - Ok(events) => mesh_llm::owner_ids_from_events(&events), +/// reporters with the current NIP-43 direct-member list. +/// +/// Returns `Err` when the relay query fails. Callers MUST distinguish this from +/// an `Ok(empty)` roster (a genuinely empty community): a failed query must +/// 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. +pub(crate) async fn resolve_trusted_owner_ids(state: &AppState) -> Result, String> { + let events = query_mesh_discovery_events(state).await?; + Ok(mesh_llm::owner_ids_from_events(&events)) +} + +/// Resolve the roster for an initial node *start*, failing closed to self-only +/// (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) -> Vec { + match resolve_trusted_owner_ids(state).await { + Ok(owners) => owners, Err(error) => { eprintln!("buzz-mesh: roster query failed; allowing only this node: {error}"); Vec::new() @@ -81,7 +141,7 @@ pub(crate) async fn restore_mesh_sharing(app: &AppHandle, state: &AppState) -> C model_id: Some(config.model_id), max_vram_gb: config.max_vram_gb, join_token: None, - trusted_owner_ids: Some(resolve_trusted_owner_ids(state).await), + trusted_owner_ids: Some(resolve_trusted_owner_ids_or_self_only(state).await), }; let started = mesh_llm::DesktopMeshRuntime::start(request) .await @@ -101,7 +161,7 @@ pub async fn mesh_start_node( // Frontend requests never carry a roster; resolve it here so every // UI-started node enforces the member allowlist. if request.trusted_owner_ids.is_none() { - request.trusted_owner_ids = Some(resolve_trusted_owner_ids(&state).await); + request.trusted_owner_ids = Some(resolve_trusted_owner_ids_or_self_only(&state).await); } let mut runtime = state.mesh_llm_runtime.lock().await; if runtime.is_some() { @@ -232,7 +292,7 @@ pub(crate) async fn ensure_client_node_for_model( model_id: None, max_vram_gb: None, join_token: Some(join_token), - trusted_owner_ids: Some(resolve_trusted_owner_ids(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 runtime.is_some() { @@ -272,14 +332,7 @@ pub(crate) async fn resolve_mesh_bootstrap_target( if model_id.is_empty() { return Ok(None); } - let events = relay::query_relay( - state, - &[ - mesh_llm::mesh_status_filter(), - mesh_llm::relay_membership_filter(), - ], - ) - .await?; + let events = query_mesh_discovery_events(state).await?; Ok(pick_serve_target_for_model( mesh_llm::availability_from_events(events).serve_targets, model_id, @@ -426,6 +479,26 @@ mod tests { } } + #[test] + fn mesh_status_cursor_uses_relay_composite_tiebreak() { + let event = nostr::EventBuilder::new(nostr::Kind::TextNote, "status") + .custom_created_at(nostr::Timestamp::from(1_234)) + .sign_with_keys(&nostr::Keys::generate()) + .expect("sign test status"); + let mut filter = mesh_llm::mesh_status_filter(); + + let cursor = advance_mesh_status_cursor(&mut filter, std::slice::from_ref(&event)) + .expect("advance status cursor"); + + assert_eq!(cursor, (1_234, event.id.to_hex())); + assert_eq!(filter["until"], serde_json::json!(1_234)); + assert_eq!(filter["before_id"], serde_json::json!(event.id.to_hex())); + assert_eq!( + filter["limit"], + serde_json::json!(mesh_llm::MESH_STATUS_PAGE_SIZE) + ); + } + #[test] fn pick_serve_target_returns_first_match_for_model() { let targets = vec![ diff --git a/desktop/src-tauri/src/commands/messages.rs b/desktop/src-tauri/src/commands/messages.rs index 283ea7ace6..1928d69d23 100644 --- a/desktop/src-tauri/src/commands/messages.rs +++ b/desktop/src-tauri/src/commands/messages.rs @@ -622,14 +622,42 @@ async fn find_managed_agent_channel_message_by_marker( fn marker_author_for_scope<'a>( marker_scope: Option<&str>, - agent_pubkey: &'a str, -) -> Option<&'a str> { + agent_pubkey: Option<&'a str>, +) -> Result, String> { match marker_scope { - Some("channel") => None, - _ => Some(agent_pubkey), + Some("channel") => Ok(None), + Some("agent") | None => agent_pubkey + .map(Some) + .ok_or_else(|| "agent pubkey is required for agent-scoped markers".to_string()), + Some(scope) => Err(format!("unsupported marker scope: {scope}")), } } +#[tauri::command] +pub async fn has_managed_agent_channel_message_marker( + channel_id: String, + marker: String, + agent_pubkey: Option, + marker_scope: Option, + state: State<'_, AppState>, +) -> Result { + uuid::Uuid::parse_str(&channel_id) + .map_err(|_| format!("invalid channel UUID: {channel_id}"))?; + let marker = marker.trim(); + if marker.is_empty() { + return Err("message marker is required".into()); + } + let agent_pubkey = agent_pubkey + .as_deref() + .map(str::trim) + .filter(|value| !value.is_empty()); + + let marker_author = marker_author_for_scope(marker_scope.as_deref(), agent_pubkey)?; + find_managed_agent_channel_message_by_marker(&state, marker_author, &channel_id, marker) + .await + .map(|event| event.is_some()) +} + fn stored_managed_agent_auth_tag(auth_tag: Option<&str>) -> Option { auth_tag .map(str::trim) @@ -663,6 +691,27 @@ fn managed_agent_submission_auth_tag( legacy_managed_agent_auth_tag(&owner_keys, agent_pubkey) } +fn build_managed_agent_channel_message( + channel_id: uuid::Uuid, + content: &str, + thread_ref: Option<&events::ThreadRef>, + mention_pubkeys: &[String], + client_tags: &[Vec], +) -> Result { + let mention_refs: Vec<&str> = mention_pubkeys.iter().map(String::as_str).collect(); + events::build_message_with_client_tags( + channel_id, + content, + thread_ref, + &mention_refs, + &[], + &[], + &[], + client_tags, + ) +} + +#[allow(clippy::too_many_arguments)] #[tauri::command] pub async fn send_managed_agent_channel_message( agent_pubkey: String, @@ -670,6 +719,9 @@ pub async fn send_managed_agent_channel_message( content: String, marker: Option, marker_scope: Option, + mention_pubkeys: Option>, + parent_event_id: Option, + additional_markers: Option>, app: AppHandle, state: State<'_, AppState>, ) -> Result { @@ -706,11 +758,15 @@ pub async fn send_managed_agent_channel_message( } let submission_auth_tag = managed_agent_submission_auth_tag(&record, &state, &keys.public_key())?; + let thread_ref = match parent_event_id.as_deref() { + Some(parent_id) => Some(resolve_thread_ref(parent_id, &state).await?), + None => None, + }; if let Some(marker) = marker.as_deref() { if let Some(existing) = find_managed_agent_channel_message_by_marker( &state, - marker_author_for_scope(marker_scope.as_deref(), &record.pubkey), + marker_author_for_scope(marker_scope.as_deref(), Some(&record.pubkey))?, &channel_id, marker, ) @@ -718,26 +774,34 @@ pub async fn send_managed_agent_channel_message( { return Ok(SendChannelMessageResponse { event_id: existing.id.to_hex(), - parent_event_id: None, - root_event_id: None, - depth: 0, + parent_event_id: thread_ref + .as_ref() + .map(|reference| reference.parent_event_id.to_hex()), + root_event_id: thread_ref + .as_ref() + .map(|reference| reference.root_event_id.to_hex()), + depth: if thread_ref.is_some() { 1 } else { 0 }, created_at: existing.created_at.as_secs() as i64, }); } } - let client_tags = marker + let mut client_tags = marker .as_deref() .map(|marker| vec![vec!["client".to_string(), marker.to_string()]]) .unwrap_or_default(); - let builder = events::build_message_with_client_tags( + for marker in additional_markers.unwrap_or_default() { + let marker = marker.trim(); + if !marker.is_empty() { + client_tags.push(vec!["client".to_string(), marker.to_string()]); + } + } + let mentions = mention_pubkeys.unwrap_or_default(); + let builder = build_managed_agent_channel_message( channel_uuid, trimmed, - None, - &[], - &[], - &[], - &[], + thread_ref.as_ref(), + &mentions, &client_tags, )?; let result = @@ -745,157 +809,13 @@ pub async fn send_managed_agent_channel_message( Ok(SendChannelMessageResponse { event_id: result.event_id, - parent_event_id: None, - root_event_id: None, - depth: 0, + parent_event_id: parent_event_id.clone(), + root_event_id: thread_ref.map(|reference| reference.root_event_id.to_hex()), + depth: if parent_event_id.is_some() { 1 } else { 0 }, created_at: chrono::Utc::now().timestamp(), }) } -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn marker_author_scope_defaults_to_agent() { - assert_eq!( - marker_author_for_scope(None, "agent-pubkey"), - Some("agent-pubkey") - ); - assert_eq!( - marker_author_for_scope(Some("agent"), "agent-pubkey"), - Some("agent-pubkey") - ); - assert_eq!( - marker_author_for_scope(Some("unknown"), "agent-pubkey"), - Some("agent-pubkey") - ); - } - - #[test] - fn marker_author_scope_can_dedupe_across_channel() { - assert_eq!( - marker_author_for_scope(Some("channel"), "agent-pubkey"), - None - ); - } - - #[test] - fn search_messages_filter_requests_prefix_mode_for_topbar_typeahead() { - let filter = build_search_messages_filter(" pro ", 12, Some("channel-1")); - - assert_eq!(filter["search"], serde_json::json!("pro")); - assert_eq!(filter["search_mode"], serde_json::json!("prefix")); - assert_eq!(filter["limit"], serde_json::json!(12)); - assert_eq!(filter["#h"], serde_json::json!(["channel-1"])); - } - - #[test] - fn channel_messages_before_filter_sends_before_id_the_relay_reads() { - // The relay bridge's `extract_before_id` reads the composite tiebreak - // from `before_id`. If this filter sent the id under any other key (an - // earlier cut used `n`), the relay would silently drop the tiebreak and - // the dense-second keyset would degrade to a bare inclusive `until` — - // re-returning the same page forever. Pin the field name here so the - // client/relay contract can't drift without a red test (the Playwright - // mock reimplements the keyset in JS and cannot catch this). - let filter = - build_channel_messages_before_filter("channel-1", 1_700_000_000, Some("ab"), 200); - - assert_eq!(filter["until"], serde_json::json!(1_700_000_000)); - assert_eq!(filter["before_id"], serde_json::json!("ab")); - assert_eq!(filter["limit"], serde_json::json!(200)); - assert_eq!(filter["#h"], serde_json::json!(["channel-1"])); - assert!( - !filter.contains_key("n"), - "tiebreak must be `before_id`, not the `n` alias the relay ignores" - ); - } - - #[test] - fn thread_replies_filter_carries_non_p_gated_kinds_to_clear_the_gate() { - // The relay bridge p-gates EVERY filter before routing - // (`p_gated_filters_authorized`): a kindless filter "could match" a - // p-gated kind, so it demands a `#p` tag we don't send -> HTTP 403, - // before the thread-subtree query runs. The headline Lane-1 fix - // (`useThreadReplies` closing the descendant gap) then fails on every - // call against a real relay. So the thread filter MUST carry `kinds`, - // and every kind MUST be non-p-gated (else the gate still fires). The - // Playwright mock does not model p-gating, so this unit test is the - // only guard against the client/relay auth contract drifting. - let filter = build_thread_replies_filter("root-hex", Some("channel-1"), 64, 200, None); - - let kinds = filter - .get("kinds") - .and_then(|v| v.as_array()) - .expect("thread filter must carry `kinds` so the p-gate passes"); - assert!(!kinds.is_empty(), "kinds must be non-empty"); - for kind in kinds { - let k = kind.as_u64().expect("kind is a number") as u32; - assert!( - !buzz_core_pkg::kind::P_GATED_KINDS.contains(&k), - "kind {k} is p-gated; a p-gated kind in the filter re-triggers the \ - 403 that this fix exists to prevent" - ); - } - assert_eq!(filter["#e"], serde_json::json!(["root-hex"])); - assert_eq!(filter["depth_limit"], serde_json::json!(64)); - assert_eq!(filter["#h"], serde_json::json!(["channel-1"])); - } - - #[test] - fn thread_replies_filter_pages_with_composite_cursor() { - // When a cursor is supplied, both the timestamp and the event-id - // tiebreak must be emitted (`thread_cursor` + `thread_cursor_id`), else - // paging degrades to timestamp-only and drops same-second replies. - let cursor = crate::models::ThreadCursor { - created_at: 1_700_000_000, - event_id: "abcd".to_string(), - }; - let filter = build_thread_replies_filter("root-hex", None, 64, 200, Some(&cursor)); - assert_eq!(filter["thread_cursor"], serde_json::json!(1_700_000_000)); - assert_eq!(filter["thread_cursor_id"], serde_json::json!("abcd")); - assert!( - !filter.contains_key("#h"), - "no channel_id -> no #h scope in the filter" - ); - } - - #[test] - fn stored_managed_agent_auth_tag_trims_blank_values() { - assert_eq!( - stored_managed_agent_auth_tag(Some(" [\"auth\",\"owner\",\"\",\"sig\"] ")), - Some("[\"auth\",\"owner\",\"\",\"sig\"]".to_string()) - ); - assert_eq!(stored_managed_agent_auth_tag(Some(" ")), None); - assert_eq!(stored_managed_agent_auth_tag(None), None); - } - - #[test] - fn legacy_managed_agent_auth_tag_verifies_for_agent_pubkey() { - let owner_keys = Keys::generate(); - let agent_keys = Keys::generate(); - - let tag = legacy_managed_agent_auth_tag(&owner_keys, &agent_keys.public_key()) - .expect("legacy auth tag should compute") - .expect("legacy auth tag should be present"); - - let owner = buzz_sdk_pkg::nip_oa::verify_auth_tag(&tag, &agent_keys.public_key()) - .expect("legacy auth tag should verify"); - assert_eq!(owner, owner_keys.public_key()); - } - - #[test] - fn legacy_managed_agent_auth_tag_skips_self_attestation() { - let owner_keys = Keys::generate(); - - let tag = legacy_managed_agent_auth_tag(&owner_keys, &owner_keys.public_key()) - .expect("self-attestation should be skipped"); - - assert_eq!(tag, None); - } -} - #[tauri::command] pub async fn add_reaction( event_id: String, @@ -1033,3 +953,6 @@ fn feed_item_from_event(ev: &nostr::Event, category: &str) -> FeedItemInfo { category: category.to_string(), } } +#[cfg(test)] +#[path = "messages_tests.rs"] +mod tests; diff --git a/desktop/src-tauri/src/commands/messages_tests.rs b/desktop/src-tauri/src/commands/messages_tests.rs new file mode 100644 index 0000000000..3c08f0f113 --- /dev/null +++ b/desktop/src-tauri/src/commands/messages_tests.rs @@ -0,0 +1,193 @@ +use super::*; + +#[test] +fn marker_author_scope_validates_scope_and_required_pubkey() { + assert_eq!( + marker_author_for_scope(None, Some("agent")), + Ok(Some("agent")) + ); + assert_eq!( + marker_author_for_scope(Some("agent"), Some("agent")), + Ok(Some("agent")) + ); + assert_eq!(marker_author_for_scope(Some("channel"), None), Ok(None)); + assert_eq!( + marker_author_for_scope(Some("agent"), None), + Err("agent pubkey is required for agent-scoped markers".to_string()) + ); + assert_eq!( + marker_author_for_scope(None, None), + Err("agent pubkey is required for agent-scoped markers".to_string()) + ); + assert_eq!( + marker_author_for_scope(Some("unexpected"), Some("agent")), + Err("unsupported marker scope: unexpected".to_string()) + ); +} + +#[test] +fn managed_agent_message_builder_adds_mentions_and_client_marker() { + let pubkey = Keys::generate().public_key().to_hex(); + let event = build_managed_agent_channel_message( + uuid::Uuid::new_v4(), + "Welcome!", + None, + std::slice::from_ref(&pubkey), + &[vec!["client".to_string(), "welcome-v1".to_string()]], + ) + .expect("message should build") + .sign_with_keys(&Keys::generate()) + .expect("message should sign"); + + assert!(event.tags.iter().any(|tag| { + let parts = tag.as_slice(); + parts.len() >= 2 && parts[0] == "p" && parts[1] == pubkey + })); + assert!(event_has_client_marker(&event, "welcome-v1")); +} + +#[test] +fn managed_agent_message_builder_can_carry_multiple_client_markers() { + let event = build_managed_agent_channel_message( + uuid::Uuid::new_v4(), + "Welcome!", + None, + &[], + &[ + vec!["client".to_string(), "opener-v1".to_string()], + vec!["client".to_string(), "closer-v1".to_string()], + ], + ) + .expect("message should build") + .sign_with_keys(&Keys::generate()) + .expect("message should sign"); + + assert!(event_has_client_marker(&event, "opener-v1")); + assert!(event_has_client_marker(&event, "closer-v1")); +} + +#[test] +fn managed_agent_message_builder_rejects_invalid_mentions() { + let error = build_managed_agent_channel_message( + uuid::Uuid::new_v4(), + "Welcome!", + None, + &["not-a-pubkey".to_string()], + &[], + ) + .expect_err("invalid mentions should fail"); + assert!(error.contains("pubkey must be a 64-character hex string")); +} +#[test] +fn search_messages_filter_requests_prefix_mode_for_topbar_typeahead() { + let filter = build_search_messages_filter(" pro ", 12, Some("channel-1")); + + assert_eq!(filter["search"], serde_json::json!("pro")); + assert_eq!(filter["search_mode"], serde_json::json!("prefix")); + assert_eq!(filter["limit"], serde_json::json!(12)); + assert_eq!(filter["#h"], serde_json::json!(["channel-1"])); +} + +#[test] +fn channel_messages_before_filter_sends_before_id_the_relay_reads() { + // The relay bridge's `extract_before_id` reads the composite tiebreak + // from `before_id`. If this filter sent the id under any other key (an + // earlier cut used `n`), the relay would silently drop the tiebreak and + // the dense-second keyset would degrade to a bare inclusive `until` — + // re-returning the same page forever. Pin the field name here so the + // client/relay contract can't drift without a red test (the Playwright + // mock reimplements the keyset in JS and cannot catch this). + let filter = build_channel_messages_before_filter("channel-1", 1_700_000_000, Some("ab"), 200); + + assert_eq!(filter["until"], serde_json::json!(1_700_000_000)); + assert_eq!(filter["before_id"], serde_json::json!("ab")); + assert_eq!(filter["limit"], serde_json::json!(200)); + assert_eq!(filter["#h"], serde_json::json!(["channel-1"])); + assert!( + !filter.contains_key("n"), + "tiebreak must be `before_id`, not the `n` alias the relay ignores" + ); +} + +#[test] +fn thread_replies_filter_carries_non_p_gated_kinds_to_clear_the_gate() { + // The relay bridge p-gates EVERY filter before routing + // (`p_gated_filters_authorized`): a kindless filter "could match" a + // p-gated kind, so it demands a `#p` tag we don't send -> HTTP 403, + // before the thread-subtree query runs. The headline Lane-1 fix + // (`useThreadReplies` closing the descendant gap) then fails on every + // call against a real relay. So the thread filter MUST carry `kinds`, + // and every kind MUST be non-p-gated (else the gate still fires). The + // Playwright mock does not model p-gating, so this unit test is the + // only guard against the client/relay auth contract drifting. + let filter = build_thread_replies_filter("root-hex", Some("channel-1"), 64, 200, None); + + let kinds = filter + .get("kinds") + .and_then(|v| v.as_array()) + .expect("thread filter must carry `kinds` so the p-gate passes"); + assert!(!kinds.is_empty(), "kinds must be non-empty"); + for kind in kinds { + let k = kind.as_u64().expect("kind is a number") as u32; + assert!( + !buzz_core_pkg::kind::P_GATED_KINDS.contains(&k), + "kind {k} is p-gated; a p-gated kind in the filter re-triggers the \ + 403 that this fix exists to prevent" + ); + } + assert_eq!(filter["#e"], serde_json::json!(["root-hex"])); + assert_eq!(filter["depth_limit"], serde_json::json!(64)); + assert_eq!(filter["#h"], serde_json::json!(["channel-1"])); +} + +#[test] +fn thread_replies_filter_pages_with_composite_cursor() { + // When a cursor is supplied, both the timestamp and the event-id + // tiebreak must be emitted (`thread_cursor` + `thread_cursor_id`), else + // paging degrades to timestamp-only and drops same-second replies. + let cursor = crate::models::ThreadCursor { + created_at: 1_700_000_000, + event_id: "abcd".to_string(), + }; + let filter = build_thread_replies_filter("root-hex", None, 64, 200, Some(&cursor)); + assert_eq!(filter["thread_cursor"], serde_json::json!(1_700_000_000)); + assert_eq!(filter["thread_cursor_id"], serde_json::json!("abcd")); + assert!( + !filter.contains_key("#h"), + "no channel_id -> no #h scope in the filter" + ); +} + +#[test] +fn stored_managed_agent_auth_tag_trims_blank_values() { + assert_eq!( + stored_managed_agent_auth_tag(Some(" [\"auth\",\"owner\",\"\",\"sig\"] ")), + Some("[\"auth\",\"owner\",\"\",\"sig\"]".to_string()) + ); + assert_eq!(stored_managed_agent_auth_tag(Some(" ")), None); + assert_eq!(stored_managed_agent_auth_tag(None), None); +} + +#[test] +fn legacy_managed_agent_auth_tag_verifies_for_agent_pubkey() { + let owner_keys = Keys::generate(); + let agent_keys = Keys::generate(); + + let tag = legacy_managed_agent_auth_tag(&owner_keys, &agent_keys.public_key()) + .expect("legacy auth tag should compute") + .expect("legacy auth tag should be present"); + + let owner = buzz_sdk_pkg::nip_oa::verify_auth_tag(&tag, &agent_keys.public_key()) + .expect("legacy auth tag should verify"); + assert_eq!(owner, owner_keys.public_key()); +} + +#[test] +fn legacy_managed_agent_auth_tag_skips_self_attestation() { + let owner_keys = Keys::generate(); + + let tag = legacy_managed_agent_auth_tag(&owner_keys, &owner_keys.public_key()) + .expect("self-attestation should be skipped"); + + assert_eq!(tag, None); +} diff --git a/desktop/src-tauri/src/commands/workspace.rs b/desktop/src-tauri/src/commands/workspace.rs index c602bdce42..cb17ad608e 100644 --- a/desktop/src-tauri/src/commands/workspace.rs +++ b/desktop/src-tauri/src/commands/workspace.rs @@ -103,6 +103,7 @@ pub async fn apply_workspace( relay_url: String, nsec: Option, repos_dir: Option, + agent_managed_profiles: Option, app: AppHandle, ) -> Result<(), String> { let restore_app = app.clone(); @@ -148,6 +149,13 @@ pub async fn apply_workspace( *keys_guard = keys; } + // Keep the backend-side reconcile guard aligned with the frontend + // experiment before launch-time restore can spawn any agents. Missing + // means the stable behavior: desktop remains authoritative. + state + .managed_agent_profile_reconcile_enabled + .store(!agent_managed_profiles.unwrap_or(false), Ordering::Release); + // ── Filesystem side-effect (non-fatal) ──────────────────────────────── // Persist the *effective* repos_dir (None when the candidate failed // validation) for the backend to read at boot, then re-point REPOS to @@ -176,18 +184,45 @@ pub async fn apply_workspace( .map_err(|e| format!("spawn_blocking failed: {e}"))??; let state = restore_app.state::(); - if state + let restore_pending = state .managed_agent_restore_pending - .swap(false, Ordering::AcqRel) + .swap(false, Ordering::AcqRel); + + // The coordinator starts before React applies the selected workspace, so + // its startup publication may have used the fallback relay and placeholder + // identity. Correct it off the command path so an unavailable relay cannot + // hold the frontend on its loading gate. On initial launch, restore MeshLLM + // first so a slow stopped-status request cannot overwrite a newly restored + // serving status, then restore managed agents after the admission identity + // has been published (or the bounded publication attempt has timed out). + #[cfg(feature = "mesh-llm")] { let app = restore_app.clone(); tauri::async_runtime::spawn(async move { let state = app.state::(); - #[cfg(feature = "mesh-llm")] - if let Err(error) = crate::commands::mesh_llm::restore_mesh_sharing(&app, &state).await - { - eprintln!("buzz-desktop: failed to restore Share Compute: {error}"); + if restore_pending { + if let Err(error) = + crate::commands::mesh_llm::restore_mesh_sharing(&app, &state).await + { + eprintln!("buzz-desktop: failed to restore Share Compute: {error}"); + } } + crate::mesh_llm::publish_current_status_once(&app, "workspace apply").await; + if restore_pending { + if let Err(error) = + restore_managed_agents_on_launch(&app, &state.shutdown_started).await + { + eprintln!("buzz-desktop: failed to restore managed agents: {error}"); + } + } + }); + } + + #[cfg(not(feature = "mesh-llm"))] + if restore_pending { + let app = restore_app.clone(); + tauri::async_runtime::spawn(async move { + let state = app.state::(); if let Err(error) = restore_managed_agents_on_launch(&app, &state.shutdown_started).await { diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 9a4eb3dcb1..dadd2198b8 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -811,6 +811,7 @@ pub fn run() { search_messages, send_channel_message, send_managed_agent_channel_message, + has_managed_agent_channel_message_marker, get_forum_posts, get_forum_thread, get_thread_replies, @@ -847,6 +848,7 @@ pub fn run() { create_managed_agent, start_managed_agent, stop_managed_agent, + set_agent_managed_profiles, set_managed_agent_start_on_app_launch, set_managed_agent_auto_restart, delete_managed_agent, diff --git a/desktop/src-tauri/src/managed_agents/runtime.rs b/desktop/src-tauri/src/managed_agents/runtime.rs index 2c15813e48..31b31e00d0 100644 --- a/desktop/src-tauri/src/managed_agents/runtime.rs +++ b/desktop/src-tauri/src/managed_agents/runtime.rs @@ -1361,6 +1361,7 @@ pub fn build_managed_agent_summary( pubkey: record.pubkey.clone(), name: record.name.clone(), persona_id: record.persona_id.clone(), + team_id: record.team_id.clone(), relay_url: record.relay_url.clone(), acp_command: record.acp_command.clone(), agent_command: effective_command, diff --git a/desktop/src-tauri/src/managed_agents/teams.rs b/desktop/src-tauri/src/managed_agents/teams.rs index 6ee8c0b2ed..937893d531 100644 --- a/desktop/src-tauri/src/managed_agents/teams.rs +++ b/desktop/src-tauri/src/managed_agents/teams.rs @@ -31,7 +31,12 @@ struct BuiltInTeam { persona_ids: &'static [&'static str], } -const BUILT_IN_TEAMS: &[BuiltInTeam] = &[]; +const BUILT_IN_TEAMS: &[BuiltInTeam] = &[BuiltInTeam { + id: "builtin-team:welcome", + name: "Welcome Team", + description: Some("A friendly starter trio ready to help you plan, create, and ship."), + persona_ids: &["builtin:fizz", "builtin:honey", "builtin:bumble"], +}]; // Built-in teams that have been retired. A stored copy that still exactly // matches its seed is purged on load (the user never touched it); customized diff --git a/desktop/src-tauri/src/managed_agents/teams_tests.rs b/desktop/src-tauri/src/managed_agents/teams_tests.rs index 7e9c7656b4..140ac3cab9 100644 --- a/desktop/src-tauri/src/managed_agents/teams_tests.rs +++ b/desktop/src-tauri/src/managed_agents/teams_tests.rs @@ -317,12 +317,62 @@ fn migration_customized_fizz_is_demoted_to_user_team() { } #[test] -fn migration_empty_store_stays_empty() { - // Fresh installs see no teams and no spurious changed flag. +fn welcome_team_is_seeded_and_idempotent() { let (records, changed) = merge_teams(Vec::new(), "2026-07-01T00:00:00Z"); + assert!(changed); + assert_eq!(records.len(), 1); + let welcome = &records[0]; + assert_eq!(welcome.id, "builtin-team:welcome"); + assert_eq!(welcome.name, "Welcome Team"); + assert_eq!( + welcome.description.as_deref(), + Some("A friendly starter trio ready to help you plan, create, and ship.") + ); + assert_eq!( + welcome.persona_ids, + vec![ + "builtin:fizz".to_string(), + "builtin:honey".to_string(), + "builtin:bumble".to_string(), + ] + ); + assert!(welcome.is_builtin); + + let expected = serde_json::to_value(&records).unwrap(); + let (records_after_second_merge, changed) = merge_teams(records, "2026-07-02T00:00:00Z"); assert!(!changed); - assert!(records.is_empty()); + assert_eq!( + serde_json::to_value(records_after_second_merge).unwrap(), + expected + ); +} + +#[test] +fn welcome_team_seed_does_not_overwrite_customization() { + let (mut records, _) = merge_teams(Vec::new(), "2026-07-01T00:00:00Z"); + let welcome = records + .iter_mut() + .find(|team| team.id == "builtin-team:welcome") + .expect("welcome team should be seeded"); + welcome.name = "My Welcome Team".to_string(); + welcome.description = Some("My customized starter team.".to_string()); + welcome.persona_ids = vec!["builtin:honey".to_string()]; + + let (records, changed) = merge_teams(records, "2026-07-02T00:00:00Z"); + + assert!(!changed); + let welcome = records + .iter() + .find(|team| team.id == "builtin-team:welcome") + .expect("customized welcome team should be preserved"); + assert_eq!(welcome.name, "My Welcome Team"); + assert_eq!( + welcome.description.as_deref(), + Some("My customized starter team.") + ); + assert_eq!(welcome.persona_ids, vec!["builtin:honey".to_string()]); + assert!(welcome.is_builtin); } // ── load_teams_readonly tests ────────────────────────────────────────── @@ -337,9 +387,9 @@ fn load_teams_readonly_absent_file_performs_no_write() { let records = load_teams_readonly(&path).unwrap(); - // Returns the merged built-in list (currently empty since BUILT_IN_TEAMS - // is empty, but the function ran merge_teams + sort_teams). - assert!(records.is_empty()); + // Returns the merged built-in list without persisting it. + assert_eq!(records.len(), 1); + assert_eq!(records[0].id, "builtin-team:welcome"); // The file must still NOT exist — no write-on-load side effect. assert!( diff --git a/desktop/src-tauri/src/managed_agents/types.rs b/desktop/src-tauri/src/managed_agents/types.rs index bb0781b643..7f69b77e0d 100644 --- a/desktop/src-tauri/src/managed_agents/types.rs +++ b/desktop/src-tauri/src/managed_agents/types.rs @@ -454,6 +454,7 @@ pub struct ManagedAgentSummary { pub pubkey: String, pub name: String, pub persona_id: Option, + pub team_id: Option, pub relay_url: String, pub acp_command: String, pub agent_command: String, diff --git a/desktop/src-tauri/src/mesh_llm/catalog.rs b/desktop/src-tauri/src/mesh_llm/catalog.rs index d5fd91b905..385971cb86 100644 --- a/desktop/src-tauri/src/mesh_llm/catalog.rs +++ b/desktop/src-tauri/src/mesh_llm/catalog.rs @@ -8,10 +8,34 @@ use serde::Serialize; use mesh_llm_client::models::catalog::{parse_size_gb, MODEL_CATALOG}; -use mesh_llm_client::network::nostr::auto_model_pack; use mesh_llm_node::models::{default_huggingface_cache_dir, scan_installed_models}; use mesh_llm_system::hardware; -use mesh_llm_system::vram::format_rated_capacity; +use mesh_llm_system::vram::{format_rated_capacity, rated_capacity_gb}; + +/// Buzz-curated tier picks. These are the models we know survive the agent +/// harness on shared compute — deliberately non-reasoning instruction models, +/// so agents stay snappy instead of burning hidden reasoning tokens. +/// +/// The large pick is resolved through mesh-llm's remote catalog +/// (huggingface.co/datasets/meshllm/catalog), so it does not need to exist in +/// the compiled `MODEL_CATALOG`; the entry is synthesized below. +const CURATED_LARGE: &str = "gemma-4-26B-A4B-it-UD-Q4_K_M"; +const CURATED_LARGE_SIZE: &str = "17GB"; +const CURATED_LARGE_FILE: &str = "gemma-4-26B-A4B-it-UD-Q4_K_M.gguf"; +const CURATED_LARGE_DESCRIPTION: &str = + "Gemma 4 26B MoE (4B active) — Buzz default for 64GB+ machines"; +const CURATED_SMALL: &str = "Gemma-4-E4B-it-Q4_K_M"; +/// Rated-capacity boundary between the two curated tiers, in GB (marketing +/// capacity — a "64GB" Mac rates as 64 even though usable AI memory is less). +const CURATED_LARGE_MIN_RATED_GB: u64 = 64; + +/// The Buzz-curated recommendation for a machine's rated memory capacity. +fn buzz_recommended_model(rated_gb: Option) -> &'static str { + match rated_gb { + Some(gb) if gb >= CURATED_LARGE_MIN_RATED_GB => CURATED_LARGE, + _ => CURATED_SMALL, + } +} /// How a model sits inside this machine's usable AI memory. /// Mirrors mesh-llm's private `fit_code_for_size_label` thresholds. @@ -57,6 +81,9 @@ pub struct MeshCatalogEntry { pub fit: ModelFit, pub installed: bool, pub recommended: bool, + /// Buzz-curated pick — known to survive the agent harness. Curated + /// entries render above the fold; everything else is "advanced". + pub curated: bool, } #[derive(Debug, Clone, Serialize)] @@ -123,6 +150,7 @@ fn build_catalog( fit: fit_code(size_gb, vram_gb), installed: is_installed(&m.file, &m.name), recommended: false, + curated: false, name: m.name.clone(), size: m.size.clone(), size_gb, @@ -131,14 +159,36 @@ fn build_catalog( }) .collect(); - let recommended = auto_model_pack(vram_gb).into_iter().next(); + // The compiled MODEL_CATALOG does not know the Buzz large pick; it + // resolves through mesh-llm's remote catalog at download time. Synthesize + // its entry so the picker can offer it. + if !entries.iter().any(|e| e.name == CURATED_LARGE) { + let size_gb = parse_size_gb(CURATED_LARGE_SIZE); + entries.push(MeshCatalogEntry { + fit: fit_code(size_gb, vram_gb), + installed: is_installed(CURATED_LARGE_FILE, CURATED_LARGE), + recommended: false, + curated: false, + name: CURATED_LARGE.to_string(), + size: CURATED_LARGE_SIZE.to_string(), + size_gb, + description: CURATED_LARGE_DESCRIPTION.to_string(), + }); + } + + let recommended = Some(buzz_recommended_model(rated_capacity_gb(vram_bytes)).to_string()); for entry in &mut entries { entry.recommended = recommended.as_deref() == Some(entry.name.as_str()); + // Both curated tiers are always offered: the recommended one for this + // machine plus the other pick (e.g. the small one as an explicit + // lighter choice on big machines). + entry.curated = entry.name == CURATED_LARGE || entry.name == CURATED_SMALL; } entries.sort_by(|a, b| { b.recommended .cmp(&a.recommended) + .then(b.curated.cmp(&a.curated)) .then(fit_rank(a.fit).cmp(&fit_rank(b.fit))) .then(b.size_gb.total_cmp(&a.size_gb)) }); @@ -191,11 +241,11 @@ mod tests { assert!(catalog.entries[0].recommended); } } - // Fit ranks must be non-decreasing after the recommended head. + // Fit ranks must be non-decreasing after the recommended/curated head. let ranks: Vec = catalog .entries .iter() - .skip_while(|e| e.recommended) + .skip_while(|e| e.recommended || e.curated) .map(|e| fit_rank(e.fit)) .collect(); assert!( @@ -205,12 +255,32 @@ mod tests { } #[test] - fn recommendation_uses_mesh_llm_auto_selection() { - let catalog = build_catalog(None, 62_000_000_000, 62.0, &[]); - assert_eq!( - catalog.recommended, - auto_model_pack(62.0).into_iter().next() - ); + fn recommendation_follows_buzz_curated_tiers() { + // 64GB+ rated machines get the large curated pick. + let large = build_catalog(None, 64_000_000_000, 64.0, &[]); + assert_eq!(large.recommended.as_deref(), Some(CURATED_LARGE)); + let big = build_catalog(None, 128_000_000_000, 128.0, &[]); + assert_eq!(big.recommended.as_deref(), Some(CURATED_LARGE)); + // Below the boundary: the small curated pick — never a reasoning + // model, never sub-4B guesswork. + let small = build_catalog(None, 32_000_000_000, 32.0, &[]); + assert_eq!(small.recommended.as_deref(), Some(CURATED_SMALL)); + let tiny = build_catalog(None, 16_000_000_000, 16.0, &[]); + assert_eq!(tiny.recommended.as_deref(), Some(CURATED_SMALL)); + } + + #[test] + fn curated_picks_lead_the_catalog() { + let catalog = build_catalog(None, 96_000_000_000, 96.0, &[]); + // Recommended curated entry first, the other curated pick second, + // advanced entries after. + assert_eq!(catalog.entries[0].name, CURATED_LARGE); + assert!(catalog.entries[0].recommended && catalog.entries[0].curated); + assert_eq!(catalog.entries[1].name, CURATED_SMALL); + assert!(catalog.entries[1].curated && !catalog.entries[1].recommended); + assert!(catalog.entries[2..].iter().all(|e| !e.curated)); + // The synthesized large pick carries a real size for fit ranking. + assert!(catalog.entries[0].size_gb > 10.0); } #[test] diff --git a/desktop/src-tauri/src/mesh_llm/coordinator.rs b/desktop/src-tauri/src/mesh_llm/coordinator.rs index 8e7b2f9536..e83bf8b280 100644 --- a/desktop/src-tauri/src/mesh_llm/coordinator.rs +++ b/desktop/src-tauri/src/mesh_llm/coordinator.rs @@ -21,6 +21,7 @@ pub const KIND_BUZZ_MESH_MEMBER_STATUS: u16 = buzz_core_pkg::kind::KIND_BOOKMARK const STATUS_D_TAG_PREFIX: &str = "buzz-mesh-member-status"; const ROSTER_POLL_INTERVAL: Duration = Duration::from_secs(60); const STATUS_PUBLISH_INTERVAL: Duration = Duration::from_secs(45); +const STATUS_PUBLISH_TIMEOUT: Duration = Duration::from_secs(10); pub struct MeshCoordinator { _status_publisher: tokio::task::JoinHandle<()>, @@ -38,21 +39,25 @@ pub async fn start_coordinator(app: AppHandle) { let publisher_app = app.clone(); let status_publisher = tokio::spawn(async move { - // Clear a stale serving status promptly after an app restart. Once the - // node is stopped, the explicit stopped event is sufficient; only a - // running node needs periodic freshness heartbeats. + // Clear a stale serving status promptly after an app restart. Keep + // publishing while stopped too: serving nodes build their admission + // allowlist from fresh member statuses, so consumer-only identities + // must remain fresh even though they advertise no serving targets. publish_current_status_once(&publisher_app, "startup").await; loop { tokio::time::sleep(STATUS_PUBLISH_INTERVAL).await; - publish_running_status_once(&publisher_app).await; + publish_current_status_once(&publisher_app, "heartbeat").await; } }); 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. + let mut pending_shrink: Option> = None; loop { tokio::time::sleep(ROSTER_POLL_INTERVAL).await; let state = roster_app.state::(); - if let Err(error) = reconcile_roster(&state).await { + if let Err(error) = reconcile_roster(&state, &mut pending_shrink).await { eprintln!("buzz-mesh: roster reconcile failed: {error}"); } } @@ -71,21 +76,108 @@ pub async fn start_coordinator(app: AppHandle) { } } -async fn reconcile_roster(state: &AppState) -> Result<(), String> { +/// Outcome of a roster reconcile decision. +#[derive(Debug, PartialEq, Eq)] +enum RosterReconcileAction { + /// Keep the running allowlist untouched (no-op, or a failure we ride out). + Keep, + /// Restart the node with a freshly resolved roster. + Restart(Vec), + /// Observed a *shrink* (or empty) once. Hold the current allowlist and + /// require the same reduced roster on the next poll before tearing down, + /// so a single transient short-read never drops a member mid-inference. + AwaitConfirm(Vec), +} + +/// Whether `fresh` removes any owner present in `current` (a shrink), as +/// opposed to purely adding owners or leaving the set unchanged. +fn roster_shrinks(current: &[String], fresh: &[String]) -> bool { + current.iter().any(|owner| !fresh.contains(owner)) +} + +/// Pure decision for `reconcile_roster`, extracted so the transient-failure +/// and hysteresis invariants are unit-testable without a live relay. +/// +/// `pending_shrink` is the reduced roster we are waiting to re-confirm (from a +/// prior poll's [`RosterReconcileAction::AwaitConfirm`]), if any. +/// +/// Rules: +/// - query failed (`Err`) → `Keep` (never de-admit on a relay blip) +/// - resolved roster == current → `Keep` (no-op) +/// - grows (only additions) → `Restart` immediately (fast admission) +/// - shrinks/empties, first observation → `AwaitConfirm` (hold, re-check next poll) +/// - shrinks/empties, confirmed → `Restart` (same reduced roster twice) +fn roster_reconcile_action( + current_owners: &[String], + pending_shrink: Option<&[String]>, + query: Result, String>, +) -> RosterReconcileAction { + let fresh = match query { + Err(error) => { + eprintln!( + "buzz-mesh: roster reconcile query failed; keeping current allowlist: {error}" + ); + return RosterReconcileAction::Keep; + } + Ok(fresh) => fresh, + }; + + if fresh == current_owners { + return RosterReconcileAction::Keep; + } + + // Growth (pure additions) is safe to apply immediately. + if !roster_shrinks(current_owners, &fresh) { + return RosterReconcileAction::Restart(fresh); + } + + // A shrink (including down to empty) must be confirmed across two + // consecutive polls with the *same* reduced roster before we tear down. + match pending_shrink { + Some(pending) if pending == fresh => RosterReconcileAction::Restart(fresh), + _ => RosterReconcileAction::AwaitConfirm(fresh), + } +} + +async fn reconcile_roster( + state: &AppState, + pending_shrink: &mut Option>, +) -> Result<(), String> { let current_request = { let runtime = state.mesh_llm_runtime.lock().await; match runtime.as_ref() { Some(runtime) => runtime.start_request().clone(), - None => return Ok(()), + None => { + *pending_shrink = None; + return Ok(()); + } } }; let Some(current_owners) = current_request.trusted_owner_ids.as_ref() else { + *pending_shrink = None; return Ok(()); }; - let fresh = crate::commands::mesh_llm::resolve_trusted_owner_ids(state).await; - if &fresh == current_owners { - return Ok(()); - } + // A failed roster query must NOT be treated as "the roster became empty": + // doing so would restart the node down to self-only and de-admit every + // 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 = crate::commands::mesh_llm::resolve_trusted_owner_ids(state).await; + let fresh = match roster_reconcile_action(current_owners, pending_shrink.as_deref(), query) { + RosterReconcileAction::Keep => { + *pending_shrink = None; + return Ok(()); + } + RosterReconcileAction::AwaitConfirm(reduced) => { + eprintln!("buzz-mesh: roster shrink observed; awaiting confirmation before restart"); + *pending_shrink = Some(reduced); + return Ok(()); + } + RosterReconcileAction::Restart(fresh) => { + *pending_shrink = None; + fresh + } + }; let mut request = current_request; request.trusted_owner_ids = Some(fresh); @@ -106,22 +198,31 @@ async fn reconcile_roster(state: &AppState) -> Result<(), String> { pub(crate) async fn publish_current_status_once(app: &AppHandle, reason: &str) { let state = app.state::(); - if let Err(error) = publish_current_status_for_state(&state).await { - eprintln!("buzz-mesh: status report after {reason} failed: {error}"); + match tokio::time::timeout( + STATUS_PUBLISH_TIMEOUT, + publish_current_status_for_state(&state), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(error)) => eprintln!("buzz-mesh: status report after {reason} failed: {error}"), + Err(_) => eprintln!("buzz-mesh: status report after {reason} timed out"), } } pub(crate) async fn publish_stopped_status_once(app: &AppHandle, reason: &str) { let state = app.state::(); - if let Err(error) = publish_stopped_status_for_state(&state).await { - eprintln!("buzz-mesh: stopped status report after {reason} failed: {error}"); - } -} - -async fn publish_running_status_once(app: &AppHandle) { - let state = app.state::(); - if let Err(error) = publish_running_status_for_state(&state).await { - eprintln!("buzz-mesh: periodic status report failed: {error}"); + match tokio::time::timeout( + STATUS_PUBLISH_TIMEOUT, + publish_stopped_status_for_state(&state), + ) + .await + { + Ok(Ok(())) => {} + Ok(Err(error)) => { + eprintln!("buzz-mesh: stopped status report after {reason} failed: {error}"); + } + Err(_) => eprintln!("buzz-mesh: stopped status report after {reason} timed out"), } } @@ -150,23 +251,6 @@ async fn publish_stopped_status_for_state(state: &AppState) -> Result<(), String publish_status_report(state, payload).await } -async fn publish_running_status_for_state(state: &AppState) -> Result<(), String> { - let identity = super::ensure_owner_identity() - .map_err(|error| format!("failed to load mesh owner identity: {error}"))?; - let mut payload = { - let runtime = state.mesh_llm_runtime.lock().await; - let Some(runtime) = runtime.as_ref() else { - return Ok(()); - }; - runtime - .status_report_payload() - .await - .map_err(|error| error.to_string())? - }; - bind_payload_to_member(state, &identity, &mut payload)?; - publish_status_report(state, payload).await -} - fn stopped_status_payload(identity: &super::identity::OwnerIdentity) -> serde_json::Value { serde_json::json!({ "ownerId": identity.owner_id, @@ -233,11 +317,97 @@ mod tests { use super::*; + // Regression: a transient roster-query failure must never restart the node + // down to self-only. Before the fix, `resolve_trusted_owner_ids` returned + // an empty Vec on error, which `reconcile_roster` read as "roster changed + // to empty" and restarted — de-admitting every other member and flapping + // the node on each relay blip. See #2000 follow-up. #[test] - fn heartbeat_leaves_room_before_status_expires() { + fn failed_roster_query_keeps_current_allowlist() { + let current = vec!["owner-a".to_string(), "owner-b".to_string()]; + let action = roster_reconcile_action(¤t, None, Err("relay returned 503".to_string())); + assert_eq!( + action, + RosterReconcileAction::Keep, + "a failed query must keep the running allowlist, never de-admit members" + ); + } + + #[test] + fn unchanged_roster_is_a_noop() { + let current = vec!["owner-a".to_string()]; + let action = roster_reconcile_action(¤t, None, Ok(vec!["owner-a".to_string()])); + assert_eq!(action, RosterReconcileAction::Keep); + } + + // Growth (pure additions) applies immediately — fast admission is fine. + #[test] + fn roster_growth_restarts_immediately() { + let current = vec!["owner-a".to_string()]; + let fresh = vec!["owner-a".to_string(), "owner-c".to_string()]; + let action = roster_reconcile_action(¤t, None, Ok(fresh.clone())); + assert_eq!(action, RosterReconcileAction::Restart(fresh)); + } + + // A shrink is NOT applied on first observation — it must be confirmed. + #[test] + fn roster_shrink_awaits_confirmation_first() { + let current = vec!["owner-a".to_string(), "owner-b".to_string()]; + let reduced = vec!["owner-a".to_string()]; + let action = roster_reconcile_action(¤t, None, Ok(reduced.clone())); + assert_eq!(action, RosterReconcileAction::AwaitConfirm(reduced)); + } + + // The same reduced roster on two consecutive polls confirms the shrink. + #[test] + fn roster_shrink_restarts_once_confirmed() { + let current = vec!["owner-a".to_string(), "owner-b".to_string()]; + let reduced = vec!["owner-a".to_string()]; + let action = roster_reconcile_action(¤t, Some(&reduced), Ok(reduced.clone())); + assert_eq!(action, RosterReconcileAction::Restart(reduced)); + } + + // A shrink that changes between polls is not confirmed — it re-holds with + // the newly observed reduced roster instead of tearing down. + #[test] + fn roster_shrink_reconfirms_when_it_changes() { + let current = vec!["a".to_string(), "b".to_string(), "c".to_string()]; + let first_reduced = vec!["a".to_string(), "b".to_string()]; + let second_reduced = vec!["a".to_string()]; + let action = + roster_reconcile_action(¤t, Some(&first_reduced), Ok(second_reduced.clone())); + assert_eq!(action, RosterReconcileAction::AwaitConfirm(second_reduced)); + } + + // A genuinely empty community (Ok(empty), distinct from a failed query) + // still shrinks to self-only — but only after confirmation. + #[test] + fn genuinely_empty_roster_awaits_then_restarts_to_self_only() { + let current = vec!["owner-a".to_string()]; + let first = roster_reconcile_action(¤t, None, Ok(Vec::new())); + assert_eq!(first, RosterReconcileAction::AwaitConfirm(Vec::new())); + let empty: Vec = Vec::new(); + let confirmed = roster_reconcile_action(¤t, Some(&empty), Ok(Vec::new())); + assert_eq!(confirmed, RosterReconcileAction::Restart(Vec::new())); + } + + // A shrink followed by recovery to the full roster cancels the teardown. + #[test] + fn roster_shrink_then_recovery_keeps_allowlist() { + let current = vec!["owner-a".to_string(), "owner-b".to_string()]; + let reduced = vec!["owner-a".to_string()]; + let held = roster_reconcile_action(¤t, None, Ok(reduced.clone())); + assert_eq!(held, RosterReconcileAction::AwaitConfirm(reduced.clone())); + let recovered = roster_reconcile_action(¤t, Some(&reduced), Ok(current.clone())); + assert_eq!(recovered, RosterReconcileAction::Keep); + } + + #[test] + fn member_heartbeat_leaves_room_before_admission_status_expires() { assert!( STATUS_PUBLISH_INTERVAL.as_secs() * 2 < super::super::discovery::STATUS_FRESHNESS_SECS ); + assert!(STATUS_PUBLISH_TIMEOUT < STATUS_PUBLISH_INTERVAL); } #[test] diff --git a/desktop/src-tauri/src/mesh_llm/discovery.rs b/desktop/src-tauri/src/mesh_llm/discovery.rs index 3907a2731e..d7dca6b11b 100644 --- a/desktop/src-tauri/src/mesh_llm/discovery.rs +++ b/desktop/src-tauri/src/mesh_llm/discovery.rs @@ -9,6 +9,7 @@ use super::{dedupe_models, MeshAvailability, MeshModelOption, MeshServeTarget, M /// than two minutes so crashed/offline devices stop contributing compute or /// admission identities without requiring a relay-side cleanup job. pub(super) const STATUS_FRESHNESS_SECS: u64 = 120; +pub(crate) const MESH_STATUS_PAGE_SIZE: usize = 100; fn status_is_fresh(event: &nostr::Event, now: u64) -> bool { event @@ -34,16 +35,22 @@ fn dedupe_targets(targets: Vec) -> Vec { /// contribute an owner id. This removes stale notes from former members and /// ignores notes from nonmembers. If the relay has no membership snapshot, the /// roster is empty and MeshLLM admission therefore remains self-only. +/// +/// Admission deliberately ignores status freshness: membership is the trust +/// boundary, and a member whose device is offline (stale status) is still a +/// member. Gating admission on freshness caused the allowlist — and therefore +/// the serving node — to churn whenever any member's app went online or +/// offline. Freshness still gates *routing* (see `availability_from_events`): +/// stale nodes are never selected as serve targets. Revocation is unaffected: +/// a member removed from the NIP-43 roster leaves the intersection at the next +/// roster poll regardless of how fresh their last status is. pub fn owner_ids_from_events(events: &[nostr::Event]) -> Vec { let Some(members) = latest_membership_list(events) else { return Vec::new(); }; - let now = nostr::Timestamp::now().as_secs(); let mut ids: Vec = events .iter() - .filter(|event| { - event.kind.as_u16() as u64 == MESH_STATUS_KIND && status_is_fresh(event, now) - }) + .filter(|event| event.kind.as_u16() as u64 == MESH_STATUS_KIND) .filter(|event| { reporter_pubkey_from_status_event(event) .is_some_and(|reporter| members.contains(&reporter.to_ascii_lowercase())) @@ -79,6 +86,27 @@ fn latest_membership_list(events: &[nostr::Event]) -> Option> { }) } +pub(crate) fn current_member_pubkeys(events: &[nostr::Event]) -> Vec { + latest_membership_list(events) + .map(BTreeSet::into_iter) + .map(Iterator::collect) + .unwrap_or_default() +} + +/// Whether the relay actually returned a NIP-43 membership snapshot (kind +/// 13534) in `events`. +/// +/// The relay publishes an explicit membership event even for a zero-member +/// community, so its presence is what makes an empty roster *authoritative*. +/// Callers use this to distinguish "the community genuinely has no members" +/// (snapshot present, zero `member` tags) from "no snapshot came back at all" +/// (a transient relay gap / replication lag). Only the former may shrink the +/// admission roster; the latter must be surfaced as an error so the reconcile +/// loop keeps the current allowlist instead of restarting to self-only. +pub(crate) fn has_membership_snapshot(events: &[nostr::Event]) -> bool { + events.iter().any(|event| event.kind.as_u16() == 13_534) +} + fn owner_id_from_status_event(event: &nostr::Event) -> Option { let content = serde_json::from_str::(&event.content).ok()?; let owner_id = content @@ -186,11 +214,12 @@ pub fn availability_from_events(events: Vec) -> MeshAvailability { .unwrap_or_default() .into_iter() .filter_map(|mut target| { - let endpoint_id = + let validated = super::transport_policy::validate_advertised_endpoint(&target.endpoint_addr) .ok()?; + target.endpoint_addr = validated.join_token; if target.endpoint_id.is_none() { - target.endpoint_id = Some(endpoint_id); + target.endpoint_id = Some(validated.endpoint_id); } if target.device_id.is_none() { target.device_id = target.endpoint_id.clone(); @@ -242,11 +271,19 @@ pub fn availability_from_events(events: Vec) -> MeshAvailability { } } +/// Status filter for admission and availability queries. +/// +/// Deliberately has no `since` bound: status events are parameterized +/// replaceable (one per member), and admission must see a member's latest +/// owner binding even when that member has been offline for longer than +/// [`STATUS_FRESHNESS_SECS`]. Freshness is applied *after* the query, and only +/// where it belongs — routing (`availability_from_events`), never admission +/// (`owner_ids_from_events`). pub fn mesh_status_filter() -> serde_json::Value { serde_json::json!({ "kinds": [MESH_STATUS_KIND], "#k": ["buzz-mesh-status"], - "limit": 100 + "limit": MESH_STATUS_PAGE_SIZE }) } @@ -300,7 +337,9 @@ pub(super) fn device_name_from_status( } fn endpoint_id_from_invite_token(invite_token: &str) -> Option { - super::transport_policy::validate_advertised_endpoint(invite_token).ok() + super::transport_policy::validate_advertised_endpoint(invite_token) + .ok() + .map(|validated| validated.endpoint_id) } fn string_value(value: &serde_json::Value, key: &str) -> Option { diff --git a/desktop/src-tauri/src/mesh_llm/mod.rs b/desktop/src-tauri/src/mesh_llm/mod.rs index a75fba1f16..80d887bec7 100644 --- a/desktop/src-tauri/src/mesh_llm/mod.rs +++ b/desktop/src-tauri/src/mesh_llm/mod.rs @@ -8,6 +8,9 @@ mod discovery; pub use discovery::{ 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, +}; use discovery::{device_name_from_status, endpoint_id_from_status, enrich_status_payload_identity}; mod catalog; @@ -269,8 +272,8 @@ async fn ensure_model_downloaded(model: &str) -> anyhow::Result<()> { } impl DesktopMeshRuntime { - pub async fn start(request: StartMeshNodeRequest) -> anyhow::Result { - validate_no_leak_request(&request)?; + pub async fn start(mut request: StartMeshNodeRequest) -> anyhow::Result { + sanitize_no_leak_request(&mut request)?; initialize_mesh_native_runtime().await?; let model_id = request .model_id @@ -419,8 +422,8 @@ impl DesktopMeshRuntime { pub async fn dial_endpoint_addr(&self, endpoint_addr: impl Into) -> anyhow::Result<()> { let endpoint_addr = endpoint_addr.into(); - validate_advertised_endpoint(&endpoint_addr)?; - self.handle.join_token(endpoint_addr).await + let validated = validate_advertised_endpoint(&endpoint_addr)?; + self.handle.join_token(validated.join_token).await } pub async fn installed_models(&self) -> anyhow::Result> { @@ -502,9 +505,9 @@ fn normalized_roster( Some(owners) } -fn validate_no_leak_request(request: &StartMeshNodeRequest) -> anyhow::Result<()> { - if let Some(join_token) = request.join_token.as_deref() { - validate_advertised_endpoint(join_token)?; +fn sanitize_no_leak_request(request: &mut StartMeshNodeRequest) -> anyhow::Result<()> { + if let Some(join_token) = request.join_token.as_mut() { + *join_token = validate_advertised_endpoint(join_token)?.join_token; } Ok(()) } diff --git a/desktop/src-tauri/src/mesh_llm/mod_tests.rs b/desktop/src-tauri/src/mesh_llm/mod_tests.rs index 49897ce65e..e169af8886 100644 --- a/desktop/src-tauri/src/mesh_llm/mod_tests.rs +++ b/desktop/src-tauri/src/mesh_llm/mod_tests.rs @@ -214,6 +214,34 @@ fn signed_membership_event(members: &[String]) -> nostr::Event { .expect("test membership event signs") } +#[test] +fn has_membership_snapshot_distinguishes_empty_from_missing() { + // A zero-member community still publishes an explicit kind:13534 event, so + // its presence — not the member count — is what makes an empty roster + // authoritative. No snapshot at all means the query was incomplete. + let zero_member_snapshot = signed_membership_event(&[]); + assert!( + super::has_membership_snapshot(std::slice::from_ref(&zero_member_snapshot)), + "an explicit zero-member snapshot counts as present" + ); + + let member = nostr::Keys::parse(&"1".repeat(64)) + .unwrap() + .public_key() + .to_hex(); + let populated = signed_membership_event(std::slice::from_ref(&member)); + assert!(super::has_membership_snapshot(std::slice::from_ref( + &populated + ))); + + // A response with only status events (or nothing) has no snapshot. + assert!(!super::has_membership_snapshot(&[])); + let status_only = signed_reporter_status(&"2".repeat(64), "owner-x"); + assert!(!super::has_membership_snapshot(std::slice::from_ref( + &status_only + ))); +} + #[test] fn owner_ids_from_events_collects_sorted_deduped_roster() { let secret_a = "1".repeat(64); @@ -388,20 +416,45 @@ fn signed_reporter_target(reporter_secret: &str, model: &str, endpoint: &str) -> } #[test] -fn stale_status_is_excluded_from_admission_and_availability() { +fn stale_status_keeps_member_admitted_but_excluded_from_routing() { let secret = "8".repeat(64); let member = nostr::Keys::parse(&secret).unwrap().public_key().to_hex(); let stale = signed_reporter_target_at(&secret, "stale-model", &test_endpoint_token(), 1_000); - // No newer event is required to age this status out. Freshness is measured - // against wall clock, so an entirely offline mesh cannot remain live forever. + // Membership is the trust boundary: a current member whose device went + // offline (stale status) must stay admitted, otherwise every app + // open/close in the community churns the allowlist and restarts serving + // nodes. Freshness still gates routing: a stale node is never selected as + // a serve target. let membership = signed_membership_event_at(std::slice::from_ref(&member), 900); let events = vec![stale, membership]; - assert!(super::owner_ids_from_events(&events).is_empty()); + assert_eq!(super::owner_ids_from_events(&events).len(), 1); let availability = super::availability_from_events(events); assert!(availability.serve_targets.is_empty()); } +#[test] +fn removed_member_is_dropped_from_admission_despite_fresh_status() { + // Revocation path: freshness must never resurrect trust. A reporter with a + // perfectly fresh status who is absent from the latest NIP-43 roster gets + // no admission entry. + let member_secret = "8".repeat(64); + let outsider_secret = "9".repeat(64); + let member = nostr::Keys::parse(&member_secret) + .unwrap() + .public_key() + .to_hex(); + let now = nostr::Timestamp::now().as_secs(); + let fresh_outsider = + signed_reporter_target_at(&outsider_secret, "model", &test_endpoint_token(), now); + // Latest roster lists only `member`; the outsider was removed (or never + // admitted). + let membership = signed_membership_event_at(std::slice::from_ref(&member), now); + let events = vec![fresh_outsider, membership]; + + assert!(super::owner_ids_from_events(&events).is_empty()); +} + #[test] fn one_endpoint_can_advertise_multiple_models() { let secret = "a".repeat(64); diff --git a/desktop/src-tauri/src/mesh_llm/transport_policy.rs b/desktop/src-tauri/src/mesh_llm/transport_policy.rs index e4e7ca9fe1..122bffd60e 100644 --- a/desktop/src-tauri/src/mesh_llm/transport_policy.rs +++ b/desktop/src-tauri/src/mesh_llm/transport_policy.rs @@ -72,7 +72,15 @@ fn parse_configured_relay_url(raw: &str) -> anyhow::Result { .map_err(|error| anyhow::anyhow!("invalid iroh relay URL {raw:?}: {error}")) } -pub(super) fn validate_advertised_endpoint(invite_token: &str) -> anyhow::Result { +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct ValidatedEndpoint { + pub endpoint_id: String, + pub join_token: String, +} + +pub(super) fn validate_advertised_endpoint( + invite_token: &str, +) -> anyhow::Result { let mode = iroh_relay_mode()?; validate_advertised_endpoint_with_mode(invite_token, &mode) } @@ -80,7 +88,7 @@ pub(super) fn validate_advertised_endpoint(invite_token: &str) -> anyhow::Result pub(super) fn validate_advertised_endpoint_with_mode( invite_token: &str, mode: &IrohRelayMode, -) -> anyhow::Result { +) -> anyhow::Result { let token = invite_token.trim(); if token.is_empty() { anyhow::bail!("mesh invite token is empty"); @@ -92,59 +100,139 @@ pub(super) fn validate_advertised_endpoint_with_mode( .decode(token) .map_err(|error| anyhow::anyhow!("invalid mesh invite encoding: {error}"))?; - let addrs = if let Ok(addr) = serde_json::from_slice::(&payload) { - vec![addr] - } else { - let signed = serde_json::from_slice::(&payload) - .map_err(|error| anyhow::anyhow!("invalid mesh invite payload: {error}"))?; - signed - .verify() - .map_err(|reason| anyhow::anyhow!("invalid signed mesh invite: {}", reason.code()))?; - if signed.serialized_addrs.is_empty() || signed.serialized_addrs.len() > MAX_BOOTSTRAP_ADDRS - { - anyhow::bail!( - "signed mesh invite must contain 1..={MAX_BOOTSTRAP_ADDRS} endpoint addresses" - ); - } - signed - .serialized_addrs - .iter() - .map(|bytes| { - serde_json::from_slice::(bytes) - .map_err(|error| anyhow::anyhow!("invalid signed endpoint address: {error}")) - }) - .collect::>>()? - }; - - for addr in &addrs { - validate_endpoint_addr(addr, mode)?; + if let Ok(mut addr) = serde_json::from_slice::(&payload) { + retain_usable_transports(&mut addr, mode)?; + let endpoint_id = addr.id.to_string(); + let join_token = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&addr)?); + return Ok(ValidatedEndpoint { + endpoint_id, + join_token, + }); } - Ok(addrs[0].id.to_string()) + + let signed = serde_json::from_slice::(&payload) + .map_err(|error| anyhow::anyhow!("invalid mesh invite payload: {error}"))?; + signed + .verify() + .map_err(|reason| anyhow::anyhow!("invalid signed mesh invite: {}", reason.code()))?; + if signed.serialized_addrs.is_empty() || signed.serialized_addrs.len() > MAX_BOOTSTRAP_ADDRS { + anyhow::bail!( + "signed mesh invite must contain 1..={MAX_BOOTSTRAP_ADDRS} endpoint addresses" + ); + } + let addrs = signed + .serialized_addrs + .iter() + .map(|bytes| { + serde_json::from_slice::(bytes) + .map_err(|error| anyhow::anyhow!("invalid signed endpoint address: {error}")) + }) + .collect::>>()?; + + // Rewriting a signed token would invalidate its signature. Keep the signed + // envelope intact only when every advertised transport is policy-approved; + // mixed signed tokens fail closed rather than leaking rejected dial targets. + for addr in &addrs { + validate_signed_transports(addr, mode)?; + } + Ok(ValidatedEndpoint { + endpoint_id: addrs[0].id.to_string(), + join_token: token.to_string(), + }) } -fn validate_endpoint_addr(addr: &EndpointAddr, mode: &IrohRelayMode) -> anyhow::Result<()> { +fn retain_usable_transports(addr: &mut EndpointAddr, mode: &IrohRelayMode) -> anyhow::Result<()> { + validate_transport_count(addr)?; + let mut rejections = Vec::new(); + addr.addrs + .retain(|transport| match validate_transport(transport, mode) { + Ok(()) => true, + Err(error) => { + rejections.push(error.to_string()); + false + } + }); + if addr.addrs.is_empty() { + anyhow::bail!( + "mesh endpoint has no usable transport address (all rejected: {})", + rejections.join("; ") + ); + } + Ok(()) +} + +fn validate_signed_transports(addr: &EndpointAddr, mode: &IrohRelayMode) -> anyhow::Result<()> { + validate_transport_count(addr)?; + let mut usable = 0usize; + for transport in &addr.addrs { + match validate_transport(transport, mode) { + Ok(()) => usable += 1, + // mesh-llm currently signs its own port-0 placeholder alongside a + // valid relay. It is non-dialable, so preserving it is safe and + // necessary for stock signed tokens to remain usable. + Err(_) if matches!(transport, TransportAddr::Ip(socket) if socket.port() == 0) => {} + Err(error) => return Err(error), + } + } + if usable == 0 { + anyhow::bail!("signed mesh endpoint has no usable transport address"); + } + Ok(()) +} + +fn validate_transport_count(addr: &EndpointAddr) -> anyhow::Result<()> { if addr.addrs.is_empty() || addr.addrs.len() > MAX_ENDPOINT_TRANSPORT_ADDRS { anyhow::bail!( "mesh endpoint must contain 1..={MAX_ENDPOINT_TRANSPORT_ADDRS} transport addresses" ); } - for transport in &addr.addrs { - match transport { - TransportAddr::Relay(relay) if relay_allowed(relay, mode) => {} - TransportAddr::Relay(relay) => { - anyhow::bail!("mesh endpoint advertises unapproved relay URL {relay}") - } - TransportAddr::Ip(socket) => validate_direct_socket(*socket)?, - _ => anyhow::bail!("mesh endpoint contains an unsupported transport address"), - } - } Ok(()) } +fn validate_transport(transport: &TransportAddr, mode: &IrohRelayMode) -> anyhow::Result<()> { + match transport { + TransportAddr::Relay(relay) if relay_allowed(relay, mode) => Ok(()), + TransportAddr::Relay(relay) => anyhow::bail!("unapproved relay URL {relay}"), + TransportAddr::Ip(socket) => validate_direct_socket(*socket), + _ => anyhow::bail!("unsupported transport address"), + } +} + +/// mesh-llm's default public relay set (`RelayPolicy::DefaultPublic` in +/// `mesh-llm-host-runtime`). A stock mesh-llm server with no custom relay +/// config advertises endpoints on exactly these relays, so buzz's `Default` +/// mode MUST accept them — otherwise shared compute rejects every out-of-the-box +/// mesh-llm serving node (they are not in iroh's own prod relay map). +/// +/// Kept in sync with `effective_relay_urls(RelayPolicy::DefaultPublic, &[])`. +const MESH_LLM_DEFAULT_RELAYS: &[&str] = &[ + "https://usw1-2.relay.michaelneale.mesh-llm.iroh.link./", + "https://aps1-1.relay.michaelneale.mesh-llm.iroh.link./", +]; + +/// Whether `relay` is one of mesh-llm's baked-in default public relays. +/// Parses each known URL to a `RelayUrl` so comparison is normalization-safe +/// (matches regardless of trailing-dot / trailing-slash formatting). +fn is_mesh_llm_default_relay(relay: &RelayUrl) -> bool { + MESH_LLM_DEFAULT_RELAYS.iter().any(|candidate| { + candidate + .parse::() + .map(|known| &known == relay) + .unwrap_or(false) + }) +} + fn relay_allowed(relay: &RelayUrl, mode: &IrohRelayMode) -> bool { match mode { IrohRelayMode::Disabled => false, - IrohRelayMode::Default => iroh::defaults::prod::default_relay_map().contains(relay), + // `Default` covers both iroh's own production relays AND mesh-llm's + // default public relays. Without the latter, a stock mesh-llm serving + // node is unreachable by default and shared compute silently fails with + // "no live member is serving this model" even though discovery found it. + IrohRelayMode::Default => { + iroh::defaults::prod::default_relay_map().contains(relay) + || is_mesh_llm_default_relay(relay) + } IrohRelayMode::Custom(urls) => urls.contains(relay), } } @@ -213,6 +301,76 @@ mod tests { assert!(validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Disabled).is_err()); } + #[test] + fn default_mode_accepts_meshllm_default_relays() { + // Regression: a stock mesh-llm serving node advertises endpoints on + // mesh-llm's OWN default public relays (not iroh's prod relay map). + // Under `Default` mode these MUST be accepted, or shared compute rejects + // every out-of-the-box mesh-llm server with "no live member is serving + // this model" even though discovery found it. See mesh-llm + // effective_relay_urls(RelayPolicy::DefaultPublic, &[]). + for relay_url in MESH_LLM_DEFAULT_RELAYS { + let relay: RelayUrl = relay_url + .parse() + .unwrap_or_else(|e| panic!("mesh-llm default relay {relay_url:?} must parse: {e}")); + assert!( + relay_allowed(&relay, &IrohRelayMode::Default), + "Default mode must accept mesh-llm default relay {relay_url}" + ); + + // And end-to-end through the advertised-endpoint validator. + let token = endpoint_token_for_test([TransportAddr::Relay(relay)]); + assert!( + validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_ok(), + "Default mode must validate an endpoint on mesh-llm default relay {relay_url}" + ); + } + } + + #[test] + fn endpoint_with_one_good_and_one_junk_candidate_is_sanitized() { + // A mesh-llm endpoint can advertise a usable relay alongside an + // unusable direct IP. Keep the endpoint reachable, but never pass the + // rejected candidate through to iroh's parallel dialer. + let good_relay: RelayUrl = MESH_LLM_DEFAULT_RELAYS[0].parse().unwrap(); + let unsafe_socket = "169.254.169.254:80".parse().unwrap(); + let token = endpoint_token_for_test([ + TransportAddr::Relay(good_relay.clone()), + TransportAddr::Ip(unsafe_socket), + ]); + let validated = + validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).unwrap(); + let payload = URL_SAFE_NO_PAD.decode(validated.join_token).unwrap(); + let sanitized: EndpointAddr = serde_json::from_slice(&payload).unwrap(); + assert_eq!( + sanitized.addrs, + [TransportAddr::Relay(good_relay)].into_iter().collect() + ); + } + + #[test] + fn endpoint_with_all_junk_candidates_is_rejected() { + // Guard: if EVERY candidate is unusable, the endpoint must still fail. + let token = endpoint_token_for_test([ + TransportAddr::Ip("180.181.228.108:0".parse().unwrap()), // port 0 + TransportAddr::Ip("127.0.0.1:9337".parse().unwrap()), // loopback + ]); + assert!( + validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_err(), + "endpoint with no usable candidate must be rejected" + ); + } + + #[test] + fn default_mode_still_rejects_unknown_relay() { + // Guard the fix doesn't over-open: a relay that is neither iroh-prod nor + // a mesh-llm default must still be rejected under Default mode. + let unknown: RelayUrl = "https://not-a-real-relay.example".parse().unwrap(); + assert!(!relay_allowed(&unknown, &IrohRelayMode::Default)); + let token = endpoint_token_for_test([TransportAddr::Relay(unknown)]); + assert!(validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).is_err()); + } + #[test] fn remote_endpoint_rejects_unsafe_direct_targets_and_oversized_tokens() { for socket in ["127.0.0.1:9337", "169.254.169.254:80", "0.0.0.0:1"] { @@ -263,10 +421,59 @@ mod tests { .expect("sign test bootstrap token"); let token = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&signed).unwrap()); assert_eq!( - validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default).unwrap(), + validate_advertised_endpoint_with_mode(&token, &IrohRelayMode::Default) + .unwrap() + .endpoint_id, endpoint.id.to_string() ); + let good_relay: RelayUrl = MESH_LLM_DEFAULT_RELAYS[0].parse().unwrap(); + let placeholder_endpoint = EndpointAddr { + id: endpoint.id, + addrs: [ + TransportAddr::Relay(good_relay.clone()), + TransportAddr::Ip("180.181.228.108:0".parse().unwrap()), + ] + .into_iter() + .collect(), + }; + let placeholder_signed = SignedBootstrapToken::sign( + vec![serde_json::to_vec(&placeholder_endpoint).unwrap()], + &signed_policy, + None, + &owner, + ) + .expect("sign placeholder test bootstrap token"); + let placeholder_token = + URL_SAFE_NO_PAD.encode(serde_json::to_vec(&placeholder_signed).unwrap()); + assert!( + validate_advertised_endpoint_with_mode(&placeholder_token, &IrohRelayMode::Default) + .is_ok(), + "signed stock token with a port-0 placeholder must remain usable" + ); + + let mixed_endpoint = EndpointAddr { + id: endpoint.id, + addrs: [ + TransportAddr::Relay(good_relay), + TransportAddr::Ip("169.254.169.254:80".parse().unwrap()), + ] + .into_iter() + .collect(), + }; + let mixed_signed = SignedBootstrapToken::sign( + vec![serde_json::to_vec(&mixed_endpoint).unwrap()], + &signed_policy, + None, + &owner, + ) + .expect("sign mixed test bootstrap token"); + let mixed_token = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&mixed_signed).unwrap()); + assert!( + validate_advertised_endpoint_with_mode(&mixed_token, &IrohRelayMode::Default).is_err(), + "signed mixed-candidate tokens must fail closed because they cannot be rewritten" + ); + let mut tampered = signed; tampered.serialized_addrs[0] = serde_json::to_vec(&EndpointAddr { id: endpoint.id, diff --git a/desktop/src-tauri/tauri.conf.json b/desktop/src-tauri/tauri.conf.json index 5ecfa89e94..77933cbe78 100644 --- a/desktop/src-tauri/tauri.conf.json +++ b/desktop/src-tauri/tauri.conf.json @@ -1,7 +1,7 @@ { "$schema": "https://schema.tauri.app/config/2", "productName": "Buzz", - "version": "0.4.10", + "version": "0.4.13", "identifier": "xyz.block.buzz.app", "build": { "beforeDevCommand": { @@ -27,7 +27,7 @@ "dragDropEnabled": false, "trafficLightPosition": { "x": 16, - "y": 24 + "y": 25 }, "backgroundThrottling": "disabled", "minWidth": 800, diff --git a/desktop/src/app/AppShell.tsx b/desktop/src/app/AppShell.tsx index 05f8e25819..5b65584e8c 100644 --- a/desktop/src/app/AppShell.tsx +++ b/desktop/src/app/AppShell.tsx @@ -4,6 +4,7 @@ import { Outlet, useLocation } from "@tanstack/react-router"; import { deriveShellRoute } from "@/app/AppShell.helpers"; import { AppShellProvider } from "@/app/AppShellContext"; +import * as BuzzTheme from "@/app/BuzzThemeSurfaces"; import { AppShellOverlays } from "@/app/AppShellOverlays"; import { AppTopChrome } from "@/app/AppTopChrome"; import { useAppNavigation } from "@/app/navigation/useAppNavigation"; @@ -762,6 +763,7 @@ export function AppShell() { isHuddleDrawerOpen && "buzz-huddle-app-surface-open", )} > + {hasCommunityRail ? ( -
+ -
+