diff --git a/crates/buzz-agent/src/config.rs b/crates/buzz-agent/src/config.rs index f06849c7d1..f3464fb903 100644 --- a/crates/buzz-agent/src/config.rs +++ b/crates/buzz-agent/src/config.rs @@ -726,6 +726,12 @@ pub struct Config { pub anthropic_api_version: String, /// OpenAI endpoint selection. See [`OpenAiApi`]. pub openai_api: OpenAiApi, + /// Prefer mesh-llm's virtual `mesh` model when the configured/effective + /// OpenAI model is `auto` and the live model catalog advertises it. + /// Set by Buzz's relay-mesh provider via + /// `BUZZ_AGENT_PREFER_MESH_FOR_AUTO=1`; other providers keep their + /// existing `auto` semantics. + pub prefer_mesh_for_auto: bool, pub hints_enabled: bool, /// Thinking/reasoning effort level. `None` = use provider default (no /// thinking config sent). Set via `BUZZ_AGENT_THINKING_EFFORT`. @@ -798,6 +804,7 @@ impl Config { base_url, anthropic_api_version: env_or("ANTHROPIC_API_VERSION", "2023-06-01"), openai_api, + prefer_mesh_for_auto: parse_env("BUZZ_AGENT_PREFER_MESH_FOR_AUTO", 0u8)? != 0, max_rounds: parse_env("BUZZ_AGENT_MAX_ROUNDS", 0)?, max_output_tokens: parse_env("BUZZ_AGENT_MAX_OUTPUT_TOKENS", 32_768)?, llm_timeout: Duration::from_secs(parse_env("BUZZ_AGENT_LLM_TIMEOUT_SECS", 240)?), @@ -844,6 +851,7 @@ impl Config { system_prompt: String::new(), anthropic_api_version: "2023-06-01".into(), openai_api: OpenAiApi::Chat, + prefer_mesh_for_auto: false, max_rounds: 0, max_output_tokens: 1, llm_timeout: Duration::from_secs(30), diff --git a/crates/buzz-agent/src/llm.rs b/crates/buzz-agent/src/llm.rs index 758be4e124..23cef24e72 100644 --- a/crates/buzz-agent/src/llm.rs +++ b/crates/buzz-agent/src/llm.rs @@ -1,8 +1,11 @@ +use std::collections::BTreeSet; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::Arc; use reqwest::Client; use serde_json::{json, Value}; +use tokio::sync::Mutex; +use tokio::time::Instant; use crate::auth::{PkceOAuthConfig, PkceOAuthTokenSource, StaticTokenSource, TokenSource}; use crate::config::{ @@ -22,11 +25,59 @@ const DATABRICKS_OAUTH_SCOPES: &[&str] = &["all-apis", "offline_access"]; const MAX_LLM_RESPONSE_BYTES: usize = 16 * 1024 * 1024; const MAX_LLM_ERROR_BODY_BYTES: usize = 4 * 1024; const STALL_NOTICE_THRESHOLD: std::time::Duration = std::time::Duration::from_secs(300); +const MESH_VIRTUAL_MODEL_ID: &str = "mesh"; +const MESH_AUTO_MODEL_ID: &str = "auto"; +const MESH_AUTO_CATALOG_TTL: std::time::Duration = std::time::Duration::from_secs(5); +const MESH_AUTO_CATALOG_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2); +const MESH_AUTO_COOLDOWN: std::time::Duration = std::time::Duration::from_secs(30); +const MESH_AUTO_ENABLE_OBSERVATIONS: u8 = 2; +const MESH_MOA_UNAVAILABLE_MESSAGE: &str = "MoA requires ≥2 models available in the mesh"; /// Parser for an OpenAI-family JSON response. Per-endpoint pair lives /// alongside its `_body` serializer. type OpenAiParse = fn(Value) -> Result; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum MeshCatalogObservation { + Available, + Unavailable, + Unknown, +} + +fn mesh_catalog_supports_collective(catalog: &Value) -> Option { + let models = catalog.get("data").and_then(Value::as_array)?; + let mut has_virtual_mesh = false; + let mut physical_models = BTreeSet::new(); + for id in models + .iter() + .filter_map(|model| model.get("id").and_then(Value::as_str)) + .map(str::trim) + .filter(|id| !id.is_empty()) + { + if id == MESH_VIRTUAL_MODEL_ID { + has_virtual_mesh = true; + } else if id != MESH_AUTO_MODEL_ID { + physical_models.insert(id.replace("@main", "")); + } + } + Some(has_virtual_mesh && physical_models.len() >= 2) +} + +fn looks_like_unstructured_tool_call(text: &str) -> bool { + let text = text.trim_start().to_ascii_lowercase(); + text.starts_with("<|tool_call") + || text.starts_with(", + consecutive_available: u8, + collective_enabled: bool, + cooldown_until: Option, +} + #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum DatabricksV2Route { OpenAiResponses, @@ -41,6 +92,10 @@ pub struct Llm { /// == Auto`. Subsequent OpenAI calls then go straight to Responses /// for the lifetime of the process. auto_upgraded: AtomicBool, + /// Hysteretic view of whether mesh-llm currently advertises its virtual + /// Mixture-of-Agents model. A TTL, confirmation count, and failure cooldown + /// let long-running agents adapt without bouncing as peers briefly flap. + mesh_auto_state: Mutex, /// Bearer-token source for OpenAI-family requests. Static for OpenAI /// (the `OPENAI_COMPAT_API_KEY` env var) and Databricks-with-token /// (the `DATABRICKS_TOKEN` env var); a refreshable PKCE engine for @@ -60,6 +115,7 @@ impl Llm { Ok(Self { http, auto_upgraded: AtomicBool::new(false), + mesh_auto_state: Mutex::new(MeshAutoState::default()), auth, }) } @@ -91,24 +147,37 @@ impl Llm { parse_anthropic(v) } Provider::OpenAi | Provider::Databricks => { - self.openai_request(cfg, effective_model, |use_responses| { - // Normalize effort for model-specific availability. Startup no longer rejects - // `max` for pure OpenAI/Databricks; this per-model table is the single authority - // — it keeps `max` for gpt-5.6, clamps `max`→`xhigh` for other OpenAI-shaped - // models, and still applies corrections like none→minimal on the gpt-5 base. - let e = effort.map(|ef| normalize_effort_for_openai_route(ef, effective_model)); - if use_responses { - ( - responses_body(cfg, system_prompt, history, tools, effective_model, e), - parse_responses as OpenAiParse, - ) - } else { - ( - openai_body(cfg, system_prompt, history, tools, effective_model, e), - parse_openai as OpenAiParse, - ) - } - }) + self.openai_request( + cfg, + effective_model, + !tools.is_empty(), + |use_responses, request_model| { + // Normalize effort for model-specific availability. Startup no longer rejects + // `max` for pure OpenAI/Databricks; this per-model table is the single authority + // — it keeps `max` for gpt-5.6, clamps `max`→`xhigh` for other OpenAI-shaped + // models, and still applies corrections like none→minimal on the gpt-5 base. + let e = + effort.map(|ef| normalize_effort_for_openai_route(ef, request_model)); + if use_responses { + ( + responses_body( + cfg, + system_prompt, + history, + tools, + request_model, + e, + ), + parse_responses as OpenAiParse, + ) + } else { + ( + openai_body(cfg, system_prompt, history, tools, request_model, e), + parse_openai as OpenAiParse, + ) + } + }, + ) .await } Provider::DatabricksV2 => { @@ -181,32 +250,37 @@ impl Llm { } Provider::OpenAi | Provider::Databricks => { let r = self - .openai_request(cfg, effective_model, |use_responses| { - if use_responses { - ( - json!({ - "model": effective_model, - "max_output_tokens": max_output_tokens, - "instructions": system_prompt, - "input": user_prompt, - }), - parse_responses as OpenAiParse, - ) - } else { - ( - json!({ - "model": effective_model, - "stream": false, - "max_completion_tokens": max_output_tokens, - "messages": [ - { "role": "system", "content": system_prompt }, - { "role": "user", "content": user_prompt }, - ], - }), - parse_openai as OpenAiParse, - ) - } - }) + .openai_request( + cfg, + effective_model, + false, + |use_responses, request_model| { + if use_responses { + ( + json!({ + "model": request_model, + "max_output_tokens": max_output_tokens, + "instructions": system_prompt, + "input": user_prompt, + }), + parse_responses as OpenAiParse, + ) + } else { + ( + json!({ + "model": request_model, + "stream": false, + "max_completion_tokens": max_output_tokens, + "messages": [ + { "role": "system", "content": system_prompt }, + { "role": "user", "content": user_prompt }, + ], + }), + parse_openai as OpenAiParse, + ) + } + }, + ) .await?; Ok(r.text) } @@ -255,49 +329,247 @@ impl Llm { async fn post_anthropic(&self, cfg: &Config, body: &Value) -> Result { let url = format!("{}/v1/messages", cfg.base_url.trim_end_matches('/')); - post(&self.http, &url, body, |r| { + post(&self.http, &url, body, false, |r| { r.header("x-api-key", &cfg.api_key) .header("anthropic-version", &cfg.anthropic_api_version) }) .await + .map_err(PostError::into_agent) } - /// OpenAI dispatch: resolve endpoint (pinned > sticky-upgraded > auto by - /// host), POST, and on `auto` retry once on Responses if the provider - /// asks for it. `build` is called with `use_responses` so callers - /// only construct the body actually needed. + /// OpenAI dispatch with Buzz's relay-mesh `auto` policy layered over the + /// normal endpoint selection. When enabled, a live virtual `mesh` model is + /// preferred; if the mesh contracts between discovery and inference, retry + /// the same request once through the router's ordinary `auto` model. async fn openai_request( &self, cfg: &Config, effective_model: &str, + tools_supplied: bool, mut build: F, ) -> Result where - F: FnMut(bool) -> (Value, OpenAiParse) + Send, + F: FnMut(bool, &str) -> (Value, OpenAiParse) + Send, + { + let request_model = self.resolve_openai_model(cfg, effective_model).await; + let adaptive_mesh = + effective_model == MESH_AUTO_MODEL_ID && request_model == MESH_VIRTUAL_MODEL_ID; + + let first = self + .openai_request_for_model(cfg, &request_model, &mut build) + .await; + match first { + Err(PostError::MeshFallback(detail)) if adaptive_mesh => { + self.cool_down_collective().await; + tracing::warn!( + configured_model = effective_model, + attempted_model = MESH_VIRTUAL_MODEL_ID, + fallback_model = MESH_AUTO_MODEL_ID, + provider_message = detail, + "relay-mesh auto: collective request failed; retrying once with auto" + ); + self.openai_request_for_model(cfg, MESH_AUTO_MODEL_ID, &mut build) + .await + .map_err(PostError::into_agent) + } + Ok(response) + if adaptive_mesh + && tools_supplied + && response.tool_calls.is_empty() + && looks_like_unstructured_tool_call(&response.text) => + { + self.cool_down_collective().await; + tracing::warn!( + configured_model = effective_model, + attempted_model = MESH_VIRTUAL_MODEL_ID, + fallback_model = MESH_AUTO_MODEL_ID, + "relay-mesh auto: collective response emitted unstructured tool markup; retrying once with auto" + ); + self.openai_request_for_model(cfg, MESH_AUTO_MODEL_ID, &mut build) + .await + .map_err(PostError::into_agent) + } + Ok(response) => Ok(response), + Err(error) => Err(error.into_agent()), + } + } + + async fn cool_down_collective(&self) { + let now = Instant::now(); + let mut state = self.mesh_auto_state.lock().await; + state.last_checked = Some(now); + state.consecutive_available = 0; + state.collective_enabled = false; + state.cooldown_until = Some(now + MESH_AUTO_COOLDOWN); + } + + /// Resolve the model for one OpenAI-family request. Explicit model choices + /// are never changed. Relay-mesh `auto` dynamically follows the short-lived + /// `/models` catalog so long-running agents can adopt or leave MoA without + /// being restarted. + async fn resolve_openai_model(&self, cfg: &Config, effective_model: &str) -> String { + if cfg.provider != Provider::OpenAi + || !cfg.prefer_mesh_for_auto + || effective_model != MESH_AUTO_MODEL_ID + { + return effective_model.to_string(); + } + + let mut state = self.mesh_auto_state.lock().await; + let now = Instant::now(); + if let Some(cooldown_until) = state.cooldown_until { + if now < cooldown_until { + return MESH_AUTO_MODEL_ID.to_string(); + } + state.cooldown_until = None; + } + if state + .last_checked + .is_some_and(|checked_at| checked_at.elapsed() < MESH_AUTO_CATALOG_TTL) + { + return if state.collective_enabled { + MESH_VIRTUAL_MODEL_ID.to_string() + } else { + MESH_AUTO_MODEL_ID.to_string() + }; + } + + let observation = self.observe_mesh_virtual_model(cfg).await; + let checked_at = Instant::now(); + state.last_checked = Some(checked_at); + match observation { + MeshCatalogObservation::Available => { + state.consecutive_available = state.consecutive_available.saturating_add(1); + if state.consecutive_available >= MESH_AUTO_ENABLE_OBSERVATIONS { + state.collective_enabled = true; + } + } + MeshCatalogObservation::Unavailable => { + if state.collective_enabled { + state.cooldown_until = Some(checked_at + MESH_AUTO_COOLDOWN); + } + state.consecutive_available = 0; + state.collective_enabled = false; + } + // A failed catalog check is not evidence that a previously stable + // mesh disappeared. Preserve the last confirmed state; inference + // itself remains authoritative and has the narrow 503 fallback. + MeshCatalogObservation::Unknown => {} + } + let request_model = if state.collective_enabled { + MESH_VIRTUAL_MODEL_ID + } else { + MESH_AUTO_MODEL_ID + }; + tracing::debug!( + configured_model = effective_model, + request_model, + "relay-mesh auto: resolved request model from live catalog" + ); + request_model.to_string() + } + + async fn observe_mesh_virtual_model(&self, cfg: &Config) -> MeshCatalogObservation { + let url = format!("{}/models", cfg.base_url.trim_end_matches('/')); + let bearer = match self.auth.bearer().await { + Ok(bearer) => bearer, + Err(error) => { + tracing::debug!( + %error, + "relay-mesh auto: catalog auth unavailable; preserving last confirmed route" + ); + return MeshCatalogObservation::Unknown; + } + }; + let response = match self + .http + .get(&url) + .bearer_auth(bearer) + .timeout(MESH_AUTO_CATALOG_TIMEOUT) + .send() + .await + { + Ok(response) => response, + Err(error) => { + tracing::debug!( + %error, + "relay-mesh auto: catalog probe failed; preserving last confirmed route" + ); + return MeshCatalogObservation::Unknown; + } + }; + if !response.status().is_success() { + tracing::debug!( + status = %response.status(), + "relay-mesh auto: catalog probe was not successful; preserving last confirmed route" + ); + return MeshCatalogObservation::Unknown; + } + match response.json::().await { + Ok(catalog) => { + let Some(available) = mesh_catalog_supports_collective(&catalog) else { + tracing::debug!( + "relay-mesh auto: catalog response has no data array; preserving last confirmed route" + ); + return MeshCatalogObservation::Unknown; + }; + if available { + MeshCatalogObservation::Available + } else { + MeshCatalogObservation::Unavailable + } + } + Err(error) => { + tracing::debug!( + %error, + "relay-mesh auto: invalid catalog response; preserving last confirmed route" + ); + MeshCatalogObservation::Unknown + } + } + } + + /// Dispatch one OpenAI request for an already-resolved model. Endpoint + /// selection remains unchanged: pinned > sticky-upgraded > host default, + /// with the existing one-shot Chat-to-Responses upgrade. + async fn openai_request_for_model( + &self, + cfg: &Config, + request_model: &str, + build: &mut F, + ) -> Result + where + F: FnMut(bool, &str) -> (Value, OpenAiParse) + Send, { let use_responses = self.auto_upgraded.load(Ordering::Relaxed) || matches!(cfg.openai_api, OpenAiApi::Responses) || matches!(cfg.openai_api, OpenAiApi::Auto) && is_openai_host(&cfg.base_url); if use_responses { - let (b, p) = build(true); - return p(self - .post_openai(cfg, "/responses", &b, effective_model) - .await?); + let (body, parse) = build(true, request_model); + return parse( + self.post_openai(cfg, "/responses", &body, request_model) + .await?, + ) + .map_err(PostError::from); } - let (b, p) = build(false); + let (body, parse) = build(false, request_model); match self - .post_openai(cfg, "/chat/completions", &b, effective_model) + .post_openai(cfg, "/chat/completions", &body, request_model) .await { - Ok(v) => p(v), - Err(e) if cfg.openai_api == OpenAiApi::Auto && self.try_upgrade(&e) => { - let (b, p) = build(true); - p(self - .post_openai(cfg, "/responses", &b, effective_model) - .await?) + Ok(value) => parse(value).map_err(PostError::from), + Err(PostError::Agent(error)) + if cfg.openai_api == OpenAiApi::Auto && self.try_upgrade(&error) => + { + let (body, parse) = build(true, request_model); + parse( + self.post_openai(cfg, "/responses", &body, request_model) + .await?, + ) + .map_err(PostError::from) } - Err(e) => Err(e), + Err(error) => Err(error), } } @@ -314,7 +586,8 @@ impl Llm { let (body, parse) = build(route); parse( self.post_openai(cfg, databricks_v2_path(route), &body, effective_model) - .await?, + .await + .map_err(PostError::into_agent)?, ) } @@ -329,7 +602,7 @@ impl Llm { path: &str, body: &Value, effective_model: &str, - ) -> Result { + ) -> Result { let (url, body_owned); let body_ref: &Value = match cfg.provider { Provider::Databricks => { @@ -354,13 +627,25 @@ impl Llm { // rejection can never suppress a later turn's legitimate retry. Both // statuses map to `LlmAuth` in `post`: a 403 is indistinguishable from // an expired-token 403 here, so we refresh once and let it propagate. - let mut bearer = self.auth.bearer().await?; + let mut bearer = self.auth.bearer().await.map_err(PostError::from)?; let mut refreshed = false; loop { - match post(&self.http, &url, body_ref, |r| r.bearer_auth(&bearer)).await { - Err(AgentError::LlmAuth(_)) if !refreshed => { + match post( + &self.http, + &url, + body_ref, + effective_model == MESH_VIRTUAL_MODEL_ID, + |r| r.bearer_auth(&bearer), + ) + .await + { + Err(PostError::Agent(AgentError::LlmAuth(_))) if !refreshed => { refreshed = true; - bearer = self.auth.refresh_now(&bearer).await?; + bearer = self + .auth + .refresh_now(&bearer) + .await + .map_err(PostError::from)?; } result => return result, } @@ -1049,12 +1334,71 @@ fn terminal_llm_error(elapsed: std::time::Duration, attempts: u32, detail: &str) )) } -async fn post(http: &Client, url: &str, body: &Value, apply: F) -> Result +/// Internal HTTP failure that preserves a mesh-specific MoA failure as a +/// typed signal. This covers both pre-inference eligibility rejection and a +/// structured `moa_failure` from workers/reducers. It is consumed inside +/// `openai_request`: an adaptive `auto` call falls back once, while an explicit +/// `mesh` call is surfaced as the ordinary LLM error callers already +/// understand. +#[derive(Debug)] +enum PostError { + Agent(AgentError), + MeshFallback(String), +} + +impl PostError { + fn into_agent(self) -> AgentError { + match self { + Self::Agent(error) => error, + Self::MeshFallback(detail) => AgentError::Llm(detail), + } + } +} + +impl From for PostError { + fn from(error: AgentError) -> Self { + Self::Agent(error) + } +} + +fn is_mesh_moa_unavailable_body(body: &str) -> bool { + serde_json::from_str::(body) + .ok() + .and_then(|value| { + value + .pointer("/error/message") + .and_then(Value::as_str) + .map(str::to_owned) + }) + .as_deref() + == Some(MESH_MOA_UNAVAILABLE_MESSAGE) +} + +fn is_mesh_moa_failure_body(body: &str) -> bool { + serde_json::from_str::(body) + .ok() + .and_then(|value| { + value + .pointer("/error/type") + .and_then(Value::as_str) + .map(str::to_owned) + }) + .as_deref() + == Some("moa_failure") +} + +async fn post( + http: &Client, + url: &str, + body: &Value, + detect_mesh_fallback: bool, + apply: F, +) -> Result where F: Fn(reqwest::RequestBuilder) -> reqwest::RequestBuilder, { - let body_bytes = - serde_json::to_vec(body).map_err(|e| AgentError::Llm(format!("serialize: {e}")))?; + let body_bytes = serde_json::to_vec(body) + .map_err(|e| PostError::Agent(AgentError::Llm(format!("serialize: {e}"))))?; let call_start = std::time::Instant::now(); for attempt in 0..MAX_RETRIES { let resp = match apply( @@ -1077,11 +1421,11 @@ where backoff_with_jitter(attempt).await; continue; } - return Err(terminal_llm_error( + return Err(PostError::Agent(terminal_llm_error( call_start.elapsed(), attempt + 1, &format!("transport: {e}"), - )); + ))); } }; let status = resp.status(); @@ -1093,9 +1437,19 @@ where // Not a stall path: auth failures are not surfaced through // terminal_llm_error, they resolve on the next call after refresh. if status == 401 || status == 403 { - return Err(AgentError::LlmAuth(read_error_body(resp).await)); + return Err(PostError::Agent(AgentError::LlmAuth( + read_error_body(resp).await, + ))); } if status.is_server_error() || status == 429 || status.as_u16() == 499 { + let body = read_error_body(resp).await; + let mesh_fallback = detect_mesh_fallback + && (status == reqwest::StatusCode::SERVICE_UNAVAILABLE + && is_mesh_moa_unavailable_body(&body) + || status.is_server_error() && is_mesh_moa_failure_body(&body)); + if mesh_fallback { + return Err(PostError::MeshFallback(format!("{status}: {body}"))); + } if attempt + 1 < MAX_RETRIES { tracing::warn!( attempt = attempt + 1, @@ -1106,33 +1460,32 @@ where backoff_with_jitter(attempt).await; continue; } - let body = read_error_body(resp).await; - return Err(terminal_llm_error( + return Err(PostError::Agent(terminal_llm_error( call_start.elapsed(), attempt + 1, &format!("exhausted retries: {status}: {body}"), - )); + ))); } // Not a stall path: the model is misconfigured, not the transport or // upstream capacity — no retry was attempted, so cumulative duration // would be misleading. if status == 404 { - return Err(AgentError::LlmModelNotFound(format!( + return Err(PostError::Agent(AgentError::LlmModelNotFound(format!( "{status}: {}", read_error_body(resp).await - ))); + )))); } if !status.is_success() { - return Err(AgentError::Llm(format!( + return Err(PostError::Agent(AgentError::Llm(format!( "{status}: {}", read_error_body(resp).await - ))); + )))); } if let Some(len) = resp.content_length() { if len as usize > MAX_LLM_RESPONSE_BYTES { - return Err(AgentError::Llm(format!( + return Err(PostError::Agent(AgentError::Llm(format!( "response too large: {len} > {MAX_LLM_RESPONSE_BYTES}" - ))); + )))); } } let mut buf: Vec = Vec::new(); @@ -1141,23 +1494,24 @@ where match stream.chunk().await { Ok(Some(chunk)) => { if buf.len() + chunk.len() > MAX_LLM_RESPONSE_BYTES { - return Err(AgentError::Llm(format!( + return Err(PostError::Agent(AgentError::Llm(format!( "response exceeded {MAX_LLM_RESPONSE_BYTES} bytes" - ))); + )))); } buf.extend_from_slice(&chunk); } Ok(None) => break, Err(e) => { - return Err(terminal_llm_error( + return Err(PostError::Agent(terminal_llm_error( call_start.elapsed(), attempt + 1, &format!("body read: {e}"), - )); + ))); } } } - return serde_json::from_slice(&buf).map_err(|e| AgentError::Llm(format!("json: {e}"))); + return serde_json::from_slice(&buf) + .map_err(|e| PostError::Agent(AgentError::Llm(format!("json: {e}")))); } unreachable!("loop always returns on its final iteration (attempt + 1 == MAX_RETRIES)"); } @@ -1249,11 +1603,602 @@ mod tests { base_url: "http://example.invalid".into(), anthropic_api_version: "2023-06-01".into(), openai_api: OpenAiApi::Chat, + prefer_mesh_for_auto: false, hints_enabled: true, thinking_effort: None, } } + #[derive(Debug, Clone)] + struct CapturedHttpRequest { + method: String, + path: String, + body: Option, + } + + #[derive(Debug)] + struct StubHttpResponse { + status: u16, + body: Value, + } + + impl StubHttpResponse { + fn ok(body: Value) -> Self { + Self { status: 200, body } + } + + fn error(status: u16, message: &str) -> Self { + Self { + status, + body: json!({ + "error": { + "message": message, + "type": "server_error", + "code": "service_unavailable", + } + }), + } + } + + fn moa_failure(status: u16, code: &str, message: &str) -> Self { + Self { + status, + body: json!({ + "choices": [{ + "finish_reason": "error", + "message": { "content": message, "role": "assistant" }, + }], + "error": { + "message": message, + "type": "moa_failure", + "code": code, + }, + "model": "mesh", + }), + } + } + } + + async fn spawn_sequence_stub( + responses: Vec, + ) -> (String, Arc>>) { + use std::collections::VecDeque; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base_url = format!("http://{}/v1", listener.local_addr().unwrap()); + let captured = Arc::new(Mutex::new(Vec::new())); + let captured_for_server = captured.clone(); + let responses = Arc::new(Mutex::new(VecDeque::from(responses))); + + tokio::spawn(async move { + loop { + let (mut socket, _) = match listener.accept().await { + Ok(connection) => connection, + Err(_) => return, + }; + let mut bytes = Vec::new(); + let mut chunk = [0u8; 4096]; + let header_end = loop { + if let Some(index) = bytes.windows(4).position(|window| window == b"\r\n\r\n") { + break index + 4; + } + match socket.read(&mut chunk).await { + Ok(0) | Err(_) => return, + Ok(read) => bytes.extend_from_slice(&chunk[..read]), + } + }; + let header_text = String::from_utf8_lossy(&bytes[..header_end]).into_owned(); + let content_length = header_text + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().ok()) + .flatten() + }) + .unwrap_or(0); + while bytes.len() < header_end + content_length { + match socket.read(&mut chunk).await { + Ok(0) | Err(_) => return, + Ok(read) => bytes.extend_from_slice(&chunk[..read]), + } + } + let mut request_line = header_text.lines().next().unwrap_or_default().split(' '); + let method = request_line.next().unwrap_or_default().to_string(); + let path = request_line.next().unwrap_or_default().to_string(); + let body = if content_length == 0 { + None + } else { + serde_json::from_slice(&bytes[header_end..header_end + content_length]).ok() + }; + captured_for_server + .lock() + .await + .push(CapturedHttpRequest { method, path, body }); + + let response = responses.lock().await.pop_front().unwrap_or_else(|| { + StubHttpResponse::error(500, "stub response sequence exhausted") + }); + let status_text = match response.status { + 200 => "OK", + 500 => "Internal Server Error", + 502 => "Bad Gateway", + 503 => "Service Unavailable", + other => panic!("unsupported stub status: {other}"), + }; + let body = response.body.to_string(); + let wire = format!( + "HTTP/1.1 {} {}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}", + response.status, + status_text, + body.len(), + body, + ); + let _ = socket.write_all(wire.as_bytes()).await; + let _ = socket.shutdown().await; + } + }); + (base_url, captured) + } + + fn model_catalog(ids: &[&str]) -> Value { + json!({ + "object": "list", + "data": ids + .iter() + .map(|id| json!({ "id": id, "object": "model" })) + .collect::>(), + }) + } + + fn chat_response(text: &str) -> Value { + json!({ + "choices": [{ + "finish_reason": "stop", + "message": { "content": text }, + }] + }) + } + + async fn complete_model( + llm: &Llm, + cfg: &Config, + model: &str, + ) -> Result { + llm.complete( + cfg, + "system", + &[HistoryItem::User("hello".into())], + &[], + model, + ) + .await + } + + async fn complete_model_with_tool( + llm: &Llm, + cfg: &Config, + model: &str, + ) -> Result { + llm.complete( + cfg, + "system", + &[HistoryItem::User("add two numbers".into())], + &[ToolDef { + name: "add_numbers".into(), + description: "Add two numbers".into(), + input_schema: json!({ + "type": "object", + "properties": { + "a": {"type": "number"}, + "b": {"type": "number"} + }, + "required": ["a", "b"] + }), + }], + model, + ) + .await + } + + async fn expire_mesh_catalog_check(llm: &Llm) { + llm.mesh_auto_state.lock().await.last_checked = + Some(Instant::now() - MESH_AUTO_CATALOG_TTL); + } + + async fn expire_mesh_auto_cooldown(llm: &Llm) { + let mut state = llm.mesh_auto_state.lock().await; + state.last_checked = Some(Instant::now() - MESH_AUTO_CATALOG_TTL); + state.cooldown_until = Some(Instant::now() - std::time::Duration::from_secs(1)); + } + + fn posted_models(requests: &[CapturedHttpRequest]) -> Vec<&str> { + requests + .iter() + .filter(|request| request.method == "POST") + .filter_map(|request| request.body.as_ref()?.get("model")?.as_str()) + .collect() + } + + #[tokio::test] + async fn mesh_auto_requires_two_stable_catalog_observations() { + let (base_url, captured) = spawn_sequence_stub(vec![ + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("direct")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("collective")), + ]) + .await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = true; + let llm = Llm::new(&config).unwrap(); + + assert_eq!( + complete_model(&llm, &config, "auto").await.unwrap().text, + "direct" + ); + expire_mesh_catalog_check(&llm).await; + assert_eq!( + complete_model(&llm, &config, "auto").await.unwrap().text, + "collective" + ); + + let requests = captured.lock().await; + assert_eq!(posted_models(&requests), vec!["auto", "mesh"]); + assert_eq!( + requests + .iter() + .filter(|request| request.path == "/v1/models") + .count(), + 2 + ); + } + + #[tokio::test] + async fn mesh_auto_does_not_enable_while_second_model_flaps() { + let (base_url, captured) = spawn_sequence_stub(vec![ + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("one")), + StubHttpResponse::ok(model_catalog(&["model-a"])), + StubHttpResponse::ok(chat_response("two")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("three")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("four")), + ]) + .await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = true; + let llm = Llm::new(&config).unwrap(); + + for _ in 0..4 { + complete_model(&llm, &config, "auto").await.unwrap(); + expire_mesh_catalog_check(&llm).await; + } + + let requests = captured.lock().await; + assert_eq!( + posted_models(&requests), + vec!["auto", "auto", "auto", "mesh"] + ); + } + + #[test] + fn collective_catalog_requires_two_distinct_physical_models() { + assert_eq!(mesh_catalog_supports_collective(&json!({})), None); + assert_eq!( + mesh_catalog_supports_collective(&model_catalog(&[])), + Some(false) + ); + assert_eq!( + mesh_catalog_supports_collective(&model_catalog(&["model-a", "mesh"])), + Some(false) + ); + assert_eq!( + mesh_catalog_supports_collective(&model_catalog(&[ + "org/model@main:Q4", + "org/model:Q4", + "mesh" + ])), + Some(false), + "two spellings of one model are not collective capacity" + ); + assert_eq!( + mesh_catalog_supports_collective(&model_catalog(&["model-a", "model-b", "mesh"])), + Some(true) + ); + } + + #[tokio::test] + async fn mesh_auto_tracks_models_appearing_disappearing_and_reappearing() { + let (base_url, captured) = spawn_sequence_stub(vec![ + StubHttpResponse::ok(model_catalog(&[])), + StubHttpResponse::ok(chat_response("zero")), + StubHttpResponse::ok(model_catalog(&["model-a", "mesh"])), + StubHttpResponse::ok(chat_response("one")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("two-first-observation")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("two-stable")), + StubHttpResponse::ok(model_catalog(&["model-a", "mesh"])), + StubHttpResponse::ok(chat_response("contracted")), + StubHttpResponse::ok(model_catalog(&[])), + StubHttpResponse::ok(chat_response("empty-again")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("rejoin-first-observation")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("rejoined-stable")), + ]) + .await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = true; + let llm = Llm::new(&config).unwrap(); + + for call in 0..8 { + if call == 5 { + expire_mesh_auto_cooldown(&llm).await; + } else if call > 0 { + expire_mesh_catalog_check(&llm).await; + } + complete_model(&llm, &config, "auto").await.unwrap(); + } + + let requests = captured.lock().await; + assert_eq!( + posted_models(&requests), + vec!["auto", "auto", "auto", "mesh", "auto", "auto", "auto", "mesh"] + ); + assert_eq!( + requests + .iter() + .filter(|request| request.path == "/v1/models") + .count(), + 8 + ); + } + + #[tokio::test] + async fn mesh_auto_falls_back_once_and_cools_down_when_mesh_contracts() { + let (base_url, captured) = spawn_sequence_stub(vec![ + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("warmup")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::error(503, MESH_MOA_UNAVAILABLE_MESSAGE), + StubHttpResponse::ok(chat_response("fallback")), + StubHttpResponse::ok(chat_response("cooldown")), + ]) + .await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = true; + let llm = Llm::new(&config).unwrap(); + + complete_model(&llm, &config, "auto").await.unwrap(); + expire_mesh_catalog_check(&llm).await; + assert_eq!( + complete_model(&llm, &config, "auto").await.unwrap().text, + "fallback" + ); + assert_eq!( + complete_model(&llm, &config, "auto").await.unwrap().text, + "cooldown" + ); + + let requests = captured.lock().await; + assert_eq!( + posted_models(&requests), + vec!["auto", "mesh", "auto", "auto"] + ); + assert_eq!( + requests + .iter() + .filter(|request| request.path == "/v1/models") + .count(), + 2, + "cooldown request must not re-probe the catalog" + ); + } + + #[tokio::test] + async fn mesh_auto_falls_back_once_when_moa_reducer_fails() { + let (base_url, captured) = spawn_sequence_stub(vec![ + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("warmup")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::moa_failure( + 502, + "all_reducers_failed", + "Reducer failed: unsupported chat template", + ), + StubHttpResponse::ok(chat_response("fallback")), + StubHttpResponse::ok(chat_response("cooldown")), + ]) + .await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = true; + let llm = Llm::new(&config).unwrap(); + + complete_model(&llm, &config, "auto").await.unwrap(); + expire_mesh_catalog_check(&llm).await; + assert_eq!( + complete_model(&llm, &config, "auto").await.unwrap().text, + "fallback" + ); + assert_eq!( + complete_model(&llm, &config, "auto").await.unwrap().text, + "cooldown" + ); + + let requests = captured.lock().await; + assert_eq!( + posted_models(&requests), + vec!["auto", "mesh", "auto", "auto"] + ); + assert_eq!( + requests + .iter() + .filter(|request| request.path == "/v1/models") + .count(), + 2, + "mesh-specific failure must enter cooldown without re-probing" + ); + } + + #[tokio::test] + async fn mesh_auto_retries_unstructured_tool_markup_through_auto() { + let (base_url, captured) = spawn_sequence_stub(vec![ + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response("warmup")), + StubHttpResponse::ok(model_catalog(&["model-a", "model-b", "mesh"])), + StubHttpResponse::ok(chat_response( + "<|tool_call>call:add_numbers{a:17,b:25}", + )), + StubHttpResponse::ok(chat_response("safe fallback")), + StubHttpResponse::ok(chat_response("cooldown")), + ]) + .await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = true; + let llm = Llm::new(&config).unwrap(); + + complete_model(&llm, &config, "auto").await.unwrap(); + expire_mesh_catalog_check(&llm).await; + assert_eq!( + complete_model_with_tool(&llm, &config, "auto") + .await + .unwrap() + .text, + "safe fallback" + ); + assert_eq!( + complete_model_with_tool(&llm, &config, "auto") + .await + .unwrap() + .text, + "cooldown" + ); + + let requests = captured.lock().await; + assert_eq!( + posted_models(&requests), + vec!["auto", "mesh", "auto", "auto"] + ); + assert_eq!( + requests + .iter() + .filter(|request| request.path == "/v1/models") + .count(), + 2, + "pseudo tool markup must enter cooldown without another catalog probe" + ); + } + + #[tokio::test] + async fn mesh_auto_catalog_failure_fails_open_to_auto() { + let (base_url, captured) = spawn_sequence_stub(vec![ + StubHttpResponse::error(500, "catalog unavailable"), + StubHttpResponse::ok(chat_response("direct")), + ]) + .await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = true; + let llm = Llm::new(&config).unwrap(); + + complete_model(&llm, &config, "auto").await.unwrap(); + let requests = captured.lock().await; + assert_eq!(posted_models(&requests), vec!["auto"]); + } + + #[tokio::test] + async fn generic_openai_auto_does_not_probe_or_select_mesh() { + let (base_url, captured) = + spawn_sequence_stub(vec![StubHttpResponse::ok(chat_response("provider auto"))]).await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = false; + let llm = Llm::new(&config).unwrap(); + + complete_model(&llm, &config, "auto").await.unwrap(); + let requests = captured.lock().await; + assert_eq!(posted_models(&requests), vec!["auto"]); + assert!(requests.iter().all(|request| request.path != "/v1/models")); + } + + #[tokio::test] + async fn explicit_models_are_never_rewritten_or_fallback_retried() { + let (base_url, captured) = spawn_sequence_stub(vec![StubHttpResponse::error( + 503, + MESH_MOA_UNAVAILABLE_MESSAGE, + )]) + .await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = true; + let llm = Llm::new(&config).unwrap(); + + let error = complete_model(&llm, &config, "mesh").await.unwrap_err(); + assert!(error.to_string().contains(MESH_MOA_UNAVAILABLE_MESSAGE)); + let requests = captured.lock().await; + assert_eq!(posted_models(&requests), vec!["mesh"]); + assert!(requests.iter().all(|request| request.path != "/v1/models")); + } + + #[tokio::test] + async fn explicit_real_model_bypasses_mesh_auto_policy() { + let (base_url, captured) = + spawn_sequence_stub(vec![StubHttpResponse::ok(chat_response("explicit"))]).await; + let mut config = cfg(Provider::OpenAi); + config.base_url = base_url; + config.prefer_mesh_for_auto = true; + let llm = Llm::new(&config).unwrap(); + + let response = complete_model(&llm, &config, "model-a").await.unwrap(); + assert_eq!(response.text, "explicit"); + let requests = captured.lock().await; + assert_eq!(posted_models(&requests), vec!["model-a"]); + assert!(requests.iter().all(|request| request.path != "/v1/models")); + } + + #[test] + fn mesh_unavailable_classifier_requires_exact_openai_error_message() { + assert!(is_mesh_moa_unavailable_body( + &StubHttpResponse::error(503, MESH_MOA_UNAVAILABLE_MESSAGE) + .body + .to_string() + )); + assert!(!is_mesh_moa_unavailable_body( + &StubHttpResponse::error(503, "some other outage") + .body + .to_string() + )); + } + + #[test] + fn mesh_failure_classifier_requires_structured_moa_failure_type() { + assert!(is_mesh_moa_failure_body( + &StubHttpResponse::moa_failure( + 502, + "all_reducers_failed", + "Reducer failed: unsupported chat template", + ) + .body + .to_string() + )); + assert!(!is_mesh_moa_failure_body( + &StubHttpResponse::error(502, "some other bad gateway") + .body + .to_string() + )); + } + fn image_history() -> Vec { vec![ HistoryItem::User("describe the image".into()), @@ -2226,7 +3171,7 @@ mod tests { .timeout(Duration::from_secs(5)) .build() .unwrap(); - let out = post(&client, &url, &serde_json::json!({}), |b| b) + let out = post(&client, &url, &serde_json::json!({}), false, |b| b) .await .expect("post should succeed after retry"); assert_eq!(out, serde_json::json!({ "ok": true })); @@ -2290,7 +3235,7 @@ mod tests { .timeout(Duration::from_secs(5)) .build() .unwrap(); - let out = post(&client, &url, &serde_json::json!({}), |b| b) + let out = post(&client, &url, &serde_json::json!({}), false, |b| b) .await .expect("post should succeed after 499 retry"); assert_eq!(out, serde_json::json!({ "ok": true })); @@ -2341,11 +3286,11 @@ mod tests { .timeout(Duration::from_secs(5)) .build() .unwrap(); - let err = post(&client, &url, &serde_json::json!({}), |b| b) + let err = post(&client, &url, &serde_json::json!({}), false, |b| b) .await .unwrap_err(); match &err { - AgentError::Llm(msg) => { + PostError::Agent(AgentError::Llm(msg)) => { assert!( msg.contains("exhausted retries") && msg.contains("499"), "expected 'exhausted retries' + '499' in error, got: {msg}" @@ -2355,7 +3300,7 @@ mod tests { "expected cumulative duration + exact attempt count, got: {msg}" ); } - other => panic!("expected AgentError::Llm, got: {other:?}"), + other => panic!("expected PostError::Agent(AgentError::Llm), got: {other:?}"), } assert_eq!( accepts.load(Ordering::SeqCst), @@ -2693,6 +3638,7 @@ mod tests { .build() .unwrap(), auto_upgraded: std::sync::atomic::AtomicBool::new(false), + mesh_auto_state: Mutex::new(MeshAutoState::default()), auth, } } @@ -2753,7 +3699,10 @@ mod tests { .post_openai(&c, "/v1/x", &json!({}), "model") .await .unwrap_err(); - assert!(matches!(err, AgentError::LlmAuth(_)), "got {err:?}"); + assert!( + matches!(err, PostError::Agent(AgentError::LlmAuth(_))), + "got {err:?}" + ); assert_eq!( auth.refreshes.load(Ordering::SeqCst), 1, @@ -2781,7 +3730,10 @@ mod tests { .post_openai(&c, "/v1/x", &json!({}), "model") .await .unwrap_err(); - assert!(matches!(err, AgentError::LlmAuth(_)), "got {err:?}"); + assert!( + matches!(err, PostError::Agent(AgentError::LlmAuth(_))), + "got {err:?}" + ); assert_eq!( auth.refreshes.load(Ordering::SeqCst), 1, diff --git a/desktop/src-tauri/Cargo.lock b/desktop/src-tauri/Cargo.lock index 080584c17f..228afc6efb 100644 --- a/desktop/src-tauri/Cargo.lock +++ b/desktop/src-tauri/Cargo.lock @@ -4935,7 +4935,7 @@ dependencies = [ [[package]] name = "mesh-llm-api-client" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "hex", "mesh-llm-client", @@ -4945,7 +4945,7 @@ dependencies = [ [[package]] name = "mesh-llm-api-server" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "mesh-llm-api-client", @@ -4956,12 +4956,12 @@ dependencies = [ [[package]] name = "mesh-llm-build-info" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" [[package]] name = "mesh-llm-client" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "async-trait", @@ -4993,7 +4993,7 @@ dependencies = [ [[package]] name = "mesh-llm-config" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "dirs", @@ -5009,7 +5009,7 @@ dependencies = [ [[package]] name = "mesh-llm-embedded-runtime" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "mesh-llm-host-runtime", @@ -5019,7 +5019,7 @@ dependencies = [ [[package]] name = "mesh-llm-events" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "clap", @@ -5031,7 +5031,7 @@ dependencies = [ [[package]] name = "mesh-llm-gpu-bench" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "cc", @@ -5044,7 +5044,7 @@ dependencies = [ [[package]] name = "mesh-llm-guardrails" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "serde", "serde_json", @@ -5053,7 +5053,7 @@ dependencies = [ [[package]] name = "mesh-llm-hardware-profile" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "mesh-llm-native-runtime", ] @@ -5061,7 +5061,7 @@ dependencies = [ [[package]] name = "mesh-llm-host-runtime" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "argon2", @@ -5154,7 +5154,7 @@ dependencies = [ [[package]] name = "mesh-llm-identity" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "argon2", "base64 0.22.1", @@ -5176,7 +5176,7 @@ dependencies = [ [[package]] name = "mesh-llm-native-runtime" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "serde", @@ -5187,7 +5187,7 @@ dependencies = [ [[package]] name = "mesh-llm-node" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "mesh-llm-types", @@ -5201,7 +5201,7 @@ dependencies = [ [[package]] name = "mesh-llm-plugin" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "async-trait", @@ -5218,7 +5218,7 @@ dependencies = [ [[package]] name = "mesh-llm-plugin-manager" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "dirs", @@ -5236,7 +5236,7 @@ dependencies = [ [[package]] name = "mesh-llm-protocol" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "hex", @@ -5249,7 +5249,7 @@ dependencies = [ [[package]] name = "mesh-llm-routing" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "iroh", ] @@ -5257,7 +5257,7 @@ dependencies = [ [[package]] name = "mesh-llm-runtime-install" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "dirs", @@ -5280,7 +5280,7 @@ dependencies = [ [[package]] name = "mesh-llm-sdk" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "mesh-llm-api-client", @@ -5295,7 +5295,7 @@ dependencies = [ [[package]] name = "mesh-llm-skills" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "dirs", @@ -5306,7 +5306,7 @@ dependencies = [ [[package]] name = "mesh-llm-system" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "chrono", @@ -5329,7 +5329,7 @@ dependencies = [ [[package]] name = "mesh-llm-types" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "hex", "serde", @@ -5340,12 +5340,12 @@ dependencies = [ [[package]] name = "mesh-llm-ui" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" [[package]] name = "mesh-mixture-of-agents" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "async-trait", "mesh-llm-guardrails", @@ -5418,7 +5418,7 @@ dependencies = [ [[package]] name = "model-artifact" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "async-trait", @@ -5429,7 +5429,7 @@ dependencies = [ [[package]] name = "model-hf" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "async-trait", @@ -5447,7 +5447,7 @@ dependencies = [ [[package]] name = "model-package" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "bytes", @@ -5467,7 +5467,7 @@ dependencies = [ [[package]] name = "model-ref" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "serde", ] @@ -5475,7 +5475,7 @@ dependencies = [ [[package]] name = "model-resolver" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "model-artifact", @@ -6504,7 +6504,7 @@ dependencies = [ [[package]] name = "openai-frontend" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "async-trait", "axum", @@ -9037,7 +9037,7 @@ checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649" [[package]] name = "skippy-cache" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "blake3", @@ -9047,7 +9047,7 @@ dependencies = [ [[package]] name = "skippy-coordinator" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "thiserror 2.0.18", ] @@ -9055,7 +9055,7 @@ dependencies = [ [[package]] name = "skippy-ffi" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "libloading 0.8.9", ] @@ -9063,12 +9063,12 @@ dependencies = [ [[package]] name = "skippy-metrics" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" [[package]] name = "skippy-protocol" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "prost", "prost-build", @@ -9079,7 +9079,7 @@ dependencies = [ [[package]] name = "skippy-runtime" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "libc", @@ -9093,7 +9093,7 @@ dependencies = [ [[package]] name = "skippy-server" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "anyhow", "async-trait", @@ -9121,7 +9121,7 @@ dependencies = [ [[package]] name = "skippy-topology" version = "0.73.1" -source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=c441ea7f328692b18b7f49ad819c7b1a603cbdcb#c441ea7f328692b18b7f49ad819c7b1a603cbdcb" +source = "git+https://github.com/Mesh-LLM/mesh-llm.git?rev=f455d493a2ae82baf2a326e2d0fda351433b4b30#f455d493a2ae82baf2a326e2d0fda351433b4b30" dependencies = [ "serde", "serde_json", diff --git a/desktop/src-tauri/Cargo.toml b/desktop/src-tauri/Cargo.toml index 0ed7417333..9df91999f6 100644 --- a/desktop/src-tauri/Cargo.toml +++ b/desktop/src-tauri/Cargo.toml @@ -90,14 +90,14 @@ buzz_persona_pkg = { package = "buzz-persona", path = "../../crates/buzz-persona buzz_sdk_pkg = { package = "buzz-sdk", path = "../../crates/buzz-sdk" } buzz_agent_pkg = { package = "buzz-agent", path = "../../crates/buzz-agent" } iroh = { version = "1.0.2", optional = true } -mesh-llm-sdk = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "c441ea7f328692b18b7f49ad819c7b1a603cbdcb", package = "mesh-llm-sdk", default-features = false, features = ["client", "serving"], optional = true } -mesh-llm-host-runtime = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "c441ea7f328692b18b7f49ad819c7b1a603cbdcb", package = "mesh-llm-host-runtime", default-features = false, features = ["dynamic-native-runtime"], optional = true } +mesh-llm-sdk = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "f455d493a2ae82baf2a326e2d0fda351433b4b30", package = "mesh-llm-sdk", default-features = false, features = ["client", "serving"], optional = true } +mesh-llm-host-runtime = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "f455d493a2ae82baf2a326e2d0fda351433b4b30", package = "mesh-llm-host-runtime", default-features = false, features = ["dynamic-native-runtime"], optional = true } # Model catalog + hardware survey for the Share-compute model picker (same # diagnose pattern as mesh-console). Lib name of mesh-llm-client is mesh_client. -mesh-llm-client = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "c441ea7f328692b18b7f49ad819c7b1a603cbdcb", package = "mesh-llm-client", optional = true } -mesh-llm-node = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "c441ea7f328692b18b7f49ad819c7b1a603cbdcb", package = "mesh-llm-node", optional = true } -mesh-llm-system = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "c441ea7f328692b18b7f49ad819c7b1a603cbdcb", package = "mesh-llm-system", optional = true } -mesh-llm-events = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "c441ea7f328692b18b7f49ad819c7b1a603cbdcb", package = "mesh-llm-events", optional = true } +mesh-llm-client = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "f455d493a2ae82baf2a326e2d0fda351433b4b30", package = "mesh-llm-client", optional = true } +mesh-llm-node = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "f455d493a2ae82baf2a326e2d0fda351433b4b30", package = "mesh-llm-node", optional = true } +mesh-llm-system = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "f455d493a2ae82baf2a326e2d0fda351433b4b30", package = "mesh-llm-system", optional = true } +mesh-llm-events = { git = "https://github.com/Mesh-LLM/mesh-llm.git", rev = "f455d493a2ae82baf2a326e2d0fda351433b4b30", package = "mesh-llm-events", optional = true } base64 = "0.22" sha2 = "0.11" tar = "0.4" diff --git a/desktop/src-tauri/src/commands/mesh_llm.rs b/desktop/src-tauri/src/commands/mesh_llm.rs index 87af9fa08f..305c54a203 100644 --- a/desktop/src-tauri/src/commands/mesh_llm.rs +++ b/desktop/src-tauri/src/commands/mesh_llm.rs @@ -1,5 +1,6 @@ use std::path::PathBuf; +use sha2::{Digest, Sha256}; use tauri::{AppHandle, Manager, State}; use crate::{app_state::AppState, mesh_llm, relay}; @@ -57,8 +58,85 @@ fn share_stop_should_teardown(mode: mesh_llm::MeshNodeMode) -> bool { matches!(mode, mesh_llm::MeshNodeMode::Serve) } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum MeshStartPlan { + Start, + RestartToReplaceClient, + RejectOccupied, +} + +fn mesh_start_plan( + requested_mode: mesh_llm::MeshNodeMode, + existing_mode: Option, +) -> MeshStartPlan { + match (requested_mode, existing_mode) { + (_, None) => MeshStartPlan::Start, + (mesh_llm::MeshNodeMode::Serve, Some(mesh_llm::MeshNodeMode::Client)) => { + MeshStartPlan::RestartToReplaceClient + } + _ => MeshStartPlan::RejectOccupied, + } +} + +fn sharing_config_from_request( + request: &mesh_llm::StartMeshNodeRequest, +) -> CmdResult { + let model_id = request + .model_id + .as_deref() + .map(str::trim) + .filter(|model_id| !model_id.is_empty()) + .ok_or_else(|| "modelId is required for serve mode".to_string())?; + Ok(MeshSharingConfig { + enabled: true, + model_id: model_id.to_string(), + max_vram_gb: request.max_vram_gb, + }) +} + +fn restarting_share_status(config: &MeshSharingConfig) -> mesh_llm::MeshNodeStatus { + mesh_llm::MeshNodeStatus { + state: mesh_llm::MeshNodeState::Starting, + mode: Some(mesh_llm::MeshNodeMode::Serve), + health: mesh_llm::MeshHealth { + status: mesh_llm::MeshHealthStatus::Degraded, + reason: Some("Buzz is restarting to switch this machine to sharing".to_string()), + }, + api_base_url: None, + console_url: None, + model_id: Some(config.model_id.clone()), + model_name: Some(config.model_id.clone()), + invite_token: None, + endpoint_id: None, + device_id: None, + device_name: None, + } +} + +fn restart_to_share( + app: &AppHandle, + config: &MeshSharingConfig, +) -> CmdResult { + save_mesh_sharing_config(app, config)?; + let status = restarting_share_status(config); + app.request_restart(); + Ok(status) +} + pub type CmdResult = Result; +fn buzz_mesh_name_for_relay(relay_url: &str) -> String { + let normalized = url::Url::parse(relay_url.trim()) + .map(|url| url.origin().ascii_serialization()) + .unwrap_or_else(|_| relay_url.trim().trim_end_matches('/').to_ascii_lowercase()); + let digest = hex::encode(Sha256::digest(normalized.as_bytes())); + format!("buzz-community-{}", &digest[..32]) +} + +fn buzz_mesh_name(state: &AppState) -> String { + buzz_mesh_name_for_relay(&relay::relay_ws_url_with_override(state)) +} + fn advance_mesh_status_cursor( filter: &mut serde_json::Value, page: &[nostr::Event], @@ -137,6 +215,81 @@ pub(crate) async fn resolve_trusted_owner_ids_or_self_only(state: &AppState) -> } } +/// Choose validated live endpoints from other runtimes in this Buzz community. +/// The stable relay-derived mesh name gives every runtime the same MeshLLM mesh +/// identity; these endpoints supply transport bootstrap only. +fn buzz_mesh_join_targets( + mut targets: Vec, + self_owner_id: &str, +) -> Vec { + targets.retain(|target| { + target.reporter_pubkey.is_some() + && target + .owner_id + .as_deref() + .is_some_and(|owner| !owner.eq_ignore_ascii_case(self_owner_id.trim())) + }); + targets.sort_by(|left, right| { + left.reporter_pubkey + .cmp(&right.reporter_pubkey) + .then_with(|| left.owner_id.cmp(&right.owner_id)) + .then_with(|| left.endpoint_id.cmp(&right.endpoint_id)) + .then_with(|| left.endpoint_addr.cmp(&right.endpoint_addr)) + .then_with(|| left.model_id.cmp(&right.model_id)) + }); + targets.dedup_by(|left, right| left.endpoint_addr == right.endpoint_addr); + targets +} + +/// Resolve the validated member endpoint this runtime should join to enter the +/// existing Buzz community mesh. `Ok(None)` means this machine is the first +/// live serving member (or is itself the shared bootstrap contact). +pub(crate) async fn resolve_buzz_mesh_join_targets( + state: &AppState, +) -> Result, String> { + let events = query_mesh_discovery_events(state).await?; + let self_owner_id = mesh_llm::ensure_owner_identity() + .map_err(|error| format!("failed to load mesh owner identity: {error}"))? + .owner_id; + Ok(buzz_mesh_join_targets( + mesh_llm::availability_from_events(events).serve_targets, + &self_owner_id, + )) +} + +/// Resolve the initial admission roster and bootstrap endpoint from one relay +/// snapshot. A node start used to repeat the full membership + status query +/// for each value, making Share Compute startup both slower and more exposed +/// to inconsistent snapshots. +async fn resolve_buzz_mesh_startup(state: &AppState) -> (Vec, Option) { + match query_mesh_discovery_events(state).await { + Ok(events) => { + let trusted_owner_ids = mesh_llm::owner_ids_from_events(&events); + let join_token = mesh_llm::ensure_owner_identity() + .ok() + .and_then(|identity| { + buzz_mesh_join_targets( + mesh_llm::availability_from_events(events).serve_targets, + &identity.owner_id, + ) + .into_iter() + .next() + }) + .map(|target| target.endpoint_addr); + (trusted_owner_ids, join_token) + } + Err(error) => { + // Initial startup fails closed to this runtime's own owner. Share + // Compute must still start for the first member and through a + // transient relay outage; the coordinator retries convergence. + eprintln!( + "buzz-mesh: startup discovery failed; allowing only this node and starting isolated for now: {error}" + ); + (Vec::new(), None) + } + } +} + pub(crate) async fn restore_mesh_sharing(app: &AppHandle, state: &AppState) -> CmdResult<()> { let Some(config) = load_mesh_sharing_config(app)? else { return Ok(()); @@ -144,6 +297,10 @@ pub(crate) async fn restore_mesh_sharing(app: &AppHandle, state: &AppState) -> C if !config.enabled || config.model_id.trim().is_empty() { return Ok(()); } + if state.mesh_llm_runtime.lock().await.is_some() { + return Ok(()); + } + let (trusted_owner_ids, join_token) = resolve_buzz_mesh_startup(state).await; let mut runtime = state.mesh_llm_runtime.lock().await; if runtime.is_some() { return Ok(()); @@ -152,8 +309,9 @@ pub(crate) async fn restore_mesh_sharing(app: &AppHandle, state: &AppState) -> C mode: mesh_llm::MeshNodeMode::Serve, model_id: Some(config.model_id), max_vram_gb: config.max_vram_gb, - join_token: None, - trusted_owner_ids: Some(resolve_trusted_owner_ids_or_self_only(state).await), + join_token, + mesh_name: Some(buzz_mesh_name(state)), + trusted_owner_ids: Some(trusted_owner_ids), }; let started = mesh_llm::DesktopMeshRuntime::start(request) .await @@ -170,37 +328,91 @@ pub async fn mesh_start_node( state: State<'_, AppState>, mut request: mesh_llm::StartMeshNodeRequest, ) -> CmdResult { - // 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_or_self_only(&state).await); + let sharing_config = if request.mode == mesh_llm::MeshNodeMode::Serve { + Some(sharing_config_from_request(&request)?) + } else { + None + }; + + // Never replace a client runtime in-process. Even a Ready SDK handle can + // finish `stop()` while its native listeners are still releasing :9337 + // and :3131; a pending client has no shutdown handle at all. Persist the + // requested serving configuration and switch roles across a controlled + // process restart, the only boundary that proves both ports are clean. + { + let runtime = state.mesh_llm_runtime.lock().await; + if let Some(existing) = runtime.as_ref() { + let plan = mesh_start_plan(request.mode, Some(existing.mode())); + match plan { + MeshStartPlan::RestartToReplaceClient => { + let config = sharing_config + .as_ref() + .ok_or_else(|| "serving configuration is unavailable".to_string())?; + drop(runtime); + return restart_to_share(&app, config); + } + MeshStartPlan::RejectOccupied => { + return Err("mesh node is already running".to_string()); + } + MeshStartPlan::Start => {} + } + } } + + // Frontend requests never carry a roster. Resolve it and the bootstrap + // endpoint from one snapshot so UI startup does not repeat relay probes. + if request.trusted_owner_ids.is_none() || request.join_token.is_none() { + let (trusted_owner_ids, join_token) = resolve_buzz_mesh_startup(&state).await; + request.trusted_owner_ids.get_or_insert(trusted_owner_ids); + if request.join_token.is_none() { + request.join_token = join_token; + } + } + request.mesh_name = Some(buzz_mesh_name(&state)); let mut runtime = state.mesh_llm_runtime.lock().await; - if runtime.is_some() { + + let plan = match runtime.as_ref() { + Some(existing) => mesh_start_plan(request.mode, Some(existing.mode())), + None => mesh_start_plan(request.mode, None), + }; + if plan == MeshStartPlan::RestartToReplaceClient { + let config = sharing_config + .as_ref() + .ok_or_else(|| "serving configuration is unavailable".to_string())?; + drop(runtime); + return restart_to_share(&app, config); + } + if plan == MeshStartPlan::RejectOccupied { return Err("mesh node is already running".to_string()); } - let saved_request = request.clone(); let started = mesh_llm::DesktopMeshRuntime::start(request) .await .map_err(|error| format!("{error:#}"))?; - let status = started - .status() - .await - .map_err(|error| format!("mesh node started but status probe failed: {error:#}"))?; + let status = match started.status().await { + Ok(status) => status, + Err(error) => { + let cleanup = started.stop().await; + if let Err(cleanup_error) = &cleanup { + eprintln!( + "buzz-mesh: started node status failed and cleanup was incomplete: {cleanup_error:#}" + ); + } + // The handle was never installed into AppState, so shutdown cannot + // see it again. Restart even when stop reported success: the + // process boundary guarantees native :9337/:3131 listeners cannot + // linger behind an untracked runtime. + drop(runtime); + app.request_restart(); + return Err(format!( + "mesh node started but status probe failed: {error:#}; Buzz is restarting to guarantee cleanup" + )); + } + }; *runtime = Some(started); drop(runtime); - if saved_request.mode == mesh_llm::MeshNodeMode::Serve { - if let Some(model_id) = saved_request.model_id.as_deref() { - save_mesh_sharing_config( - &app, - &MeshSharingConfig { - enabled: true, - model_id: model_id.to_string(), - max_vram_gb: saved_request.max_vram_gb, - }, - )?; - } + if let Some(config) = sharing_config.as_ref() { + save_mesh_sharing_config(&app, config)?; } mesh_llm::publish_current_status_once(&app, "start").await; Ok(status) @@ -398,12 +610,21 @@ pub(crate) async fn ensure_client_node_for_model( mode: mesh_llm::MeshNodeMode::Client, model_id: None, max_vram_gb: None, - join_token: Some(join_token), + join_token: Some(join_token.clone()), + mesh_name: Some(buzz_mesh_name(state)), 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() { - return Err("mesh node changed while starting Buzz shared compute client".to_string()); + if let Some(existing) = runtime.as_ref() { + // Another GUI agent may have won the startup race while this caller + // was resolving membership. The runtime is machine-scoped, not + // agent-scoped: join its selected endpoint into the existing node and + // let every caller reuse the same local ingress. + existing + .dial_endpoint_addr(join_token) + .await + .map_err(|error| format!("mesh dial failed: {error}"))?; + return existing.status().await.map_err(|error| error.to_string()); } let started = mesh_llm::DesktopMeshRuntime::start(start) .await @@ -644,266 +865,5 @@ pub async fn mesh_model_catalog() -> CmdResult { } #[cfg(all(test, feature = "mesh-llm"))] -mod tests { - use super::*; - use crate::app_state::build_app_state; - - fn target(model_id: &str, endpoint_addr: &str) -> mesh_llm::MeshServeTarget { - mesh_llm::MeshServeTarget { - model_id: model_id.to_string(), - model_name: None, - endpoint_addr: endpoint_addr.to_string(), - node_name: None, - capacity: None, - endpoint_id: None, - device_id: None, - device_name: None, - } - } - - #[test] - fn readiness_failure_is_catalog_sync_when_model_never_visible() { - assert_eq!( - classify_mesh_readiness_failure(false), - MeshReadinessFailure::CatalogNeverSynced - ); - } - - #[test] - fn readiness_failure_is_routing_when_model_was_visible() { - assert_eq!( - classify_mesh_readiness_failure(true), - MeshReadinessFailure::RoutingNeverCompleted - ); - } - - #[test] - fn readiness_messages_are_distinct_and_actionable() { - let catalog = mesh_readiness_failure_message( - MeshReadinessFailure::CatalogNeverSynced, - "auto", - "HTTP 429", - ); - let routing = mesh_readiness_failure_message( - MeshReadinessFailure::RoutingNeverCompleted, - "auto", - "HTTP 503", - ); - // Distinct diagnoses, each names the model and carries the raw detail. - assert_ne!(catalog, routing); - assert!(catalog.contains("network path")); - assert!(catalog.contains("HTTP 429")); - assert!(routing.contains("did not complete")); - assert!(routing.contains("HTTP 503")); - } - - #[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![ - target("model-a", "addr-a"), - target("model-b", "addr-b1"), - target("model-b", "addr-b2"), - ]; - // Matches by model id and returns the first such target. - assert_eq!( - pick_serve_target_for_model(targets, "model-b").map(|t| t.endpoint_addr), - Some("addr-b1".to_string()) - ); - } - - #[test] - fn pick_serve_target_normalizes_main_revision() { - let targets = vec![target("org/model@main:q4", "addr")]; - assert_eq!( - pick_serve_target_for_model(targets, "org/model:q4").map(|target| target.endpoint_addr), - Some("addr".to_string()) - ); - } - - #[test] - fn pick_serve_target_auto_takes_any_live_target() { - let targets = vec![target("model-a", "addr-a"), target("model-b", "addr-b")]; - // "auto" delegates model choice to the mesh router; any live target - // is a valid bootstrap peer (first one wins). - assert_eq!( - pick_serve_target_for_model(targets, crate::mesh_llm::AUTO_MODEL_ID) - .map(|t| t.endpoint_addr), - Some("addr-a".to_string()) - ); - // But auto with zero live targets still falls closed. - assert_eq!( - pick_serve_target_for_model(Vec::new(), crate::mesh_llm::AUTO_MODEL_ID), - None - ); - } - - #[test] - fn pick_serve_target_none_when_model_not_hosted() { - let targets = vec![target("model-a", "addr-a")]; - // No live target serves this model -> caller falls closed. - assert_eq!(pick_serve_target_for_model(targets, "model-missing"), None); - } - - #[test] - fn share_stop_tears_down_serve_but_not_client() { - // Stopping "Share compute" tears down a serve node (we were sharing) - // but must leave a client node alone (we are consuming a peer). This is - // the backend half of the toggle-on regression: a client node occupies - // the single slot and reports state:"running", and the stop path must - // not kill it. - assert!( - share_stop_should_teardown(mesh_llm::MeshNodeMode::Serve), - "serve node is our sharing runtime; stop must tear it down" - ); - assert!( - !share_stop_should_teardown(mesh_llm::MeshNodeMode::Client), - "client node is a consume session; stop must NOT tear it down" - ); - } - - #[test] - fn client_status_serializes_with_running_state_and_client_mode() { - // Contract pin for the TS mock (e2eBridge.ts) and the frontend - // predicate: a consuming node serializes as - // {"state":"running","mode":"client"}. If serde renaming drifts, the - // hand-written mock shape and `deriveMeshShareToggle` would silently - // stop matching the real IPC payload. - let status = mesh_llm::MeshNodeStatus { - state: mesh_llm::MeshNodeState::Running, - mode: Some(mesh_llm::MeshNodeMode::Client), - // `MeshHealth::ok()` is module-private; build via the public fields. - health: mesh_llm::MeshHealth { - status: mesh_llm::MeshHealthStatus::Ok, - reason: None, - }, - api_base_url: Some("http://127.0.0.1:9337/v1".to_string()), - console_url: None, - model_id: None, - model_name: None, - invite_token: None, - endpoint_id: None, - device_id: None, - device_name: None, - }; - let value = serde_json::to_value(&status).expect("serialize mesh status"); - assert_eq!(value["state"], serde_json::json!("running")); - assert_eq!(value["mode"], serde_json::json!("client")); - } - - #[tokio::test] - async fn cold_client_preflight_requires_explicit_target() { - let state = build_app_state(); - let error = ensure_client_node_for_model(&state, "demo/model", None) - .await - .expect_err("cold relay-mesh preflight must not auto-pick a target"); - assert_eq!(error, RELAY_MESH_RUNTIME_NO_TARGET); - } - - /// Acceptance-critical regression for dropping the serve-vs-client guard. - /// - /// Before this change, `ensure_client_node_for_model` hard-errored whenever - /// the running runtime was in `Serve` mode ("stop sharing before using - /// Buzz shared compute as a client"). That forbade exactly what a user should be - /// able to do: host model A while pointing an agent at a different model B - /// through the same `9337` ingress. - /// - /// This test starts a real serve runtime and asserts that a follow-up - /// preflight for a *different* model and no explicit target still reuses the - /// existing runtime. Cold starts without a target are rejected before mesh-llm - /// startup; running runtimes are already joined to whatever target the - /// frontend selected earlier. - /// - /// Hardware-gated (`#[ignore]`): loads a real model. Run with: - /// cargo test -p buzz-desktop --features mesh-llm \ - /// ensure_serve_runtime_serves_other_model -- --ignored --nocapture - #[test] - #[ignore = "loads a real model; run manually with --ignored"] - fn ensure_serve_runtime_serves_other_model() { - std::thread::Builder::new() - .name("mesh-hardware-acceptance".to_string()) - .stack_size(mesh_llm::MESH_WORKER_STACK_SIZE) - .spawn(|| { - let runtime = tokio::runtime::Builder::new_multi_thread() - .worker_threads(2) - .thread_stack_size(mesh_llm::MESH_WORKER_STACK_SIZE) - .enable_all() - .build() - .expect("build mesh acceptance runtime"); - runtime.block_on(async { - const HOSTED_MODEL: &str = "jc-builds/SmolLM2-135M-Instruct-Q4_K_M-GGUF:Q4_K_M"; - const OTHER_MODEL: &str = "some/other-model-not-hosted-locally:Q4_K_M"; - - let state = build_app_state(); - - // Start a serve runtime hosting HOSTED_MODEL — this is the "Share - // compute" path. - let serve = - mesh_llm::DesktopMeshRuntime::start(mesh_llm::StartMeshNodeRequest { - mode: mesh_llm::MeshNodeMode::Serve, - model_id: Some(HOSTED_MODEL.to_string()), - max_vram_gb: None, - join_token: None, - trusted_owner_ids: None, - }) - .await - .expect("serve runtime should start"); - - let serve_status = serve.status().await.expect("serve status"); - let serve_base = serve_status.api_base_url.clone(); - assert_eq!(serve_status.mode, Some(mesh_llm::MeshNodeMode::Serve)); - - { - let mut runtime = state.mesh_llm_runtime.lock().await; - *runtime = Some(serve); - } - - // Preflight for a DIFFERENT model with no explicit target. Old code: - // Err(...sharing compute...). New code: reuse the running ingress. - let status = ensure_client_node_for_model(&state, OTHER_MODEL, None) - .await - .expect("serve runtime must not reject a different-model preflight"); - - // It returns the SAME running node — agents keep using A's 9337, and - // the router decides routability for OTHER_MODEL per request. - assert_eq!( - status.mode, - Some(mesh_llm::MeshNodeMode::Serve), - "preflight should reuse the existing serve runtime, not spin up a client" - ); - assert_eq!( - status.api_base_url, serve_base, - "agent must be pointed at the existing serve node's ingress" - ); - - // Clean up the runtime. - let taken = state.mesh_llm_runtime.lock().await.take(); - if let Some(runtime) = taken { - let _ = runtime.stop().await; - } - }); - }) - .expect("spawn mesh acceptance thread") - .join() - .expect("mesh acceptance thread panicked"); - } -} +#[path = "mesh_llm_tests.rs"] +mod tests; diff --git a/desktop/src-tauri/src/commands/mesh_llm_tests.rs b/desktop/src-tauri/src/commands/mesh_llm_tests.rs new file mode 100644 index 0000000000..ccc5287d62 --- /dev/null +++ b/desktop/src-tauri/src/commands/mesh_llm_tests.rs @@ -0,0 +1,499 @@ +use super::*; +use crate::app_state::build_app_state; + +fn target(model_id: &str, endpoint_addr: &str) -> mesh_llm::MeshServeTarget { + mesh_llm::MeshServeTarget { + model_id: model_id.to_string(), + model_name: None, + endpoint_addr: endpoint_addr.to_string(), + reporter_pubkey: None, + owner_id: None, + node_name: None, + capacity: None, + endpoint_id: None, + device_id: None, + device_name: None, + } +} + +fn reported_target( + reporter_pubkey: &str, + model_id: &str, + endpoint_addr: &str, +) -> mesh_llm::MeshServeTarget { + let mut target = target(model_id, endpoint_addr); + target.reporter_pubkey = Some(reporter_pubkey.to_string()); + target.owner_id = Some(reporter_pubkey.to_string()); + target +} + +#[test] +fn buzz_mesh_join_uses_the_same_live_member_from_every_other_node() { + let targets = vec![ + reported_target("member-c", "model-c", "addr-c"), + reported_target("member-a", "model-a", "addr-a"), + reported_target("member-b", "model-b", "addr-b"), + ]; + + assert_eq!( + buzz_mesh_join_targets(targets.clone(), "member-b") + .into_iter() + .next() + .map(|target| target.endpoint_addr), + Some("addr-a".to_string()) + ); + assert_eq!( + buzz_mesh_join_targets(targets, "member-c") + .into_iter() + .next() + .map(|target| target.endpoint_addr), + Some("addr-a".to_string()) + ); +} + +#[test] +fn buzz_mesh_bootstrap_member_does_not_dial_itself() { + let targets = vec![ + reported_target("member-b", "model-b", "addr-b"), + reported_target("member-a", "model-a", "addr-a"), + ]; + + assert_eq!( + buzz_mesh_join_targets(targets, "MEMBER-A") + .into_iter() + .next(), + Some(reported_target("member-b", "model-b", "addr-b")) + ); +} + +#[test] +fn buzz_mesh_join_ignores_targets_without_a_validated_reporter() { + let targets = vec![ + target("unbound-model", "unbound-addr"), + reported_target("member-b", "model-b", "addr-b"), + ]; + + assert_eq!( + buzz_mesh_join_targets(targets, "member-c") + .into_iter() + .next() + .map(|target| target.endpoint_addr), + Some("addr-b".to_string()) + ); +} + +#[test] +fn buzz_mesh_join_keeps_other_device_with_the_same_member_key() { + let mut self_target = reported_target("same-member", "model-a", "self-addr"); + self_target.owner_id = Some("owner-self".to_string()); + let mut other_device = reported_target("same-member", "model-b", "other-addr"); + other_device.owner_id = Some("owner-other".to_string()); + + assert_eq!( + buzz_mesh_join_targets(vec![self_target, other_device], "owner-self") + .into_iter() + .next() + .map(|target| target.endpoint_addr), + Some("other-addr".to_string()) + ); +} + +#[test] +fn buzz_mesh_name_is_stable_and_does_not_expose_the_relay() { + let first = buzz_mesh_name_for_relay("WSS://EXAMPLE.COM/"); + let second = buzz_mesh_name_for_relay("wss://example.com:443/some/path?ignored=yes"); + let other_relay = buzz_mesh_name_for_relay("wss://other.example.com"); + + assert_eq!(first, second); + assert_ne!(first, other_relay); + assert!(first.starts_with("buzz-community-")); + assert!(!first.contains("example")); +} + +#[test] +fn readiness_failure_is_catalog_sync_when_model_never_visible() { + assert_eq!( + classify_mesh_readiness_failure(false), + MeshReadinessFailure::CatalogNeverSynced + ); +} + +#[test] +fn readiness_failure_is_routing_when_model_was_visible() { + assert_eq!( + classify_mesh_readiness_failure(true), + MeshReadinessFailure::RoutingNeverCompleted + ); +} + +#[test] +fn readiness_messages_are_distinct_and_actionable() { + let catalog = mesh_readiness_failure_message( + MeshReadinessFailure::CatalogNeverSynced, + "auto", + "HTTP 429", + ); + let routing = mesh_readiness_failure_message( + MeshReadinessFailure::RoutingNeverCompleted, + "auto", + "HTTP 503", + ); + // Distinct diagnoses, each names the model and carries the raw detail. + assert_ne!(catalog, routing); + assert!(catalog.contains("network path")); + assert!(catalog.contains("HTTP 429")); + assert!(routing.contains("did not complete")); + assert!(routing.contains("HTTP 503")); +} + +#[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![ + target("model-a", "addr-a"), + target("model-b", "addr-b1"), + target("model-b", "addr-b2"), + ]; + // Matches by model id and returns the first such target. + assert_eq!( + pick_serve_target_for_model(targets, "model-b").map(|t| t.endpoint_addr), + Some("addr-b1".to_string()) + ); +} + +#[test] +fn pick_serve_target_normalizes_main_revision() { + let targets = vec![target("org/model@main:q4", "addr")]; + assert_eq!( + pick_serve_target_for_model(targets, "org/model:q4").map(|target| target.endpoint_addr), + Some("addr".to_string()) + ); +} + +#[test] +fn pick_serve_target_auto_takes_any_live_target() { + let targets = vec![target("model-a", "addr-a"), target("model-b", "addr-b")]; + // "auto" delegates model choice to the mesh router; any live target + // is a valid bootstrap peer (first one wins). + assert_eq!( + pick_serve_target_for_model(targets, crate::mesh_llm::AUTO_MODEL_ID) + .map(|t| t.endpoint_addr), + Some("addr-a".to_string()) + ); + // But auto with zero live targets still falls closed. + assert_eq!( + pick_serve_target_for_model(Vec::new(), crate::mesh_llm::AUTO_MODEL_ID), + None + ); +} + +#[test] +fn pick_serve_target_none_when_model_not_hosted() { + let targets = vec![target("model-a", "addr-a")]; + // No live target serves this model -> caller falls closed. + assert_eq!(pick_serve_target_for_model(targets, "model-missing"), None); +} + +#[test] +fn share_stop_tears_down_serve_but_not_client() { + // Stopping "Share compute" tears down a serve node (we were sharing) + // but must leave a client node alone (we are consuming a peer). This is + // the backend half of the toggle-on regression: a client node occupies + // the single slot and reports state:"running", and the stop path must + // not kill it. + assert!( + share_stop_should_teardown(mesh_llm::MeshNodeMode::Serve), + "serve node is our sharing runtime; stop must tear it down" + ); + assert!( + !share_stop_should_teardown(mesh_llm::MeshNodeMode::Client), + "client node is a consume session; stop must NOT tear it down" + ); +} + +#[test] +fn share_start_restarts_to_replace_only_client_runtimes() { + assert_eq!( + mesh_start_plan(mesh_llm::MeshNodeMode::Serve, None), + MeshStartPlan::Start + ); + assert_eq!( + mesh_start_plan( + mesh_llm::MeshNodeMode::Serve, + Some(mesh_llm::MeshNodeMode::Client), + ), + MeshStartPlan::RestartToReplaceClient + ); + assert_eq!( + mesh_start_plan( + mesh_llm::MeshNodeMode::Serve, + Some(mesh_llm::MeshNodeMode::Serve), + ), + MeshStartPlan::RejectOccupied + ); + assert_eq!( + mesh_start_plan( + mesh_llm::MeshNodeMode::Client, + Some(mesh_llm::MeshNodeMode::Client), + ), + MeshStartPlan::RejectOccupied + ); +} + +#[test] +fn client_status_serializes_with_running_state_and_client_mode() { + // Contract pin for the TS mock (e2eBridge.ts) and the frontend + // predicate: a consuming node serializes as + // {"state":"running","mode":"client"}. If serde renaming drifts, the + // hand-written mock shape and `deriveMeshShareToggle` would silently + // stop matching the real IPC payload. + let status = mesh_llm::MeshNodeStatus { + state: mesh_llm::MeshNodeState::Running, + mode: Some(mesh_llm::MeshNodeMode::Client), + // `MeshHealth::ok()` is module-private; build via the public fields. + health: mesh_llm::MeshHealth { + status: mesh_llm::MeshHealthStatus::Ok, + reason: None, + }, + api_base_url: Some("http://127.0.0.1:9337/v1".to_string()), + console_url: None, + model_id: None, + model_name: None, + invite_token: None, + endpoint_id: None, + device_id: None, + device_name: None, + }; + let value = serde_json::to_value(&status).expect("serialize mesh status"); + assert_eq!(value["state"], serde_json::json!("running")); + assert_eq!(value["mode"], serde_json::json!("client")); +} + +#[tokio::test] +async fn cold_client_preflight_requires_explicit_target() { + let state = build_app_state(); + let error = ensure_client_node_for_model(&state, "demo/model", None) + .await + .expect_err("cold relay-mesh preflight must not auto-pick a target"); + assert_eq!(error, RELAY_MESH_RUNTIME_NO_TARGET); +} + +/// Acceptance-critical regression for dropping the serve-vs-client guard. +/// +/// Before this change, `ensure_client_node_for_model` hard-errored whenever +/// the running runtime was in `Serve` mode ("stop sharing before using +/// Buzz shared compute as a client"). That forbade exactly what a user should be +/// able to do: host model A while pointing an agent at a different model B +/// through the same `9337` ingress. +/// +/// This test starts a real serve runtime and asserts that a follow-up +/// preflight for a *different* model and no explicit target still reuses the +/// existing runtime. Cold starts without a target are rejected before mesh-llm +/// startup; running runtimes are already joined to whatever target the +/// frontend selected earlier. +/// +/// Hardware-gated (`#[ignore]`): loads a real model. Run with: +/// cargo test -p buzz-desktop --features mesh-llm \ +/// ensure_serve_runtime_serves_other_model -- --ignored --nocapture +#[test] +#[ignore = "loads a real model; run manually with --ignored"] +fn ensure_serve_runtime_serves_other_model() { + std::thread::Builder::new() + .name("mesh-hardware-acceptance".to_string()) + .stack_size(mesh_llm::MESH_WORKER_STACK_SIZE) + .spawn(|| { + let runtime = tokio::runtime::Builder::new_multi_thread() + .worker_threads(2) + .thread_stack_size(mesh_llm::MESH_WORKER_STACK_SIZE) + .enable_all() + .build() + .expect("build mesh acceptance runtime"); + runtime.block_on(async { + const DEFAULT_HOSTED_MODEL: &str = + "jc-builds/SmolLM2-135M-Instruct-Q4_K_M-GGUF:Q4_K_M"; + const OTHER_MODEL: &str = "some/other-model-not-hosted-locally:Q4_K_M"; + let hosted_model = std::env::var("BUZZ_MESH_TEST_MODEL") + .unwrap_or_else(|_| DEFAULT_HOSTED_MODEL.to_string()); + + let state = build_app_state(); + + // Start a serve runtime hosting HOSTED_MODEL — this is the "Share + // compute" path. + let serve = + mesh_llm::DesktopMeshRuntime::start(mesh_llm::StartMeshNodeRequest { + mode: mesh_llm::MeshNodeMode::Serve, + model_id: Some(hosted_model.clone()), + max_vram_gb: None, + join_token: None, + mesh_name: None, + trusted_owner_ids: None, + }) + .await + .expect("serve runtime should start"); + + let serve_status = serve.status().await.expect("serve status"); + let serve_base = serve_status + .api_base_url + .clone() + .expect("serve runtime must expose its local API base"); + assert_eq!(serve_status.mode, Some(mesh_llm::MeshNodeMode::Serve)); + + { + let mut runtime = state.mesh_llm_runtime.lock().await; + *runtime = Some(serve); + } + + // Concurrent GUI agent starts all reuse this one machine-scoped + // runtime. None may create a per-agent client/serve node. + let preflights = tokio::join!( + ensure_client_node_for_model(&state, OTHER_MODEL, None), + ensure_client_node_for_model(&state, OTHER_MODEL, None), + ensure_client_node_for_model(&state, OTHER_MODEL, None), + ensure_client_node_for_model(&state, OTHER_MODEL, None), + ); + let statuses = [ + preflights.0, + preflights.1, + preflights.2, + preflights.3, + ] + .into_iter() + .collect::, _>>() + .expect("all agent preflights must reuse the serve runtime"); + + // It returns the SAME running node — agents keep using A's 9337, and + // the router decides routability for OTHER_MODEL per request. + for status in statuses { + assert_eq!( + status.mode, + Some(mesh_llm::MeshNodeMode::Serve), + "preflight should reuse the existing serve runtime, not spin up a client" + ); + assert_eq!( + status.api_base_url.as_deref(), + Some(serve_base.as_str()), + "every agent must be pointed at the machine's existing ingress" + ); + } + + // A standalone Share Compute runtime must advertise exactly + // one physical model and serve inference through `auto`. + let http = reqwest::Client::new(); + let catalog_deadline = + tokio::time::Instant::now() + std::time::Duration::from_secs(120); + let catalog = loop { + let body = http + .get(format!("{serve_base}/models")) + .send() + .await + .expect("query single-node catalog") + .error_for_status() + .expect("single-node catalog status") + .json::() + .await + .expect("parse single-node catalog"); + let physical_count = body["data"] + .as_array() + .map(|models| { + models + .iter() + .filter_map(|model| model["id"].as_str()) + .filter(|model| *model != "mesh") + .count() + }) + .unwrap_or_default(); + if physical_count > 0 { + break body; + } + assert!( + tokio::time::Instant::now() < catalog_deadline, + "single Share Compute model never became ready: {body}" + ); + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + }; + let physical_models = catalog["data"] + .as_array() + .expect("catalog data") + .iter() + .filter_map(|model| model["id"].as_str()) + .filter(|model| *model != "mesh") + .collect::>(); + assert_eq!( + physical_models.len(), + 1, + "single Share Compute node must advertise one physical model: {catalog}" + ); + assert!( + physical_models[0].contains(hosted_model.split(':').next().unwrap_or("")), + "catalog must contain the hosted model: {catalog}" + ); + + let inference_deadline = + tokio::time::Instant::now() + std::time::Duration::from_secs(120); + let inference = loop { + let response = http + .post(format!("{serve_base}/chat/completions")) + .json(&serde_json::json!({ + "model": "auto", + "messages": [{ + "role": "user", + "content": "Reply with exactly BUZZ_SINGLE_SHARE_OK and nothing else." + }], + "max_tokens": 512, + "temperature": 0 + })) + .send() + .await + .expect("single-node auto inference"); + if response.status().is_success() { + break response + .json::() + .await + .expect("parse single-node inference"); + } + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + assert!( + tokio::time::Instant::now() < inference_deadline, + "single-node auto inference never became ready: HTTP {status}: {body}" + ); + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + }; + let answer = inference["choices"][0]["message"]["content"] + .as_str() + .unwrap_or_default(); + assert!( + !answer.trim().is_empty(), + "single-node auto must produce visible output: {inference}" + ); + + // Clean up the runtime. + let taken = state.mesh_llm_runtime.lock().await.take(); + if let Some(runtime) = taken { + let _ = runtime.stop().await; + } + }); + }) + .expect("spawn mesh acceptance thread") + .join() + .expect("mesh acceptance thread panicked"); +} diff --git a/desktop/src-tauri/src/managed_agents/relay_mesh.rs b/desktop/src-tauri/src/managed_agents/relay_mesh.rs index 934155e881..7a6f5b094a 100644 --- a/desktop/src-tauri/src/managed_agents/relay_mesh.rs +++ b/desktop/src-tauri/src/managed_agents/relay_mesh.rs @@ -2,6 +2,8 @@ pub const RELAY_MESH_API_BASE_URL: &str = "http://127.0.0.1:9337/v1"; pub const RELAY_MESH_API_KEY_PLACEHOLDER: &str = "buzz-mesh-local"; pub const RELAY_MESH_PROVIDER_ID: &str = "relay-mesh"; pub const RELAY_MESH_AUTO_MODEL_ID: &str = "auto"; +#[cfg(feature = "mesh-llm")] +pub const RELAY_MESH_PREFER_MESH_FOR_AUTO_ENV: &str = "BUZZ_AGENT_PREFER_MESH_FOR_AUTO"; /// Translate the native Buzz shared compute provider into the OpenAI-compatible /// transport understood by buzz-agent. These are derived runtime details, not @@ -32,6 +34,14 @@ pub fn apply_relay_mesh_env( RELAY_MESH_API_KEY_PLACEHOLDER.to_string(), ); env.insert("OPENAI_COMPAT_API".to_string(), "chat".to_string()); + // Buzz owns the meaning of relay-mesh `auto`: buzz-agent dynamically uses + // mesh-llm's virtual Mixture-of-Agents model whenever the live catalog says + // at least two distinct models are available, and otherwise keeps the + // router's normal single-model `auto` behavior. + env.insert( + RELAY_MESH_PREFER_MESH_FOR_AUTO_ENV.to_string(), + "1".to_string(), + ); // Keep the requested response inside smaller local-model context windows, // and spend that budget on an answer/tool call instead of hidden reasoning. // Without both settings Qwen3 either fails the router's fit check at the @@ -66,5 +76,10 @@ mod tests { env.get("BUZZ_AGENT_THINKING_EFFORT").map(String::as_str), Some("none") ); + assert_eq!( + env.get(RELAY_MESH_PREFER_MESH_FOR_AUTO_ENV) + .map(String::as_str), + Some("1") + ); } } diff --git a/desktop/src-tauri/src/mesh_llm/coordinator.rs b/desktop/src-tauri/src/mesh_llm/coordinator.rs index 930149d0a5..1e279353b6 100644 --- a/desktop/src-tauri/src/mesh_llm/coordinator.rs +++ b/desktop/src-tauri/src/mesh_llm/coordinator.rs @@ -27,11 +27,17 @@ const STATUS_PUBLISH_TIMEOUT: Duration = Duration::from_secs(10); /// not hammer discovery every tick, but a recovered peer is noticed quickly. const INGRESS_WATCHDOG_BASE: Duration = Duration::from_secs(15); const INGRESS_WATCHDOG_MAX: Duration = Duration::from_secs(120); +/// A Share Compute node may start before another member's signed status reaches +/// the relay. Recheck promptly so simultaneous starts converge into one Buzz +/// 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<()>, _ingress_watchdog: tokio::task::JoinHandle<()>, + _mesh_join_watcher: tokio::task::JoinHandle<()>, } /// Start the runtime-owned status publisher and admission-roster watcher. @@ -62,12 +68,25 @@ pub async fn start_coordinator(app: AppHandle) { 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, &mut pending_shrink).await { + if let Err(error) = reconcile_roster(&roster_app, &mut pending_shrink).await { eprintln!("buzz-mesh: roster reconcile failed: {error}"); } } }); + let join_app = app.clone(); + let mesh_join_watcher = tokio::spawn(async move { + let mut sleep_for = MESH_JOIN_POLL_INTERVAL; + loop { + tokio::time::sleep(sleep_for).await; + match reconcile_buzz_mesh_join(&join_app).await { + Ok(()) => sleep_for = MESH_JOIN_POLL_INTERVAL, + Err(error) => { + eprintln!("buzz-mesh: community mesh join reconcile failed: {error}"); + sleep_for = (sleep_for * 2).min(MESH_JOIN_RETRY_MAX); + } + } + } + }); // Brad #2304 / #2062: ensure_relay_mesh_for_record only runs on explicit // start + launch restore. After launch, local buzz-agent processes talk @@ -98,12 +117,83 @@ pub async fn start_coordinator(app: AppHandle) { _status_publisher: status_publisher, _roster_watcher: roster_watcher, _ingress_watchdog: ingress_watchdog, + _mesh_join_watcher: mesh_join_watcher, }); } else { status_publisher.abort(); roster_watcher.abort(); ingress_watchdog.abort(); + mesh_join_watcher.abort(); + } +} + +/// Join an isolated runtime to the existing Buzz community mesh. The relay is +/// discovery only: the selected endpoint is member-signed and validated, then +/// MeshLLM establishes the encrypted peer transport itself. +async fn reconcile_buzz_mesh_join(app: &AppHandle) -> Result<(), String> { + let state = app.state::(); + let peer_ids = { + let runtime = state.mesh_llm_runtime.lock().await; + let Some(runtime) = runtime.as_ref() else { + return Ok(()); + }; + let payload = runtime + .status_report_payload() + .await + .map_err(|error| error.to_string())?; + visible_peer_ids(&payload) + }; + + let targets = crate::commands::mesh_llm::resolve_buzz_mesh_join_targets(&state).await?; + let Some(target) = targets + .into_iter() + .find(|target| !target_is_visible(target, &peer_ids)) + else { + return Ok(()); + }; + + let runtime = state.mesh_llm_runtime.lock().await; + let Some(runtime) = runtime.as_ref() else { + return Ok(()); + }; + let payload = runtime + .status_report_payload() + .await + .map_err(|error| error.to_string())?; + if target_is_visible(&target, &visible_peer_ids(&payload)) { + return Ok(()); + } + runtime + .dial_endpoint_addr(target.endpoint_addr) + .await + .map_err(|error| format!("mesh join failed: {error:#}")) +} + +fn visible_peer_ids(payload: &serde_json::Value) -> Vec { + payload + .get("peers") + .and_then(serde_json::Value::as_array) + .into_iter() + .flatten() + .filter_map(|peer| peer.get("id").and_then(serde_json::Value::as_str)) + .map(|id| id.trim().to_ascii_lowercase()) + .filter(|id| !id.is_empty()) + .collect() +} + +fn target_is_visible(target: &crate::mesh_llm::MeshServeTarget, peer_ids: &[String]) -> bool { + let Some(endpoint_id) = target.endpoint_id.as_deref() else { + return false; + }; + let endpoint_id = endpoint_id.trim().to_ascii_lowercase(); + if endpoint_id.is_empty() { + return false; } + peer_ids.iter().any(|peer_id| { + let peer_id = peer_id.trim().to_ascii_lowercase(); + !peer_id.is_empty() + && (endpoint_id.starts_with(&peer_id) || peer_id.starts_with(&endpoint_id)) + }) } /// Outcome of a roster reconcile decision. @@ -170,9 +260,10 @@ fn roster_reconcile_action( } async fn reconcile_roster( - state: &AppState, + app: &AppHandle, pending_shrink: &mut Option>, ) -> Result<(), String> { + let state = app.state::(); let current_request = { let runtime = state.mesh_llm_runtime.lock().await; match runtime.as_ref() { @@ -192,7 +283,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 = crate::commands::mesh_llm::resolve_trusted_owner_ids(state).await; + 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; @@ -209,8 +300,27 @@ async fn reconcile_roster( } }; - let mut request = current_request; + let mut request = current_request.clone(); request.trusted_owner_ids = Some(fresh); + // Bootstrap endpoints are live device state, not configuration. The + // endpoint used at the previous start may belong to the member that just + // left or to a device whose iroh identity rotated while offline. Resolve a + // fresh validated peer for this restart; starting isolated is safe because + // the join watcher will converge it when a member next publishes. + request.join_token = match crate::commands::mesh_llm::resolve_buzz_mesh_join_targets(&state) + .await + { + Ok(targets) => targets + .into_iter() + .next() + .map(|target| target.endpoint_addr), + Err(error) => { + eprintln!( + "buzz-mesh: could not refresh bootstrap endpoint for roster restart; starting isolated: {error}" + ); + None + } + }; let mut guard = state.mesh_llm_runtime.lock().await; let startup_pending = match guard.as_ref() { Some(runtime) => runtime.is_starting().await, @@ -222,12 +332,29 @@ async fn reconcile_roster( ); return Ok(()); } + if guard + .as_ref() + .is_some_and(|runtime| runtime.start_request() != ¤t_request) + { + // The runtime changed while relay discovery was in flight. Its own + // request is now authoritative; the next poll will reconcile that + // runtime instead of tearing down a fresh replacement from this stale + // snapshot. + return Ok(()); + } let Some(running) = guard.take() else { return Ok(()); }; eprintln!("buzz-mesh: membership roster changed; restarting mesh node with fresh allowlist"); if let Err(error) = running.stop().await { - eprintln!("buzz-mesh: stopping mesh node for roster restart failed: {error}"); + drop(guard); + eprintln!( + "buzz-mesh: stopping mesh node for roster restart failed; restarting Buzz instead of racing the occupied ingress: {error}" + ); + app.request_restart(); + return Err(format!( + "mesh node shutdown failed during roster change: {error}" + )); } let replacement = crate::mesh_llm::DesktopMeshRuntime::start(request) .await @@ -357,6 +484,51 @@ mod tests { use super::*; + fn join_target(endpoint_id: Option<&str>) -> crate::mesh_llm::MeshServeTarget { + crate::mesh_llm::MeshServeTarget { + model_id: "model".to_string(), + model_name: None, + endpoint_addr: "iroh://example".to_string(), + reporter_pubkey: Some("member".to_string()), + owner_id: Some("owner".to_string()), + node_name: None, + capacity: None, + endpoint_id: endpoint_id.map(str::to_string), + device_id: None, + device_name: None, + } + } + + #[test] + fn visible_peer_ids_reads_the_sdk_status_shape() { + assert_eq!( + visible_peer_ids(&serde_json::json!({ + "peers": [{"id": " ABC123 "}, {"id": "def456"}, {"name": "ignored"}] + })), + vec!["abc123".to_string(), "def456".to_string()] + ); + } + + #[test] + fn target_visibility_accepts_sdk_short_endpoint_ids() { + let target = join_target(Some("ABC1234567890")); + + assert!(target_is_visible(&target, &["abc123".to_string()])); + assert!(!target_is_visible(&target, &["def456".to_string()])); + } + + #[test] + fn target_without_an_endpoint_id_is_never_treated_as_connected() { + assert!(!target_is_visible( + &join_target(None), + &["abc123".to_string()] + )); + assert!(!target_is_visible( + &join_target(Some(" ")), + &["abc123".to_string()] + )); + } + // 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 diff --git a/desktop/src-tauri/src/mesh_llm/discovery.rs b/desktop/src-tauri/src/mesh_llm/discovery.rs index d7dca6b11b..2df4a7f8c5 100644 --- a/desktop/src-tauri/src/mesh_llm/discovery.rs +++ b/desktop/src-tauri/src/mesh_llm/discovery.rs @@ -5,9 +5,10 @@ use sha2::{Digest, Sha256}; use super::{dedupe_models, MeshAvailability, MeshModelOption, MeshServeTarget, MESH_STATUS_KIND}; -/// Running-node status notes are refreshed every 45 seconds. Ignore notes older -/// than two minutes so crashed/offline devices stop contributing compute or -/// admission identities without requiring a relay-side cleanup job. +/// Running-node status notes are refreshed every 45 seconds. Routing ignores +/// notes older than two minutes so crashed/offline devices stop contributing +/// compute without a relay-side cleanup job. Admission intentionally does not: +/// Buzz membership, rather than device liveness, is the trust boundary. pub(super) const STATUS_FRESHNESS_SECS: u64 = 120; pub(crate) const MESH_STATUS_PAGE_SIZE: usize = 100; @@ -199,13 +200,14 @@ pub fn availability_from_events(events: Vec) -> MeshAvailability { let Ok(content) = serde_json::from_str::(&event.content) else { continue; }; - if owner_id_from_status_event(&event).is_none() { + let Some(owner_id) = owner_id_from_status_event(&event) else { continue; - } + }; if !endpoint_binding_is_valid(&event, &content) { continue; } saw_valid_status = true; + let reporter_pubkey = event.pubkey.to_hex().to_ascii_lowercase(); let mut serve_targets = content .get("serveTargets") .or_else(|| content.get("serve_targets")) @@ -218,6 +220,8 @@ pub fn availability_from_events(events: Vec) -> MeshAvailability { super::transport_policy::validate_advertised_endpoint(&target.endpoint_addr) .ok()?; target.endpoint_addr = validated.join_token; + target.reporter_pubkey = Some(reporter_pubkey.clone()); + target.owner_id = Some(owner_id.clone()); if target.endpoint_id.is_none() { target.endpoint_id = Some(validated.endpoint_id); } diff --git a/desktop/src-tauri/src/mesh_llm/mod.rs b/desktop/src-tauri/src/mesh_llm/mod.rs index c041aa07a9..6e3ab4b28b 100644 --- a/desktop/src-tauri/src/mesh_llm/mod.rs +++ b/desktop/src-tauri/src/mesh_llm/mod.rs @@ -29,6 +29,9 @@ pub(crate) use recovery::{ MeshRuntimeRecovery, }; +mod usage; +pub use usage::{serving_usage_from_payload, MeshServingUsage}; + mod transport_policy; #[cfg(test)] use transport_policy::iroh_relay_mode_from; @@ -84,6 +87,15 @@ pub struct MeshServeTarget { pub model_id: String, pub model_name: Option, pub endpoint_addr: String, + /// Buzz member that signed the discovery note containing this target. + /// Populated after signature/membership validation; never trusted from the + /// note payload itself. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reporter_pubkey: Option, + /// Per-runtime MeshLLM owner identity verified by the signed Buzz status. + /// Distinguishes two devices logged into the same Buzz member account. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub owner_id: Option, pub node_name: Option, pub capacity: Option, #[serde(default)] @@ -184,6 +196,10 @@ pub struct StartMeshNodeRequest { pub max_vram_gb: Option, #[serde(default)] pub join_token: Option, + /// Stable, relay-scoped mesh name injected by the Buzz backend. It is not + /// accepted from the frontend and contains no relay address. + #[serde(default, skip_deserializing)] + pub mesh_name: Option, /// Mesh owner ids admitted to this node (the member roster from /// member-signed discovery notes). `None` = caller did not resolve a roster /// (tests, direct invocations): the node runs without allowlist @@ -214,81 +230,6 @@ pub struct MeshNodeStatus { pub device_name: Option, } -/// Host-side "who is using the compute I'm sharing" snapshot. -/// -/// Read-only projection of the serving node's own runtime metrics (the same -/// `routing_metrics` / `inflight_requests` the SDK already exposes on the local -/// console). No new trust surface: it reads the node's own status payload. -/// -/// The local/remote/endpoint attempt split is what distinguishes *my own* -/// agent (local) from *another member consuming my compute* (remote/endpoint). -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] -#[serde(rename_all = "camelCase")] -pub struct MeshServingUsage { - /// Requests being served right now. - pub inflight: u64, - /// Highest concurrent in-flight seen this session. - pub peak_inflight: u64, - /// Total requests routed through this node. - pub requests_served: u64, - /// Completion tokens produced. - pub tokens_served: u64, - /// Recent decode throughput. - pub tokens_per_second: f64, - /// Requests served for this machine's own agents. - pub local_attempts: u64, - /// Requests served for a remote peer (someone else consuming my compute). - pub remote_attempts: u64, - /// Requests served via an advertised endpoint (also a remote consumer). - pub endpoint_attempts: u64, - /// Other nodes currently visible as peers. - pub peers: u64, -} - -impl MeshServingUsage { - /// True when at least one request has been served for a non-local consumer. - #[cfg(test)] - pub fn has_remote_consumers(&self) -> bool { - self.remote_attempts > 0 || self.endpoint_attempts > 0 - } -} - -/// Pure extractor: project a raw SDK status payload into [`MeshServingUsage`]. -/// -/// Every field is read defensively (missing → 0) so an SDK shape change -/// degrades to "no usage shown" rather than an error. Kept pure so it can be -/// unit-tested against a captured payload without a live runtime. -pub fn serving_usage_from_payload(payload: &serde_json::Value) -> MeshServingUsage { - let u64_at = |v: &serde_json::Value| v.as_u64().unwrap_or(0); - let rm = payload.get("routing_metrics"); - let local = rm.and_then(|m| m.get("local_node")); - let get_u64 = |obj: Option<&serde_json::Value>, key: &str| { - obj.and_then(|o| o.get(key)).map(u64_at).unwrap_or(0) - }; - MeshServingUsage { - inflight: local - .and_then(|l| l.get("current_inflight_requests")) - .map(u64_at) - .or_else(|| payload.get("inflight_requests").map(u64_at)) - .unwrap_or(0), - peak_inflight: get_u64(local, "peak_inflight_requests"), - requests_served: get_u64(rm, "request_count"), - tokens_served: get_u64(rm, "completion_tokens_observed"), - tokens_per_second: rm - .and_then(|m| m.get("avg_tokens_per_second")) - .and_then(serde_json::Value::as_f64) - .unwrap_or(0.0), - local_attempts: get_u64(local, "local_attempt_count"), - remote_attempts: get_u64(local, "remote_attempt_count"), - endpoint_attempts: get_u64(local, "endpoint_attempt_count"), - peers: payload - .get("peers") - .and_then(serde_json::Value::as_array) - .map(|a| a.len() as u64) - .unwrap_or(0), - } -} - pub fn stopped_status() -> MeshNodeStatus { MeshNodeStatus { state: MeshNodeState::Off, @@ -423,6 +364,9 @@ impl DesktopMeshRuntime { .discovery_mode(MeshDiscoveryMode::Nostr) .startup_timeout(MESH_STARTUP_TIMEOUT) .console_ui(true); + if let Some(mesh_name) = request.mesh_name.as_deref() { + builder = builder.mesh_name(mesh_name); + } builder = match iroh_relay_mode()? { IrohRelayMode::Disabled => builder.disable_iroh_relays(true), IrohRelayMode::Default => builder.disable_iroh_relays(false), @@ -460,6 +404,9 @@ impl DesktopMeshRuntime { .discovery_mode(MeshDiscoveryMode::Nostr) .startup_timeout(MESH_CLIENT_MANAGEMENT_TIMEOUT) .console_ui(true); + if let Some(mesh_name) = request.mesh_name.as_deref() { + builder = builder.mesh_name(mesh_name); + } builder = match iroh_relay_mode()? { IrohRelayMode::Disabled => builder.disable_iroh_relays(true), IrohRelayMode::Default => builder.disable_iroh_relays(false), @@ -646,6 +593,8 @@ impl DesktopMeshRuntime { model_id: model.id, model_name: model.name, endpoint_addr: endpoint_addr.clone(), + reporter_pubkey: None, + owner_id: None, node_name: payload .get("node_id") .and_then(serde_json::Value::as_str) diff --git a/desktop/src-tauri/src/mesh_llm/mod_tests.rs b/desktop/src-tauri/src/mesh_llm/mod_tests.rs index 1f2c6abb44..0b726c264f 100644 --- a/desktop/src-tauri/src/mesh_llm/mod_tests.rs +++ b/desktop/src-tauri/src/mesh_llm/mod_tests.rs @@ -11,6 +11,7 @@ fn pending_client_runtime( model_id: None, max_vram_gb: None, join_token: Some("initial-token".to_string()), + mesh_name: None, trusted_owner_ids: None, }; super::DesktopMeshRuntime { @@ -692,10 +693,7 @@ fn serving_usage_extracts_local_and_remote_attempts() { assert_eq!(usage.local_attempts, 4); assert_eq!(usage.remote_attempts, 0); assert_eq!(usage.endpoint_attempts, 0); - assert!( - !usage.has_remote_consumers(), - "all-local traffic is not a remote consumer" - ); + assert_eq!(usage.remote_attempts + usage.endpoint_attempts, 0); } #[test] @@ -715,10 +713,7 @@ fn serving_usage_flags_remote_consumer() { assert_eq!(usage.remote_attempts, 6); assert_eq!(usage.endpoint_attempts, 1); assert_eq!(usage.peers, 2); - assert!( - usage.has_remote_consumers(), - "remote/endpoint attempts mean someone else is using my compute" - ); + assert!(usage.remote_attempts + usage.endpoint_attempts > 0); } #[test] @@ -726,5 +721,5 @@ fn serving_usage_defaults_to_zero_on_missing_fields() { // SDK shape drift must degrade to "no usage" not panic. let usage = super::serving_usage_from_payload(&json!({})); assert_eq!(usage, super::MeshServingUsage::default()); - assert!(!usage.has_remote_consumers()); + assert_eq!(usage.remote_attempts + usage.endpoint_attempts, 0); } diff --git a/desktop/src-tauri/src/mesh_llm/recovery.rs b/desktop/src-tauri/src/mesh_llm/recovery.rs index 29ed8a13fe..ce6d495a47 100644 --- a/desktop/src-tauri/src/mesh_llm/recovery.rs +++ b/desktop/src-tauri/src/mesh_llm/recovery.rs @@ -243,6 +243,11 @@ pub(crate) async fn rearm_relay_mesh_for_running_agents(app: &AppHandle) -> Resu let _rearm_guard = state.mesh_recovery.rearm_lock.lock().await; let recovery = recover_stale_mesh_runtime(&state, MeshRecoveryUrgency::Watchdog).await; let active_pubkeys = active_managed_agent_pubkeys(&state); + // Mesh participation is resolved through the same definition-authoritative + // path as spawn/restore (#1968): definition → global fallback. A linked + // instance's own bytes never contribute. + let personas = crate::managed_agents::load_personas(app).unwrap_or_default(); + let global = crate::managed_agents::load_global_agent_config(app).unwrap_or_default(); match recovery { MeshRuntimeRecovery::Live @@ -250,10 +255,9 @@ pub(crate) async fn rearm_relay_mesh_for_running_agents(app: &AppHandle) -> Resu | MeshRuntimeRecovery::Replaced => return Ok(()), MeshRuntimeRecovery::RestartRequired => { let records = crate::managed_agents::load_managed_agents(app).unwrap_or_default(); - if !records - .iter() - .any(|record| is_running_relay_mesh_agent(record, &active_pubkeys)) - { + if !records.iter().any(|record| { + running_relay_mesh_model_id(record, &active_pubkeys, &personas, &global).is_some() + }) { // A foreground save may still be bringing up its first ingress. // Only an already-running consumer justifies an automatic app // relaunch from the background watchdog. @@ -272,10 +276,9 @@ pub(crate) async fn rearm_relay_mesh_for_running_agents(app: &AppHandle) -> Resu } MeshRuntimeRecovery::Absent => { let records = crate::managed_agents::load_managed_agents(app).unwrap_or_default(); - if !records - .iter() - .any(|record| is_running_relay_mesh_agent(record, &active_pubkeys)) - { + if !records.iter().any(|record| { + running_relay_mesh_model_id(record, &active_pubkeys, &personas, &global).is_some() + }) { return Ok(()); } } @@ -285,11 +288,20 @@ pub(crate) async fn rearm_relay_mesh_for_running_agents(app: &AppHandle) -> Resu let records = crate::managed_agents::load_managed_agents(app).unwrap_or_default(); let mesh_records: Vec<_> = records .into_iter() - .filter(|record| is_running_relay_mesh_agent(record, &active_pubkeys)) + .filter_map(|record| { + running_relay_mesh_model_id(&record, &active_pubkeys, &personas, &global) + .map(|mesh_model_id| (record, mesh_model_id)) + }) .collect(); let mut first_error = None; - for record in &mesh_records { - match crate::commands::mesh_llm::ensure_relay_mesh_for_record(app, record, false).await { + for (record, mesh_model_id) in &mesh_records { + match crate::commands::mesh_llm::ensure_relay_mesh_for_record( + app, + Some(mesh_model_id.as_str()), + false, + ) + .await + { Ok(()) => { if let Err(error) = clear_mesh_last_error_if_set(app, &record.pubkey) { eprintln!("buzz-mesh: failed to clear recovery error: {error}"); @@ -322,16 +334,28 @@ fn active_managed_agent_pubkeys(state: &AppState) -> HashSet { .unwrap_or_default() } -fn is_running_relay_mesh_agent( +/// Effective mesh model for a record that is actively running, or `None` +/// when the record is not a running relay-mesh consumer. Resolution goes +/// through `resolve_effective_relay_mesh_model_id` (definition → global +/// fallback, #1968) so the watchdog agrees with spawn/restore about which +/// agents are mesh-backed. +fn running_relay_mesh_model_id( record: &crate::managed_agents::ManagedAgentRecord, active_pubkeys: &HashSet, -) -> bool { - record.backend == crate::managed_agents::BackendKind::Local - && crate::managed_agents::relay_mesh_model_id(record).is_some() + personas: &[crate::managed_agents::AgentDefinition], + global: &crate::managed_agents::GlobalAgentConfig, +) -> Option { + let running = record.backend == crate::managed_agents::BackendKind::Local && active_pubkeys.contains(&record.pubkey.to_ascii_lowercase()) && record .runtime_pid - .is_none_or(crate::managed_agents::process_is_running) + .is_none_or(crate::managed_agents::process_is_running); + if !running { + return None; + } + crate::managed_agents::effective_config::resolve_effective_relay_mesh_model_id( + record, personas, global, + ) } fn persist_mesh_last_error(app: &AppHandle, pubkey: &str, error: &str) -> Result<(), String> { @@ -458,29 +482,45 @@ mod tests { #[test] fn only_running_relay_mesh_agents_trigger_rearm() { + let personas: Vec = Vec::new(); + let global = crate::managed_agents::GlobalAgentConfig::default(); + let empty = active_set(&[]); - assert!(!is_running_relay_mesh_agent( + assert!(running_relay_mesh_model_id( &mesh_record("stopped", Some(std::process::id())), - &empty - )); + &empty, + &personas, + &global, + ) + .is_none()); let active = active_set(&["live"]); - assert!(is_running_relay_mesh_agent( - &mesh_record("live", Some(std::process::id())), - &active - )); - assert!(is_running_relay_mesh_agent( - &mesh_record("live", None), - &active - )); + assert_eq!( + running_relay_mesh_model_id( + &mesh_record("live", Some(std::process::id())), + &active, + &personas, + &global, + ) + .as_deref(), + Some("Qwen3") + ); + assert_eq!( + running_relay_mesh_model_id(&mesh_record("live", None), &active, &personas, &global) + .as_deref(), + Some("Qwen3") + ); let mut non_mesh = mesh_record("plain", Some(std::process::id())); non_mesh.env_vars.clear(); non_mesh.relay_mesh = None; - assert!(!is_running_relay_mesh_agent( + assert!(running_relay_mesh_model_id( &non_mesh, - &active_set(&["plain"]) - )); + &active_set(&["plain"]), + &personas, + &global, + ) + .is_none()); } #[test] diff --git a/desktop/src-tauri/src/mesh_llm/usage.rs b/desktop/src-tauri/src/mesh_llm/usage.rs new file mode 100644 index 0000000000..830f224ef5 --- /dev/null +++ b/desktop/src-tauri/src/mesh_llm/usage.rs @@ -0,0 +1,68 @@ +use serde::{Deserialize, Serialize}; + +/// Host-side "who is using the compute I'm sharing" snapshot. +/// +/// Read-only projection of the serving node's own runtime metrics (the same +/// `routing_metrics` / `inflight_requests` the SDK already exposes on the local +/// console). No new trust surface: it reads the node's own status payload. +/// +/// The local/remote/endpoint attempt split is what distinguishes *my own* +/// agent (local) from *another member consuming my compute* (remote/endpoint). +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +#[serde(rename_all = "camelCase")] +pub struct MeshServingUsage { + /// Requests being served right now. + pub inflight: u64, + /// Highest concurrent in-flight seen this session. + pub peak_inflight: u64, + /// Total requests routed through this node. + pub requests_served: u64, + /// Completion tokens produced. + pub tokens_served: u64, + /// Recent decode throughput. + pub tokens_per_second: f64, + /// Requests served for this machine's own agents. + pub local_attempts: u64, + /// Requests served for a remote peer (someone else consuming my compute). + pub remote_attempts: u64, + /// Requests served via an advertised endpoint (also a remote consumer). + pub endpoint_attempts: u64, + /// Other nodes currently visible as peers. + pub peers: u64, +} + +/// Pure extractor: project a raw SDK status payload into [`MeshServingUsage`]. +/// +/// Every field is read defensively (missing → 0) so an SDK shape change +/// degrades to "no usage shown" rather than an error. Kept pure so it can be +/// unit-tested against a captured payload without a live runtime. +pub fn serving_usage_from_payload(payload: &serde_json::Value) -> MeshServingUsage { + let u64_at = |v: &serde_json::Value| v.as_u64().unwrap_or(0); + let rm = payload.get("routing_metrics"); + let local = rm.and_then(|m| m.get("local_node")); + let get_u64 = |obj: Option<&serde_json::Value>, key: &str| { + obj.and_then(|o| o.get(key)).map(u64_at).unwrap_or(0) + }; + MeshServingUsage { + inflight: local + .and_then(|l| l.get("current_inflight_requests")) + .map(u64_at) + .or_else(|| payload.get("inflight_requests").map(u64_at)) + .unwrap_or(0), + peak_inflight: get_u64(local, "peak_inflight_requests"), + requests_served: get_u64(rm, "request_count"), + tokens_served: get_u64(rm, "completion_tokens_observed"), + tokens_per_second: rm + .and_then(|m| m.get("avg_tokens_per_second")) + .and_then(serde_json::Value::as_f64) + .unwrap_or(0.0), + local_attempts: get_u64(local, "local_attempt_count"), + remote_attempts: get_u64(local, "remote_attempt_count"), + endpoint_attempts: get_u64(local, "endpoint_attempt_count"), + peers: payload + .get("peers") + .and_then(serde_json::Value::as_array) + .map(|a| a.len() as u64) + .unwrap_or(0), + } +} diff --git a/desktop/src/features/agents/ui/PersonaModelField.tsx b/desktop/src/features/agents/ui/PersonaModelField.tsx index bfbdb2c83d..79e3cb7bac 100644 --- a/desktop/src/features/agents/ui/PersonaModelField.tsx +++ b/desktop/src/features/agents/ui/PersonaModelField.tsx @@ -92,7 +92,8 @@ export function PersonaModelField({ ) : null} {showSharedComputeAutoHint ? (

- Buzz will choose an available shared model when the agent starts. + Auto uses Mesh collective intelligence when two or more models stay + available, otherwise it chooses one available model.

) : null} {modelDiscoveryStatus ? ( diff --git a/desktop/src/features/agents/ui/agentConfigControls.test.mjs b/desktop/src/features/agents/ui/agentConfigControls.test.mjs index 5f3ef718bc..6d0cc0be19 100644 --- a/desktop/src/features/agents/ui/agentConfigControls.test.mjs +++ b/desktop/src/features/agents/ui/agentConfigControls.test.mjs @@ -44,6 +44,16 @@ test("an explicit inherited default label wins over harness discovery", () => { ); }); +test("shared compute describes Auto's collective behavior", () => { + assert.equal( + resolveDefaultModelLabel({ + discoveredModelOptions: null, + isSharedCompute: true, + }), + "Auto (collective when available)", + ); +}); + // ── appendNoModelsSentinel ───────────────────────────────────────────────────── test("appendNoModelsSentinel_emptyOptionsDiscoveryFinished_addsDisabledRow", () => { diff --git a/desktop/src/features/agents/ui/agentConfigControls.tsx b/desktop/src/features/agents/ui/agentConfigControls.tsx index 8c2f5b6744..1a431d1f91 100644 --- a/desktop/src/features/agents/ui/agentConfigControls.tsx +++ b/desktop/src/features/agents/ui/agentConfigControls.tsx @@ -326,7 +326,9 @@ export function resolveDefaultModelLabel({ return ( defaultModelLabel ?? discoveredModelOptions?.find((option) => option.id.trim() === "")?.label ?? - (isSharedCompute ? "Default (auto)" : getDefaultLlmModelLabel(globalModel)) + (isSharedCompute + ? "Auto (collective when available)" + : getDefaultLlmModelLabel(globalModel)) ); } diff --git a/desktop/src/features/agents/ui/relayMeshModelPicker.test.mjs b/desktop/src/features/agents/ui/relayMeshModelPicker.test.mjs index 3dd1b7c7b3..ce002dfe61 100644 --- a/desktop/src/features/agents/ui/relayMeshModelPicker.test.mjs +++ b/desktop/src/features/agents/ui/relayMeshModelPicker.test.mjs @@ -9,7 +9,7 @@ import { AUTO_MODEL_DROPDOWN_VALUE } from "./agentConfigOptions.tsx"; const fallback = [{ id: "", label: "Default model" }]; const live = [ - { id: "", label: "Default (auto)" }, + { id: "", label: "Auto (collective when available)" }, { id: "mesh/model", label: "mesh/model" }, ]; @@ -35,7 +35,9 @@ test("Buzz shared compute fallback is Default auto while normal providers remain model: "", provider: "relay-mesh", }); - assert.deepEqual(mesh.options, [{ id: "", label: "Default (auto)" }]); + assert.deepEqual(mesh.options, [ + { id: "", label: "Auto (collective when available)" }, + ]); const openai = relayMeshModelPickerState({ discoveredOptions: null, @@ -57,7 +59,9 @@ test("Buzz shared compute keeps Default auto when discovery is empty", () => { provider: "relay-mesh", }); - assert.deepEqual(state.options, [{ id: "", label: "Default (auto)" }]); + assert.deepEqual(state.options, [ + { id: "", label: "Auto (collective when available)" }, + ]); assert.equal(state.selectValue, AUTO_MODEL_DROPDOWN_VALUE); assert.equal(state.showCustomInput, false); }); @@ -71,6 +75,6 @@ test("Buzz shared compute dropdown contains Default plus live models and no cust }); assert.deepEqual( options.map((option) => option.label), - ["Default (auto)", "mesh/model"], + ["Auto (collective when available)", "mesh/model"], ); }); diff --git a/desktop/src/features/agents/ui/relayMeshModelPicker.ts b/desktop/src/features/agents/ui/relayMeshModelPicker.ts index 2e3cd93eb2..dee401824a 100644 --- a/desktop/src/features/agents/ui/relayMeshModelPicker.ts +++ b/desktop/src/features/agents/ui/relayMeshModelPicker.ts @@ -12,7 +12,10 @@ function withSharedComputeAutoOption( options: readonly PersonaModelOption[], ): readonly PersonaModelOption[] { const modelOptions = options.filter((option) => option.id.trim() !== ""); - return [{ id: "", label: "Default (auto)" }, ...modelOptions]; + return [ + { id: "", label: "Auto (collective when available)" }, + ...modelOptions, + ]; } export function relayMeshModelPickerState({ diff --git a/desktop/src/features/mesh-compute/shareToggleState.ts b/desktop/src/features/mesh-compute/shareToggleState.ts index e0a9b1fc44..58c81139a1 100644 --- a/desktop/src/features/mesh-compute/shareToggleState.ts +++ b/desktop/src/features/mesh-compute/shareToggleState.ts @@ -21,13 +21,14 @@ export type MeshShareToggleModel = { isSharing: boolean; /** * A client-mode runtime occupies the single slot (this machine is consuming - * a peer's compute). The Share switch must read off + disabled while true. + * a peer's compute). The Share switch reads off while true, but the member + * may replace this client with a serve runtime by turning sharing on. */ isConsuming: boolean; /** * ANY runtime occupies the single slot (serve or client, healthy or failed). - * A fresh `mesh_start_node` fails with "already running" while true, so the - * switch must not offer a start — only a stop of an existing serve node. + * This remains useful for disabling edits around an existing serve runtime; + * the backend supports the intentional client-to-serve replacement. */ slotOccupied: boolean; }; diff --git a/desktop/src/features/mesh-compute/ui/MeshComputeSettingsCard.tsx b/desktop/src/features/mesh-compute/ui/MeshComputeSettingsCard.tsx index 30c543e245..76f86f8054 100644 --- a/desktop/src/features/mesh-compute/ui/MeshComputeSettingsCard.tsx +++ b/desktop/src/features/mesh-compute/ui/MeshComputeSettingsCard.tsx @@ -124,22 +124,21 @@ export function MeshComputeSettingsCard() { }; }, []); - // Mirror the running node's modelId back into the field so the card shows - // what's ACTUALLY being served — not just on a fresh load (empty field) but - // whenever the served model differs from the field (e.g. the member switched - // models and the node restarted on the new one). This is safe from clobbering - // a mid-edit because the field is disabled whenever the slot is occupied - // (`controlsDisabled`), so the user can't be typing while a node is running. + // Mirror only a SERVE runtime's model into the field. A client reports the + // remote model it is consuming; copying that value here both loses the + // member's local sharing choice and can cause this machine to download the + // remote member's much larger model when sharing is enabled. React.useEffect(() => { if ( status?.state === "running" && + status.mode === "serve" && status.modelId && status.modelId !== modelInput ) { setModelInput(status.modelId); writeDraft(MODEL_DRAFT_STORAGE_KEY, status.modelId); } - }, [status?.state, status?.modelId, modelInput]); + }, [status?.state, status?.mode, status?.modelId, modelInput]); // The Share toggle reflects ONLY serve-mode occupancy. A client-mode runtime // (this machine consuming a peer's compute) shares the single runtime slot @@ -151,14 +150,12 @@ export function MeshComputeSettingsCard() { // actively sharing (serve mode). Read-only; reads the node's own metrics. const servingUsage = useMeshServingUsage(isSharing); const servingIndicator = deriveServingIndicator(servingUsage, isSharing); - // Any occupying runtime (serve or client, healthy or failed) locks the model - // inputs and blocks a fresh start — stop it before reconfiguring. - const controlsDisabled = slotOccupied || actionInFlight; + // A consuming client may be intentionally replaced by a serve runtime, so + // keep the local model controls available in client mode. Serve and unknown + // occupants remain locked until stopped/recovered. + const controlsDisabled = actionInFlight || (slotOccupied && !isConsuming); const refClass = classifyModelRef(modelInput); - const canStart = - refClass.kind !== "unknown" && - !actionInFlight && - status?.state !== "starting"; + const canStart = refClass.kind !== "unknown" && !actionInFlight; async function handleToggle(next: boolean) { // Never let the Share switch tear down a consume session. The switch is @@ -260,13 +257,15 @@ export function MeshComputeSettingsCard() { checked={isSharing} data-testid="mesh-share-compute-toggle" disabled={ - // When the slot is occupied, the switch is only actionable if - // WE are sharing (so it can stop a serve node — even a failed - // one). Any other occupant (consuming, or an unexpected - // modeless-running node) would make a fresh start throw "already - // running", so keep it disabled. When the slot is empty, gate on - // a valid model ref. - actionInFlight || (slotOccupied ? !isSharing : !canStart) + // A serve node can always be stopped. An off node or consuming + // client can start sharing once a valid local model is selected. + // Unknown occupants remain protected from replacement. + actionInFlight || + (isSharing + ? false + : slotOccupied && !isConsuming + ? true + : !canStart) } id="mesh-share-compute-toggle" onCheckedChange={handleToggle} @@ -563,14 +562,13 @@ function StatusLine({ return

Stopping…

; } // A client-mode runtime owns the single slot: this machine is consuming a - // peer's compute, not sharing. Explain why Share is off + disabled instead - // of showing the misleading "Sharing … with relay members" serve copy. The - // client node is app-session-lived infra with no user stop control, so the - // copy states the fact rather than promising an action that doesn't exist. + // peer's compute, not sharing. The switch stays off, but remains available + // so the member can replace the client with a serving runtime. if (isConsuming) { return (

