diff --git a/desktop/scripts/check-file-sizes.mjs b/desktop/scripts/check-file-sizes.mjs index 96f97126e0..f295f95fbc 100644 --- a/desktop/scripts/check-file-sizes.mjs +++ b/desktop/scripts/check-file-sizes.mjs @@ -400,7 +400,10 @@ const overrides = new Map([ // transition lock doc broadened to cover all protected-PID transitions, and // clear_agent_session_caches (per-pubkey retain) added alongside the // per-key clear. Load-bearing identity-contract change; queued to split. - ["src-tauri/src/app_state.rs", 1081], + // +4 (1081 -> 1085): mesh recovery keeps one app-scoped state object beside + // the embedded runtime and coordinator. Probe/re-arm logic lives in + // mesh_llm/recovery.rs rather than growing AppState or command modules. + ["src-tauri/src/app_state.rs", 1085], // multi-slot splitting + no-op suppression (#1309): the ReadStateManager // class grew from ~700 lines to ~1019 with the addition of // splitContextsIntoBudgetedSlots (pure fn + 5 tests), publishSplitSlots, diff --git a/desktop/src-tauri/src/app_state.rs b/desktop/src-tauri/src/app_state.rs index 778bd5fb3f..94d162e620 100644 --- a/desktop/src-tauri/src/app_state.rs +++ b/desktop/src-tauri/src/app_state.rs @@ -108,6 +108,8 @@ pub struct AppState { /// In-process mesh-llm node started by Buzz Desktop. #[cfg(feature = "mesh-llm")] pub mesh_llm_runtime: AsyncMutex>, + #[cfg(feature = "mesh-llm")] + pub mesh_recovery: crate::mesh_llm::MeshRecoveryState, /// Runtime-owned shared-compute coordinator. It publishes member-signed /// discovery status and reconciles MeshLLM's admission roster; MeshLLM /// itself owns direct QUIC/iroh connection establishment. @@ -223,6 +225,8 @@ pub fn build_app_state() -> AppState { #[cfg(feature = "mesh-llm")] mesh_llm_runtime: AsyncMutex::new(None), #[cfg(feature = "mesh-llm")] + mesh_recovery: crate::mesh_llm::MeshRecoveryState::default(), + #[cfg(feature = "mesh-llm")] mesh_coordinator: AsyncMutex::new(None), pending_owned_channels: Mutex::new(std::collections::HashSet::new()), } diff --git a/desktop/src-tauri/src/commands/mesh_llm.rs b/desktop/src-tauri/src/commands/mesh_llm.rs index 9a0a3c32eb..adcdeb7ec1 100644 --- a/desktop/src-tauri/src/commands/mesh_llm.rs +++ b/desktop/src-tauri/src/commands/mesh_llm.rs @@ -157,7 +157,7 @@ pub(crate) async fn restore_mesh_sharing(app: &AppHandle, state: &AppState) -> C }; let started = mesh_llm::DesktopMeshRuntime::start(request) .await - .map_err(|error| format!("failed to restore Share Compute: {error}"))?; + .map_err(|error| format!("failed to restore Share Compute: {error:#}"))?; *runtime = Some(started); drop(runtime); mesh_llm::publish_current_status_once(app, "restore").await; @@ -183,11 +183,11 @@ pub async fn mesh_start_node( let saved_request = request.clone(); let started = mesh_llm::DesktopMeshRuntime::start(request) .await - .map_err(|error| error.to_string())?; + .map_err(|error| format!("{error:#}"))?; let status = started .status() .await - .map_err(|error| format!("mesh node started but status probe failed: {error}"))?; + .map_err(|error| format!("mesh node started but status probe failed: {error:#}"))?; *runtime = Some(started); drop(runtime); if saved_request.mode == mesh_llm::MeshNodeMode::Serve { @@ -407,11 +407,11 @@ pub(crate) async fn ensure_client_node_for_model( } let started = mesh_llm::DesktopMeshRuntime::start(start) .await - .map_err(|error| format!("mesh client failed to start: {error}"))?; + .map_err(|error| format!("mesh client failed to start: {error:#}"))?; let status = started .status() .await - .map_err(|error| format!("mesh client started but status probe failed: {error}"))?; + .map_err(|error| format!("mesh client started but status probe failed: {error:#}"))?; *runtime = Some(started); Ok(status) } @@ -490,9 +490,47 @@ pub(crate) async fn ensure_relay_mesh_for_record( }; // A local serve/client runtime already owns the OpenAI ingress and its // router can resolve both `auto` and explicit remote models. Do not require - // a separate relay-advertised target in that case. + // a separate relay-advertised target in that case — BUT only trust it when + // the ingress is actually alive. A runtime that exited/wedged after launch + // leaves `mesh_llm_runtime = Some` pointing at a dead `:9337` ingress, so a + // blind `wait_for_mesh_inference` would just time out and the agent would + // stay silent (#2062). Probe first; if the ingress is dead, drop the stale + // runtime and fall through to re-arm it. The mesh coordinator watchdog also + // calls this path after eviction so recovery is not start-only (Brad #2304). if state.mesh_llm_runtime.lock().await.is_some() { - return wait_for_mesh_inference(&model_id).await; + match mesh_llm::recover_stale_mesh_runtime( + &state, + mesh_llm::MeshRecoveryUrgency::Foreground, + ) + .await + { + mesh_llm::MeshRuntimeRecovery::Live => { + return wait_for_mesh_inference(&model_id).await; + } + mesh_llm::MeshRuntimeRecovery::Evicted | mesh_llm::MeshRuntimeRecovery::Absent => {} + mesh_llm::MeshRuntimeRecovery::Debouncing => { + return Err( + "Buzz shared compute ingress is temporarily unresponsive; recovery is already scheduled. Try again shortly." + .to_string(), + ); + } + mesh_llm::MeshRuntimeRecovery::ReleasePending => { + return Err( + "Buzz shared compute is still shutting down its previous local ingress. Try again shortly." + .to_string(), + ); + } + mesh_llm::MeshRuntimeRecovery::Replaced => { + return wait_for_mesh_inference(&model_id).await; + } + mesh_llm::MeshRuntimeRecovery::RestartRequired => { + app.request_restart(); + return Err( + "Buzz shared compute startup lost its local ingress before shutdown control became available. Buzz is restarting to recover it." + .to_string(), + ); + } + } } let target = match resolve_mesh_bootstrap_target(&state, &model_id).await { Ok(Some(target)) => target, @@ -509,6 +547,15 @@ pub(crate) async fn ensure_relay_mesh_for_record( } }; + // Serve→Client re-arm transition (micspiral review #3, intentional-by-design): + // if the dead ingress belonged to a *serve* node with running consumer + // agents, this re-arms it as a Client (`MeshNodeMode::Client`). That is the + // correct/safe recovery here — config-backed serve restoration is + // `restore_mesh_sharing`'s job (`MeshNodeMode::Serve`), and + // `ensure_client_node_for_model` reuses any live runtime of *either* mode + // (the router resolves per-request), so it only cold-starts a Client when + // there is genuinely no runtime. Falling back to Client if a serve node + // crashed under local pressure is a desirable fail-safe, not a regression. ensure_client_node_for_model(&state, &model_id, Some(target.endpoint_addr)).await?; wait_for_mesh_inference(&model_id).await } diff --git a/desktop/src-tauri/src/mesh_llm/coordinator.rs b/desktop/src-tauri/src/mesh_llm/coordinator.rs index e83bf8b280..930149d0a5 100644 --- a/desktop/src-tauri/src/mesh_llm/coordinator.rs +++ b/desktop/src-tauri/src/mesh_llm/coordinator.rs @@ -22,10 +22,16 @@ const STATUS_D_TAG_PREFIX: &str = "buzz-mesh-member-status"; const ROSTER_POLL_INTERVAL: Duration = Duration::from_secs(60); const STATUS_PUBLISH_INTERVAL: Duration = Duration::from_secs(45); const STATUS_PUBLISH_TIMEOUT: Duration = Duration::from_secs(10); +/// Post-launch ingress liveness / re-arm for #2062. Bounded backoff: base 15s, +/// doubles after consecutive failures up to 120s so a sticky offline peer does +/// not hammer discovery every tick, but a recovered peer is noticed quickly. +const INGRESS_WATCHDOG_BASE: Duration = Duration::from_secs(15); +const INGRESS_WATCHDOG_MAX: Duration = Duration::from_secs(120); pub struct MeshCoordinator { _status_publisher: tokio::task::JoinHandle<()>, _roster_watcher: tokio::task::JoinHandle<()>, + _ingress_watchdog: tokio::task::JoinHandle<()>, } /// Start the runtime-owned status publisher and admission-roster watcher. @@ -63,16 +69,40 @@ pub async fn start_coordinator(app: AppHandle) { } }); + // Brad #2304 / #2062: ensure_relay_mesh_for_record only runs on explicit + // start + launch restore. After launch, local buzz-agent processes talk + // directly to :9337; there is no desktop "turn dispatch" hook. This + // watchdog is the post-launch seam: probe ingress, drop a zombie handle, + // re-arm via ensure_relay_mesh_for_record, surface last_error on failure. + let ingress_app = app.clone(); + let ingress_watchdog = tokio::spawn(async move { + let mut sleep_for = INGRESS_WATCHDOG_BASE; + loop { + tokio::time::sleep(sleep_for).await; + match crate::mesh_llm::rearm_relay_mesh_for_running_agents(&ingress_app).await { + Ok(()) => { + sleep_for = INGRESS_WATCHDOG_BASE; + } + Err(error) => { + eprintln!("buzz-mesh: ingress re-arm watchdog: {error}"); + sleep_for = (sleep_for * 2).min(INGRESS_WATCHDOG_MAX); + } + } + } + }); + let state = app.state::(); let mut guard = state.mesh_coordinator.lock().await; if guard.is_none() { *guard = Some(MeshCoordinator { _status_publisher: status_publisher, _roster_watcher: roster_watcher, + _ingress_watchdog: ingress_watchdog, }); } else { status_publisher.abort(); roster_watcher.abort(); + ingress_watchdog.abort(); } } @@ -182,6 +212,16 @@ async fn reconcile_roster( let mut request = current_request; request.trusted_owner_ids = Some(fresh); let mut guard = state.mesh_llm_runtime.lock().await; + let startup_pending = match guard.as_ref() { + Some(runtime) => runtime.is_starting().await, + None => false, + }; + if startup_pending { + eprintln!( + "buzz-mesh: membership roster changed while client management startup is pending; deferring restart" + ); + return Ok(()); + } let Some(running) = guard.take() else { return Ok(()); }; @@ -191,7 +231,7 @@ async fn reconcile_roster( } let replacement = crate::mesh_llm::DesktopMeshRuntime::start(request) .await - .map_err(|error| format!("mesh node restart after roster change failed: {error}"))?; + .map_err(|error| format!("mesh node restart after roster change failed: {error:#}"))?; *guard = Some(replacement); Ok(()) } diff --git a/desktop/src-tauri/src/mesh_llm/mod.rs b/desktop/src-tauri/src/mesh_llm/mod.rs index acece498c0..c041aa07a9 100644 --- a/desktop/src-tauri/src/mesh_llm/mod.rs +++ b/desktop/src-tauri/src/mesh_llm/mod.rs @@ -22,6 +22,13 @@ pub use identity::ensure_owner_identity; mod progress; pub use progress::install_progress_sink; +mod recovery; +pub use recovery::MeshRecoveryState; +pub(crate) use recovery::{ + rearm_relay_mesh_for_running_agents, recover_stale_mesh_runtime, MeshRecoveryUrgency, + MeshRuntimeRecovery, +}; + mod transport_policy; #[cfg(test)] use transport_policy::iroh_relay_mode_from; @@ -48,6 +55,17 @@ const MESH_IROH_RELAYS_ENV: &str = "BUZZ_MESH_IROH_RELAYS"; /// First model load can include a multi-GB download plus Metal warmup; the /// SDK default (30s) times out long before that. Matches mesh-console. const MESH_STARTUP_TIMEOUT: Duration = Duration::from_secs(180); +/// The pinned SDK defines startup readiness as the management API on `:3131`. +/// Buzz defines client readiness by the OpenAI ingress agents consume on +/// `:9337`, so it supervises client startup independently and gives the SDK a +/// long management deadline. A live ingress is therefore not torn down merely +/// because the optional management API is delayed; Buzz's ingress watchdog is +/// responsible for aborting and re-arming genuinely dead attempts. +const MESH_CLIENT_MANAGEMENT_TIMEOUT: Duration = Duration::from_secs(365 * 24 * 60 * 60); +/// Bound explicit and watchdog-driven shutdowns. `EmbeddedNodeHandle::stop` +/// sends the shutdown signal before awaiting the runtime thread, so dropping +/// that wait after the deadline still leaves a graceful shutdown in flight. +const MESH_STOP_TIMEOUT: Duration = Duration::from_secs(12); /// Sentinel model id meaning "let the mesh router pick". mesh-llm's OpenAI /// ingress auto-routes `"model": "auto"` to a context-compatible live target /// (`resolve_auto_routed_model`), so agents don't have to name a model and @@ -229,6 +247,7 @@ pub struct MeshServingUsage { impl MeshServingUsage { /// True when at least one request has been served for a non-local consumer. + #[cfg(test)] pub fn has_remote_consumers(&self) -> bool { self.remote_attempts > 0 || self.endpoint_attempts > 0 } @@ -286,9 +305,31 @@ pub fn stopped_status() -> MeshNodeStatus { } } +/// Monotonic id source so callers can compare runtime *identity* across an +/// `.await` point. The re-arm watchdog must not evict a fresh replacement that +/// a concurrent stop/start swapped in while the ingress probe was in flight +/// (Brad #2304 race), so it captures the id before probing and only evicts if +/// the same handle is still installed on lock reacquire. +static MESH_RUNTIME_ID_SEQ: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); + +enum DesktopMeshHandle { + /// The pinned SDK does not return its handle until `:3131/api/status` is + /// available. Keep that wait off the agent-save path while Buzz supervises + /// the real `:9337` ingress independently. + Starting { + task: tokio::task::JoinHandle>, + queued_join_tokens: Vec, + }, + Ready(EmbeddedNodeHandle), + Failed(String), +} + pub struct DesktopMeshRuntime { - handle: EmbeddedNodeHandle, + id: u64, + handle: tokio::sync::Mutex, mode: MeshNodeMode, + api_base_url: String, + console_url: String, model_id: Option, model_name: Option, /// The request this node was started with. Kept so the coordinator can @@ -362,6 +403,8 @@ impl DesktopMeshRuntime { ensure_model_downloaded(model).await?; } } + let api_port = mesh_api_port()?; + let console_port = mesh_console_port()?; let handle = match request.mode { MeshNodeMode::Serve => { let model = model_id @@ -369,8 +412,8 @@ impl DesktopMeshRuntime { .ok_or_else(|| anyhow::anyhow!("modelId is required for serve mode"))?; let mut builder = serve::EmbeddedServeConfig::builder() .model(model) - .api_port(mesh_api_port()?) - .console_port(mesh_console_port()?) + .api_port(api_port) + .console_port(console_port) // No-leak invariants: never publish mesh presence, never // auto-discover other meshes, no public Nostr relays. // Iroh relays are transport-only and enabled by default @@ -405,17 +448,17 @@ impl DesktopMeshRuntime { .trust_policy(TrustPolicy::Allowlist) .trust_owners(owners); } - serve::start(builder.build()).await? + DesktopMeshHandle::Ready(serve::start(builder.build()).await?) } MeshNodeMode::Client => { let mut builder = client::EmbeddedClientConfig::builder() - .api_port(mesh_api_port()?) - .console_port(mesh_console_port()?) + .api_port(api_port) + .console_port(console_port) // Same no-leak invariants as serve mode above. .publish(false) .auto_join(false) .discovery_mode(MeshDiscoveryMode::Nostr) - .startup_timeout(MESH_STARTUP_TIMEOUT) + .startup_timeout(MESH_CLIENT_MANAGEMENT_TIMEOUT) .console_ui(true); builder = match iroh_relay_mode()? { IrohRelayMode::Disabled => builder.disable_iroh_relays(true), @@ -437,19 +480,33 @@ impl DesktopMeshRuntime { .trust_policy(TrustPolicy::Allowlist) .trust_owners(owners); } - client::start(builder.build()).await? + let config = builder.build(); + DesktopMeshHandle::Starting { + task: tokio::spawn(async move { client::start(config).await }), + queued_join_tokens: Vec::new(), + } } }; Ok(Self { - handle, + id: MESH_RUNTIME_ID_SEQ.fetch_add(1, std::sync::atomic::Ordering::Relaxed), + handle: tokio::sync::Mutex::new(handle), mode: request.mode, + api_base_url: format!("http://127.0.0.1:{api_port}/v1"), + console_url: format!("http://127.0.0.1:{console_port}"), model_id, model_name, start_request: request, }) } + /// Process-unique identity for this runtime instance. Used by the re-arm + /// watchdog to detect a concurrent handle swap across the ingress probe + /// `.await` so it never evicts a fresh replacement runtime (Brad #2304). + pub fn id(&self) -> u64 { + self.id + } + /// The request this node was started with (roster drift detection). pub fn start_request(&self) -> &StartMeshNodeRequest { &self.start_request @@ -463,13 +520,117 @@ impl DesktopMeshRuntime { self.mode } + async fn promote_finished_startup(handle: &mut DesktopMeshHandle) { + let startup_finished = matches!( + handle, + DesktopMeshHandle::Starting { task, .. } if task.is_finished() + ); + if !startup_finished { + return; + } + + let previous = std::mem::replace( + handle, + DesktopMeshHandle::Failed("mesh client startup state was unavailable".to_string()), + ); + let (task, queued_join_tokens) = match previous { + DesktopMeshHandle::Starting { + task, + queued_join_tokens, + } => (task, queued_join_tokens), + other => { + *handle = other; + return; + } + }; + *handle = match task.await { + Ok(Ok(ready)) => { + for token in queued_join_tokens { + if let Err(error) = ready.join_token(token).await { + eprintln!( + "buzz-mesh: failed to apply a dial request queued during startup: {error:#}" + ); + } + } + DesktopMeshHandle::Ready(ready) + } + Ok(Err(error)) => { + DesktopMeshHandle::Failed(format!("embedded mesh client startup failed: {error:#}")) + } + Err(error) if error.is_cancelled() => { + DesktopMeshHandle::Failed("embedded mesh client startup was cancelled".to_string()) + } + Err(error) => DesktopMeshHandle::Failed(format!( + "embedded mesh client startup task failed: {error}" + )), + }; + } + + pub async fn is_starting(&self) -> bool { + let mut handle = self.handle.lock().await; + Self::promote_finished_startup(&mut handle).await; + matches!(*handle, DesktopMeshHandle::Starting { .. }) + } + pub async fn status(&self) -> anyhow::Result { - let status = self.handle.status().await?; - self.status_from_sdk(status) + let mut handle = self.handle.lock().await; + Self::promote_finished_startup(&mut handle).await; + match &*handle { + DesktopMeshHandle::Ready(ready) => { + let status = ready.status().await; + drop(handle); + match status { + Ok(status) => self.status_from_sdk(status), + Err(error) + if self.mode == MeshNodeMode::Client + && recovery::mesh_ingress_is_live_at(&self.api_base_url).await => + { + Ok(self.ingress_only_client_status(&format!( + "management status is unavailable: {error:#}" + ))) + } + Err(error) => Err(error), + } + } + DesktopMeshHandle::Starting { .. } => { + drop(handle); + if recovery::mesh_ingress_is_live_at(&self.api_base_url).await { + Ok(self.ingress_only_client_status("management status is still starting")) + } else { + Ok(self.starting_client_status()) + } + } + DesktopMeshHandle::Failed(error) => { + let error = error.clone(); + Err(anyhow::anyhow!(error)) + } + } } pub async fn status_report_payload(&self) -> anyhow::Result { - let status = self.handle.status().await?; + let mut handle = self.handle.lock().await; + Self::promote_finished_startup(&mut handle).await; + let ready = match &*handle { + DesktopMeshHandle::Ready(ready) => ready, + DesktopMeshHandle::Starting { .. } if self.mode == MeshNodeMode::Client => { + // Consumer identities must keep publishing owner bindings while + // the SDK is still waiting for its management API. Serving + // peers derive their admission roster from these fresh member + // notes; withholding heartbeats here would make an otherwise + // working `:9337` client age out of the host's allowlist. + return Ok(serde_json::json!({ + "serveTargets": [], + "models": [], + })); + } + DesktopMeshHandle::Starting { .. } => { + anyhow::bail!("mesh management status is not ready") + } + DesktopMeshHandle::Failed(error) => { + anyhow::bail!("mesh runtime failed: {error}") + } + }; + let status = ready.status().await?; let mut payload = status.payload; enrich_status_payload_identity(&mut payload, status.invite_token.as_deref()); if let Ok(identity) = ensure_owner_identity() { @@ -504,18 +665,43 @@ impl DesktopMeshRuntime { /// Read-only host-side usage snapshot from the node's own runtime metrics. pub async fn serving_usage(&self) -> anyhow::Result { - let status = self.handle.status().await?; + let mut handle = self.handle.lock().await; + Self::promote_finished_startup(&mut handle).await; + let DesktopMeshHandle::Ready(ready) = &*handle else { + anyhow::bail!("mesh management status is not ready"); + }; + let status = ready.status().await?; Ok(serving_usage_from_payload(&status.payload)) } pub async fn dial_endpoint_addr(&self, endpoint_addr: impl Into) -> anyhow::Result<()> { let endpoint_addr = endpoint_addr.into(); let validated = validate_advertised_endpoint(&endpoint_addr)?; - self.handle.join_token(validated.join_token).await + let mut handle = self.handle.lock().await; + Self::promote_finished_startup(&mut handle).await; + match &mut *handle { + DesktopMeshHandle::Ready(ready) => ready.join_token(validated.join_token).await, + DesktopMeshHandle::Starting { + queued_join_tokens, .. + } => { + if !queued_join_tokens.contains(&validated.join_token) { + queued_join_tokens.push(validated.join_token); + } + Ok(()) + } + DesktopMeshHandle::Failed(error) => { + anyhow::bail!("cannot dial from failed mesh client: {error}") + } + } } pub async fn installed_models(&self) -> anyhow::Result> { - let status = self.handle.status().await?; + let mut handle = self.handle.lock().await; + Self::promote_finished_startup(&mut handle).await; + let DesktopMeshHandle::Ready(ready) = &*handle else { + anyhow::bail!("mesh management status is not ready"); + }; + let status = ready.status().await?; Ok(models_from_status_payload(Some(&status.payload))) } @@ -543,8 +729,60 @@ impl DesktopMeshRuntime { }) } + fn ingress_only_client_status(&self, reason: &str) -> MeshNodeStatus { + MeshNodeStatus { + state: MeshNodeState::Running, + mode: Some(MeshNodeMode::Client), + health: MeshHealth::degraded(format!("OpenAI ingress is live; {reason}")), + api_base_url: Some(self.api_base_url.clone()), + console_url: Some(self.console_url.clone()), + model_id: self.model_id.clone(), + model_name: self.model_name.clone(), + invite_token: None, + endpoint_id: None, + device_id: None, + device_name: None, + } + } + + fn starting_client_status(&self) -> MeshNodeStatus { + MeshNodeStatus { + state: MeshNodeState::Starting, + mode: Some(MeshNodeMode::Client), + health: MeshHealth::degraded("OpenAI ingress and management status are still starting"), + api_base_url: Some(self.api_base_url.clone()), + console_url: Some(self.console_url.clone()), + model_id: self.model_id.clone(), + model_name: self.model_name.clone(), + invite_token: None, + endpoint_id: None, + device_id: None, + device_name: None, + } + } + pub async fn stop(self) -> anyhow::Result<()> { - self.handle.stop().await + match self.handle.into_inner() { + DesktopMeshHandle::Ready(ready) => { + match tokio::time::timeout(MESH_STOP_TIMEOUT, ready.stop()).await { + Ok(result) => result, + Err(_) => anyhow::bail!( + "timed out after {}s waiting for embedded mesh runtime to stop", + MESH_STOP_TIMEOUT.as_secs() + ), + } + } + DesktopMeshHandle::Starting { task, .. } => { + // The pinned SDK has not exposed its shutdown handle yet. + // Abort only the Buzz-side waiter; normal callers never replace + // a pending runtime, and app shutdown/restart then terminates + // the process-owned embedded thread. Recovery requests that + // controlled restart instead of racing a second runtime. + task.abort(); + Ok(()) + } + DesktopMeshHandle::Failed(_) => Ok(()), + } } } diff --git a/desktop/src-tauri/src/mesh_llm/mod_tests.rs b/desktop/src-tauri/src/mesh_llm/mod_tests.rs index a8b78bc190..1f2c6abb44 100644 --- a/desktop/src-tauri/src/mesh_llm/mod_tests.rs +++ b/desktop/src-tauri/src/mesh_llm/mod_tests.rs @@ -3,6 +3,71 @@ use super::find_progressish_reason; use serde_json::json; +fn pending_client_runtime( + task: tokio::task::JoinHandle>, +) -> super::DesktopMeshRuntime { + let request = super::StartMeshNodeRequest { + mode: super::MeshNodeMode::Client, + model_id: None, + max_vram_gb: None, + join_token: Some("initial-token".to_string()), + trusted_owner_ids: None, + }; + super::DesktopMeshRuntime { + id: 7, + handle: tokio::sync::Mutex::new(super::DesktopMeshHandle::Starting { + task, + queued_join_tokens: Vec::new(), + }), + mode: super::MeshNodeMode::Client, + api_base_url: "http://127.0.0.1:1/v1".to_string(), + console_url: "http://127.0.0.1:2".to_string(), + model_id: None, + model_name: None, + start_request: request, + } +} + +#[tokio::test] +async fn pending_client_status_does_not_wait_for_management_timeout() { + let task = tokio::spawn(async { + std::future::pending::>().await + }); + let runtime = pending_client_runtime(task); + + let status = tokio::time::timeout(std::time::Duration::from_secs(1), runtime.status()) + .await + .expect("status should not wait for the SDK management probe") + .expect("pending client should have a synthetic status"); + assert_eq!(status.state, super::MeshNodeState::Starting); + assert_eq!(status.mode, Some(super::MeshNodeMode::Client)); + let report = runtime + .status_report_payload() + .await + .expect("pending clients must keep publishing admission identity heartbeats"); + assert_eq!(report["serveTargets"], json!([])); + assert_eq!(report["models"], json!([])); + + tokio::time::timeout(std::time::Duration::from_secs(1), runtime.stop()) + .await + .expect("stopping a pending client should abort its SDK task") + .expect("pending client stop should succeed"); +} + +#[tokio::test] +async fn failed_client_startup_is_promoted_without_losing_the_runtime_slot() { + let task = tokio::spawn(async { anyhow::bail!("controlled startup failure") }); + let runtime = pending_client_runtime(task); + tokio::task::yield_now().await; + + let error = runtime + .status() + .await + .expect_err("finished failed startup should be surfaced"); + assert!(error.to_string().contains("controlled startup failure")); + assert!(!runtime.is_starting().await); +} + #[test] fn progressish_reads_typed_phase_not_whole_tree() { assert_eq!( diff --git a/desktop/src-tauri/src/mesh_llm/recovery.rs b/desktop/src-tauri/src/mesh_llm/recovery.rs new file mode 100644 index 0000000000..29ed8a13fe --- /dev/null +++ b/desktop/src-tauri/src/mesh_llm/recovery.rs @@ -0,0 +1,493 @@ +use std::collections::HashSet; +use std::sync::atomic::{AtomicU32, AtomicU64, Ordering}; +use std::time::Duration; + +use tauri::{AppHandle, Manager}; + +use crate::app_state::AppState; + +const INGRESS_PROBE_TIMEOUT: Duration = Duration::from_secs(3); +const INGRESS_CONNECT_TIMEOUT: Duration = Duration::from_millis(300); +const INGRESS_RELEASE_TIMEOUT: Duration = Duration::from_secs(3); +const STALE_STOP_TIMEOUT: Duration = Duration::from_secs(12); +const DEAD_PROBE_EVICT_THRESHOLD: u32 = 2; + +/// Sentinel prefix on errors owned by this recovery path. Recovery clears only +/// these errors and never an unrelated agent failure. +pub(crate) const MESH_REARM_ERROR_SENTINEL: &str = "[buzz-mesh-rearm] "; + +/// App-scoped recovery coordination. The runtime id binds a dead-probe streak +/// to one specific handle, while `rearm_lock` prevents overlapping watchdog +/// passes from starting competing replacements. +pub struct MeshRecoveryState { + probe_runtime_id: AtomicU64, + dead_probes: AtomicU32, + rearm_lock: tokio::sync::Mutex<()>, +} + +impl Default for MeshRecoveryState { + fn default() -> Self { + Self { + probe_runtime_id: AtomicU64::new(0), + dead_probes: AtomicU32::new(0), + rearm_lock: tokio::sync::Mutex::new(()), + } + } +} + +impl MeshRecoveryState { + fn reset_probe_streak(&self) { + self.probe_runtime_id.store(0, Ordering::Relaxed); + self.dead_probes.store(0, Ordering::Relaxed); + } + + fn record_dead_probe(&self, runtime_id: u64) -> u32 { + if self.probe_runtime_id.swap(runtime_id, Ordering::Relaxed) != runtime_id { + self.dead_probes.store(1, Ordering::Relaxed); + return 1; + } + self.dead_probes.fetch_add(1, Ordering::Relaxed) + 1 + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum MeshIngressProbe { + Live, + PortClosed, + Unhealthy, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum MeshRecoveryUrgency { + Foreground, + Watchdog, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum MeshRuntimeRecovery { + Absent, + Live, + Debouncing, + Evicted, + ReleasePending, + Replaced, + RestartRequired, +} + +/// Probe the contract agents actually consume. A successful `/v1/models` +/// response proves the local OpenAI ingress is accepting requests; the +/// management console on `:3131` is intentionally not part of this health +/// decision. +pub(crate) async fn probe_mesh_ingress() -> MeshIngressProbe { + probe_mesh_ingress_at(crate::managed_agents::RELAY_MESH_API_BASE_URL).await +} + +pub(crate) async fn mesh_ingress_is_live_at(api_base_url: &str) -> bool { + probe_mesh_ingress_at(api_base_url).await == MeshIngressProbe::Live +} + +pub(crate) async fn probe_mesh_ingress_at(api_base_url: &str) -> MeshIngressProbe { + let base = api_base_url.trim_end_matches('/'); + let client = match reqwest::Client::builder() + .timeout(INGRESS_PROBE_TIMEOUT) + .build() + { + Ok(client) => client, + Err(_) => return MeshIngressProbe::Unhealthy, + }; + if client + .get(format!("{base}/models")) + .bearer_auth(crate::managed_agents::RELAY_MESH_API_KEY_PLACEHOLDER) + .send() + .await + .is_ok_and(|response| response.status().is_success()) + { + return MeshIngressProbe::Live; + } + if ingress_port_is_bound(api_base_url).await { + MeshIngressProbe::Unhealthy + } else { + MeshIngressProbe::PortClosed + } +} + +async fn ingress_port_is_bound(api_base_url: &str) -> bool { + let Ok(url) = url::Url::parse(api_base_url) else { + return false; + }; + let Some(host) = url.host_str() else { + return false; + }; + let Some(port) = url.port_or_known_default() else { + return false; + }; + matches!( + tokio::time::timeout( + INGRESS_CONNECT_TIMEOUT, + tokio::net::TcpStream::connect((host, port)), + ) + .await, + Ok(Ok(_)) + ) +} + +async fn wait_for_ingress_release() -> bool { + let deadline = tokio::time::Instant::now() + INGRESS_RELEASE_TIMEOUT; + loop { + if !ingress_port_is_bound(crate::managed_agents::RELAY_MESH_API_BASE_URL).await { + return true; + } + if tokio::time::Instant::now() >= deadline { + return false; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } +} + +fn should_evict_after_probe( + urgency: MeshRecoveryUrgency, + probe: MeshIngressProbe, + consecutive: u32, +) -> bool { + urgency == MeshRecoveryUrgency::Foreground && probe == MeshIngressProbe::PortClosed + || consecutive >= DEAD_PROBE_EVICT_THRESHOLD +} + +/// Probe and, when justified, remove one stale runtime. A closed port is +/// decisive for a foreground agent start; watchdog and ambiguous/unhealthy +/// ports require consecutive failures to avoid restarting on a transient load +/// spike. +pub(crate) async fn recover_stale_mesh_runtime( + state: &AppState, + urgency: MeshRecoveryUrgency, +) -> MeshRuntimeRecovery { + let (candidate_id, startup_in_progress) = match state.mesh_llm_runtime.lock().await.as_ref() { + Some(runtime) => (runtime.id(), runtime.is_starting().await), + None => { + state.mesh_recovery.reset_probe_streak(); + // A cancelled SDK startup can outlive its Buzz-side task briefly + // because the embedded runtime runs on its own thread. Never start + // a replacement merely because the tracked handle is gone: first + // prove the old ingress is either still useful or has released the + // port. This closes the port-conflict loop in #2304. + return match probe_mesh_ingress().await { + MeshIngressProbe::Live => MeshRuntimeRecovery::Live, + MeshIngressProbe::PortClosed => MeshRuntimeRecovery::Absent, + MeshIngressProbe::Unhealthy => MeshRuntimeRecovery::ReleasePending, + }; + } + }; + let probe = probe_mesh_ingress().await; + if probe == MeshIngressProbe::Live { + state.mesh_recovery.reset_probe_streak(); + return MeshRuntimeRecovery::Live; + } + let consecutive = state.mesh_recovery.record_dead_probe(candidate_id); + if startup_in_progress && consecutive < DEAD_PROBE_EVICT_THRESHOLD { + eprintln!( + "buzz-mesh: ingress is not live on the first startup probe; allowing the supervised SDK task one watchdog interval" + ); + return MeshRuntimeRecovery::Debouncing; + } + if !should_evict_after_probe(urgency, probe, consecutive) { + eprintln!( + "buzz-mesh: ingress probe {probe:?} ({consecutive}/{DEAD_PROBE_EVICT_THRESHOLD}); debouncing" + ); + return MeshRuntimeRecovery::Debouncing; + } + + // The pinned SDK does not yield its control handle until the management + // API is ready. Dropping its still-pending start future would detach the + // embedded runtime thread without sending a shutdown request, so Buzz must + // not evict it and race a replacement onto the same ports. A controlled + // app relaunch is the only process-owned cleanup boundary in this state. + if startup_in_progress { + state.mesh_recovery.reset_probe_streak(); + return MeshRuntimeRecovery::RestartRequired; + } + + let stale = { + let mut guard = state.mesh_llm_runtime.lock().await; + if guard.as_ref().map(|runtime| runtime.id()) != Some(candidate_id) { + state.mesh_recovery.reset_probe_streak(); + return MeshRuntimeRecovery::Replaced; + } + guard.take() + }; + let Some(stale) = stale else { + return MeshRuntimeRecovery::Replaced; + }; + state.mesh_recovery.reset_probe_streak(); + + match tokio::time::timeout(STALE_STOP_TIMEOUT, stale.stop()).await { + Ok(Ok(())) => {} + Ok(Err(error)) => eprintln!("buzz-mesh: stale runtime stop failed: {error:#}"), + Err(_) => eprintln!( + "buzz-mesh: stale runtime stop exceeded {}s; waiting for port release", + STALE_STOP_TIMEOUT.as_secs() + ), + } + if wait_for_ingress_release().await { + MeshRuntimeRecovery::Evicted + } else { + eprintln!( + "buzz-mesh: old runtime still owns the local ingress; deferring replacement to avoid a port-conflict loop" + ); + MeshRuntimeRecovery::ReleasePending + } +} + +/// Post-launch recovery for actively running relay-mesh agents. +pub(crate) async fn rearm_relay_mesh_for_running_agents(app: &AppHandle) -> Result<(), String> { + let state = app.state::(); + let _rearm_guard = state.mesh_recovery.rearm_lock.lock().await; + let recovery = recover_stale_mesh_runtime(&state, MeshRecoveryUrgency::Watchdog).await; + let active_pubkeys = active_managed_agent_pubkeys(&state); + + match recovery { + MeshRuntimeRecovery::Live + | MeshRuntimeRecovery::Debouncing + | MeshRuntimeRecovery::Replaced => return Ok(()), + MeshRuntimeRecovery::RestartRequired => { + let records = crate::managed_agents::load_managed_agents(app).unwrap_or_default(); + if !records + .iter() + .any(|record| is_running_relay_mesh_agent(record, &active_pubkeys)) + { + // A foreground save may still be bringing up its first ingress. + // Only an already-running consumer justifies an automatic app + // relaunch from the background watchdog. + return Ok(()); + } + eprintln!( + "buzz-mesh: supervised client startup lost its ingress before the SDK exposed a shutdown handle; restarting Buzz" + ); + app.request_restart(); + return Ok(()); + } + MeshRuntimeRecovery::ReleasePending => { + return Err(format!( + "{MESH_REARM_ERROR_SENTINEL}old local mesh ingress is still shutting down" + )); + } + MeshRuntimeRecovery::Absent => { + let records = crate::managed_agents::load_managed_agents(app).unwrap_or_default(); + if !records + .iter() + .any(|record| is_running_relay_mesh_agent(record, &active_pubkeys)) + { + return Ok(()); + } + } + MeshRuntimeRecovery::Evicted => {} + } + + let records = crate::managed_agents::load_managed_agents(app).unwrap_or_default(); + let mesh_records: Vec<_> = records + .into_iter() + .filter(|record| is_running_relay_mesh_agent(record, &active_pubkeys)) + .collect(); + let mut first_error = None; + for record in &mesh_records { + match crate::commands::mesh_llm::ensure_relay_mesh_for_record(app, record, false).await { + Ok(()) => { + if let Err(error) = clear_mesh_last_error_if_set(app, &record.pubkey) { + eprintln!("buzz-mesh: failed to clear recovery error: {error}"); + } + } + Err(error) => { + let message = format!( + "{MESH_REARM_ERROR_SENTINEL}Buzz shared compute offline — failed to re-arm local ingress for this agent: {error}" + ); + if let Err(persist_error) = persist_mesh_last_error(app, &record.pubkey, &message) { + eprintln!("buzz-mesh: failed to persist recovery error: {persist_error}"); + } + first_error.get_or_insert(message); + } + } + } + first_error.map_or(Ok(()), Err) +} + +fn active_managed_agent_pubkeys(state: &AppState) -> HashSet { + state + .managed_agent_processes + .lock() + .map(|guard| { + guard + .keys() + .map(|key| key.pubkey.to_ascii_lowercase()) + .collect() + }) + .unwrap_or_default() +} + +fn is_running_relay_mesh_agent( + record: &crate::managed_agents::ManagedAgentRecord, + active_pubkeys: &HashSet, +) -> bool { + record.backend == crate::managed_agents::BackendKind::Local + && crate::managed_agents::relay_mesh_model_id(record).is_some() + && active_pubkeys.contains(&record.pubkey.to_ascii_lowercase()) + && record + .runtime_pid + .is_none_or(crate::managed_agents::process_is_running) +} + +fn persist_mesh_last_error(app: &AppHandle, pubkey: &str, error: &str) -> Result<(), String> { + let state = app.state::(); + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| format!("failed to acquire managed agents store lock: {e}"))?; + let mut records = crate::managed_agents::load_managed_agents(app)?; + let record = crate::managed_agents::find_managed_agent_mut(&mut records, pubkey)?; + record.last_error = Some(error.to_string()); + record.updated_at = crate::util::now_iso(); + crate::managed_agents::save_managed_agents(app, &records) +} + +fn clear_mesh_last_error_if_set(app: &AppHandle, pubkey: &str) -> Result<(), String> { + let state = app.state::(); + let _store_guard = state + .managed_agents_store_lock + .lock() + .map_err(|e| format!("failed to acquire managed agents store lock: {e}"))?; + let mut records = crate::managed_agents::load_managed_agents(app)?; + let record = crate::managed_agents::find_managed_agent_mut(&mut records, pubkey)?; + if !record + .last_error + .as_deref() + .is_some_and(|error| error.starts_with(MESH_REARM_ERROR_SENTINEL)) + { + return Ok(()); + } + record.last_error = None; + record.updated_at = crate::util::now_iso(); + crate::managed_agents::save_managed_agents(app, &records) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn mesh_record( + pubkey: &str, + runtime_pid: Option, + ) -> crate::managed_agents::ManagedAgentRecord { + let mut record = crate::managed_agents::AgentDefinition { + id: pubkey.to_string(), + display_name: pubkey.to_string(), + avatar_url: None, + system_prompt: String::new(), + runtime: None, + model: None, + provider: None, + name_pool: Vec::new(), + is_builtin: false, + is_active: true, + source_team: None, + source_team_persona_slug: None, + env_vars: std::collections::BTreeMap::from([ + ("BUZZ_AGENT_PROVIDER".to_string(), "openai".to_string()), + ( + "OPENAI_COMPAT_BASE_URL".to_string(), + "http://127.0.0.1:9337/v1/".to_string(), + ), + ("OPENAI_COMPAT_MODEL".to_string(), "Qwen3".to_string()), + ( + "OPENAI_COMPAT_API_KEY".to_string(), + crate::managed_agents::RELAY_MESH_API_KEY_PLACEHOLDER.to_string(), + ), + ]), + respond_to: None, + respond_to_allowlist: Vec::new(), + parallelism: None, + created_at: "2026-01-01T00:00:00Z".to_string(), + updated_at: "2026-01-01T00:00:00Z".to_string(), + } + .into_agent_record(); + record.pubkey = pubkey.to_string(); + record.backend = crate::managed_agents::BackendKind::Local; + record.runtime_pid = runtime_pid; + record + } + + fn active_set(pubkeys: &[&str]) -> HashSet { + pubkeys + .iter() + .map(|pubkey| pubkey.to_ascii_lowercase()) + .collect() + } + + #[tokio::test] + async fn closed_port_is_distinct_from_unhealthy_bound_port() { + assert_eq!( + probe_mesh_ingress_at("http://127.0.0.1:1/v1").await, + MeshIngressProbe::PortClosed + ); + } + + #[test] + fn foreground_closed_port_evicts_without_debounce() { + assert!(should_evict_after_probe( + MeshRecoveryUrgency::Foreground, + MeshIngressProbe::PortClosed, + 1 + )); + assert!(!should_evict_after_probe( + MeshRecoveryUrgency::Watchdog, + MeshIngressProbe::PortClosed, + 1 + )); + } + + #[test] + fn probe_streak_is_scoped_to_runtime_identity() { + let state = MeshRecoveryState::default(); + assert_eq!(state.record_dead_probe(7), 1); + assert_eq!(state.record_dead_probe(7), 2); + assert_eq!(state.record_dead_probe(8), 1); + } + + #[test] + fn explicit_stop_budget_accommodates_sdk_forced_shutdown() { + assert!(STALE_STOP_TIMEOUT >= Duration::from_secs(10)); + assert!(STALE_STOP_TIMEOUT <= Duration::from_secs(15)); + } + + #[test] + fn only_running_relay_mesh_agents_trigger_rearm() { + let empty = active_set(&[]); + assert!(!is_running_relay_mesh_agent( + &mesh_record("stopped", Some(std::process::id())), + &empty + )); + + let active = active_set(&["live"]); + assert!(is_running_relay_mesh_agent( + &mesh_record("live", Some(std::process::id())), + &active + )); + assert!(is_running_relay_mesh_agent( + &mesh_record("live", None), + &active + )); + + let mut non_mesh = mesh_record("plain", Some(std::process::id())); + non_mesh.env_vars.clear(); + non_mesh.relay_mesh = None; + assert!(!is_running_relay_mesh_agent( + &non_mesh, + &active_set(&["plain"]) + )); + } + + #[test] + fn recovery_error_sentinel_does_not_match_unrelated_errors() { + assert!( + format!("{MESH_REARM_ERROR_SENTINEL}offline").starts_with(MESH_REARM_ERROR_SENTINEL) + ); + assert!(!"user note: shared compute config".starts_with(MESH_REARM_ERROR_SENTINEL)); + } +} diff --git a/scripts/bundle-sidecars.sh b/scripts/bundle-sidecars.sh index be37cbce0d..07de477405 100755 --- a/scripts/bundle-sidecars.sh +++ b/scripts/bundle-sidecars.sh @@ -35,6 +35,14 @@ fi mkdir -p "$BINARIES_DIR" for bin in "${SIDECARS[@]}"; do - cp "$SRC_DIR/${bin}${EXE}" "$BINARIES_DIR/${bin}-${TARGET}${EXE}" + destination="$BINARIES_DIR/${bin}-${TARGET}${EXE}" + cp "$SRC_DIR/${bin}${EXE}" "$destination" + + # cp preserves the mode of an existing destination on macOS. Generated + # sidecar placeholders may not be executable, so make the bundled Unix + # binaries executable explicitly. + if [[ -z "$EXE" ]]; then + chmod 755 "$destination" + fi done echo "Sidecars bundled for $TARGET"