feat(#2289): debounced SweepHealth banner tracker, wire knowledge pull

hive-c0re/src/lib.rs

@@ -45,7 +45,7 @@ pub mod workers;
 // Root re-exports: keep every pre-grouping `crate::<module>` /
 // `hive_c0re::<module>` 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,
 };

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() => {

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;

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<Duration>,
+}
+
+/// 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<Instant>,
+    guard: Option<WarningGuard>,
+}
+
+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<String> {
+        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");
+    }
+}