feat(#2289): push-based server-warning registry with RAII guard

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};
+pub use stats::{container_stats, hive_stats, host_stats, warnings};
 pub use stores::{
     approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts,
 };

hive-c0re/src/main.rs

@@ -13,8 +13,8 @@ use hive_sh4re::{HostRequest, HostResponse};
 use hive_c0re::coordinator::{Coordinator, HiveEnv, ServeConfig};
 use hive_c0re::{
     agent_sockets, auto_update, broker, client, crash_watch, dashboard, dashboard_events, forge,
-    job_queue, knowledge, matrix, mcp_sockets, migrate, reminder_scheduler,
-    scheduled_prompts_worker, server, socket_server,
+    host_stats, job_queue, knowledge, matrix, mcp_sockets, migrate, reminder_scheduler,
+    scheduled_prompts_worker, server, socket_server, warnings,
 };
 
 #[derive(Parser)]
@@ -340,6 +340,27 @@ async fn cmd_serve(
             }
         }
     });
+    // Disk-pressure watch: raise a `disk_pressure` banner warning while the
+    // host nix store is over threshold and clear it when back under, via the
+    // push-based warnings registry. Replaces the old per-`/api/state`
+    // `statvfs`. ~60s cadence; first tick fires immediately. The held guard
+    // lives for the task's lifetime — dropping it (on shutdown) clears the
+    // banner.
+    let mut disk_shutdown = coord.shutdown_rx();
+    tokio::spawn(async move {
+        let mut disk_guard: Option<warnings::WarningGuard> = None;
+        let interval = std::time::Duration::from_mins(1);
+        loop {
+            host_stats::refresh_disk_warning(&mut disk_guard);
+            tokio::select! {
+                () = tokio::time::sleep(interval) => {}
+                _ = disk_shutdown.changed() => {
+                    tracing::info!("disk-pressure watch: shutdown signal received");
+                    break;
+                }
+            }
+        }
+    });
     // Matrix user sweep: same shape — ensure every container has
     // an account on the local matrix-tuwunel homeserver with an
     // access_token persisted to `<state>/matrix-token`. No-op when

hive-c0re/src/stats/host_stats.rs

@@ -8,10 +8,12 @@
 //! operator *before* an ENOSPC, not after.
 //!
 //! [`server_warnings`] is the public surface: it returns a flat list of
-//! [`ServerWarning`]s for `/api/state`. The dashboard renders whatever it
-//! returns, coloured by `level`, so adding a new system warning (memory
-//! pressure, a failed unit, …) is a backend-only change — no frontend
-//! edit. Keep producers cheap; this runs on every `/api/state` assembly.
+//! [`ServerWarning`]s for `/api/state`. It is now just a snapshot of the
+//! push-based [`crate::warnings`] registry — producers raise/clear their
+//! own warnings via an RAII guard rather than being re-probed on every
+//! render. The disk-pressure check is reworked onto that model here:
+//! [`refresh_disk_warning`] drives a held guard from a periodic task (see
+//! `spawn` in `main.rs`) instead of a `statvfs` per `/api/state`.
 
 use std::collections::HashMap;
 
