From 28dbb529c0fc338c997c4ff03d5045ea2926e63e Mon Sep 17 00:00:00 2001 From: damocles Date: Fri, 10 Jul 2026 14:11:41 +0200 Subject: [PATCH] feat(#2289): debounced SweepHealth banner tracker, wire knowledge pull --- hive-c0re/src/lib.rs | 2 +- hive-c0re/src/main.rs | 30 +++- hive-c0re/src/stats/mod.rs | 1 + hive-c0re/src/stats/sweep_health.rs | 208 ++++++++++++++++++++++++++++ 4 files changed, 236 insertions(+), 5 deletions(-) create mode 100644 hive-c0re/src/stats/sweep_health.rs diff --git a/hive-c0re/src/lib.rs b/hive-c0re/src/lib.rs index e75d197e..6c1f717e 100644 --- a/hive-c0re/src/lib.rs +++ b/hive-c0re/src/lib.rs @@ -45,7 +45,7 @@ pub mod workers; // Root re-exports: keep every pre-grouping `crate::` / // `hive_c0re::` path compiling without touching consumers. pub use agent_config::{capabilities, limits, tool_groups, topology}; -pub use stats::{container_stats, hive_stats, host_stats, warnings}; +pub use stats::{container_stats, hive_stats, host_stats, sweep_health, warnings}; pub use stores::{ approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts, }; diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index c2d4e180..12cbac7a 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -14,7 +14,7 @@ use hive_c0re::coordinator::{Coordinator, HiveEnv, ServeConfig}; use hive_c0re::{ agent_sockets, auto_update, broker, client, crash_watch, dashboard, dashboard_events, forge, host_stats, job_queue, knowledge, matrix, mcp_sockets, migrate, reminder_scheduler, - scheduled_prompts_worker, server, socket_server, warnings, + scheduled_prompts_worker, server, socket_server, sweep_health, warnings, }; #[derive(Parser)] @@ -321,16 +321,38 @@ async fn cmd_serve( let mut knowledge_shutdown = coord.shutdown_rx(); tokio::spawn(async move { // Initial pull — reconcile any commits that landed while c0re - // was offline. + // was offline. Not fed to the health tracker: a startup miss is + // expected (the clone may not exist yet) and is logged at debug. if let Err(e) = knowledge::pull().await { tracing::debug!(error = ?e, "knowledge: startup pull skipped (no clone yet?)"); } + // Persistent-failure → banner. An hourly sweep that keeps failing for + // several hours means the operator's `/knowledge` is drifting; raise a + // warn banner after 3 consecutive misses so a one-off network blip + // self-heals on the next tick without ever bannering. Cleared on the + // next successful pull. + let mut health = sweep_health::SweepHealth::new("knowledge_pull", "warn", 3); let interval = std::time::Duration::from_hours(1); loop { tokio::select! { () = tokio::time::sleep(interval) => { - if let Err(e) = knowledge::pull().await { - tracing::warn!(error = ?e, "knowledge: periodic pull failed"); + match knowledge::pull().await { + Ok(()) => health.record_ok(), + Err(e) => { + tracing::warn!(error = ?e, "knowledge: periodic pull failed"); + let err = format!("{e:#}"); + health.record_err(|ctx| { + let age = ctx.since_last_ok.map_or_else( + || "no success this session".to_owned(), + |d| format!("last ok {} ago", sweep_health::fmt_age(d)), + ); + format!( + "knowledge repo pull failing ({} consecutive, {age}) \ + — /knowledge is stale until it recovers: {err}", + ctx.consecutive + ) + }); + } } } _ = knowledge_shutdown.changed() => { diff --git a/hive-c0re/src/stats/mod.rs b/hive-c0re/src/stats/mod.rs index 37ea752c..8b9d5ec8 100644 --- a/hive-c0re/src/stats/mod.rs +++ b/hive-c0re/src/stats/mod.rs @@ -6,4 +6,5 @@ pub mod container_stats; pub mod hive_stats; pub mod host_stats; +pub mod sweep_health; pub mod warnings; diff --git a/hive-c0re/src/stats/sweep_health.rs b/hive-c0re/src/stats/sweep_health.rs new file mode 100644 index 00000000..2dceb863 --- /dev/null +++ b/hive-c0re/src/stats/sweep_health.rs @@ -0,0 +1,208 @@ +//! Debounced sweep-failure tracking on top of the [`crate::warnings`] +//! registry. +//! +//! Background converge sweeps (knowledge pull, matrix / forge provisioning, +//! gateway nginx writes, per-agent branch protection, …) historically +//! `warn!`'d to the journal and moved on, so a persistent failure was +//! invisible to the operator. [`SweepHealth`] routes that failure to the +//! dashboard banner instead — but *debounced*, per the routing guidance in +//! the tracker issue: +//! +//! - a transient miss (container down mid-sweep, a network blip) shouldn't +//! flap a banner on the first failure → `threshold > 1` raises only after +//! that many **consecutive** misses; +//! - a won't-fix-itself failure (missing team, bad config, auth rejection) +//! should banner immediately → `threshold == 1`. +//! +//! The warning clears the moment the sweep next succeeds. A sweep owns one +//! `SweepHealth`, calls [`record_ok`](SweepHealth::record_ok) on success and +//! [`record_err`](SweepHealth::record_err) on failure, and the held +//! [`WarningGuard`] does the rest (drop = clear). + +use std::time::{Duration, Instant}; + +use crate::warnings::{WarningGuard, set_warning}; + +/// Per-sweep failure context handed to the [`record_err`](SweepHealth::record_err) +/// message builder so the banner text can include how long the sweep has been +/// failing and since when it last worked. +#[derive(Debug, Clone, Copy)] +pub struct SweepFailure { + /// Number of consecutive failures so far (≥ `threshold` when the banner + /// is raised). + pub consecutive: u32, + /// Time since the last recorded success, if the sweep has ever succeeded + /// in this process's lifetime. `None` = never succeeded yet. + pub since_last_ok: Option, +} + +/// A debounced link between one background sweep and the warning banner. +/// Not `Sync`-shared: each sweep task owns its own instance (the underlying +/// registry is the shared, thread-safe part). +pub struct SweepHealth { + kind: &'static str, + level: &'static str, + threshold: u32, + misses: u32, + last_ok: Option, + guard: Option, +} + +impl SweepHealth { + /// Track sweep `kind` at severity `level` (`"warn"` / `"crit"`), raising + /// the banner after `threshold` consecutive failures. `threshold` is + /// floored at 1 (0 would banner before any failure). + #[must_use] + pub fn new(kind: &'static str, level: &'static str, threshold: u32) -> Self { + Self { + kind, + level, + threshold: threshold.max(1), + misses: 0, + last_ok: None, + guard: None, + } + } + + /// Record a successful sweep: reset the consecutive-miss counter and drop + /// the guard (clearing any banner entry). + pub fn record_ok(&mut self) { + self.misses = 0; + self.last_ok = Some(Instant::now()); + // Dropping the guard removes the warning from the registry. + self.guard = None; + } + + /// Record a failed sweep. Bumps the consecutive-miss counter; once it + /// reaches `threshold`, raises (or refreshes) the banner warning built by + /// `message`. Below the threshold this is a cheap no-op — `message` is + /// only invoked when the banner is actually shown, so callers can format + /// lazily. + pub fn record_err(&mut self, message: impl FnOnce(SweepFailure) -> String) { + self.misses = self.misses.saturating_add(1); + if self.misses < self.threshold { + return; + } + let ctx = SweepFailure { + consecutive: self.misses, + since_last_ok: self.last_ok.map(|t| t.elapsed()), + }; + let msg = message(ctx); + match &self.guard { + Some(g) => g.update(self.level, msg), + None => self.guard = Some(set_warning(self.kind, self.level, msg)), + } + } + + /// Whether the banner warning is currently raised for this sweep. + #[must_use] + pub fn is_warning(&self) -> bool { + self.guard.is_some() + } +} + +/// Compact human-readable age (e.g. `"45s"`, `"12m"`, `"3h"`, `"2d"`) for a +/// banner "last succeeded N ago" suffix. Coarse on purpose — the banner wants +/// a glanceable magnitude, not precision. +#[must_use] +pub fn fmt_age(d: Duration) -> String { + let s = d.as_secs(); + if s < 60 { + format!("{s}s") + } else if s < 3600 { + format!("{}m", s / 60) + } else if s < 86_400 { + format!("{}h", s / 3600) + } else { + format!("{}d", s / 86_400) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::warnings::snapshot; + + fn banner(kind: &str) -> Option { + snapshot() + .into_iter() + .find(|w| w.kind == kind) + .map(|w| w.message) + } + + #[test] + fn below_threshold_does_not_banner() { + let mut h = SweepHealth::new("sh_below", "warn", 3); + h.record_err(|_| "boom".to_owned()); + h.record_err(|_| "boom".to_owned()); + assert!(!h.is_warning()); + assert!(banner("sh_below").is_none()); + } + + #[test] + fn raises_at_threshold_and_clears_on_ok() { + let mut h = SweepHealth::new("sh_raise", "warn", 2); + h.record_err(|c| format!("fail #{}", c.consecutive)); + assert!(banner("sh_raise").is_none(), "one miss < threshold"); + h.record_err(|c| format!("fail #{}", c.consecutive)); + assert_eq!(banner("sh_raise").as_deref(), Some("fail #2")); + assert!(h.is_warning()); + h.record_ok(); + assert!(!h.is_warning()); + assert!(banner("sh_raise").is_none(), "success clears the banner"); + } + + #[test] + fn threshold_one_banners_immediately() { + let mut h = SweepHealth::new("sh_immediate", "crit", 1); + h.record_err(|_| "gone".to_owned()); + assert_eq!(banner("sh_immediate").as_deref(), Some("gone")); + h.record_ok(); + assert!(banner("sh_immediate").is_none()); + } + + #[test] + fn success_resets_consecutive_counter() { + let mut h = SweepHealth::new("sh_reset", "warn", 2); + h.record_err(|_| "x".to_owned()); // miss 1 + h.record_ok(); // reset + h.record_err(|c| format!("c={}", c.consecutive)); // miss 1 again, < threshold + assert!(banner("sh_reset").is_none(), "counter reset after ok"); + h.record_err(|c| format!("c={}", c.consecutive)); // miss 2 → raise + assert_eq!(banner("sh_reset").as_deref(), Some("c=2")); + h.record_ok(); + } + + #[test] + fn refresh_updates_message_while_held() { + let mut h = SweepHealth::new("sh_refresh", "warn", 1); + h.record_err(|c| format!("miss {}", c.consecutive)); + assert_eq!(banner("sh_refresh").as_deref(), Some("miss 1")); + h.record_err(|c| format!("miss {}", c.consecutive)); + assert_eq!(banner("sh_refresh").as_deref(), Some("miss 2")); + h.record_ok(); + } + + #[test] + fn since_last_ok_is_some_after_a_success() { + let mut h = SweepHealth::new("sh_age", "warn", 1); + h.record_err(|c| { + assert!(c.since_last_ok.is_none(), "never succeeded yet"); + "first".to_owned() + }); + h.record_ok(); + h.record_err(|c| { + assert!(c.since_last_ok.is_some(), "has a last-ok timestamp now"); + "again".to_owned() + }); + h.record_ok(); + } + + #[test] + fn fmt_age_buckets() { + assert_eq!(fmt_age(Duration::from_secs(5)), "5s"); + assert_eq!(fmt_age(Duration::from_secs(90)), "1m"); + assert_eq!(fmt_age(Duration::from_hours(2)), "2h"); + assert_eq!(fmt_age(Duration::from_secs(200_000)), "2d"); + } +}