diff --git a/crates/buzz-relay/examples/mesh_agent_e2e.rs b/crates/buzz-relay/examples/mesh_agent_e2e.rs index db0fc6090b..eced61a8ad 100644 --- a/crates/buzz-relay/examples/mesh_agent_e2e.rs +++ b/crates/buzz-relay/examples/mesh_agent_e2e.rs @@ -5,10 +5,9 @@ //! //! Permutations: //! P1 explicit-model chat — agent pinned to the served model id replies. -//! P2 auto-model chat — agent sends `model: "auto"`; mesh router picks. -//! P3 context-fit regression — an oversized output budget (150k tokens) -//! must FAIL with the router's context error (proves the router's fit -//! gate — the failure mode the 1024 preset cap protects against). +//! P2 virtual-mesh chat — one served model exercises single-model fallback. +//! P3 generous output ceiling succeeds (Mesh #1350); oversized input still +//! fails context admission via the real HTTP router, before inference. //! P4 agentic tool use — agent + buzz-dev-mcp writes a file on disk. //! //! The serve node is the same `mesh_llm_sdk::serve` path Share-compute uses @@ -136,13 +135,11 @@ async fn run() -> anyhow::Result<()> { ), } - // P3: regression — an output budget no served model's context can hold - // must be rejected by the router with the context-fit error (the failure - // mode that broke relay-mesh agents when buzz-agent's default 32768 - // budget met a 32k-context model). 150k output: passes buzz-agent's own - // config validation (must stay under its 200k max_context_tokens) but - // with the router's +25% margin overflows even 128k-context models like - // GLM-4.7-Flash. + // Mesh v0.76.0-rc9 (9f192c9), routing_rank.rs:43-58 reserves + // min(completion ceiling, 512) decode headroom, NOT the entire ceiling + // (#1350). buzz-agent llm.rs sends max_completion_tokens on the chat API. + // Keep the 150k ceiling as a positive regression, plus a negative input + // fit probe below. Do not accept generic service failures as context errors. let r = agent_chat( &base, "mesh", @@ -152,18 +149,13 @@ async fn run() -> anyhow::Result<()> { ) .await; match r { - Ok(text) => record( - "P3 oversized-budget must fail", - false, - format!("unexpectedly succeeded: {text}"), - ), - Err(e) => { - let msg = e.to_string(); - let is_context_503 = msg.contains("503") - || msg.contains("service_unavailable") - || msg.contains("context-compatible"); - record("P3 oversized-budget must fail", is_context_503, msg); - } + Ok(text) => record("P3a generous output ceiling", text.trim() == "PONG", text), + Err(e) => record("P3a generous output ceiling", false, e.to_string()), + } + let r = oversized_input_probe(&http, &base, &served_id).await; + match r { + Ok(detail) => record("P3b oversized input rejected", true, detail), + Err(e) => record("P3b oversized input rejected", false, e.to_string()), } // P4: agentic tool use via buzz-dev-mcp — write a real file inside the @@ -171,13 +163,13 @@ async fn run() -> anyhow::Result<()> { // nonexistent absolute paths outside that root. let marker_name = format!("mesh-e2e-{}.txt", std::process::id()); let prompt = format!( - "Use your developer tools to create {marker_name} in the current working directory containing exactly the text BUZZ_OK (no quotes, no newline commentary). Then confirm." + "Call dev__shell exactly once with command `{}` and no workdir override. This is the only authorized operation. Then confirm in text; do not run any other tools.", marker_command(&marker_name) ); let mcp = vec![("dev".to_string(), repo_bin("buzz-dev-mcp")?)]; let (r, marker) = agent_chat_with_marker(&base, "mesh", None, &prompt, &mcp, &marker_name).await; let file_ok = std::fs::read_to_string(&marker) - .map(|c| c.contains("BUZZ_OK")) + .map(|c| c == "BUZZ_OK") .unwrap_or(false); match r { Ok(text) => record( @@ -189,7 +181,7 @@ async fn run() -> anyhow::Result<()> { format!("no file at {}; agent said: {text}", marker.display()) }, ), - Err(e) => record("P4 agentic tool use", file_ok, format!("agent error: {e}")), + Err(e) => record("P4 agentic tool use", false, format!("agent error: {e}")), } let _ = std::fs::remove_file(&marker); @@ -237,7 +229,8 @@ async fn agent_chat( mcp_servers: &[(String, String)], ) -> anyhow::Result { let (result, _) = - agent_chat_in_isolated_home(base, model, max_output_tokens, prompt, mcp_servers).await; + agent_chat_in_isolated_home(base, model, max_output_tokens, prompt, mcp_servers, None) + .await; result } @@ -249,8 +242,15 @@ async fn agent_chat_with_marker( mcp_servers: &[(String, String)], marker_name: &str, ) -> (anyhow::Result, std::path::PathBuf) { - let (result, home) = - agent_chat_in_isolated_home(base, model, max_output_tokens, prompt, mcp_servers).await; + let (result, home) = agent_chat_in_isolated_home( + base, + model, + max_output_tokens, + prompt, + mcp_servers, + Some(marker_name), + ) + .await; (result, home.join(marker_name)) } @@ -260,19 +260,25 @@ async fn agent_chat_in_isolated_home( max_output_tokens: Option<&str>, prompt: &str, mcp_servers: &[(String, String)], + marker_name: Option<&str>, ) -> (anyhow::Result, std::path::PathBuf) { let agent = match repo_bin("buzz-agent") { Ok(agent) => agent, Err(error) => return (Err(error), std::path::PathBuf::new()), }; // Isolated HOME: no skills, no AGENTS.md chain, no keychain, tiny prompt. - let home = std::env::temp_dir().join(format!("mesh-e2e-home-{}", std::process::id())); - if let Err(error) = std::fs::create_dir_all(&home) { - return (Err(error.into()), home); - } + // Fresh per leg: stale markers and prior agent state can never pass P4. + // Retain the directory for diagnostics (also when P4 fails). + let home = match tempfile::Builder::new().prefix("mesh-e2e-home-").tempdir() { + Ok(dir) => dir.keep(), + Err(error) => return (Err(error.into()), std::path::PathBuf::new()), + }; + eprintln!("[e2e] isolated home: {}", home.display()); let mut command = Command::new(&agent); command + .current_dir(&home) + .kill_on_drop(true) .env_clear() .env("PATH", std::env::var("PATH").unwrap_or_default()) .env("HOME", &home) @@ -291,10 +297,8 @@ async fn agent_chat_in_isolated_home( // Pinning a value here would test a config the product does not ship. .stdin(Stdio::piped()) .stdout(Stdio::piped()) - .stderr(Stdio::null()); - // P3 deliberately overrides the production default to exercise the - // router's context-fit rejection. Normal and tool turns leave it unset, - // matching the desktop provider path. + .stderr(Stdio::inherit()); + // P3a exercises a generous ceiling, not an admission reservation. if let Some(value) = max_output_tokens { command.env("BUZZ_AGENT_MAX_OUTPUT_TOKENS", value); } @@ -303,7 +307,7 @@ async fn agent_chat_in_isolated_home( Err(error) => return (Err(error.into()), home), }; - let result = drive_acp(&mut child, prompt, mcp_servers, &home).await; + let result = drive_acp(&mut child, prompt, mcp_servers, &home, marker_name).await; let _ = child.kill().await; (result, home) } @@ -313,6 +317,7 @@ async fn drive_acp( prompt: &str, mcp_servers: &[(String, String)], cwd: &std::path::Path, + marker_name: Option<&str>, ) -> anyhow::Result { let mut stdin = child .stdin @@ -341,7 +346,68 @@ async fn drive_acp( .as_bytes(), ) .await?; - stdin + let mut session_id: Option = None; + let mut agent_text = String::new(); + let mut marker_approved = false; + let deadline = tokio::time::Instant::now() + Duration::from_secs(600); + + loop { + let line = tokio::time::timeout_at(deadline, lines.next_line()) + .await + .map_err(|_| anyhow::anyhow!("agent timed out; text so far: {agent_text}"))?? + .ok_or_else(|| anyhow::anyhow!("agent closed stdout; text so far: {agent_text}"))?; + let Ok(msg) = serde_json::from_str::(&line) else { + continue; + }; + // Requests have their own id namespace; handle before response ids. + if msg["method"] == "session/request_permission" { + let response = permission_response( + &msg, + session_id.as_deref(), + marker_name, + &mut marker_approved, + ); + eprintln!( + "[e2e] permission: {} -> {}", + msg["params"]["toolCall"], response["result"] + ); + stdin.write_all(send(response).as_bytes()).await?; + continue; + } + if msg.get("method").is_some() && msg.get("id").is_some() { + stdin + .write_all( + send(serde_json::json!({ + "jsonrpc": "2.0", "id": msg["id"], + "error": {"code": -32601, "message": "Unsupported harness method"} + })) + .as_bytes(), + ) + .await?; + continue; + } + // Collect any streamed agent text from session/update notifications. + if msg.get("method").and_then(|m| m.as_str()) == Some("session/update") { + if msg["params"]["sessionId"].as_str() == session_id.as_deref() { + let update = &msg["params"]["update"]; + if matches!( + update["sessionUpdate"].as_str(), + Some("tool_call" | "tool_call_update") + ) { + eprintln!("[e2e] tool update: {update}"); + } + collect_text(update, &mut agent_text); + } + continue; + } + match msg.get("id").and_then(|i| i.as_i64()) { + Some(1) => { + anyhow::ensure!(msg.get("error").is_none(), "initialize failed: {msg}"); + anyhow::ensure!( + msg["result"]["protocolVersion"] == 1, + "expected ACP v1: {msg}" + ); + stdin .write_all( send(serde_json::json!({ "jsonrpc": "2.0", "id": 2, "method": "session/new", @@ -354,25 +420,7 @@ async fn drive_acp( .as_bytes(), ) .await?; - - let mut session_id: Option = None; - let mut agent_text = String::new(); - let deadline = tokio::time::Instant::now() + Duration::from_secs(600); - - loop { - let line = tokio::time::timeout_at(deadline, lines.next_line()) - .await - .map_err(|_| anyhow::anyhow!("agent timed out; text so far: {agent_text}"))?? - .ok_or_else(|| anyhow::anyhow!("agent closed stdout; text so far: {agent_text}"))?; - let Ok(msg) = serde_json::from_str::(&line) else { - continue; - }; - // Collect any streamed agent text from session/update notifications. - if msg.get("method").and_then(|m| m.as_str()) == Some("session/update") { - collect_text(&msg["params"]["update"], &mut agent_text); - continue; - } - match msg.get("id").and_then(|i| i.as_i64()) { + } Some(2) => { if let Some(err) = msg.get("error") { anyhow::bail!("session/new failed: {err}"); @@ -400,6 +448,14 @@ async fn drive_acp( if let Some(err) = msg.get("error") { anyhow::bail!("session/prompt failed: {err}"); } + anyhow::ensure!( + msg["result"]["stopReason"] == "end_turn", + "unexpected prompt stop: {msg}; text: {agent_text}" + ); + anyhow::ensure!( + marker_name.is_none() || marker_approved, + "marker permission was never approved; text: {agent_text}" + ); return Ok(agent_text.trim().to_string()); } _ => {} @@ -407,25 +463,116 @@ async fn drive_acp( } } -/// Recursively harvest "text" string fields out of a session/update payload. +/// Only user-visible assistant chunks count, never reasoning or tool content. fn collect_text(value: &serde_json::Value, out: &mut String) { - match value { - serde_json::Value::Object(map) => { - for (k, v) in map { - if k == "text" { - if let Some(s) = v.as_str() { - out.push_str(s); - } - } else { - collect_text(v, out); - } - } + if value["sessionUpdate"] == "agent_message_chunk" && value["content"]["type"] == "text" { + if let Some(text) = value["content"]["text"].as_str() { + out.push_str(text); } - serde_json::Value::Array(items) => { - for item in items { - collect_text(item, out); - } - } - _ => {} } } + +fn marker_command(name: &str) -> String { + format!("printf %s BUZZ_OK > {name}") +} + +/// Test-client-only grant: one exact shell invocation, in a fresh cwd. +/// No command parsing, prefix matches, persistent grants, or alternate tools. +fn permission_response( + request: &serde_json::Value, + session: Option<&str>, + marker: Option<&str>, + approved: &mut bool, +) -> serde_json::Value { + let p = &request["params"]; + let tool = &p["toolCall"]; // We negotiate v1 only; fail closed on other shapes. + let input = &tool["rawInput"]; + let safe_marker = marker.filter(|name| { + !name.is_empty() + && name + .bytes() + .all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'.') + }); + let exact_input = safe_marker.is_some_and(|name| { + input.as_object().is_some_and(|o| { + o.keys() + .all(|k| k == "command" || k == "timeout_ms" || k == "workdir") + // MCP treats absent/null workdir as its default cwd; no overrides. + && (input.get("workdir").is_none() || input["workdir"].is_null()) + && input["command"] == marker_command(name) + && (input.get("timeout_ms").is_none() + || input["timeout_ms"] + .as_u64() + .is_some_and(|n| n > 0 && n <= 120_000)) + }) + }); + let allow = !*approved + && session.is_some() + && p["sessionId"].as_str() == session + && tool["title"] == "dev__shell" + && exact_input; + let option = p["options"] + .as_array() + .and_then(|options| options.iter().find(|o| o["kind"] == "allow_once")) + .and_then(|o| o["optionId"].as_str()); + let outcome = if let Some(id) = option.filter(|_| allow) { + *approved = true; + serde_json::json!({"outcome": "selected", "optionId": id}) + } else { + serde_json::json!({"outcome": "cancelled"}) + }; + serde_json::json!({"jsonrpc": "2.0", "id": request["id"], "result": {"outcome": outcome}}) +} + +async fn oversized_input_probe( + http: &reqwest::Client, + base: &str, + model: &str, +) -> anyhow::Result { + // Use advertised runtime capacity, not native model capacity. Explicit + // model routing avoids conflating MoA selection failure with context fit. + let models: serde_json::Value = http + .get(format!("{base}/models")) + .send() + .await? + .error_for_status()? + .json() + .await?; + let entry = models["data"] + .as_array() + .and_then(|a| a.iter().find(|m| m["id"] == model)) + .ok_or_else(|| anyhow::anyhow!("served model absent: {models}"))?; + let metadata = &entry["metadata"]; + let context = metadata["max_context_length"] + .as_u64() + .or_else(|| metadata["context_length"].as_u64()) + .ok_or_else(|| anyhow::anyhow!("no advertised runtime context: {entry}"))?; + // Bound allocation/request size; fail rather than guess for huge models. + anyhow::ensure!( + context > 0 && context <= 262_144, + "unsupported smoke context: {context}" + ); + // rc9 estimates serialized input bytes / 4, retains ALL prompt tokens. + let body = serde_json::json!({"model": model, "stream": false, + "max_completion_tokens": 1, + "messages": [{"role": "user", "content": "x".repeat((context as usize + 1024) * 4)}]}); + let response = http + .post(format!("{base}/chat/completions")) + .timeout(Duration::from_secs(30)) + .json(&body) + .send() + .await?; + let status = response.status(); + let text = response.text().await?; + anyhow::ensure!( + status == reqwest::StatusCode::SERVICE_UNAVAILABLE + && text.contains("no context-compatible target"), + "expected context-fit 503 for runtime context {context}, got {status}: {}", + text.chars().take(2048).collect::() + ); + Ok(format!("runtime context {context}; {status}: {text}")) +} + +#[cfg(test)] +#[path = "mesh_agent_e2e/tests.rs"] +mod tests; diff --git a/crates/buzz-relay/examples/mesh_agent_e2e/tests.rs b/crates/buzz-relay/examples/mesh_agent_e2e/tests.rs new file mode 100644 index 0000000000..d7413f1e9f --- /dev/null +++ b/crates/buzz-relay/examples/mesh_agent_e2e/tests.rs @@ -0,0 +1,185 @@ +use super::*; +use serde_json::{json, Value}; + +fn request() -> Value { + json!({"jsonrpc":"2.0", "id":3, "method":"session/request_permission", + "params":{"sessionId":"s", "toolCall":{"title":"dev__shell", + "rawInput":{"command":marker_command("marker.txt")}}, + "options":[{"kind":"allow_always","optionId":"persistent"}, + {"kind":"allow_once","optionId":"one-shot"}]}}) +} + +#[test] +fn exact_marker_grants_offered_once_option_and_preserves_id() { + let mut approved = false; + let response = permission_response(&request(), Some("s"), Some("marker.txt"), &mut approved); + assert!(approved); + assert_eq!(response["id"], 3); + assert_eq!( + response["result"]["outcome"], + json!({"outcome":"selected","optionId":"one-shot"}) + ); + assert_eq!( + permission_response(&request(), Some("s"), Some("marker.txt"), &mut approved)["result"] + ["outcome"]["outcome"], + "cancelled" + ); +} + +#[test] +fn hardware_payload_allows_null_workdir_but_rejects_non_null_values() { + let mut r = request(); + r["params"]["toolCall"]["rawInput"] = json!({ + "command": "printf %s BUZZ_OK > marker.txt", + "timeout_ms": 120000, + "workdir": null + }); + let mut approved = false; + let response = permission_response(&r, Some("s"), Some("marker.txt"), &mut approved); + assert!(approved); + assert_eq!(response["id"], 3); + assert_eq!( + response["result"]["outcome"], + json!({"outcome":"selected","optionId":"one-shot"}) + ); + assert_eq!( + permission_response(&r, Some("s"), Some("marker.txt"), &mut approved)["result"]["outcome"] + ["outcome"], + "cancelled" + ); + for workdir in [ + json!(""), + json!("/tmp"), + json!("."), + json!(false), + json!(0), + json!([]), + json!({}), + ] { + r["params"]["toolCall"]["rawInput"]["workdir"] = workdir; + let mut approved = false; + let response = permission_response(&r, Some("s"), Some("marker.txt"), &mut approved); + assert_eq!(response["result"]["outcome"]["outcome"], "cancelled", "{r}"); + assert!(!approved); + } +} + +#[test] +fn permission_fails_closed_on_scope_command_options_and_protocol() { + let mut mutations = Vec::new(); + for command in [ + "printf %s BUZZ_OK > other.txt", + "printf %s BUZZ_OK > marker.txt; pwd", + "printf %s BUZZ_OK > ../marker.txt", + "echo BUZZ_OK > marker.txt", + ] { + let mut r = request(); + r["params"]["toolCall"]["rawInput"]["command"] = json!(command); + mutations.push(r); + } + for (pointer, value) in [ + ("/params/sessionId", json!("other")), + ("/params/toolCall/title", json!("other__shell")), + ( + "/params/toolCall/rawInput", + json!({"command":marker_command("marker.txt"),"workdir":"/tmp"}), + ), + ( + "/params/toolCall/rawInput", + json!({"command":marker_command("marker.txt"),"timeout_ms":999999}), + ), + ( + "/params/options", + json!([{"kind":"allow_always","optionId":"always"}]), + ), + ("/params/toolCall", Value::Null), + ] { + let mut r = request(); + *r.pointer_mut(pointer).unwrap() = value; + mutations.push(r); + } + for r in mutations { + let mut approved = false; + let response = permission_response(&r, Some("s"), Some("marker.txt"), &mut approved); + assert_eq!(response["result"]["outcome"]["outcome"], "cancelled", "{r}"); + assert!(!approved); + } + for (session, marker) in [ + (None, Some("marker.txt")), + (Some("s"), None), + (Some("s"), Some("../marker.txt")), + ] { + assert_eq!( + permission_response(&request(), session, marker, &mut false)["result"]["outcome"] + ["outcome"], + "cancelled" + ); + } +} + +#[test] +fn collector_excludes_thoughts_tool_results_and_nontext_content() { + let mut text = String::new(); + for update in [ + "agent_thought_chunk", + "tool_call", + "tool_call_update", + "user_message_chunk", + ] { + collect_text( + &json!({"sessionUpdate":update,"content":{"type":"text","text":"PONG"}}), + &mut text, + ); + } + collect_text( + &json!({"sessionUpdate":"agent_message_chunk","content":{"type":"image","text":"PONG"}}), + &mut text, + ); + assert!(text.is_empty()); + for chunk in ["PO", "NG"] { + collect_text( + &json!({"sessionUpdate":"agent_message_chunk","content":{"type":"text","text":chunk}}), + &mut text, + ); + } + assert_eq!(text, "PONG"); +} + +/// Real stdio framing through drive_acp, no model, MCP or native runtime. +/// The permission request intentionally collides with our prompt response id. +#[tokio::test] +async fn acp_permission_request_is_not_mistaken_for_prompt_response() { + let script = r#" +import json, sys +def read(): return json.loads(sys.stdin.readline()) +def send(v): print(json.dumps(v), flush=True) +assert read()['method'] == 'initialize' +send({'id':1,'result':{'protocolVersion':1}}) +assert read()['method'] == 'session/new' +send({'id':2,'result':{'sessionId':'s'}}) +assert read()['method'] == 'session/prompt' +send({'id':3,'method':'session/request_permission','params':{'sessionId':'s','toolCall':{'title':'dev__shell','rawInput':{'command':'printf %s BUZZ_OK > marker.txt','timeout_ms':120000,'workdir':None}},'options':[{'kind':'allow_once','optionId':'yes'}]}}) +r = read() +assert r['id'] == 3 and r['result']['outcome'] == {'outcome':'selected','optionId':'yes'} +for kind,text in [('agent_thought_chunk','NOT VISIBLE'),('agent_message_chunk','PONG')]: + send({'method':'session/update','params':{'sessionId':'s','update':{'sessionUpdate':kind,'content':{'type':'text','text':text}}}}) +send({'id':3,'result':{'stopReason':'end_turn'}}) +"#; + let mut child = Command::new("python3") + .args(["-c", script]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::inherit()) + .kill_on_drop(true) + .spawn() + .unwrap(); + let home = tempfile::tempdir().unwrap(); + let result = tokio::time::timeout( + Duration::from_secs(10), + drive_acp(&mut child, "test", &[], home.path(), Some("marker.txt")), + ) + .await + .unwrap(); + assert_eq!(result.unwrap(), "PONG"); + assert!(child.wait().await.unwrap().success()); +} diff --git a/desktop/src-tauri/src/commands/mesh_llm.rs b/desktop/src-tauri/src/commands/mesh_llm.rs index 7356cd7fc0..a4f5863509 100644 --- a/desktop/src-tauri/src/commands/mesh_llm.rs +++ b/desktop/src-tauri/src/commands/mesh_llm.rs @@ -176,7 +176,7 @@ fn advance_mesh_status_cursor( Ok(cursor) } -async fn query_mesh_discovery_events_at( +pub(super) async fn query_mesh_discovery_events_at( state: &AppState, relay_url: &str, ) -> Result, String> { @@ -484,7 +484,7 @@ pub async fn mesh_start_node( let started = mesh_llm::DesktopMeshRuntime::start(request) .await .map_err(|error| format!("{error:#}"))?; - let status = match started.status().await { + let mut status = match started.status().await { Ok(status) => status, Err(error) => { let cleanup = started.stop().await; @@ -520,11 +520,23 @@ pub async fn mesh_start_node( // (leaves a warming node alone) can loop a slow-but-alive node, and an // unstartable config fails earlier in `start()`. Probe is informational. save_mesh_sharing_config(&app, config)?; - if let Err(error) = wait_for_mesh_inference(&config.model_id).await { - eprintln!( - "buzz-mesh: node started but inference is not ready yet ({error}); \ - leaving it to warm up (Share Compute stays armed for next launch)" - ); + match wait_for_mesh_inference(&config.model_id).await { + Ok(()) => { + // The status captured immediately after runtime installation can + // still say `starting`. Re-read after the readiness probe so the + // command response reflects the successful boot the user waited + // for instead of pinning the UI's transition indicator. + let runtime = state.mesh_llm_runtime.lock().await; + if let Some(running) = runtime.as_ref() { + status = running.status().await.map_err(|error| error.to_string())?; + } + } + Err(error) => { + eprintln!( + "buzz-mesh: node started but inference is not ready yet ({error}); \ + leaving it to warm up (Share Compute stays armed for next launch)" + ); + } } } mesh_llm::publish_current_status_once(&app, "start").await; diff --git a/desktop/src-tauri/src/commands/mesh_snapshot.rs b/desktop/src-tauri/src/commands/mesh_snapshot.rs new file mode 100644 index 0000000000..389e228439 --- /dev/null +++ b/desktop/src-tauri/src/commands/mesh_snapshot.rs @@ -0,0 +1,45 @@ +//! `mesh_snapshot` — community-wide shared-compute snapshot for the UI. +//! +//! Separate file from `commands/mesh_llm.rs` deliberately: that module is at +//! the 1000-line file ceiling enforced by `desktop/scripts/check-file-sizes.mjs`. +//! +//! Read-only. Queries the same member-status + membership filters routing uses, +//! then projects them through [`mesh_llm::snapshot_from_events`], which applies +//! the membership/binding/freshness rules. This never selects a serve target — +//! routing stays with `availability_from_events`. + +use tauri::State; + +use crate::app_state::AppState; +use crate::mesh_llm; + +type CmdResult = Result; + +/// Snapshot of who is sharing compute in this community right now. +/// +/// An empty mesh is a normal state, not an error: the snapshot carries a +/// `reason` string instead of failing, so the card can render an honest empty +/// state. A genuine relay/transport failure still returns `Err`. +#[tauri::command] +pub async fn mesh_snapshot(state: State<'_, AppState>) -> CmdResult { + // Reuse discovery's author-scoped, composite-cursor pagination. A direct + // two-filter query stops at the status filter's 100-event page size and + // silently undercounts larger communities. + let relay_url = crate::relay::relay_ws_url_with_override(&state); + let events = super::mesh_llm::query_mesh_discovery_events_at(&state, &relay_url) + .await + .map_err(|error| format!("Shared compute status query failed: {error}"))?; + + // Identify this member's own device so the card can say "including yours". + // A missing/locked identity is not fatal here — the snapshot is still + // useful, just without self-attribution. + let self_pubkey = state + .signing_keys() + .ok() + .map(|keys| keys.public_key().to_hex()); + + Ok(mesh_llm::snapshot_from_events( + events, + self_pubkey.as_deref(), + )) +} diff --git a/desktop/src-tauri/src/commands/mod.rs b/desktop/src-tauri/src/commands/mod.rs index c8184a0103..b56cfafc94 100644 --- a/desktop/src-tauri/src/commands/mod.rs +++ b/desktop/src-tauri/src/commands/mod.rs @@ -43,6 +43,8 @@ mod media_voice_note; pub(crate) mod mesh_llm; #[cfg(feature = "mesh-llm")] pub(crate) mod mesh_readiness; +#[cfg(feature = "mesh-llm")] +mod mesh_snapshot; mod messages; mod notifications; mod observer_archive; @@ -106,6 +108,8 @@ pub use media_fetch_cancellation::*; pub use media_raw::*; #[cfg(feature = "mesh-llm")] pub use mesh_llm::*; +#[cfg(feature = "mesh-llm")] +pub use mesh_snapshot::*; pub use messages::*; pub use notifications::*; pub use observer_archive::*; diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 43d7b03857..fe8fdfe6c7 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -728,6 +728,7 @@ pub fn run() { mesh_serving_usage, mesh_installed_models, mesh_model_catalog, + mesh_snapshot, update_managed_agent, discover_backend_providers, probe_backend_provider, diff --git a/desktop/src-tauri/src/mesh_llm/discovery.rs b/desktop/src-tauri/src/mesh_llm/discovery.rs index 2df4a7f8c5..8629ed62c9 100644 --- a/desktop/src-tauri/src/mesh_llm/discovery.rs +++ b/desktop/src-tauri/src/mesh_llm/discovery.rs @@ -12,7 +12,7 @@ use super::{dedupe_models, MeshAvailability, MeshModelOption, MeshServeTarget, M pub(super) const STATUS_FRESHNESS_SECS: u64 = 120; pub(crate) const MESH_STATUS_PAGE_SIZE: usize = 100; -fn status_is_fresh(event: &nostr::Event, now: u64) -> bool { +pub(super) fn status_is_fresh(event: &nostr::Event, now: u64) -> bool { event .created_at .as_secs() @@ -63,7 +63,7 @@ pub fn owner_ids_from_events(events: &[nostr::Event]) -> Vec { ids } -fn latest_membership_list(events: &[nostr::Event]) -> Option> { +pub(super) fn latest_membership_list(events: &[nostr::Event]) -> Option> { events .iter() .filter(|event| event.kind.as_u16() == 13_534) @@ -108,7 +108,7 @@ pub(crate) fn has_membership_snapshot(events: &[nostr::Event]) -> bool { events.iter().any(|event| event.kind.as_u16() == 13_534) } -fn owner_id_from_status_event(event: &nostr::Event) -> Option { +pub(super) fn owner_id_from_status_event(event: &nostr::Event) -> Option { let content = serde_json::from_str::(&event.content).ok()?; let owner_id = content .get("ownerId") @@ -136,7 +136,7 @@ fn owner_id_from_status_event(event: &nostr::Event) -> Option { Some(owner_id.to_string()) } -fn endpoint_binding_is_valid(event: &nostr::Event, content: &serde_json::Value) -> bool { +pub(super) fn endpoint_binding_is_valid(event: &nostr::Event, content: &serde_json::Value) -> bool { let Some(endpoint_tokens) = super::identity::advertised_endpoint_tokens(content) else { return false; }; diff --git a/desktop/src-tauri/src/mesh_llm/mod.rs b/desktop/src-tauri/src/mesh_llm/mod.rs index daab3c4b1c..57b9783f55 100644 --- a/desktop/src-tauri/src/mesh_llm/mod.rs +++ b/desktop/src-tauri/src/mesh_llm/mod.rs @@ -13,6 +13,9 @@ pub(crate) use discovery::{ }; use discovery::{device_name_from_status, endpoint_id_from_status, enrich_status_payload_identity}; +mod snapshot; +pub use snapshot::{snapshot_from_events, MeshSnapshot}; + mod catalog; pub(crate) use catalog::canonical_curated_model_id; pub use catalog::{model_catalog, MeshModelCatalog}; diff --git a/desktop/src-tauri/src/mesh_llm/snapshot.rs b/desktop/src-tauri/src/mesh_llm/snapshot.rs new file mode 100644 index 0000000000..b76b4bd12d --- /dev/null +++ b/desktop/src-tauri/src/mesh_llm/snapshot.rs @@ -0,0 +1,352 @@ +//! Community-wide mesh snapshot — an **allowlisted** projection of the member +//! status notes already on the relay. +//! +//! Every sharing member already publishes its whole SDK status payload as +//! kind:30003 content (see `mod.rs::status_report_payload`), but +//! `availability_from_events` reads only `serveTargets`/`models` and discards +//! the rest. This module projects the few extra fields the Share-compute UI +//! needs — per-device shared capacity, a coarse state, and a stable label — +//! and **nothing else**. +//! +//! Deliberately narrow, for two reasons: +//! 1. Routing must keep using [`super::availability_from_events`]. This +//! snapshot is presentation-only and must never select a serve target. +//! 2. The published payload also carries a hardware fingerprint, peer +//! topology, and operational counters. Naming each field we surface keeps +//! the UI from quietly growing into the rest of it. +//! +//! Capacity uses the payload's `my_vram_gb`, which mesh-llm derives from a +//! `vram_bytes` already clamped to the member's own `--max-vram` cap. Raw +//! `gpus[].vram_bytes` is *not* capped, so summing GPUs would advertise more +//! than the member agreed to share. + +use std::collections::BTreeMap; + +use serde::{Deserialize, Serialize}; + +use super::discovery::{ + device_name_from_status, endpoint_binding_is_valid, endpoint_id_from_status, + latest_membership_list, owner_id_from_status_event, status_is_fresh, STATUS_FRESHNESS_SECS, +}; +use super::{MeshServeTarget, MESH_STATUS_KIND}; + +/// Coarse per-device state. Intentionally not the SDK's `NodeState`: the UI +/// only needs "is this device giving compute, warming up, or just consuming?" +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum MeshDeviceState { + /// Advertising at least one routable model. + Serving, + /// Serve-mode, model not ready yet. + Loading, + /// Serve-mode participant with no model work (contributes capacity only). + Standby, + /// Client-mode: consuming someone else's compute, not sharing. + Consuming, +} + +/// One device visible on the mesh. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(rename_all = "camelCase")] +pub struct MeshSnapshotDevice { + /// Stable endpoint/device id, when the note carries one. + pub device_id: Option, + /// Human label: hostname when published, else a short endpoint id. + pub label: String, + /// Shared AI memory in GB, honoring the member's own cap. `None` when the + /// member is not currently running a node (a stopped member publishes only + /// its owner binding). + pub capacity_gb: Option, + /// Models this device is advertising right now. + pub models: Vec, + pub state: MeshDeviceState, + /// Whether this is the local member's own device. + pub is_self: bool, + /// Community member who signed this status note. Safe for profile lookup; + /// endpoint tokens, owner signatures, and raw hardware remain excluded. + pub member_pubkey: Option, + /// Relay event creation time, in Unix seconds. + pub reported_at: Option, + /// Approximate hosted-model footprint reported by MeshLLM, when available. + pub model_size_gb: Option, +} + +/// What the community's shared compute looks like right now. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +#[serde(rename_all = "camelCase")] +pub struct MeshSnapshot { + /// Devices advertising at least one routable model. + pub sharing_device_count: usize, + /// Unique current members with at least one ready serving device. + pub contributor_member_count: usize, + /// Summed shared capacity across sharing devices. `None` when no sharing + /// device published a figure — the UI then shows a count with no GB, never + /// a misleading `0 GB`. + pub shared_capacity_gb: Option, + /// Summed model footprint only when every sharing device reports it. + /// `None` preserves mixed-version honesty instead of presenting a partial sum. + pub allocated_capacity_gb: Option, + /// Distinct models ready to use, deduped across devices. + pub models: Vec, + /// Per-device rows, for the topology view. Sharing devices first. + pub devices: Vec, + /// True when the local member's own device is among the sharing devices. + pub includes_self: bool, + /// Members in the community right now (NIP-43 roster size). + /// + /// The denominator for "N of M sharing". Deliberately NOT used to estimate + /// unshared capacity: a member who never starts a node publishes no status + /// note, so their hardware is unknown by design — they never consented to + /// disclose it. Ghost nodes in the topology are a count, never GB. + pub member_count: usize, + /// When this projection was evaluated, in Unix seconds. + pub observed_at: u64, + /// Maximum age of a relay-reported serving status. + pub freshness_seconds: u64, + /// Why the snapshot is empty, when it is. Never a hard error: an empty + /// mesh is a normal, expected state. + pub reason: Option, +} + +impl MeshSnapshot { + fn empty(reason: impl Into) -> Self { + Self { + observed_at: nostr::Timestamp::now().as_secs(), + freshness_seconds: STATUS_FRESHNESS_SECS, + reason: Some(reason.into()), + ..Default::default() + } + } +} + +fn f64_field(content: &serde_json::Value, key: &str) -> Option { + content + .get(key) + .and_then(serde_json::Value::as_f64) + .filter(|value| value.is_finite() && *value > 0.0) +} + +/// Coarse state from the published `node_state`, falling back to whether the +/// note actually advertises routable targets. +fn device_state(content: &serde_json::Value, has_targets: bool) -> MeshDeviceState { + let raw = content + .get("node_state") + .and_then(serde_json::Value::as_str) + .map(str::trim) + .unwrap_or_default() + .to_ascii_lowercase(); + match raw.as_str() { + "serving" => MeshDeviceState::Serving, + "loading" => MeshDeviceState::Loading, + "client" => MeshDeviceState::Consuming, + "standby" => MeshDeviceState::Standby, + // Unknown/absent: trust the routable evidence rather than guessing. + _ if has_targets => MeshDeviceState::Serving, + _ => MeshDeviceState::Standby, + } +} + +/// Project validated member status notes into a presentation snapshot. +/// +/// Applies the same trust rules as routing — current NIP-43 membership, owner +/// binding, endpoint binding, and [`status_is_fresh`] — so a stale or +/// non-member note can never inflate the headline. Pure and total so it can be +/// unit-tested without a relay. +pub fn snapshot_from_events( + events: Vec, + self_member_pubkey: Option<&str>, +) -> MeshSnapshot { + if events.is_empty() { + return MeshSnapshot::empty("No shared compute has been published yet"); + } + let Some(members) = latest_membership_list(&events) else { + return MeshSnapshot::empty("Waiting for the current member roster"); + }; + + let self_pubkey = self_member_pubkey.map(str::to_ascii_lowercase); + let now = nostr::Timestamp::now().as_secs(); + // Status notes are replaceable per owner id, so key on that: two devices + // signed in as the same member each publish their own note. + let mut by_owner = BTreeMap::::new(); + + for event in events { + if event.kind.as_u16() as u64 != MESH_STATUS_KIND + || !status_is_fresh(&event, now) + || !members.contains(&event.pubkey.to_hex().to_ascii_lowercase()) + { + continue; + } + let Ok(content) = serde_json::from_str::(&event.content) else { + continue; + }; + let Some(owner_id) = owner_id_from_status_event(&event) else { + continue; + }; + if !endpoint_binding_is_valid(&event, &content) { + continue; + } + + // Reuse the routing-side target parse so "sharing" here means exactly + // what it means to the router: a validated, advertisable endpoint. + let targets = content + .get("serveTargets") + .or_else(|| content.get("serve_targets")) + .cloned() + .and_then(|value| serde_json::from_value::>(value).ok()) + .unwrap_or_default() + .into_iter() + .filter(|target| { + super::transport_policy::validate_advertised_endpoint(&target.endpoint_addr).is_ok() + }) + .collect::>(); + + let mut models = targets + .iter() + .map(|target| target.model_id.clone()) + .collect::>(); + models.sort(); + models.dedup(); + + let endpoint_id = endpoint_id_from_status(&content, None); + let is_self = self_pubkey + .as_deref() + .is_some_and(|pubkey| event.pubkey.to_hex().to_ascii_lowercase() == pubkey); + + by_owner.insert( + owner_id, + MeshSnapshotDevice { + label: device_name_from_status(&content, endpoint_id.as_deref()) + .unwrap_or_else(|| "Unknown device".to_string()), + device_id: endpoint_id, + capacity_gb: f64_field(&content, "my_vram_gb"), + state: device_state(&content, !models.is_empty()), + models, + is_self, + member_pubkey: Some(event.pubkey.to_hex()), + reported_at: Some(event.created_at.as_secs()), + model_size_gb: f64_field(&content, "model_size_gb"), + }, + ); + } + + let mut devices = by_owner.into_values().collect::>(); + // Sharing devices first, then the local device, then by label so the list + // is stable across polls (no jitter in the topology view). + devices.sort_by(|a, b| { + let sharing = |d: &MeshSnapshotDevice| d.state == MeshDeviceState::Serving; + sharing(b) + .cmp(&sharing(a)) + .then_with(|| b.is_self.cmp(&a.is_self)) + .then_with(|| a.label.cmp(&b.label)) + }); + + let sharing = devices + .iter() + .filter(|device| device.state == MeshDeviceState::Serving) + .collect::>(); + let sharing_device_count = sharing.len(); + let contributor_member_count = sharing + .iter() + .filter_map(|device| device.member_pubkey.as_ref()) + .collect::>() + .len(); + let includes_self = sharing.iter().any(|device| device.is_self); + // Sum only reported figures; keep `None` when nobody reported one so the + // UI can drop the number instead of printing `0 GB`. + let shared_capacity_gb = sharing + .iter() + .filter_map(|device| device.capacity_gb) + .fold(None::, |acc, gb| Some(acc.unwrap_or(0.0) + gb)); + let allocated_capacity_gb = (!sharing.is_empty() + && sharing.iter().all(|device| device.model_size_gb.is_some())) + .then(|| { + sharing + .iter() + .filter_map(|device| device.model_size_gb) + .sum() + }); + let mut models = sharing + .iter() + .flat_map(|device| device.models.iter().cloned()) + .collect::>(); + models.sort(); + models.dedup(); + + let reason = if sharing_device_count == 0 { + Some("No one is sharing compute yet".to_string()) + } else { + None + }; + + MeshSnapshot { + member_count: members.len(), + contributor_member_count, + sharing_device_count, + shared_capacity_gb, + allocated_capacity_gb, + models, + devices, + includes_self, + observed_at: now, + freshness_seconds: STATUS_FRESHNESS_SECS, + reason, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn empty_events_report_a_reason_not_an_error() { + let snapshot = snapshot_from_events(Vec::new(), None); + assert_eq!(snapshot.sharing_device_count, 0); + assert_eq!(snapshot.shared_capacity_gb, None); + assert!(snapshot.reason.is_some()); + assert_eq!(snapshot.freshness_seconds, STATUS_FRESHNESS_SECS); + assert!(snapshot.observed_at > 0); + } + + #[test] + fn missing_membership_snapshot_is_not_treated_as_an_empty_mesh() { + // A relay gap must not read as "nobody is sharing". + let event = nostr::EventBuilder::new(nostr::Kind::Custom(30003), "{}") + .sign_with_keys(&nostr::Keys::generate()) + .expect("test event signs"); + let snapshot = snapshot_from_events(vec![event], None); + assert_eq!( + snapshot.reason.as_deref(), + Some("Waiting for the current member roster") + ); + } + + #[test] + fn unknown_node_state_falls_back_to_routable_evidence() { + assert_eq!( + device_state(&serde_json::json!({}), true), + MeshDeviceState::Serving + ); + assert_eq!( + device_state(&serde_json::json!({}), false), + MeshDeviceState::Standby + ); + assert_eq!( + device_state(&serde_json::json!({"node_state": "client"}), false), + MeshDeviceState::Consuming + ); + } + + #[test] + fn capacity_ignores_zero_and_nonfinite_reports() { + // A node that has not surveyed yet publishes 0.0; that must not count + // as a real reported capacity. + assert_eq!( + f64_field(&serde_json::json!({"my_vram_gb": 0.0}), "my_vram_gb"), + None + ); + assert_eq!( + f64_field(&serde_json::json!({"my_vram_gb": 36.5}), "my_vram_gb"), + Some(36.5) + ); + } +} diff --git a/desktop/src-tauri/src/mesh_llm_stubs.rs b/desktop/src-tauri/src/mesh_llm_stubs.rs index e8c13f48ea..5b93cf8c76 100644 --- a/desktop/src-tauri/src/mesh_llm_stubs.rs +++ b/desktop/src-tauri/src/mesh_llm_stubs.rs @@ -42,3 +42,8 @@ pub async fn mesh_installed_models( pub async fn mesh_model_catalog() -> CmdResult { Err("mesh-llm feature not enabled".to_string()) } + +#[tauri::command] +pub async fn mesh_snapshot(_state: State<'_, AppState>) -> CmdResult { + Err("mesh-llm feature not enabled".to_string()) +} diff --git a/desktop/src/features/agents/AGENTS.md b/desktop/src/features/agents/AGENTS.md index 7f1f73b175..67eba2c60f 100644 --- a/desktop/src/features/agents/AGENTS.md +++ b/desktop/src/features/agents/AGENTS.md @@ -381,6 +381,8 @@ buzz messages send --channel --reply-to \ - Rust: `definition_validation` and inbound persona tests pin the shared Unicode/control-character policy at local, import, publish, and sync gates. +16. **Community-mesh agent creation uses the normal create dialog with a named preset.** The Compute settings CTA requests the `community-mesh` create preset; the app-level create-dialog owner resolves it to Buzz Agent with the `relay-mesh` provider and `auto` model, then passes those values through `AgentDialog`. Keep this as a preset of the canonical create flow—never add a second mesh-specific form or bypass normal runtime catalog validation and persistence. + ## Keep this file true **If you change how agent configuration is modeled, rendered, persisted, diff --git a/desktop/src/features/agents/openCreateAgentEvent.ts b/desktop/src/features/agents/openCreateAgentEvent.ts index 78bf59eb9d..4504dbfb77 100644 --- a/desktop/src/features/agents/openCreateAgentEvent.ts +++ b/desktop/src/features/agents/openCreateAgentEvent.ts @@ -3,6 +3,7 @@ const OPEN_CREATE_AGENT_EVENT = "buzz:open-create-agent"; export type OpenCreateAgentOptions = { channelId?: string; channelName?: string; + preset?: "community-mesh"; }; let pendingOpenCreateAgent: OpenCreateAgentOptions | null = null; diff --git a/desktop/src/features/agents/ui/RequestedAgentCreateDialogs.tsx b/desktop/src/features/agents/ui/RequestedAgentCreateDialogs.tsx index 4b04de638d..f63b598776 100644 --- a/desktop/src/features/agents/ui/RequestedAgentCreateDialogs.tsx +++ b/desktop/src/features/agents/ui/RequestedAgentCreateDialogs.tsx @@ -5,6 +5,7 @@ import { subscribeOpenCreateAgent, type OpenCreateAgentOptions, } from "@/features/agents/openCreateAgentEvent"; +import type { CreatePersonaInput } from "@/shared/api/types"; import { AgentDialog } from "./AgentDialog"; import { usePersonaActions } from "./usePersonaActions"; @@ -16,9 +17,23 @@ export function RequestedAgentCreateDialogs() { name: string; } | null>(null); const [isOpen, setIsOpen] = React.useState(false); + const [initialValues, setInitialValues] = + React.useState(null); const openCreate = React.useEffectEvent((options: OpenCreateAgentOptions) => { personas.prepareCreate(); + setInitialValues( + options.preset === "community-mesh" + ? { + displayName: "Community agent", + avatarUrl: "", + systemPrompt: "", + runtime: "buzz-agent", + provider: "relay-mesh", + model: "auto", + } + : null, + ); setTargetChannel( options.channelId && options.channelName ? { id: options.channelId, name: options.channelName } @@ -42,12 +57,14 @@ export function RequestedAgentCreateDialogs() { ? personas.createPersonaMutation.error : null } + initialValues={initialValues} isDefinitionPending={personas.isPending} mode="definition" onOpenChange={(open) => { if (!open) { setIsOpen(false); setTargetChannel(null); + setInitialValues(null); } }} onSubmitDefinition={(input, intent, backendIntent) => diff --git a/desktop/src/features/mesh-compute/communityComputeFixtures.ts b/desktop/src/features/mesh-compute/communityComputeFixtures.ts new file mode 100644 index 0000000000..e88f672680 --- /dev/null +++ b/desktop/src/features/mesh-compute/communityComputeFixtures.ts @@ -0,0 +1,234 @@ +import type { + CommunityComputeDeviceInput, + CommunityComputeSnapshotInput, +} from "./communityComputeMapModel"; + +export type CommunityComputeTutorialFixtureId = + | "empty" + | "single-device" + | "small-company" + | "large-dense-500"; + +export type CommunityComputeTutorialFixture = { + id: CommunityComputeTutorialFixtureId; + title: string; + description: string; + snapshot: CommunityComputeSnapshotInput; +}; + +const MODEL_REFS = [ + "unsloth/gemma-4-26B-A4B-it-GGUF:UD-Q4_K_M", + "Qwen/Qwen3.5-35B-A3B-GGUF:Q4_K_M", + "mistralai/Ministral-3-14B-Instruct-GGUF:Q4_K_M", + "bartowski/Llama-3.3-70B-Instruct-GGUF:Q4_K_M", +] as const; + +export const EMPTY_COMMUNITY_COMPUTE_FIXTURE: CommunityComputeTutorialFixture = + { + id: "empty", + title: "New community", + description: "No member has shared compute yet.", + snapshot: { + sharingDeviceCount: 0, + contributorMemberCount: 0, + sharedCapacityGb: null, + allocatedCapacityGb: null, + models: [], + devices: [], + includesSelf: false, + memberCount: 8, + reason: "No sharing devices found", + }, + }; + +export const SINGLE_DEVICE_COMMUNITY_COMPUTE_FIXTURE: CommunityComputeTutorialFixture = + { + id: "single-device", + title: "First deployment", + description: "One member is sharing a single workstation.", + snapshot: snapshotFromDevices( + [ + device({ + index: 0, + id: "studio-workstation", + label: "Morgan’s Studio", + capacityGb: 24, + modelSizeGb: 18, + isSelf: true, + }), + ], + 1, + ), + }; + +export const SMALL_COMPANY_COMMUNITY_COMPUTE_FIXTURE: CommunityComputeTutorialFixture = + { + id: "small-company", + title: "Small company", + description: "A mixed fleet across a small distributed team.", + snapshot: snapshotFromDevices( + [ + ...[ + { capacityGb: 24, modelSizeGb: 18 }, + { capacityGb: 24, modelSizeGb: 18 }, + { capacityGb: 36, modelSizeGb: 28 }, + { capacityGb: 36, modelSizeGb: 28 }, + { capacityGb: 48, modelSizeGb: 36 }, + { capacityGb: 48, modelSizeGb: 36 }, + { capacityGb: 64, modelSizeGb: 48 }, + { capacityGb: 64, modelSizeGb: 48 }, + { capacityGb: 96, modelSizeGb: 72 }, + { capacityGb: 96, modelSizeGb: 72 }, + ].map(({ capacityGb, modelSizeGb }, index) => + device({ + index, + id: `team-macbook-${String(index + 1).padStart(2, "0")}`, + label: `Team MacBook ${String(index + 1).padStart(2, "0")}`, + capacityGb, + modelSizeGb, + isSelf: index === 0, + }), + ), + device({ + index: 10, + id: "intensive-model-host-a", + label: "Intensive model host A", + capacityGb: 768, + modelRef: "unsloth/Kimi-K2-Thinking-GGUF:UD-Q4_K_XL", + modelSizeGb: 646, + }), + device({ + index: 11, + id: "intensive-model-host-b", + label: "Intensive model host B", + capacityGb: 1_440, + modelRef: "unsloth/DeepSeek-V3.2-GGUF:UD-Q4_K_XL", + modelSizeGb: 1_000, + }), + ], + 24, + ), + }; + +/** Deterministic stress fixture using only the production snapshot contract. */ +export const LARGE_DENSE_COMMUNITY_COMPUTE_FIXTURE: CommunityComputeTutorialFixture = + { + id: "large-dense-500", + title: "Large company", + description: + "Five hundred deterministic reporting devices for density testing.", + snapshot: createLargeDenseSnapshot(500), + }; + +export const COMMUNITY_COMPUTE_TUTORIAL_FIXTURES: Readonly< + Record +> = { + empty: EMPTY_COMMUNITY_COMPUTE_FIXTURE, + "single-device": SINGLE_DEVICE_COMMUNITY_COMPUTE_FIXTURE, + "small-company": SMALL_COMPANY_COMMUNITY_COMPUTE_FIXTURE, + "large-dense-500": LARGE_DENSE_COMMUNITY_COMPUTE_FIXTURE, +}; + +export function createLargeDenseSnapshot( + deviceCount = 500, +): CommunityComputeSnapshotInput { + const count = Math.max(0, Math.floor(deviceCount)); + const capacities = [8, 16, 24, 32, 48, 64, 96, 128] as const; + const devices = Array.from({ length: count }, (_, index) => { + const capacityGb = capacities[index % capacities.length]; + const state = + index % 41 === 0 ? "loading" : index % 17 === 0 ? "standby" : "serving"; + const largeIndex = [49, 149, 249, 349, 449, 499].indexOf(index); + const largeSizes = [170, 230, 350, 480, 700, 1_000] as const; + const largeCapacities = [192, 256, 384, 512, 768, 1_440] as const; + const modelSizeGb = + largeIndex >= 0 + ? largeSizes[largeIndex] + : roundCapacity(capacityGb * 0.72); + return device({ + index, + id: `dense-device-${String(index + 1).padStart(4, "0")}`, + label: + largeIndex >= 0 + ? `${largeCapacities[largeIndex]} GB model host` + : `Compute ${String(index + 1).padStart(4, "0")}`, + capacityGb: largeIndex >= 0 ? largeCapacities[largeIndex] : capacityGb, + modelRef: + largeIndex >= 0 + ? `community/Large-${modelSizeGb}GB-Instruct:FP8` + : undefined, + modelSizeGb, + state, + models: state === "serving" ? undefined : [], + isSelf: index === 0, + }); + }); + return snapshotFromDevices(devices, Math.ceil(count * 1.37)); +} + +type DeviceOptions = { + index: number; + id: string; + label: string; + capacityGb: number; + modelRef?: string; + modelSizeGb?: number; + models?: string[]; + state?: CommunityComputeDeviceInput["state"]; + isSelf?: boolean; +}; + +function device(options: DeviceOptions): CommunityComputeDeviceInput { + const model = + options.modelRef ?? MODEL_REFS[options.index % MODEL_REFS.length]; + return { + deviceId: options.id, + label: options.label, + capacityGb: options.capacityGb, + models: options.models ?? [model], + state: options.state ?? "serving", + isSelf: options.isSelf ?? false, + memberPubkey: `tutorial-member-${String(options.index + 1).padStart(4, "0")}`, + reportedAt: 1_786_032_000, + modelSizeGb: options.modelSizeGb ?? null, + }; +} + +function roundCapacity(value: number): number { + return Math.round(value * 10) / 10; +} + +function snapshotFromDevices( + devices: CommunityComputeDeviceInput[], + memberCount: number, +): CommunityComputeSnapshotInput { + const serving = devices.filter( + (item) => item.state === "serving" && item.models.length > 0, + ); + const models = [...new Set(serving.flatMap((item) => item.models))].sort(); + const sharedCapacityGb = completeSum(serving.map((item) => item.capacityGb)); + const allocatedCapacityGb = completeSum( + serving.map((item) => item.modelSizeGb ?? null), + ); + return { + sharingDeviceCount: serving.length, + contributorMemberCount: new Set( + serving.map((item) => item.memberPubkey ?? item.deviceId), + ).size, + sharedCapacityGb, + allocatedCapacityGb, + models, + devices, + includesSelf: serving.some((item) => item.isSelf), + memberCount, + observedAt: 1_786_032_000, + freshnessSeconds: 120, + reason: serving.length === 0 ? "No sharing devices found" : null, + }; +} + +function completeSum(values: readonly (number | null)[]): number | null { + return values.length > 0 && values.every((value) => value !== null) + ? values.reduce((total, value) => total + (value ?? 0), 0) + : null; +} diff --git a/desktop/src/features/mesh-compute/communityComputeHexLayout.test.mjs b/desktop/src/features/mesh-compute/communityComputeHexLayout.test.mjs new file mode 100644 index 0000000000..ee14e9032d --- /dev/null +++ b/desktop/src/features/mesh-compute/communityComputeHexLayout.test.mjs @@ -0,0 +1,358 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + hexCoordinateKey, + hexDistance, + hexNeighbours, + hexTerritoryBoundaryEdges, + hexTerritoryInteriorEdges, + isContiguousHexTerritory, + layoutCommunityComputeHexTerritories, +} from "./communityComputeHexLayout.ts"; +import { LARGE_DENSE_COMMUNITY_COMPUTE_FIXTURE } from "./communityComputeFixtures.ts"; +import { deriveCommunityComputeMapModel } from "./communityComputeMapModel.ts"; + +function assertStructuralInvariants(layout) { + const occupied = new Set(layout.cells.map(hexCoordinateKey)); + assert.equal(occupied.size, layout.cells.length, "cells must not overlap"); + assert.ok( + layout.territories.every((territory) => + isContiguousHexTerritory(territory.cells), + ), + "every territory must be contiguous", + ); + if (layout.cells.length > 0) { + const reached = new Set([hexCoordinateKey(layout.cells[0])]); + const queue = [layout.cells[0]]; + for (let cursor = 0; cursor < queue.length; cursor += 1) { + for (const neighbour of hexNeighbours(queue[cursor])) { + const key = hexCoordinateKey(neighbour); + if (occupied.has(key) && !reached.has(key)) { + reached.add(key); + queue.push(neighbour); + } + } + } + assert.equal( + reached.size, + occupied.size, + "occupied mesh must be connected", + ); + } +} + +function organicGaps(layout) { + const occupied = new Set(layout.cells.map(hexCoordinateKey)); + const empty = new Map(); + for (const cell of layout.cells) { + for (const neighbour of hexNeighbours(cell)) { + const key = hexCoordinateKey(neighbour); + if (!occupied.has(key)) empty.set(key, neighbour); + } + } + return [...empty.values()].filter( + (cell) => + hexNeighbours(cell).filter((neighbour) => + occupied.has(hexCoordinateKey(neighbour)), + ).length >= 2, + ); +} + +function assertBalancedSparseFrontier(layout, deploymentCount) { + const points = layout.cells.map((cell) => ({ + x: Math.sqrt(3) * (cell.q + cell.r / 2), + y: 1.5 * cell.r, + })); + const width = + Math.max(...points.map(({ x }) => x)) - + Math.min(...points.map(({ x }) => x)) + + Math.sqrt(3); + const height = + Math.max(...points.map(({ y }) => y)) - + Math.min(...points.map(({ y }) => y)) + + 2; + const aspectRatio = width / height; + assert.ok( + aspectRatio >= 0.8 && aspectRatio <= 1.3, + `${deploymentCount} deployments have unbalanced aspect ratio ${aspectRatio}`, + ); + + const radius = Math.max( + ...layout.cells.map((cell) => hexDistance(cell, { q: 0, r: 0 })), + ); + const directionalRadii = Array(6).fill(0); + for (const cell of layout.cells) { + const projections = [ + cell.q, + -cell.r, + -cell.q - cell.r, + -cell.q, + cell.r, + cell.q + cell.r, + ]; + for (const [index, projection] of projections.entries()) { + directionalRadii[index] = Math.max(directionalRadii[index], projection); + } + } + assert.ok( + Math.max(...directionalRadii) - Math.min(...directionalRadii) <= 2, + `${deploymentCount} deployments must reach every side evenly`, + ); + + const outerOccupancy = + layout.cells.filter((cell) => hexDistance(cell, { q: 0, r: 0 }) === radius) + .length / + (6 * radius); + assert.ok( + outerOccupancy >= 0.1 && outerOccupancy <= 0.55, + `${deploymentCount} deployments need a sparse, notched outer edge (${outerOccupancy})`, + ); +} + +function deployments(count) { + const buckets = [1, 2, 4, 7]; + return Array.from({ length: count }, (_, index) => ({ + id: `deployment-${String(index + 1).padStart(4, "0")}`, + cellCount: buckets[index % buckets.length], + })); +} + +test("layout is deterministic and independent of input order", () => { + const input = deployments(80); + const first = layoutCommunityComputeHexTerritories(input); + const second = layoutCommunityComputeHexTerritories([...input].reverse()); + assert.deepEqual(first, second); +}); + +test("growing layouts stay balanced with sparse outer frontiers", () => { + for (const count of [24, 250, 1_000]) { + const layout = layoutCommunityComputeHexTerritories(deployments(count)); + assertStructuralInvariants(layout); + assertBalancedSparseFrontier(layout, count); + assert.ok( + organicGaps(layout).length > 0, + `${count} deployments should leave deterministic pockets or edge gaps`, + ); + } +}); + +test("canonical territory order remains stable when previous placement is reused", () => { + const input = deployments(60); + const initial = layoutCommunityComputeHexTerritories(input); + const next = layoutCommunityComputeHexTerritories([...input].reverse(), { + previous: initial, + }); + assert.deepEqual( + next.territories.map(({ deploymentId }) => deploymentId), + initial.territories.map(({ deploymentId }) => deploymentId), + ); +}); + +test("payload changes preserve positions because IDs own the slots", () => { + const initial = deployments(40); + const changed = initial.map((deployment) => ({ + ...deployment, + label: `Changed ${deployment.id}`, + health: "degraded", + })); + assert.deepEqual( + layoutCommunityComputeHexTerritories(initial), + layoutCommunityComputeHexTerritories(changed), + ); +}); + +test("surviving IDs keep positions as deployments are added and removed", () => { + const initialInput = deployments(40); + const initial = layoutCommunityComputeHexTerritories(initialInput); + const nextInput = [ + ...initialInput.filter((_, index) => index % 4 !== 0), + ...deployments(6).map((item, index) => ({ + ...item, + id: `new-deployment-${index}`, + })), + ]; + const next = layoutCommunityComputeHexTerritories(nextInput, { + previous: initial, + }); + const initialCenters = new Map( + initial.territories.map((territory) => [ + territory.deploymentId, + territory.center, + ]), + ); + + for (const territory of next.territories) { + const oldCenter = initialCenters.get(territory.deploymentId); + if (oldCenter !== undefined) { + assert.deepEqual(territory.center, oldCenter); + } + } + assert.equal( + new Set(next.cells.map(hexCoordinateKey)).size, + next.cells.length, + ); +}); + +test("every supported territory shape is contiguous", () => { + const layout = layoutCommunityComputeHexTerritories([ + { id: "one", cellCount: 1 }, + { id: "two", cellCount: 2 }, + { id: "four", cellCount: 4 }, + { id: "seven", cellCount: 7 }, + ]); + const expected = new Map([ + ["one", 1], + ["two", 2], + ["four", 4], + ["seven", 7], + ]); + + for (const territory of layout.territories) { + assert.equal(territory.cells.length, expected.get(territory.deploymentId)); + assert.ok( + isContiguousHexTerritory(territory.cells), + `${territory.deploymentId} must be contiguous`, + ); + } +}); + +test("fused boundaries omit interior edges for 1, 2, 4, and 7 cells", () => { + const shapes = [ + [{ q: 0, r: 0 }], + [ + { q: 0, r: 0 }, + { q: 1, r: 0 }, + ], + [ + { q: 0, r: 0 }, + { q: 1, r: 0 }, + { q: 1, r: -1 }, + { q: 0, r: -1 }, + ], + [{ q: 0, r: 0 }, ...hexNeighbours({ q: 0, r: 0 })], + ]; + + for (const [index, cells] of shapes.entries()) { + const sharedPairs = + cells.reduce( + (total, cell) => + total + + hexNeighbours(cell).filter((neighbour) => + cells.some( + (candidate) => + hexCoordinateKey(candidate) === hexCoordinateKey(neighbour), + ), + ).length, + 0, + ) / 2; + const boundary = hexTerritoryBoundaryEdges(cells); + assert.equal(boundary.length, 6 * cells.length - 2 * sharedPairs); + + const endpointDegree = new Map(); + for (const edge of boundary) { + assert.notDeepEqual(edge.start, edge.end); + for (const point of [edge.start, edge.end]) { + const key = `${point.x.toFixed(8)},${point.y.toFixed(8)}`; + endpointDegree.set(key, (endpointDegree.get(key) ?? 0) + 1); + } + } + assert.ok( + [...endpointDegree.values()].every((degree) => degree === 2), + `shape ${index} boundary must form closed loops without dangling edges`, + ); + } + + const edgeKey = (edge) => + [edge.start, edge.end] + .map((point) => `${point.x.toFixed(8)},${point.y.toFixed(8)}`) + .sort() + .join("|"); + assert.deepEqual( + hexTerritoryBoundaryEdges(shapes[3]).map(edgeKey).sort(), + hexTerritoryBoundaryEdges([...shapes[3]].reverse()) + .map(edgeKey) + .sort(), + "edge membership must not depend on cell order", + ); +}); + +test("interior edges contain each shared device boundary exactly once", () => { + const cells = [{ q: 0, r: 0 }, ...hexNeighbours({ q: 0, r: 0 })]; + const interior = hexTerritoryInteriorEdges(cells); + const edgeKey = (edge) => + [edge.start, edge.end] + .map((point) => `${point.x.toFixed(8)},${point.y.toFixed(8)}`) + .sort() + .join("|"); + + assert.equal(interior.length, 12); + assert.equal(new Set(interior.map(edgeKey)).size, interior.length); + assert.deepEqual( + interior.map(edgeKey).sort(), + hexTerritoryInteriorEdges([...cells].reverse()) + .map(edgeKey) + .sort(), + "shared edge membership must not depend on cell order", + ); + assert.deepEqual(hexTerritoryInteriorEdges([{ q: 0, r: 0 }]), []); +}); + +test("territories never emit duplicate occupied cells", () => { + const layout = layoutCommunityComputeHexTerritories(deployments(250)); + const coordinates = layout.cells.map(hexCoordinateKey); + assert.equal(new Set(coordinates).size, coordinates.length); + assert.equal( + new Set(layout.cells.map((cell) => cell.id)).size, + layout.cells.length, + ); +}); + +test("duplicate deployment IDs are rejected rather than overlapped", () => { + assert.throws( + () => + layoutCommunityComputeHexTerritories([ + { id: "same", cellCount: 1 }, + { id: "same", cellCount: 7 }, + ]), + /Duplicate community compute deployment ID: same/, + ); +}); + +test("empty layout has no fabricated bounds", () => { + assert.deepEqual(layoutCommunityComputeHexTerritories([]), { + territories: [], + cells: [], + bounds: null, + }); +}); + +test("500-deployment fixture lays out uniquely and contiguously", () => { + const model = deriveCommunityComputeMapModel( + LARGE_DENSE_COMMUNITY_COMPUTE_FIXTURE.snapshot, + ); + const input = model.deployments.map(({ id, cellCount }) => ({ + id, + cellCount, + })); + const layout = layoutCommunityComputeHexTerritories(input); + + assert.equal(layout.territories.length, model.deployments.length); + assert.equal( + layout.cells.length, + model.deployments.reduce( + (total, deployment) => total + deployment.cellCount, + 0, + ), + ); + assert.equal( + new Set(layout.cells.map(hexCoordinateKey)).size, + layout.cells.length, + ); + assert.ok( + layout.territories.every((territory) => + isContiguousHexTerritory(territory.cells), + ), + ); + assert.ok(layout.bounds !== null); +}); diff --git a/desktop/src/features/mesh-compute/communityComputeHexLayout.ts b/desktop/src/features/mesh-compute/communityComputeHexLayout.ts new file mode 100644 index 0000000000..aac860d99a --- /dev/null +++ b/desktop/src/features/mesh-compute/communityComputeHexLayout.ts @@ -0,0 +1,448 @@ +import type { HexCellCount } from "./communityComputeMapModel"; + +/** Axial coordinates for a pointy-top hex grid. */ +export type HexCoordinate = { + q: number; + r: number; +}; + +/** Cartesian point in pointy-top hex geometry. */ +export type HexPoint = { + x: number; + y: number; +}; + +/** One exposed side of a fused hex territory boundary. */ +export type HexBoundaryEdge = { + start: HexPoint; + end: HexPoint; +}; + +export type CommunityComputeLayoutInput = { + id: string; + cellCount: HexCellCount; +}; + +export type CommunityComputeHexCell = HexCoordinate & { + id: string; + deploymentId: string; + index: number; +}; + +export type CommunityComputeHexTerritory = { + deploymentId: string; + center: HexCoordinate; + cells: readonly CommunityComputeHexCell[]; +}; + +export type CommunityComputeHexBounds = { + minQ: number; + maxQ: number; + minR: number; + maxR: number; +}; + +export type CommunityComputeHexLayout = { + territories: readonly CommunityComputeHexTerritory[]; + cells: readonly CommunityComputeHexCell[]; + bounds: CommunityComputeHexBounds | null; +}; + +export type CommunityComputeHexLayoutOptions = { + /** + * Reuse centres for IDs that are still present. Supplying the last layout + * keeps a live map steady while deployments arrive, leave, or change size. + */ + previous?: CommunityComputeHexLayout | null; +}; + +const HEX_DIRECTIONS: readonly HexCoordinate[] = [ + { q: 1, r: 0 }, + { q: 1, r: -1 }, + { q: 0, r: -1 }, + { q: -1, r: 0 }, + { q: -1, r: 1 }, + { q: 0, r: 1 }, +]; + +/** + * Lay deployments out as one organic, edge-connected honeycomb. + * + * Input order never affects the result. Stable hashes choose territory + * orientation, while a deterministic low-discrepancy permutation spreads each + * ring's placements around the whole current frontier. Completed inner rings + * remain dense; only the growing outer layers carry deliberate notches. With a + * previous layout, collision-free surviving shapes retain their exact cells; + * new shapes take the first non-overlapping frontier position that shares an + * edge with the occupied mesh. + */ +export function layoutCommunityComputeHexTerritories( + deployments: readonly CommunityComputeLayoutInput[], + options: CommunityComputeHexLayoutOptions = {}, +): CommunityComputeHexLayout { + const seenIds = new Set(); + for (const deployment of deployments) { + if (seenIds.has(deployment.id)) { + throw new Error( + `Duplicate community compute deployment ID: ${deployment.id}`, + ); + } + seenIds.add(deployment.id); + } + + const ordered = [...deployments].sort( + (left, right) => + stableHash(left.id) - stableHash(right.id) || + left.id.localeCompare(right.id), + ); + // Previous cells are retained only when they remain collision-free. New + // territories search the compact spiral for the first candidate that both + // fits and touches the occupied frontier. + const previousById = new Map( + (options.previous?.territories ?? []).map((territory) => [ + territory.deploymentId, + territory, + ]), + ); + const retainableIds = connectedRetainedTerritoryIds(ordered, previousById); + const occupied = new Set(); + const territories: CommunityComputeHexTerritory[] = []; + const placementOrder = [...ordered].sort( + (left, right) => + Number(retainableIds.has(right.id)) - + Number(retainableIds.has(left.id)) || + ordered.indexOf(left) - ordered.indexOf(right), + ); + + for (const deployment of placementOrder) { + const previous = retainableIds.has(deployment.id) + ? previousById.get(deployment.id) + : undefined; + const retained = + previous?.cells.length === deployment.cellCount + ? previous.cells.map(({ q, r }) => ({ q, r })) + : null; + const retainedFits = retained?.every( + (cell) => !occupied.has(hexCoordinateKey(cell)), + ); + const orientation = stableHash(`${deployment.id}:orientation`) % 6; + let center = retainedFits ? previous?.center : undefined; + let coordinates = retainedFits ? retained : null; + + if (!coordinates || !center) { + for (let slot = 0; ; slot += 1) { + const candidate = balancedSpiralCoordinate(slot); + const offsets = territoryOffsets(deployment.cellCount, orientation); + const candidateCells = offsets.map((offset) => ({ + q: candidate.q + offset.q, + r: candidate.r + offset.r, + })); + const fits = candidateCells.every( + (cell) => !occupied.has(hexCoordinateKey(cell)), + ); + const touches = + occupied.size === 0 || + candidateCells.some((cell) => + hexNeighbours(cell).some((neighbour) => + occupied.has(hexCoordinateKey(neighbour)), + ), + ); + if (fits && touches) { + center = candidate; + coordinates = candidateCells; + break; + } + } + } + + const cells = coordinates.map(({ q, r }, index) => { + occupied.add(hexCoordinateKey({ q, r })); + return { + id: `${deployment.id}@${q},${r}`, + deploymentId: deployment.id, + index, + q, + r, + }; + }); + territories.push({ deploymentId: deployment.id, center, cells }); + } + territories.sort( + (left, right) => + ordered.findIndex((item) => item.id === left.deploymentId) - + ordered.findIndex((item) => item.id === right.deploymentId), + ); + const cells = territories.flatMap((territory) => territory.cells); + + return { territories, cells, bounds: findBounds(cells) }; +} + +/** The six axial neighbours of a cell, in stable clockwise order. */ +export function hexNeighbours(coordinate: HexCoordinate): HexCoordinate[] { + return HEX_DIRECTIONS.map((direction) => ({ + q: coordinate.q + direction.q, + r: coordinate.r + direction.r, + })); +} + +/** + * Return only the exposed unit-hex sides of a territory. + * + * Coordinates use pointy-top axial geometry. Shared sides are omitted, so the + * result can be stroked as one fused territory rather than as overlapping + * individual hexagons. Edge direction is stable and follows each cell's + * perimeter; callers may chain matching endpoints when an ordered path is + * required. + */ +export function hexTerritoryBoundaryEdges( + cells: readonly HexCoordinate[], +): HexBoundaryEdge[] { + const occupied = new Set(cells.map(hexCoordinateKey)); + const edges: HexBoundaryEdge[] = []; + + for (const cell of cells) { + for (let direction = 0; direction < HEX_DIRECTIONS.length; direction += 1) { + const neighbour = { + q: cell.q + HEX_DIRECTIONS[direction].q, + r: cell.r + HEX_DIRECTIONS[direction].r, + }; + if (!occupied.has(hexCoordinateKey(neighbour))) { + edges.push(hexEdge(cell, direction)); + } + } + } + + return edges; +} + +/** + * Return each side shared by two cells in a territory exactly once. + * + * Fused model territories do not render these edges. Split deployments use + * them to show the device boundaries without drawing inset outlines around + * every cell or accidentally doubling a shared stroke. + */ +export function hexTerritoryInteriorEdges( + cells: readonly HexCoordinate[], +): HexBoundaryEdge[] { + const occupied = new Set(cells.map(hexCoordinateKey)); + const edges: HexBoundaryEdge[] = []; + + for (const cell of cells) { + const cellKey = hexCoordinateKey(cell); + for (let direction = 0; direction < HEX_DIRECTIONS.length; direction += 1) { + const neighbour = { + q: cell.q + HEX_DIRECTIONS[direction].q, + r: cell.r + HEX_DIRECTIONS[direction].r, + }; + const neighbourKey = hexCoordinateKey(neighbour); + if (occupied.has(neighbourKey) && cellKey < neighbourKey) { + edges.push(hexEdge(cell, direction)); + } + } + } + + return edges; +} + +function hexEdge(cell: HexCoordinate, direction: number): HexBoundaryEdge { + const center = { + x: Math.sqrt(3) * (cell.q + cell.r / 2), + y: 1.5 * cell.r, + }; + const corners = Array.from({ length: 6 }, (_, index) => { + const angle = ((-30 + index * 60) * Math.PI) / 180; + return { + x: normalizeHexPointComponent(center.x + Math.cos(angle)), + y: normalizeHexPointComponent(center.y + Math.sin(angle)), + }; + }); + return { + start: corners[(6 - direction) % 6], + end: corners[(7 - direction) % 6], + }; +} + +function normalizeHexPointComponent(value: number): number { + return Math.abs(value) < Number.EPSILON * 8 ? 0 : value; +} + +/** Axial hex distance, useful to consumers for borders and transitions. */ +export function hexDistance(left: HexCoordinate, right: HexCoordinate): number { + const dq = left.q - right.q; + const dr = left.r - right.r; + return Math.max(Math.abs(dq), Math.abs(dr), Math.abs(dq + dr)); +} + +/** True when every cell can be reached from every other through shared edges. */ +export function isContiguousHexTerritory( + cells: readonly HexCoordinate[], +): boolean { + if (cells.length === 0) { + return false; + } + const remaining = new Set(cells.map(hexCoordinateKey)); + const queue: HexCoordinate[] = [cells[0]]; + remaining.delete(hexCoordinateKey(cells[0])); + for (let cursor = 0; cursor < queue.length; cursor += 1) { + for (const neighbour of hexNeighbours(queue[cursor])) { + const key = hexCoordinateKey(neighbour); + if (remaining.delete(key)) { + queue.push(neighbour); + } + } + } + return remaining.size === 0; +} + +export function hexCoordinateKey(coordinate: HexCoordinate): string { + return `${coordinate.q},${coordinate.r}`; +} + +/** Keep only the edge-connected component of valid surviving territories. */ +function connectedRetainedTerritoryIds( + ordered: readonly CommunityComputeLayoutInput[], + previousById: ReadonlyMap, +): Set { + const valid = ordered + .map((deployment) => previousById.get(deployment.id)) + .filter( + (territory): territory is CommunityComputeHexTerritory => + territory !== undefined && + territory.cells.length === + ordered.find((item) => item.id === territory.deploymentId)?.cellCount, + ); + if (valid.length === 0) return new Set(); + + const owner = new Map(); + for (const territory of valid) { + for (const cell of territory.cells) { + const key = hexCoordinateKey(cell); + if (owner.has(key)) return new Set(); + owner.set(key, territory.deploymentId); + } + } + const retained = new Set([valid[0].deploymentId]); + const queue = [valid[0].deploymentId]; + const byId = new Map( + valid.map((territory) => [territory.deploymentId, territory]), + ); + for (let cursor = 0; cursor < queue.length; cursor += 1) { + const territory = byId.get(queue[cursor]); + if (!territory) continue; + for (const cell of territory.cells) { + for (const neighbour of hexNeighbours(cell)) { + const neighbourOwner = owner.get(hexCoordinateKey(neighbour)); + if (neighbourOwner && !retained.has(neighbourOwner)) { + retained.add(neighbourOwner); + queue.push(neighbourOwner); + } + } + } + } + return retained; +} + +function territoryOffsets( + count: HexCellCount, + orientation: number, +): HexCoordinate[] { + const center = { q: 0, r: 0 }; + if (count === 1) { + return [center]; + } + if (count === 2) { + return [center, HEX_DIRECTIONS[orientation]]; + } + if (count === 4) { + return [ + center, + HEX_DIRECTIONS[orientation], + HEX_DIRECTIONS[(orientation + 1) % 6], + HEX_DIRECTIONS[(orientation + 2) % 6], + ]; + } + return [center, ...HEX_DIRECTIONS]; +} + +/** + * Zero-based spiral slot: origin, then complete radius-one, radius-two, … rings. + * Runtime is O(radius), which is O(sqrt(n)); even at dense fixture sizes the + * largest radius remains small. + */ +function balancedSpiralCoordinate(index: number): HexCoordinate { + if (index === 0) { + return { q: 0, r: 0 }; + } + + let radius = 1; + let ringStart = 1; + while (index >= ringStart + 6 * radius) { + ringStart += 6 * radius; + radius += 1; + } + + const ringLength = 6 * radius; + const ringIndex = index - ringStart; + // Multiplication by a number coprime with the ring length is a permutation, + // not sampling: every slot is eventually visited. A step near half a ring + // alternates opposite sides, preventing a partially filled layer from + // producing the one-sided growth of a clockwise spiral. + let step = Math.max(1, Math.floor(ringLength / 2) - 1); + while (greatestCommonDivisor(step, ringLength) !== 1) step -= 1; + const offsetSeed = stableHash(`ring:${radius}`) % ringLength; + let offset = (offsetSeed + ringIndex * step) % ringLength; + let coordinate = { q: -radius, r: radius }; + for (const direction of HEX_DIRECTIONS) { + const steps = Math.min(radius, offset); + coordinate = { + q: coordinate.q + direction.q * steps, + r: coordinate.r + direction.r * steps, + }; + offset -= steps; + if (offset === 0) { + break; + } + } + return coordinate; +} + +function greatestCommonDivisor(left: number, right: number): number { + let a = left; + let b = right; + while (b !== 0) { + const remainder = a % b; + a = b; + b = remainder; + } + return a; +} + +function findBounds( + cells: readonly HexCoordinate[], +): CommunityComputeHexBounds | null { + if (cells.length === 0) { + return null; + } + let minQ = cells[0].q; + let maxQ = cells[0].q; + let minR = cells[0].r; + let maxR = cells[0].r; + for (const cell of cells.slice(1)) { + minQ = Math.min(minQ, cell.q); + maxQ = Math.max(maxQ, cell.q); + minR = Math.min(minR, cell.r); + maxR = Math.max(maxR, cell.r); + } + return { minQ, maxQ, minR, maxR }; +} + +/** FNV-1a over UTF-16 code units; deterministic on every JS runtime. */ +function stableHash(value: string): number { + let hash = 0x811c9dc5; + for (let index = 0; index < value.length; index += 1) { + hash ^= value.charCodeAt(index); + hash = Math.imul(hash, 0x01000193); + } + return hash >>> 0; +} diff --git a/desktop/src/features/mesh-compute/communityComputeMapModel.test.mjs b/desktop/src/features/mesh-compute/communityComputeMapModel.test.mjs new file mode 100644 index 0000000000..e9c3ed32b2 --- /dev/null +++ b/desktop/src/features/mesh-compute/communityComputeMapModel.test.mjs @@ -0,0 +1,206 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + bucketHexCellCount, + capacityGbToHexCellCount, + deriveCommunityComputeMapModel, +} from "./communityComputeMapModel.ts"; +import { + COMMUNITY_COMPUTE_TUTORIAL_FIXTURES, + LARGE_DENSE_COMMUNITY_COMPUTE_FIXTURE, + createLargeDenseSnapshot, +} from "./communityComputeFixtures.ts"; + +function device(overrides = {}) { + return { + deviceId: "device-1", + label: "Studio", + capacityGb: 32, + models: ["gemma"], + state: "serving", + isSelf: false, + memberPubkey: "member-1", + reportedAt: 1_786_032_000, + modelSizeGb: 18, + ...overrides, + }; +} + +function snapshot(overrides = {}) { + return { + sharingDeviceCount: 1, + contributorMemberCount: 1, + sharedCapacityGb: 32, + allocatedCapacityGb: 18, + models: ["gemma"], + devices: [device()], + includesSelf: false, + memberCount: 4, + reason: null, + ...overrides, + }; +} + +test("cell buckets are exactly 1, 2, 4, and 7", () => { + assert.deepEqual( + [-1, 1, 1.1, 2, 2.1, 4, 4.1, 999, Number.NaN].map(bucketHexCellCount), + [1, 1, 2, 2, 4, 4, 7, 7, 1], + ); + assert.deepEqual( + [null, 128, 128.1, 256, 256.1, 512, 512.1, 1000].map( + capacityGbToHexCellCount, + ), + [1, 1, 2, 2, 4, 4, 7, 7], + ); +}); + +test("only ready serving models become deployments", () => { + const model = deriveCommunityComputeMapModel( + snapshot({ + sharingDeviceCount: 1, + devices: [ + device({ deviceId: "a", models: ["gemma", "qwen"] }), + device({ deviceId: "b", state: "loading", models: [] }), + device({ deviceId: "c", state: "standby", models: [] }), + device({ deviceId: "d", state: "consuming", models: [] }), + ], + }), + ); + + assert.equal(model.status, "ready"); + assert.deepEqual( + model.deployments.map((deployment) => deployment.modelId), + ["gemma", "qwen"], + ); + assert.equal(model.kpis.deploymentCount, 2); + assert.equal(model.kpis.contributorMemberCount, 1); + assert.equal(model.health.counts.healthy, 1); + assert.equal(model.health.counts.warming, 1); + assert.equal(model.health.counts.idle, 1); +}); + +test("one territory is produced per model and per device replica", () => { + const model = deriveCommunityComputeMapModel( + snapshot({ + devices: [ + device({ deviceId: "a", models: ["qwen", "gemma"] }), + device({ deviceId: "b", memberPubkey: "member-2", models: ["gemma"] }), + ], + }), + ); + assert.equal(model.deployments.length, 3); + assert.equal( + model.deployments.filter((deployment) => deployment.modelId === "gemma") + .length, + 2, + ); + assert.equal(new Set(model.deployments.map(({ id }) => id)).size, 3); +}); + +test("reported model footprint controls area without fixture overrides", () => { + const model = deriveCommunityComputeMapModel( + snapshot({ devices: [device({ capacityGb: 768, modelSizeGb: 96 })] }), + ); + assert.equal(model.deployments[0].cellCount, 1); + + const large = deriveCommunityComputeMapModel( + snapshot({ devices: [device({ capacityGb: 768, modelSizeGb: 646 })] }), + ); + assert.equal(large.deployments[0].cellCount, 7); +}); + +test("loading, consumer-only, and empty snapshots render no territories", () => { + for (const state of ["loading", "standby", "consuming"]) { + const model = deriveCommunityComputeMapModel( + snapshot({ devices: [device({ state, models: [] })] }), + ); + assert.equal(model.status, "empty"); + assert.equal(model.deployments.length, 0); + } + assert.equal(deriveCommunityComputeMapModel(null).status, "loading"); + assert.equal( + deriveCommunityComputeMapModel( + COMMUNITY_COMPUTE_TUTORIAL_FIXTURES.empty.snapshot, + ).status, + "empty", + ); +}); + +test("projection is deterministic and independent of device/model order", () => { + const devices = [ + device({ deviceId: "z", label: "Zulu", models: ["qwen", "gemma"] }), + device({ deviceId: "a", label: "Alpha", models: ["mistral"] }), + ]; + const forward = deriveCommunityComputeMapModel(snapshot({ devices })); + const reverse = deriveCommunityComputeMapModel( + snapshot({ + devices: [...devices] + .reverse() + .map((item) => ({ ...item, models: [...item.models].reverse() })), + }), + ); + assert.deepEqual(forward, reverse); +}); + +test("tutorial fixtures use only production fields and production sizing", () => { + assert.deepEqual(Object.keys(COMMUNITY_COMPUTE_TUTORIAL_FIXTURES).sort(), [ + "empty", + "large-dense-500", + "single-device", + "small-company", + ]); + const allowed = new Set([ + "deviceId", + "label", + "capacityGb", + "models", + "state", + "isSelf", + "memberPubkey", + "reportedAt", + "modelSizeGb", + ]); + for (const fixture of Object.values(COMMUNITY_COMPUTE_TUTORIAL_FIXTURES)) { + for (const item of fixture.snapshot.devices) { + assert.deepEqual( + Object.keys(item).filter((key) => !allowed.has(key)), + [], + ); + } + } + const small = deriveCommunityComputeMapModel( + COMMUNITY_COMPUTE_TUTORIAL_FIXTURES["small-company"].snapshot, + ); + assert.equal(small.deployments.length, 12); + assert.equal( + small.deployments.filter(({ deviceLabel }) => + deviceLabel.startsWith("Team MacBook"), + ).length, + 10, + ); + assert.equal( + small.deployments.filter(({ cellCount }) => cellCount === 1).length, + 10, + ); + assert.equal( + small.deployments.filter(({ cellCount }) => cellCount === 7).length, + 2, + ); +}); + +test("large dense fixture is deterministic and excludes non-serving rows", () => { + const fixture = LARGE_DENSE_COMMUNITY_COMPUTE_FIXTURE.snapshot; + assert.deepEqual(fixture, createLargeDenseSnapshot(500)); + assert.equal(fixture.devices.length, 500); + const servingCount = fixture.devices.filter( + (item) => item.state === "serving" && item.models.length > 0, + ).length; + const model = deriveCommunityComputeMapModel(fixture); + assert.equal(model.deployments.length, servingCount); + assert.equal(model.kpis.deploymentCount, servingCount); + assert.equal( + new Set(model.deployments.map(({ id }) => id)).size, + servingCount, + ); +}); diff --git a/desktop/src/features/mesh-compute/communityComputeMapModel.ts b/desktop/src/features/mesh-compute/communityComputeMapModel.ts new file mode 100644 index 0000000000..663b957d5f --- /dev/null +++ b/desktop/src/features/mesh-compute/communityComputeMapModel.ts @@ -0,0 +1,365 @@ +import type { + MeshDeviceState, + MeshSnapshot, + MeshSnapshotDevice, +} from "@/shared/api/tauriMesh"; + +/** The only territory sizes rendered by the community compute map. */ +export const HEX_CELL_BUCKETS = [1, 2, 4, 7] as const; + +export type HexCellCount = (typeof HEX_CELL_BUCKETS)[number]; + +/** + * The Community view intentionally consumes only fields in the production + * snapshot contract. Tutorial fixtures use this same shape so examples cannot + * quietly grow capabilities that live Compute does not have. + */ +export type CommunityComputeDeviceInput = MeshSnapshotDevice; + +/** A readonly, structurally compatible form of the current mesh snapshot. */ +export type CommunityComputeSnapshotInput< + TDevice extends MeshSnapshotDevice = CommunityComputeDeviceInput, +> = Omit & { + readonly devices: readonly TDevice[]; +}; + +export type CommunityComputeDeploymentHealth = + | "healthy" + | "warming" + | "idle" + | "degraded" + | "offline" + | "unknown"; + +export type CommunityComputeMapHealth = + | "healthy" + | "attention" + | "critical" + | "empty" + | "unknown"; + +export type CommunityComputeDeployment< + TDevice extends MeshSnapshotDevice = CommunityComputeDeviceInput, +> = { + /** Stable map identity. Never derived from the device's array position. */ + id: string; + /** Model shown as the territory's primary identity. */ + label: string; + modelId: string; + deviceLabel: string; + capacityGb: number | null; + models: readonly string[]; + state: MeshDeviceState; + isSelf: boolean; + health: CommunityComputeDeploymentHealth; + healthReason: string | null; + cellCount: HexCellCount; + /** Preserves additive backend data for a future detail projection. */ + source: Readonly; +}; + +export type CommunityComputeKpis = { + memberCount: number; + contributorMemberCount: number; + deploymentCount: number; + activeDeploymentCount: number; + sharedCapacityGb: number | null; + allocatedCapacityGb: number | null; + allocatedPercent: number | null; + modelCount: number; +}; + +export type CommunityComputeHealthSummary = { + status: CommunityComputeMapHealth; + label: string; + reason: string | null; + counts: Record; +}; + +export type CommunityComputeMapModel< + TDevice extends MeshSnapshotDevice = CommunityComputeDeviceInput, +> = { + status: "loading" | "ready" | "empty"; + kpis: CommunityComputeKpis; + health: CommunityComputeHealthSummary; + deployments: readonly CommunityComputeDeployment[]; +}; + +/** + * Round a desired territory footprint up to one of the four supported shapes. + * Invalid and non-positive weights still get one cell so a deployment cannot + * disappear from the map. + */ +export function bucketHexCellCount( + weight: number | null | undefined, +): HexCellCount { + if (weight === null || weight === undefined || !Number.isFinite(weight)) { + return 1; + } + const normalized = Math.max(1, Math.ceil(weight)); + return HEX_CELL_BUCKETS.find((bucket) => normalized <= bucket) ?? 7; +} + +/** + * Turn model memory into a deliberately coarse footprint. + * + * A 128 GB Mac is the common mesh workhorse, so every model that fits on one is + * one cell. Multi-cell territories are reserved for genuinely multi-GPU scale + * deployments; this avoids making ordinary local models look fragmented. + */ +export function capacityGbToHexCellCount( + capacityGb: number | null | undefined, +): HexCellCount { + if (capacityGb === null || capacityGb === undefined || capacityGb <= 128) { + return 1; + } + if (capacityGb <= 256) { + return 2; + } + if (capacityGb <= 512) { + return 4; + } + return 7; +} + +const EMPTY_HEALTH_COUNTS: Record = { + healthy: 0, + warming: 0, + idle: 0, + degraded: 0, + offline: 0, + unknown: 0, +}; + +/** Project a backend snapshot into the complete, render-agnostic map model. */ +export function deriveCommunityComputeMapModel< + TDevice extends MeshSnapshotDevice = CommunityComputeDeviceInput, +>( + snapshot: CommunityComputeSnapshotInput | null, +): CommunityComputeMapModel { + if (snapshot === null) { + return { + status: "loading", + kpis: { + memberCount: 0, + contributorMemberCount: 0, + deploymentCount: 0, + activeDeploymentCount: 0, + sharedCapacityGb: null, + allocatedCapacityGb: null, + allocatedPercent: null, + modelCount: 0, + }, + health: { + status: "unknown", + label: "Checking community compute", + reason: null, + counts: { ...EMPTY_HEALTH_COUNTS }, + }, + deployments: [], + }; + } + + const deployments = projectDeployments(snapshot.devices); + const healthStates = snapshot.devices + .filter((device) => device.state !== "consuming") + .map((device) => healthFromState(device.state)); + const counts = countHealth(healthStates); + const models = new Set(snapshot.models); + for (const deployment of deployments) { + for (const model of deployment.models) { + models.add(model); + } + } + + const contributorMemberCount = + snapshot.contributorMemberCount ?? + new Set( + deployments + .filter((deployment) => deployment.state === "serving") + .map( + (deployment) => + deployment.source.memberPubkey?.trim() || deployment.id, + ), + ).size; + const sharedCapacityGb = finiteNonNegative(snapshot.sharedCapacityGb); + const reportedAllocated = finiteNonNegative(snapshot.allocatedCapacityGb); + const servingDeployments = deployments.filter( + (deployment) => deployment.state === "serving", + ); + const inferredAllocations = servingDeployments.map((deployment) => + finiteNonNegative(deployment.source.modelSizeGb), + ); + const allocatedCapacityGb = + reportedAllocated ?? + (inferredAllocations.length > 0 && + inferredAllocations.every((value) => value !== null) + ? inferredAllocations.reduce((total, value) => total + (value ?? 0), 0) + : null); + + return { + status: deployments.length === 0 ? "empty" : "ready", + kpis: { + memberCount: nonNegativeInteger(snapshot.memberCount), + contributorMemberCount, + deploymentCount: deployments.length, + activeDeploymentCount: counts.healthy + counts.warming, + sharedCapacityGb, + allocatedCapacityGb, + allocatedPercent: + allocatedCapacityGb !== null && + sharedCapacityGb !== null && + sharedCapacityGb > 0 + ? Math.min(100, (allocatedCapacityGb / sharedCapacityGb) * 100) + : null, + modelCount: models.size, + }, + health: summarizeHealth(counts, healthStates, snapshot.reason), + deployments, + }; +} + +function projectDeployments( + devices: readonly TDevice[], +): CommunityComputeDeployment[] { + const canonical = devices + .filter((device) => device.state === "serving" && device.models.length > 0) + .flatMap((device) => + [...new Set(device.models)].map((model) => ({ + device: { ...device, models: [model] } as TDevice, + identity: deploymentIdentity(device, model), + })), + ) + .sort( + (left, right) => + left.identity.localeCompare(right.identity) || + deviceFingerprint(left.device).localeCompare( + deviceFingerprint(right.device), + ), + ); + const occurrences = new Map(); + + return canonical.map(({ device, identity }) => { + const occurrence = occurrences.get(identity) ?? 0; + occurrences.set(identity, occurrence + 1); + const models = [...new Set(device.models)].sort(); + return { + id: occurrence === 0 ? identity : `${identity}#${occurrence + 1}`, + label: models[0] ?? device.label, + modelId: models[0] ?? "Model not reported", + deviceLabel: device.label, + capacityGb: finiteNonNegative(device.capacityGb), + models, + state: device.state, + isSelf: device.isSelf, + health: healthFromState(device.state), + healthReason: null, + cellCount: capacityGbToHexCellCount( + device.modelSizeGb ?? device.capacityGb, + ), + source: device, + }; + }); +} + +function deploymentIdentity( + device: MeshSnapshotDevice, + modelId: string, +): string { + const deviceId = cleanOptionalText(device.deviceId); + const model = encodeURIComponent(modelId); + return deviceId === null + ? `derived:${encodeURIComponent(deviceFingerprint(device))}:${model}` + : `deployment:${encodeURIComponent(deviceId)}:${model}`; +} + +function deviceFingerprint(device: MeshSnapshotDevice): string { + return JSON.stringify([ + device.label, + device.capacityGb, + [...device.models].sort(), + device.state, + device.isSelf, + ]); +} + +function healthFromState( + state: MeshDeviceState, +): CommunityComputeDeploymentHealth { + switch (state) { + case "serving": + case "consuming": + return "healthy"; + case "loading": + return "warming"; + case "standby": + return "idle"; + } +} + +function countHealth( + states: readonly CommunityComputeDeploymentHealth[], +): Record { + const counts = { ...EMPTY_HEALTH_COUNTS }; + for (const state of states) { + counts[state] += 1; + } + return counts; +} + +function summarizeHealth( + counts: Record, + states: readonly CommunityComputeDeploymentHealth[], + snapshotReason: string | null, +): CommunityComputeHealthSummary { + if (states.length === 0) { + return { + status: "empty", + label: "No deployments", + reason: cleanOptionalText(snapshotReason), + counts, + }; + } + + const unavailable = counts.offline + counts.degraded; + const reason = null; + + if (counts.offline === states.length) { + return { status: "critical", label: "Compute unavailable", reason, counts }; + } + if (unavailable > 0) { + return { + status: "attention", + label: `${unavailable} deployment${unavailable === 1 ? "" : "s"} need attention`, + reason, + counts, + }; + } + if (counts.unknown === states.length) { + return { status: "unknown", label: "Health unknown", reason: null, counts }; + } + return { + status: "healthy", + label: "No reported issues", + reason: null, + counts, + }; +} + +function finiteNonNegative(value: number | null | undefined): number | null { + return value !== null && + value !== undefined && + Number.isFinite(value) && + value >= 0 + ? value + : null; +} + +function nonNegativeInteger(value: number): number { + return Number.isFinite(value) ? Math.max(0, Math.floor(value)) : 0; +} + +function cleanOptionalText(value: string | null | undefined): string | null { + const cleaned = value?.trim(); + return cleaned ? cleaned : null; +} diff --git a/desktop/src/features/mesh-compute/hooks/useMeshNodeStatus.ts b/desktop/src/features/mesh-compute/hooks/useMeshNodeStatus.ts index c9194a0e05..2bc632d84b 100644 --- a/desktop/src/features/mesh-compute/hooks/useMeshNodeStatus.ts +++ b/desktop/src/features/mesh-compute/hooks/useMeshNodeStatus.ts @@ -17,42 +17,54 @@ export function useMeshNodeStatus(): { status: MeshNodeStatus | null; error: string | null; refresh: () => void; + update: (status: MeshNodeStatus) => void; } { const [status, setStatus] = React.useState(null); const [error, setError] = React.useState(null); + const requestSequence = React.useRef(0); const fetchOnce = React.useCallback(() => { - let cancelled = false; - (async () => { + const sequence = ++requestSequence.current; + void (async () => { try { const value = await meshNodeStatus(); - if (!cancelled) { + if (sequence === requestSequence.current) { setStatus(value); setError(null); } } catch (err) { - if (!cancelled) { + if (sequence === requestSequence.current) { setError(err instanceof Error ? err.message : String(err)); } } })(); - return () => { - cancelled = true; - }; }, []); - React.useEffect(() => fetchOnce(), [fetchOnce]); + const update = React.useCallback((value: MeshNodeStatus) => { + // Invalidate any slower request started before this authoritative command + // result so it cannot restore the preceding lifecycle state. + requestSequence.current += 1; + setStatus(value); + setError(null); + }, []); + + React.useEffect(() => { + fetchOnce(); + }, [fetchOnce]); // Fast poll while in a transitioning state; slow poll while steady or off. React.useEffect(() => { const transitioning = status?.state === "starting" || status?.state === "stopping"; - const interval = transitioning ? 750 : 4000; + // Keep reconciling aggressively while a start command is still waiting on + // model readiness. The command can take minutes and its eventual response + // is not the only signal that the runtime crossed to running. + const interval = transitioning ? 500 : 4000; const handle = window.setInterval(() => { fetchOnce(); }, interval); return () => window.clearInterval(handle); }, [status?.state, fetchOnce]); - return { status, error, refresh: fetchOnce }; + return { status, error, refresh: fetchOnce, update }; } diff --git a/desktop/src/features/mesh-compute/hooks/useMeshSnapshot.ts b/desktop/src/features/mesh-compute/hooks/useMeshSnapshot.ts new file mode 100644 index 0000000000..85de3d794d --- /dev/null +++ b/desktop/src/features/mesh-compute/hooks/useMeshSnapshot.ts @@ -0,0 +1,62 @@ +import * as React from "react"; + +import { meshSnapshot } from "@/shared/api/tauriMesh"; +import type { MeshSnapshot } from "@/shared/api/tauriMesh"; + +/** + * Poll the community shared-compute snapshot. + * + * Deliberately slow compared to `useMeshNodeStatus` (750ms/4000ms): that hook + * watches a LOCAL lifecycle that must feel responsive, while this one reads + * RELAY state that only changes as fast as the 45s member heartbeat, with a + * 120s freshness window. Polling faster would just add relay queries without + * adding information. + * + * An empty mesh is not an error — the snapshot carries its own `reason`. `error` + * here means the query itself failed (offline relay, stub build). + */ +const SNAPSHOT_POLL_INTERVAL_MS = 30_000; + +export function useMeshSnapshot(options?: { refreshKey?: number }): { + snapshot: MeshSnapshot | null; + error: string | null; + refresh: () => void; +} { + const [snapshot, setSnapshot] = React.useState(null); + const [error, setError] = React.useState(null); + const refreshKey = options?.refreshKey ?? 0; + + // Guards against a slow in-flight response overwriting a newer one. + const requestSeq = React.useRef(0); + + const fetchOnce = React.useCallback(() => { + const seq = ++requestSeq.current; + (async () => { + try { + const value = await meshSnapshot(); + if (seq === requestSeq.current) { + setSnapshot(value); + setError(null); + } + } catch (err) { + if (seq === requestSeq.current) { + setError(err instanceof Error ? err.message : String(err)); + } + } + })(); + }, []); + + // Re-fetch immediately when the local node transitions (refreshKey), so + // turning sharing on updates the community view without waiting for a poll. + // biome-ignore lint/correctness/useExhaustiveDependencies: refreshKey is an intentional trigger, not a value read in the body — bumping it re-queries the community snapshot right after a local node transition instead of waiting out the 30s poll + React.useEffect(() => { + fetchOnce(); + }, [fetchOnce, refreshKey]); + + React.useEffect(() => { + const handle = window.setInterval(fetchOnce, SNAPSHOT_POLL_INTERVAL_MS); + return () => window.clearInterval(handle); + }, [fetchOnce]); + + return { snapshot, error, refresh: fetchOnce }; +} diff --git a/desktop/src/features/mesh-compute/ui/CommunityComputeTerritoryMap.tsx b/desktop/src/features/mesh-compute/ui/CommunityComputeTerritoryMap.tsx new file mode 100644 index 0000000000..91dd44d2ab --- /dev/null +++ b/desktop/src/features/mesh-compute/ui/CommunityComputeTerritoryMap.tsx @@ -0,0 +1,342 @@ +import * as React from "react"; + +import { cn } from "@/shared/lib/cn"; +import { + hexTerritoryBoundaryEdges, + layoutCommunityComputeHexTerritories, + type CommunityComputeHexCell, +} from "../communityComputeHexLayout"; +import type { + CommunityComputeDeployment, + CommunityComputeMapModel, +} from "../communityComputeMapModel"; + +export type CommunityComputeTerritoryMapProps = { + model: CommunityComputeMapModel; + className?: string; +}; + +type Point = { x: number; y: number }; + +type ViewBox = { + x: number; + y: number; + width: number; + height: number; +}; + +// Deliberately overlap polygons, then trim their outside with the territory +// edge. This avoids SVG antialias seams inside one model footprint. +const FUSED_TILE_RADIUS = 1.025; +const TILE_EDGE_WIDTH = 0.1; +const SQRT_THREE = Math.sqrt(3); + +/** + * Responsive SVG territory map. Each deployment owns one stable, contiguous + * cluster of discrete hexes; health changes never move that cluster. + */ +export function CommunityComputeTerritoryMap({ + model, + className, +}: CommunityComputeTerritoryMapProps) { + const deploymentById = React.useMemo( + () => + new Map( + model.deployments.map((deployment) => [deployment.id, deployment]), + ), + [model.deployments], + ); + const previousLayoutRef = React.useRef | null>(null); + const layout = React.useMemo( + () => + layoutCommunityComputeHexTerritories( + model.deployments.map(({ id, cellCount }) => ({ id, cellCount })), + { previous: previousLayoutRef.current }, + ), + [model.deployments], + ); + React.useEffect(() => { + previousLayoutRef.current = layout; + }, [layout]); + const viewBox = computeViewBox(layout.cells); + const [hoveredId, setHoveredId] = React.useState(null); + const lowerTimerRef = React.useRef | null>( + null, + ); + const hoveredDeployment = hoveredId + ? (deploymentById.get(hoveredId) ?? null) + : null; + const isDenseMap = model.deployments.length >= 50; + + React.useEffect( + () => () => { + if (lowerTimerRef.current) clearTimeout(lowerTimerRef.current); + }, + [], + ); + + function raiseTerritory(deploymentId: string, territoryElement: SVGGElement) { + if (lowerTimerRef.current) clearTimeout(lowerTimerRef.current); + // SVG paints in DOM order. Move the existing node instead of rebuilding + // children so hover stays smooth as territories scale. + territoryElement.parentNode?.appendChild(territoryElement); + setHoveredId(deploymentId); + } + + function lowerTerritory() { + if (lowerTimerRef.current) clearTimeout(lowerTimerRef.current); + // Let the detail card linger briefly while the territory settles. + lowerTimerRef.current = setTimeout(() => setHoveredId(null), 280); + } + + if (model.status === "loading") { + return ( +
+
+

