mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
fix(desktop): supervise and re-arm relay-mesh runtime (#2823)
Refs #2062 This carries forward the relay-mesh recovery work from #2304 by @Bartok9 (cherry-picked with original authorship/sign-offs) and adds the startup/readiness supervision and packaging fixes found while validating it against a real two-machine Buzz setup. ## From #2304 - Watch the local OpenAI ingress (`:9337`) after launch and re-arm a stale relay-mesh runtime. - Require consecutive failed probes before eviction so a transient inference stall does not cause a cold restart. - Bound stale-runtime shutdown, preserve runtime identity across the asynchronous probe, and never evict a concurrent replacement. - Re-arm only for agents that are actually running; deliberately stopped agents stay stopped. - Persist an actionable sentinel error under the managed-agent store lock and clear only that error after recovery. - Treat serve-to-client fallback as an intentional fail-safe; configured serve restoration remains on its existing path. ## Added here - Use the inference ingress (`:9337`), rather than management port `:3131`, as Buzz's client-readiness boundary. A usable client no longer fails or holds agent-save open merely because management startup is still pending. - Supervise the embedded SDK startup asynchronously, publish a pending status while management is unavailable, and keep its mesh identity alive. - Avoid racing a replacement while SDK startup still owns the embedded runtime. If that pending startup later loses ingress while a running agent still needs it, request a controlled Buzz restart to reclaim the otherwise-unreachable SDK thread. - Defer roster-driven replacement while client management startup is pending. - Keep post-launch recovery in a dedicated module so the mesh entry point remains within the desktop file-size gate. - Explicitly mark generated Unix sidecars executable. On macOS, copying over an existing non-executable destination preserved its old mode, causing packaged `buzz-acp`, `buzz-agent`, and tool sidecars to be reported as missing. ## Validation Automated: - `just ci` — passed, including formatting, Clippy with warnings denied, desktop/web/mobile checks and tests, and builds. - Full Tauri mesh-feature suite — 1,702 passed, 0 failed, 15 ignored. - Mesh-feature Clippy with `-D warnings` — passed. - Release macOS app bundle with `mesh-llm` — built successfully; every bundled sidecar passed executable-mode and deep code-signature verification. Live two-machine E2E: - M5: released Buzz serving `unsloth/gemma-4-26B-A4B-it-GGUF:UD-Q4_K_M`. - Mac mini: this branch's packaged Buzz running a saved `buzz-agent`/relay-mesh agent. - Confirmed `:9337` accepted inference and the saved ACP harness started with no error while `:3131` was still unavailable. - Exact inference succeeded before restart (`CORRECTED-MINI-E2E-OK`). - Bespoke Buzz shut down cleanly in 2.3s, then restored ingress and the saved harness in 18.8s while `:3131` was still unavailable. - Exact inference succeeded after restart (`AFTER-RESTART-E2E-OK`). - A real Buzz `@C55` message traversed desktop → `buzz-acp` → `buzz-agent` → mini `:9337` → M5 compute and published the requested reply successfully. --------- Signed-off-by: Bartok9 <danielrpike9@gmail.com> Signed-off-by: Michael Neale <michael.neale@gmail.com> Co-authored-by: Bartok9 <danielrpike9@gmail.com> Co-authored-by: Michael Neale <michael.neale@gmail.com>
This commit is contained in:
Mic Neale
co-authored by
Bartok9
Michael Neale
Bartok9
Michael Neale
parent
8eb6e3eb60
commit
aa51dab9da
@@ -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,
|
||||
|
||||
@@ -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<Option<crate::mesh_llm::DesktopMeshRuntime>>,
|
||||
#[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()),
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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::<AppState>();
|
||||
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(())
|
||||
}
|
||||
|
||||
@@ -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<anyhow::Result<EmbeddedNodeHandle>>,
|
||||
queued_join_tokens: Vec<String>,
|
||||
},
|
||||
Ready(EmbeddedNodeHandle),
|
||||
Failed(String),
|
||||
}
|
||||
|
||||
pub struct DesktopMeshRuntime {
|
||||
handle: EmbeddedNodeHandle,
|
||||
id: u64,
|
||||
handle: tokio::sync::Mutex<DesktopMeshHandle>,
|
||||
mode: MeshNodeMode,
|
||||
api_base_url: String,
|
||||
console_url: String,
|
||||
model_id: Option<String>,
|
||||
model_name: Option<String>,
|
||||
/// 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<MeshNodeStatus> {
|
||||
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<serde_json::Value> {
|
||||
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<MeshServingUsage> {
|
||||
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<String>) -> 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<Vec<MeshModelOption>> {
|
||||
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(()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -3,6 +3,71 @@
|
||||
use super::find_progressish_reason;
|
||||
use serde_json::json;
|
||||
|
||||
fn pending_client_runtime(
|
||||
task: tokio::task::JoinHandle<anyhow::Result<mesh_llm_sdk::EmbeddedNodeHandle>>,
|
||||
) -> 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::<anyhow::Result<mesh_llm_sdk::EmbeddedNodeHandle>>().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!(
|
||||
|
||||
@@ -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::<AppState>();
|
||||
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<String> {
|
||||
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<String>,
|
||||
) -> 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::<AppState>();
|
||||
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::<AppState>();
|
||||
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<u32>,
|
||||
) -> 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<String> {
|
||||
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));
|
||||
}
|
||||
}
|
||||
@@ -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"
|
||||
|
||||
Reference in New Issue
Block a user