@@ -37,36 +39,62 @@ pub struct ServerWarning {
 const DISK_WARN_PCT: f64 = 85.0;
 const DISK_CRIT_PCT: f64 = 95.0;
 
-/// Collect the current server-level warnings for the dashboard banner.
-/// Each producer pushes zero or more [`ServerWarning`]s; the frontend
-/// renders whatever this returns. Cheap to call on every `/api/state`
-/// assembly (currently a single `statvfs`).
+/// Snapshot the current server-level warnings for the dashboard banner —
+/// a cheap read of the push-based [`crate::warnings`] registry. Producers
+/// (the disk watch below, background converge sweeps, …) raise and clear
+/// their own entries via [`crate::warnings::set_warning`]; nothing is
+/// re-probed here.
 #[must_use]
 pub fn server_warnings() -> Vec<ServerWarning> {
-    let mut out = Vec::new();
-    if let Some(d) = nix_disk_usage()
-        && d.used_pct >= DISK_WARN_PCT
-    {
-        #[allow(
-            clippy::cast_precision_loss,
-            reason = "byte counts stay well under f64's 2^53 exact-integer range, so this GiB conversion loses no precision"
-        )]
-        let free_gib = d.free_bytes as f64 / (1024.0 * 1024.0 * 1024.0);
-        out.push(ServerWarning {
-            kind: "disk_pressure",
-            level: if d.used_pct >= DISK_CRIT_PCT {
-                "crit"
-            } else {
-                "warn"
-            },
-            message: format!(
-                "host nix store {:.0}% full ({free_gib:.1} GiB free) \
-                 — garbage-collect the store before it runs out of space",
-                d.used_pct
-            ),
-        });
+    crate::warnings::snapshot()
+}
+
+/// Pure disk-pressure decision: given usage, return the banner
+/// `(level, message)` or `None` when below the warn threshold. Split from
+/// the probe + guard plumbing so the threshold + formatting is unit-tested
+/// without a real `statvfs`.
+fn disk_pressure_warning(used_pct: f64, free_bytes: u64) -> Option<(&'static str, String)> {
+    if used_pct < DISK_WARN_PCT {
+        return None;
+    }
+    #[allow(
+        clippy::cast_precision_loss,
+        reason = "byte counts stay well under f64's 2^53 exact-integer range, so this GiB conversion loses no precision"
+    )]
+    let free_gib = free_bytes as f64 / (1024.0 * 1024.0 * 1024.0);
+    let level = if used_pct >= DISK_CRIT_PCT {
+        "crit"
+    } else {
+        "warn"
+    };
+    let message = format!(
+        "host nix store {used_pct:.0}% full ({free_gib:.1} GiB free) \
+         — garbage-collect the store before it runs out of space"
+    );
+    Some((level, message))
+}
+
+/// One tick of the disk-pressure watch: probe the store and reconcile the
+/// held warning `guard` — raise or refresh it while usage is over
+/// threshold, drop it (clearing the banner) once back under. Driven by the
+/// periodic disk-watch task spawned in `main.rs`, replacing the old
+/// per-render `statvfs` in [`server_warnings`].
+pub fn refresh_disk_warning(guard: &mut Option<crate::warnings::WarningGuard>) {
+    let decision = nix_disk_usage().and_then(|d| disk_pressure_warning(d.used_pct, d.free_bytes));
+    match decision {
+        Some((level, message)) => match guard {
+            Some(g) => g.update(level, message),
+            None => {
+                *guard = Some(crate::warnings::set_warning(
+                    "disk_pressure",
+                    level,
+                    message,
+                ));
+            }
+        },
+        // Back under threshold (or probe failed) — drop the guard to clear.
+        None => *guard = None,
     }
-    out
 }
 
 /// Agent-state warnings derived from the live container snapshot the