+ Mapping community compute… +

+
+ ); + } + + if (model.status === "empty" || viewBox === null) { + return ( +
+ +

No compute territories yet

+

+ Shared compute appears here when members make it available to the + community. +

+
+ ); + } + + return ( +
+
+ + {layout.territories.map((territory) => { + const deployment = deploymentById.get(territory.deploymentId); + if (!deployment) return null; + return ( + // SVG groups have no semantic element for hover/focus detail disclosure. + // biome-ignore lint/a11y/noStaticElementInteractions: keyboard focus mirrors hover without making the territory an action. + + raiseTerritory(deployment.id, event.currentTarget) + } + onMouseEnter={(event) => + raiseTerritory(deployment.id, event.currentTarget) + } + onMouseLeave={lowerTerritory} + style={territoryHoverStyle( + territory.cells, + viewBox, + isDenseMap, + )} + tabIndex={0} + > + {territoryAccessibleLabel(deployment)} + + + ); + })} + + {hoveredDeployment ? ( +
+

+ {shortModelName(hoveredDeployment.modelId)} +

+

+ {hoveredDeployment.deviceLabel} ·{" "} + {formatCapacity(hoveredDeployment.capacityGb)} +

+

+ Available to the community +

+
+ ) : null} +
+
+ ); +} + +function TerritoryShape({ + cells, +}: { + cells: readonly CommunityComputeHexCell[]; +}) { + const boundary = hexTerritoryBoundaryEdges(cells); + return ( + + {cells.map((cell) => { + const center = axialToPoint(cell.q, cell.r); + return ( + + ); + })} + + + ); +} + +function territoryHoverStyle( + cells: readonly CommunityComputeHexCell[], + viewBox: ViewBox, + dense: boolean, +): React.CSSProperties { + const centers = cells.map((cell) => axialToPoint(cell.q, cell.r)); + const minX = Math.min(...centers.map(({ x }) => x)) - 1; + const maxX = Math.max(...centers.map(({ x }) => x)) + 1; + const minY = Math.min(...centers.map(({ y }) => y)) - 1; + const maxY = Math.max(...centers.map(({ y }) => y)) + 1; + const territorySpan = Math.max(maxX - minX, maxY - minY); + const mapSpan = Math.min(viewBox.width, viewBox.height); + // Normalize every hovered territory toward the same visual footprint. Dense + // maps therefore expand dramatically while already-large sparse territories + // receive only a restrained lift. + const targetSpan = mapSpan * (dense ? 0.18 : 0.16); + const scale = Math.max(1.06, Math.min(8, targetSpan / territorySpan)); + return { "--mesh-hover-scale": scale } as React.CSSProperties; +} + +function edgesToPath( + edges: ReturnType, +): string { + return edges + .map( + (edge) => + `M ${edge.start.x} ${edge.start.y} L ${edge.end.x} ${edge.end.y}`, + ) + .join(" "); +} + +function EmptyHexes() { + const centers = [ + { x: 2, y: 1.75 }, + { x: 3.58, y: 2.65 }, + { x: 2, y: 3.55 }, + { x: 0.42, y: 2.65 }, + ]; + return ( + + ); +} + +function axialToPoint(q: number, r: number): Point { + return { x: SQRT_THREE * (q + r / 2), y: 1.5 * r }; +} + +function hexPoints(center: Point, radius: number): string { + return Array.from({ length: 6 }, (_, index) => { + const angle = ((60 * index - 30) * Math.PI) / 180; + return `${center.x + radius * Math.cos(angle)},${center.y + radius * Math.sin(angle)}`; + }).join(" "); +} + +function computeViewBox( + cells: readonly CommunityComputeHexCell[], +): ViewBox | null { + if (cells.length === 0) return null; + const centers = cells.map((cell) => axialToPoint(cell.q, cell.r)); + const minX = Math.min(...centers.map(({ x }) => x)) - 1.4; + const maxX = Math.max(...centers.map(({ x }) => x)) + 1.4; + const minY = Math.min(...centers.map(({ y }) => y)) - 1.4; + const maxY = Math.max(...centers.map(({ y }) => y)) + 1.4; + return { + x: minX, + y: minY, + width: Math.max(2.8, maxX - minX), + height: Math.max(2.8, maxY - minY), + }; +} + +function shortModelName(modelId: string): string { + const tail = modelId.split("/").at(-1) ?? modelId; + return tail.length > 34 ? `${tail.slice(0, 31)}…` : tail; +} + +function territoryAccessibleLabel( + deployment: CommunityComputeDeployment, +): string { + const capacity = formatCapacity(deployment.capacityGb); + return `${deployment.modelId} on ${deployment.deviceLabel}, ${capacity}`; +} + +function formatCapacity(value: number | null): string { + return value === null + ? "Capacity not reported" + : `${Math.round(value)} GB shared`; +} + +function testId(value: string): string { + return encodeURIComponent(value).replaceAll("%", "-"); +} diff --git a/desktop/src/features/mesh-compute/ui/MeshComputeSettingsCard.tsx b/desktop/src/features/mesh-compute/ui/MeshComputeSettingsCard.tsx index c27c5eef3a..bb5590b3e2 100644 --- a/desktop/src/features/mesh-compute/ui/MeshComputeSettingsCard.tsx +++ b/desktop/src/features/mesh-compute/ui/MeshComputeSettingsCard.tsx @@ -1,6 +1,5 @@ import * as React from "react"; -import { ChevronDown } from "lucide-react"; -import { AnimatePresence, motion, useReducedMotion } from "motion/react"; +import { ChevronDown, Hexagon, Sparkles } from "lucide-react"; import { Switch } from "@/shared/ui/switch"; import { cn } from "@/shared/lib/cn"; @@ -29,6 +28,12 @@ import type { } from "@/shared/api/tauriMesh"; import { SettingsOptionGroup } from "@/features/settings/ui/SettingsOptionGroup"; import { SettingsSectionHeader } from "@/features/settings/ui/SettingsSectionHeader"; +import { requestOpenCreateAgent } from "@/features/agents/openCreateAgentEvent"; +import { useCommunities } from "@/features/communities/useCommunities"; +import { Button } from "@/shared/ui/button"; +import { useMeshSnapshot } from "../hooks/useMeshSnapshot"; +import { deriveCommunityComputeMapModel } from "../communityComputeMapModel"; +import { CommunityComputeTerritoryMap } from "./CommunityComputeTerritoryMap"; import { classifyModelRef } from "../classifyModelRef"; import { downloadPercent, @@ -52,11 +57,6 @@ const MESH_SELECT_TRIGGER_CLASS = cn( "h-11 px-3 py-2 leading-6 hover:bg-muted/40 focus:bg-muted/40 [&>svg]:text-muted-foreground/60", ); -const SHARE_COMPUTE_REVEAL_TRANSITION = { - duration: 0.22, - ease: [0.23, 1, 0.32, 1], -} as const; - function readDraft(key: string): string { try { return window.localStorage.getItem(key) ?? ""; @@ -85,8 +85,11 @@ function writeDraft(key: string, value: string): void { * exposing implementation protocols or raw mesh controls. */ export function MeshComputeSettingsCard() { - const shouldReduceMotion = useReducedMotion(); - const { status, error, refresh } = useMeshNodeStatus(); + const { activeCommunity } = useCommunities(); + const communityName = activeCommunity?.name ?? "this community"; + const { status, error, refresh, update } = useMeshNodeStatus(); + const [snapshotRefreshKey, setSnapshotRefreshKey] = React.useState(0); + const { snapshot } = useMeshSnapshot({ refreshKey: snapshotRefreshKey }); const [installedModels, setInstalledModels] = React.useState< MeshModelOption[] >([]); @@ -187,6 +190,17 @@ export function MeshComputeSettingsCard() { const refClass = classifyModelRef(modelInput); const canStart = refClass.kind !== "unknown" && !actionInFlight; const showSharingControls = isSharing || pendingAction === "start"; + // Live runtime state wins over the still-pending command promise. Starting a + // node waits on an additional inference-readiness probe, so the command can + // remain in flight after status already proves this machine is running. Do + // not leave the visual state spinning during that window. + const isSuccessfullySharing = + status?.mode === "serve" && status.state === "running"; + const isTransitioning = + !isSuccessfullySharing && + (pendingAction !== null || + status?.state === "starting" || + status?.state === "stopping"); async function handleToggle(next: boolean) { // Never let the Share switch tear down a consume session. The switch is @@ -203,7 +217,7 @@ export function MeshComputeSettingsCard() { if (next) { const maxVram = maxVramGb.trim() === "" ? undefined : Number.parseFloat(maxVramGb); - await meshStartNode({ + const nextStatus = await meshStartNode({ mode: "serve", modelId: modelInput.trim() || undefined, maxVramGb: @@ -211,10 +225,13 @@ export function MeshComputeSettingsCard() { ? maxVram : undefined, }); + update(nextStatus); } else { - await meshStopNode(); + const nextStatus = await meshStopNode(); + update(nextStatus); } refresh(); + setSnapshotRefreshKey((current) => current + 1); } catch (err) { setActionError(err instanceof Error ? err.message : String(err)); } finally { @@ -227,8 +244,8 @@ export function MeshComputeSettingsCard() { return (
{error ? ( @@ -248,20 +265,68 @@ export function MeshComputeSettingsCard() {
-
- - {!isSharing ? ( - - ) : null} + {isSuccessfullySharing ? ( + + + + ) : null} + {isTransitioning ? ( + + Preparing shared compute + + + + ) : ( + + )} + +
+ + {!isSuccessfullySharing ? ( + + ) : ( +

+ This machine is actively sharing compute on {communityName} + ’s mesh. +

+ )} +
+ {showSharingControls ? ( +
+

Status

+
+ + {servingIndicator.show ? ( +

+ {servingIndicator.label} + {servingIndicator.detail + ? ` · ${servingIndicator.detail}` + : null} +

+ ) : null} +
+
+ ) : null} {status?.consoleUrl ? (

Debug console:{" "} @@ -347,59 +439,47 @@ export function MeshComputeSettingsCard() { ) : null}

- - {showSharingControls ? ( - -
-

Status

-
- - {servingIndicator.show ? ( -

- {servingIndicator.label} - {servingIndicator.detail ? ( - - {" "} - · {servingIndicator.detail} - - ) : null} -

- ) : null} -
-
- - ) : null} - + {(snapshot?.contributorMemberCount ?? 0) >= 6 ? ( +
+
+

Community mesh

+

+ {snapshot?.contributorMemberCount} people are contributing + compute. +

+
+ +
+ ) : null}
+ +
+
+
+

+ Put the community mesh to work +

+

+ Anyone can use compute shared by the community—even if this + machine can’t contribute. Name your agent and customize its + instructions; we’ll handle the compute setup. +

+
+ +
+
); } @@ -496,7 +576,9 @@ function MeshModelPicker({ seen.add(entry.name); return { disabled: entry.fit === "too_large", - label: , + label: ( + + ), value: entry.name, }; }); @@ -523,7 +605,7 @@ function MeshModelPicker({ ...localOptions, { label: "Custom model…", value: CUSTOM_MODEL_DROPDOWN_VALUE }, ]; - }, [catalog?.entries, installedModels]); + }, [catalog?.entries, catalog?.gpuName, installedModels]); const knownModel = options.some((option) => option.value === model.trim()); const showCustomModelInput = isCustomModelEditing || (model.trim().length > 0 && !knownModel); @@ -578,16 +660,19 @@ function MeshModelPicker({ className="text-sm font-normal text-muted-foreground/70" data-settings-subcopy > - {catalog - ? `Recommended for this machine${catalog.gpuName ? ` (${catalog.gpuName}, ${catalog.vramDisplay} AI memory)` : ""}.` - : "Choose a model or enter a model reference or local file."}{" "} - Buzz downloads remote models when sharing starts. + Buzz downloads the selected model when sharing starts.

); } -function MeshModelOptionLabel({ entry }: { entry: MeshCatalogEntry }) { +function MeshModelOptionLabel({ + entry, + hardware, +}: { + entry: MeshCatalogEntry; + hardware?: string | null; +}) { return (
{entry.name} @@ -596,8 +681,8 @@ function MeshModelOptionLabel({ entry }: { entry: MeshCatalogEntry }) { {FIT_LABEL[entry.fit]} {entry.recommended ? ( - - Recommended + + Recommended for {hardware ?? "this machine"} ) : null} {entry.installed ? ( diff --git a/desktop/src/shared/api/tauriMesh.ts b/desktop/src/shared/api/tauriMesh.ts index 8a0153123f..1890e76650 100644 --- a/desktop/src/shared/api/tauriMesh.ts +++ b/desktop/src/shared/api/tauriMesh.ts @@ -112,3 +112,35 @@ export type MeshModelCatalog = { export async function meshModelCatalog(): Promise { return await invokeTauri("mesh_model_catalog"); } + +export type MeshDeviceState = "serving" | "loading" | "standby" | "consuming"; + +export type MeshSnapshotDevice = { + deviceId: string | null; + label: string; + capacityGb: number | null; + models: string[]; + state: MeshDeviceState; + isSelf: boolean; + memberPubkey?: string | null; + reportedAt?: number | null; + modelSizeGb?: number | null; +}; + +export type MeshSnapshot = { + sharingDeviceCount: number; + contributorMemberCount?: number; + sharedCapacityGb: number | null; + allocatedCapacityGb?: number | null; + models: string[]; + devices: MeshSnapshotDevice[]; + includesSelf: boolean; + observedAt?: number; + freshnessSeconds?: number; + memberCount: number; + reason: string | null; +}; + +export async function meshSnapshot(): Promise { + return await invokeTauri("mesh_snapshot"); +} diff --git a/desktop/src/shared/styles/globals/animations.css b/desktop/src/shared/styles/globals/animations.css index 5a0723cdaa..bc248a0523 100644 --- a/desktop/src/shared/styles/globals/animations.css +++ b/desktop/src/shared/styles/globals/animations.css @@ -930,3 +930,14 @@ animation: none; } } + +/* The shape stays fixed while a short dark stroke travels around its edge. */ +@keyframes mesh-compute-hex-trace { + to { + stroke-dashoffset: -100; + } +} + +.mesh-compute-hex-trace { + animation: mesh-compute-hex-trace 1.1s linear infinite; +} diff --git a/desktop/tests/e2e/mesh-compute.spec.ts b/desktop/tests/e2e/mesh-compute.spec.ts index c55b9db5d9..e7dd937d4b 100644 --- a/desktop/tests/e2e/mesh-compute.spec.ts +++ b/desktop/tests/e2e/mesh-compute.spec.ts @@ -26,9 +26,6 @@ test("Share compute chooses a model before sharing", async ({ page }) => { const model = page.getByTestId("mesh-share-compute-model"); await expect(card).not.toContainText("Not sharing right now"); - await expect( - page.getByTestId("mesh-share-compute-options-motion"), - ).toHaveCount(0); await expect( page.getByTestId("mesh-share-compute-sharing-status"), ).toHaveCount(0); @@ -39,17 +36,20 @@ test("Share compute chooses a model before sharing", async ({ page }) => { await page.getByLabel("Custom model reference").fill(modelRef); await toggle.click(); - await expect( - page.getByTestId("mesh-share-compute-options-motion"), - ).toBeVisible(); + await expect(toggle).toBeChecked(); + await expect(card).toContainText("You’re sharing compute"); + await expect(card).toContainText("This machine is actively sharing compute"); + await expect(model).toBeVisible(); + await expect(card).toContainText( + "Buzz downloads the selected model when sharing starts", + ); + + // Runtime/model details are intentionally secondary to the successful + // sharing state and remain available under Advanced. + await page.getByTestId("mesh-share-compute-advanced-toggle").click(); await expect( page.getByTestId("mesh-share-compute-sharing-status"), ).toBeVisible(); - await expect(model).toBeVisible(); - await expect(card).toContainText( - "Buzz downloads remote models when sharing starts", - ); - await expect(toggle).toBeChecked(); await expect( page.getByTestId("mesh-share-compute-sharing-status"), ).toContainText("SmolLM2 135M with relay members"); @@ -74,9 +74,6 @@ test("Share compute chooses a model before sharing", async ({ page }) => { await toggle.click(); await expect(toggle).not.toBeChecked(); await expect(card).not.toContainText("Not sharing right now"); - await expect( - page.getByTestId("mesh-share-compute-options-motion"), - ).toHaveCount(0); await expect( page.getByTestId("mesh-share-compute-sharing-status"), ).toHaveCount(0); @@ -123,9 +120,6 @@ test("a consuming client can switch to sharing its saved local model", async ({ ); await expect(card).toContainText("Buzz may briefly restart"); await expect(toggle).not.toBeChecked(); - await expect( - page.getByTestId("mesh-share-compute-options-motion"), - ).toHaveCount(0); await expect(toggle).toBeEnabled(); const customModel = page.getByLabel("Custom model reference"); await expect(customModel).toHaveValue(localModel);