- This machine is currently using another member's shared compute. + This machine is currently using another member's shared compute. Turn on + sharing to switch to the selected local model; Buzz may briefly restart.

); } diff --git a/desktop/tests/e2e/mesh-compute.spec.ts b/desktop/tests/e2e/mesh-compute.spec.ts index 28c4ee488a..7c360368a2 100644 --- a/desktop/tests/e2e/mesh-compute.spec.ts +++ b/desktop/tests/e2e/mesh-compute.spec.ts @@ -5,19 +5,20 @@ import { openSettings } from "../helpers/settings"; type E2eWindow = Window & { __BUZZ_E2E_COMMANDS__?: string[]; + __BUZZ_E2E_COMMAND_PAYLOADS__?: Array<{ + command: string; + payload: { request?: { mode?: string; modelId?: string } } | null; + }>; __BUZZ_E2E_SET_MESH__?: (mesh: { nodeState?: "off" | "running"; nodeMode?: "serve" | "client" | null; }) => void; }; -test.beforeEach(async ({ page }) => { - await installMockBridge(page); -}); - test("Share compute has a clear empty state and starts and stops sharing", async ({ page, }) => { + await installMockBridge(page); await page.goto("/"); await openSettings(page, "compute"); @@ -56,14 +57,20 @@ test("Share compute has a clear empty state and starts and stops sharing", async .toContain("mesh_stop_node"); }); -test("consuming a peer's compute does NOT light the Share toggle", async ({ +test("a consuming client can switch to sharing its saved local model", async ({ page, }) => { // Regression: consuming someone else's shared compute starts a client-mode // node in the single runtime slot, which reports state:"running". The Share - // toggle keyed off state alone and lit up — and clicking it would have torn - // down the unrelated consume session. It must stay off + disabled, explain - // why, and issue no stop command. + // toggle keyed off state alone and lit up. A later guard overcorrected by + // disabling the switch and copying the remote model over the local sharing + // choice. Keep the switch off, preserve the local model, then replace the + // client with one serve start (never a stop command). + const localModel = "hf://demo/local-small-model:Q4_K_M"; + await page.addInitScript((model) => { + window.localStorage.setItem("buzz.mesh-compute.share.model.v1", model); + }, localModel); + await installMockBridge(page); await page.goto("/"); // The mesh seed hook is installed when the mock bridge boots; calling it // before then silently no-ops (optional chaining) and the seed is lost. @@ -80,19 +87,26 @@ test("consuming a peer's compute does NOT light the Share toggle", async ({ const card = page.getByTestId("settings-mesh-share-compute"); const toggle = page.getByTestId("mesh-share-compute-toggle"); + const model = page.getByTestId("mesh-share-compute-model"); await expect(card).toContainText( "This machine is currently using another member's shared compute", ); + await expect(card).toContainText("Buzz may briefly restart"); await expect(toggle).not.toBeChecked(); - await expect(toggle).toBeDisabled(); + await expect(model).toBeEnabled(); + await expect(model).toHaveValue(localModel); + await expect(toggle).toBeEnabled(); + await toggle.click(); + await expect(toggle).toBeChecked(); - // The switch is disabled, so a click can't fire onCheckedChange — but assert - // the destructive command never went out regardless. - const stopIssued = await page.evaluate(() => - ((window as E2eWindow).__BUZZ_E2E_COMMANDS__ ?? []).includes( - "mesh_stop_node", - ), - ); - expect(stopIssued).toBe(false); + const commands = await page.evaluate(() => ({ + names: (window as E2eWindow).__BUZZ_E2E_COMMANDS__ ?? [], + payloads: (window as E2eWindow).__BUZZ_E2E_COMMAND_PAYLOADS__ ?? [], + })); + expect(commands.names).not.toContain("mesh_stop_node"); + expect(commands.payloads).toContainEqual({ + command: "mesh_start_node", + payload: { request: { mode: "serve", modelId: localModel } }, + }); }); diff --git a/desktop/tests/e2e/smoke.spec.ts b/desktop/tests/e2e/smoke.spec.ts index afe58d3713..ff38d55415 100644 --- a/desktop/tests/e2e/smoke.spec.ts +++ b/desktop/tests/e2e/smoke.spec.ts @@ -156,7 +156,7 @@ test("Buzz shared compute explains automatic model selection", async ({ await expect(page.locator("#persona-model")).toContainText("Automatic"); await expect( page.getByText( - "Buzz will choose an available shared model when the agent starts.", + "Auto uses Mesh collective intelligence when two or more models stay available, otherwise it chooses one available model.", ), ).toBeVisible(); await expect(page.locator("#persona-custom-model")).toHaveCount(0);