@@ -194,6 +222,25 @@ fn disk_usage(path: &str) -> Option<DiskUsage> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn disk_below_threshold_is_none() {
+        assert!(disk_pressure_warning(84.9, 100 << 30).is_none());
+    }
+
+    #[test]
+    fn disk_warn_band_is_amber() {
+        let (level, msg) = disk_pressure_warning(90.0, 20 << 30).expect("warn");
+        assert_eq!(level, "warn");
+        assert!(msg.contains("90% full"));
+        assert!(msg.contains("20.0 GiB free"));
+    }
+
+    #[test]
+    fn disk_crit_band_is_red() {
+        let (level, _) = disk_pressure_warning(95.0, 1 << 30).expect("crit");
+        assert_eq!(level, "crit");
+    }
+
     fn cv(name: &str, needs_login: bool) -> ContainerView {
         ContainerView {
             name: name.to_owned(),

hive-c0re/src/stats/mod.rs

@@ -6,3 +6,4 @@
 pub mod container_stats;
 pub mod hive_stats;
 pub mod host_stats;
+pub mod warnings;

hive-c0re/src/stats/warnings.rs

@@ -0,0 +1,258 @@
+//! Process-global server-warning registry backing the dashboard banner.
+//!
+//! The dashboard shows a top-of-page banner for hive-level degradation
+//! (disk pressure, a failed converge sweep, agents that can't log in, …).
+//! Historically each warning was *pulled*: [`host_stats::server_warnings`]
+//! recomputed the whole list on every `/api/state` by re-probing (a
+//! `statvfs`, a scan of the container snapshot). Background sweeps that
+//! failed had nowhere to record it, so they just `warn!`'d to the journal
+//! and vanished from anywhere an operator looks.
+//!
+//! This module replaces the pull model with a **push** one: a single
+//! `Mutex<BTreeMap<kind, …>>` any subsystem writes into. [`snapshot`] is a
+//! cheap read of that map. The default API is RAII: [`set_warning`] returns
+//! a [`WarningGuard`] whose `Drop` removes the key, so a producer holds the
+//! guard exactly as long as the condition is true and lets it drop to clear
+//! — no manual dismiss, no stale banner. A failing periodic sweep does
+//! `let _w = set_warning(...)` in its failure branch and drops it on the
+//! next success.
+//!
+//! Concurrency: all mutation is behind the one mutex. Each guard carries a
+//! unique id and only touches its own map entry (`Drop` / [`update`]
+//! no-op once a later `set_warning` for the same kind has superseded it),
+//! so a re-`set_warning` for the same kind is never clobbered when the
+//! stale guard later drops.
+//!
+//! [`host_stats::server_warnings`]: crate::host_stats::server_warnings
+//! [`update`]: WarningGuard::update
+
+use std::collections::BTreeMap;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Mutex, OnceLock};
+
+use crate::host_stats::ServerWarning;
+
+/// Monotonic guard-id source. Each [`set_warning`] mints a fresh id so a
+/// guard's `Drop` can tell "my entry" from "an entry a later guard replaced
+/// under the same kind".
+static NEXT_ID: AtomicU64 = AtomicU64::new(1);
+
+/// One registry slot: the warning payload plus the id of the guard that
+/// currently owns the `kind`.
+struct Slot {
+    level: &'static str,
+    message: String,
+    owner: u64,
+}
+
+/// The one process-global registry. `BTreeMap` (not `HashMap`) so
+/// [`snapshot`] yields a deterministic, kind-sorted banner order across
+/// renders.
+fn registry() -> &'static Mutex<BTreeMap<&'static str, Slot>> {
+    static REG: OnceLock<Mutex<BTreeMap<&'static str, Slot>>> = OnceLock::new();
+    REG.get_or_init(|| Mutex::new(BTreeMap::new()))
+}
+
+/// Lock the registry, recovering the guard even if a previous holder
+/// panicked mid-mutation — a poisoned lock must not wedge every future
+/// warning update.
+fn lock() -> std::sync::MutexGuard<'static, BTreeMap<&'static str, Slot>> {
+    registry()
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+/// Raise (or replace) the warning under `kind` and return a guard that
+/// clears it on drop. This is the default way to surface a hive-level
+/// degradation: hold the returned guard while the condition is true, and
+/// let it drop (or call `drop(guard)`) the moment it clears.
+///
+/// Re-calling `set_warning` for a `kind` that already has a live guard
+/// replaces the payload and takes ownership of the entry (fresh id): the
+/// superseded guard's later `Drop`/`update` become no-ops, so a producer
+/// can freely re-arm without racing its own previous guard.
+#[must_use = "dropping the guard immediately clears the warning; bind it for as long as the condition holds"]
+pub fn set_warning(
+    kind: &'static str,
+    level: &'static str,
+    message: impl Into<String>,
+) -> WarningGuard {
+    let id = NEXT_ID.fetch_add(1, Ordering::Relaxed);
+    lock().insert(
+        kind,
+        Slot {
+            level,
+            message: message.into(),
+            owner: id,
+        },
+    );
+    WarningGuard { kind, id }
+}
+
+/// Snapshot of the currently-active warnings, kind-sorted. Cheap read
+/// behind the registry mutex — safe to call on every `/api/state`.
+#[must_use]
+pub fn snapshot() -> Vec<ServerWarning> {
+    lock()
+        .iter()
+        .map(|(kind, slot)| ServerWarning {
+            kind,
+            level: slot.level,
+            message: slot.message.clone(),
+        })
+        .collect()
+}
+
+/// RAII handle for one active warning. Drop clears the `kind` from the
+/// banner. Obtained from [`set_warning`].
+#[must_use = "dropping the guard immediately clears the warning; bind it for as long as the condition holds"]
+pub struct WarningGuard {
+    kind: &'static str,
+    id: u64,
+}
+
+impl WarningGuard {
+    /// Refresh the held warning's `level` + `message` in place — e.g. a
+    /// disk-usage percentage ticking up, or a `warn` escalating to `crit`.
+    /// A no-op once a later [`set_warning`] for the same kind has superseded
+    /// this guard (its entry is owned by the newer id).
+    pub fn update(&self, level: &'static str, message: impl Into<String>) {
+        let mut reg = lock();
+        if let Some(slot) = reg.get_mut(self.kind)
+            && slot.owner == self.id
+        {
+            slot.level = level;
+            slot.message = message.into();
+        }
+    }
+}
+
+impl Drop for WarningGuard {
+    fn drop(&mut self) {
+        let mut reg = lock();
+        // Only remove if we still own the slot — a later `set_warning` for
+        // the same kind takes over the entry (fresh owner id), so this
+        // superseded guard's drop must not clobber the live one.
+        let ours = reg.get(self.kind).is_some_and(|slot| slot.owner == self.id);
+        if ours {
+            reg.remove(self.kind);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Find the snapshot entry for `kind`, if present. Tests filter by
+    /// their own unique kinds instead of asserting global emptiness, so
+    /// they stay correct under cargo's parallel test runner (the registry
+    /// is process-global and shared across tests).
+    fn find(kind: &str) -> Option<ServerWarning> {
+        snapshot().into_iter().find(|w| w.kind == kind)
+    }
+
+    #[test]
+    fn set_surfaces_in_snapshot_then_drop_clears() {
+        {
+            let _g = set_warning("t_setdrop", "warn", "boom");
+            let w = find("t_setdrop").expect("warning present while guard held");
+            assert_eq!(w.level, "warn");
+            assert_eq!(w.message, "boom");
+        }
+        assert!(find("t_setdrop").is_none(), "dropping the guard clears it");
+    }
+
+    #[test]
+    fn update_refreshes_level_and_message_in_place() {
+        let g = set_warning("t_update", "warn", "85% full");
+        g.update("crit", "96% full");
+        let w = find("t_update").expect("present");
+        assert_eq!(w.level, "crit");
+        assert_eq!(w.message, "96% full");
+        drop(g);
+        assert!(find("t_update").is_none());
+    }
+
+    #[test]
+    fn resetting_same_kind_survives_stale_guard_drop() {
+        let a = set_warning("t_reset", "warn", "first");
+        let b = set_warning("t_reset", "crit", "second");
+        // `b` took over the entry; dropping the superseded `a` must not
+        // remove `b`'s live warning.
+        drop(a);
+        let w = find("t_reset").expect("b's warning survives a's drop");
+        assert_eq!(w.level, "crit");
+        assert_eq!(w.message, "second");
+        // A stale guard's update is also a no-op — prove by dropping b and
+        // confirming the kind is gone (only the owner clears it).
+        drop(b);
+        assert!(find("t_reset").is_none());
+    }
+
+    #[test]
+    fn stale_guard_update_is_noop() {
+        let a = set_warning("t_staleupd", "warn", "old");
+        let b = set_warning("t_staleupd", "warn", "new");
+        a.update("crit", "should not apply");
+        let w = find("t_staleupd").expect("present");
+        assert_eq!(w.message, "new", "stale guard update ignored");
+        drop(b);
+        drop(a);
+        assert!(find("t_staleupd").is_none());
+    }
+
+    #[test]
+    fn distinct_kinds_coexist_and_snapshot_is_sorted() {
+        let _z = set_warning("t_zzz", "warn", "z");
+        let _a = set_warning("t_aaa", "warn", "a");
+        let kinds: Vec<&str> = snapshot()
+            .iter()
+            .map(|w| w.kind)
+            .filter(|k| k.starts_with("t_zzz") || k.starts_with("t_aaa"))
+            .collect();
+        assert_eq!(kinds, ["t_aaa", "t_zzz"], "kind-sorted order");
+    }
+
+    #[test]
+    fn concurrent_unique_kinds_never_race_and_clear_out() {
+        // Each thread owns a distinct kind and hammers set/update/drop in a
+        // loop. The shared resource under test is the one mutex; a data
+        // race would trip under the thread sanitizer / show a wrong final
+        // state. After every guard has dropped, none of the kinds remain.
+        const KINDS: [&str; 6] = ["cc0", "cc1", "cc2", "cc3", "cc4", "cc5"];
+        std::thread::scope(|s| {
+            for kind in KINDS {
+                s.spawn(move || {
+                    for i in 0..500 {
+                        let g = set_warning(kind, "warn", format!("iter {i}"));
+                        g.update("crit", format!("iter {i} upd"));
+                        drop(g);
+                    }
+                });
+            }
+        });
+        for kind in KINDS {
+            assert!(find(kind).is_none(), "{kind} cleared after all drops");
+        }
+    }
+
+    #[test]
+    fn concurrent_same_kind_resolves_to_absent_after_all_drop() {
+        // Many threads contend on ONE kind, each holding its own guard
+        // briefly. The id-guard invariant means whichever set ran last owns
+        // the entry, and once every guard has dropped the kind is absent —
+        // no stale guard leaves a phantom warning behind.
+        std::thread::scope(|s| {
+            for t in 0..8 {
+                s.spawn(move || {
+                    for i in 0..300 {
+                        let g = set_warning("cc_shared", "warn", format!("{t}:{i}"));
+                        drop(g);
+                    }
+                });
+            }
+        });
+        assert!(find("cc_shared").is_none());
+    }
+}