From c1523e8acf9726412ee0790c8413aca3b27387dc Mon Sep 17 00:00:00 2001 From: Huskies Agent Date: Fri, 17 Jul 2026 19:18:37 +0000 Subject: [PATCH] huskies: merge 1200 story Low-disk warning: the fleet tells the operator before the disk takes it down --- server/src/agent_mode/mod.rs | 19 + server/src/chat/transport/matrix/health.rs | 21 ++ server/src/config/mod.rs | 61 ++++ server/src/gateway/tests.rs | 170 +++++++++ server/src/gateway_relay.rs | 52 +++ server/src/io/watcher/events.rs | 24 ++ server/src/service/disk_watch/io.rs | 144 ++++++++ server/src/service/disk_watch/mod.rs | 330 ++++++++++++++++++ server/src/service/events/buffer.rs | 26 ++ server/src/service/gateway/io.rs | 89 +++++ server/src/service/gateway/polling.rs | 36 +- server/src/service/mod.rs | 2 + server/src/service/notifications/events.rs | 11 + server/src/service/notifications/format.rs | 97 +++++ .../src/service/notifications/io/listener.rs | 56 ++- server/src/service/notifications/mod.rs | 3 +- server/src/service/status/format.rs | 16 + server/src/service/status/mod.rs | 20 ++ server/src/service/ws/message/convert.rs | 3 + server/src/startup/tick_loop.rs | 25 ++ server/src/worktree/cleanup.rs | 1 + server/src/worktree/create.rs | 1 + server/src/worktree/remove.rs | 1 + server/src/worktree/sweep.rs | 1 + 24 files changed, 1206 insertions(+), 3 deletions(-) create mode 100644 server/src/service/disk_watch/io.rs create mode 100644 server/src/service/disk_watch/mod.rs diff --git a/server/src/agent_mode/mod.rs b/server/src/agent_mode/mod.rs index 8bef0a18..b4ad8981 100644 --- a/server/src/agent_mode/mod.rs +++ b/server/src/agent_mode/mod.rs @@ -213,6 +213,15 @@ pub async fn run( // Track which stories we've claimed so we can detect conflicts. let mut our_claims: HashMap = HashMap::new(); + // Low-disk-space watchdog (story 1200 AC1): tracks rate-limit/recovery + // state across loop iterations. Thresholds come from the config loaded + // at startup; host_id identifies this sled in chat messages and the + // gateway dedupe key. + let mut disk_watch_state = crate::service::disk_watch::DiskWatchState::default(); + let disk_watch_host_id = + crdt_state::our_node_id().unwrap_or_else(|| "unknown-host".to_string()); + let disk_watch_status = agents.status_broadcaster(); + // Main loop: heartbeat, scan, claim, detect conflicts. let mut interval = tokio::time::interval(std::time::Duration::from_secs(SCAN_INTERVAL_SECS)); loop { @@ -221,6 +230,16 @@ pub async fn run( // Write heartbeat. write_heartbeat(&rendezvous_url, port); + // Low-disk-space check (story 1200 AC1): every tick period. + crate::service::disk_watch::io::check_and_notify( + &project_root, + &config.disk_watch, + &mut disk_watch_state, + &watcher_tx, + &disk_watch_status, + &disk_watch_host_id, + ); + // Scan CRDT for claimable work. scan_and_claim(&agents, &project_root, &mut our_claims).await; diff --git a/server/src/chat/transport/matrix/health.rs b/server/src/chat/transport/matrix/health.rs index 49f4b773..10193cc2 100644 --- a/server/src/chat/transport/matrix/health.rs +++ b/server/src/chat/transport/matrix/health.rs @@ -478,6 +478,7 @@ pub async fn run_health_check(ctx: &BotContext) -> String { lines.push(sync_line); lines.push(creds_line); lines.push(hash_line); + lines.push(check_disk_space()); let lines = truncate_lines(lines); lines @@ -487,6 +488,26 @@ pub async fn run_health_check(ctx: &BotContext) -> String { .join("\n") } +/// Report current free disk space on `/workspace` (story 1200 AC5). +fn check_disk_space() -> HealthLine { + match crate::service::disk_watch::io::free_space_bytes(std::path::Path::new("/workspace")) { + Ok(free_bytes) => { + let free_gb = free_bytes as f64 / 1_000_000_000.0; + HealthLine { + subsystem: "disk".to_string(), + status: Status::Pass, + detail: Some(format!("{free_gb:.1}GB free")), + hint: None, + } + } + Err(_) => HealthLine::warn( + "disk", + "unable to read free space", + "check /workspace mount", + ), + } +} + // ── Utilities ──────────────────────────────────────────────────────────────── /// Shorten a long error string to the first 60 characters for compact display. diff --git a/server/src/config/mod.rs b/server/src/config/mod.rs index 6fbf209b..5f58b2e5 100644 --- a/server/src/config/mod.rs +++ b/server/src/config/mod.rs @@ -19,6 +19,9 @@ pub struct ProjectConfig { pub agent: Vec, #[serde(default)] pub watcher: WatcherConfig, + /// Configuration for the low-disk-space watchdog (story 1200). + #[serde(default)] + pub disk_watch: DiskWatchConfig, /// Project-wide default QA mode: "server", "agent", or "human". /// Per-story `qa` front matter overrides this. Default: "server". #[serde(default = "default_qa")] @@ -245,6 +248,58 @@ fn default_max_mesh_peers() -> usize { 3 } +/// Configuration for the low-disk-space watchdog's free-space thresholds. +/// +/// Sleds check free space on the `/workspace` filesystem each tick and +/// compare it against `warn_gb` / `critical_gb`, rate-limiting repeat +/// notifications per level to `rate_limit_secs` and sending a single +/// recovery notice once free space climbs back above +/// `warn_gb * (1 + recovery_margin_pct / 100)`. +#[derive(Debug, Clone, Deserialize, PartialEq)] +pub struct DiskWatchConfig { + /// Free space (in GB) below which a "warn" notification fires. Default: 50. + #[serde(default = "default_disk_warn_gb")] + pub warn_gb: u64, + /// Free space (in GB) below which a "critical" notification fires. Default: 20. + #[serde(default = "default_disk_critical_gb")] + pub critical_gb: u64, + /// Minimum time (in seconds) between repeat notifications at the same + /// level while the condition persists. Default: 21600 (6 hours). + #[serde(default = "default_disk_rate_limit_secs")] + pub rate_limit_secs: u64, + /// Percentage above `warn_gb` free space must climb before a single + /// recovery notice is sent. Default: 10. + #[serde(default = "default_disk_recovery_margin_pct")] + pub recovery_margin_pct: u64, +} + +impl Default for DiskWatchConfig { + fn default() -> Self { + Self { + warn_gb: default_disk_warn_gb(), + critical_gb: default_disk_critical_gb(), + rate_limit_secs: default_disk_rate_limit_secs(), + recovery_margin_pct: default_disk_recovery_margin_pct(), + } + } +} + +fn default_disk_warn_gb() -> u64 { + 50 +} + +fn default_disk_critical_gb() -> u64 { + 20 +} + +fn default_disk_rate_limit_secs() -> u64 { + 6 * 60 * 60 +} + +fn default_disk_recovery_margin_pct() -> u64 { + 10 +} + /// Configuration for a project component (name, path, setup/teardown commands). /// /// `deny_unknown_fields` turns a top-level setting that lands inside a @@ -347,6 +402,8 @@ struct LegacyProjectConfig { agent: Option, #[serde(default)] watcher: WatcherConfig, + #[serde(default)] + disk_watch: DiskWatchConfig, #[serde(default = "default_qa")] default_qa: String, #[serde(default)] @@ -385,6 +442,7 @@ impl Default for ProjectConfig { runtime: None, }], watcher: WatcherConfig::default(), + disk_watch: DiskWatchConfig::default(), default_qa: default_qa(), default_coder_model: None, max_coders: None, @@ -475,6 +533,7 @@ impl ProjectConfig { component: legacy.component, agent: vec![agent], watcher: legacy.watcher, + disk_watch: legacy.disk_watch, default_qa: legacy.default_qa, default_coder_model: legacy.default_coder_model, max_coders: legacy.max_coders, @@ -516,6 +575,7 @@ impl ProjectConfig { component: legacy.component, agent: vec![agent], watcher: legacy.watcher, + disk_watch: legacy.disk_watch, default_qa: legacy.default_qa, default_coder_model: legacy.default_coder_model, max_coders: legacy.max_coders, @@ -545,6 +605,7 @@ impl ProjectConfig { component: legacy.component, agent: Vec::new(), watcher: legacy.watcher, + disk_watch: legacy.disk_watch, default_qa: legacy.default_qa, default_coder_model: legacy.default_coder_model, max_coders: legacy.max_coders, diff --git a/server/src/gateway/tests.rs b/server/src/gateway/tests.rs index 30732a45..6fb5b96d 100644 --- a/server/src/gateway/tests.rs +++ b/server/src/gateway/tests.rs @@ -513,6 +513,176 @@ async fn broadcaster_forwarder_resubscribes_on_lag() { ); } +/// Story 1200 AC4: identical disk-space warnings arriving from different +/// sleds within the rate window must collapse into a single forwarded chat +/// message. +#[tokio::test] +async fn broadcaster_forwarder_dedupes_identical_disk_warnings_from_different_sleds() { + use crate::chat::{ChatTransport, MessageId}; + use crate::service::events::StoredEvent; + use async_trait::async_trait; + + type CallLog = Arc>>; + + struct MockTransport { + calls: CallLog, + } + + #[async_trait] + impl ChatTransport for MockTransport { + async fn send_message( + &self, + room_id: &str, + plain: &str, + _html: &str, + ) -> Result { + self.calls + .lock() + .unwrap() + .push((room_id.to_string(), plain.to_string())); + Ok("id".to_string()) + } + + async fn edit_message( + &self, + _room_id: &str, + _id: &str, + _plain: &str, + _html: &str, + ) -> Result<(), String> { + Ok(()) + } + + async fn send_typing(&self, _room_id: &str, _typing: bool) -> Result<(), String> { + Ok(()) + } + } + + let calls: CallLog = Arc::new(std::sync::Mutex::new(Vec::new())); + let transport = Arc::new(MockTransport { + calls: Arc::clone(&calls), + }); + + let (tx, rx) = + tokio::sync::broadcast::channel::(16); + gateway::spawn_gateway_broadcaster_forwarder( + transport as Arc, + vec!["!room:example.org".to_string()], + rx, + ); + + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + + let warning = |host_id: &str| crate::service::gateway::GatewayStatusEvent { + project: host_id.to_string(), + event: StoredEvent::DiskSpaceWarning { + level: "warn".to_string(), + free_bytes: 45_000_000_000, + target_bytes: 10_000_000_000, + worktrees_bytes: 5_000_000_000, + host_id: host_id.to_string(), + timestamp_ms: 100, + }, + }; + + // Two different sleds both observe the same "warn" level within the + // dedupe window — only the first should be forwarded. + tx.send(warning("sled-a")).unwrap(); + tx.send(warning("sled-b")).unwrap(); + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let messages = calls.lock().unwrap(); + assert_eq!( + messages.len(), + 1, + "Expected identical disk warnings from different sleds to dedupe to one message" + ); +} + +/// Non-disk-space events (e.g. stage transitions) must never be deduped, even +/// when several arrive back-to-back — only disk-space warnings/recoveries +/// share a dedupe key (story 1200 AC4). +#[tokio::test] +async fn broadcaster_forwarder_does_not_dedupe_non_disk_events() { + use crate::chat::{ChatTransport, MessageId}; + use crate::service::events::StoredEvent; + use async_trait::async_trait; + + type CallLog = Arc>>; + + struct MockTransport { + calls: CallLog, + } + + #[async_trait] + impl ChatTransport for MockTransport { + async fn send_message( + &self, + room_id: &str, + plain: &str, + _html: &str, + ) -> Result { + self.calls + .lock() + .unwrap() + .push((room_id.to_string(), plain.to_string())); + Ok("id".to_string()) + } + + async fn edit_message( + &self, + _room_id: &str, + _id: &str, + _plain: &str, + _html: &str, + ) -> Result<(), String> { + Ok(()) + } + + async fn send_typing(&self, _room_id: &str, _typing: bool) -> Result<(), String> { + Ok(()) + } + } + + let calls: CallLog = Arc::new(std::sync::Mutex::new(Vec::new())); + let transport = Arc::new(MockTransport { + calls: Arc::clone(&calls), + }); + + let (tx, rx) = + tokio::sync::broadcast::channel::(16); + gateway::spawn_gateway_broadcaster_forwarder( + transport as Arc, + vec!["!room:example.org".to_string()], + rx, + ); + + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + + let transition = |n: u64| crate::service::gateway::GatewayStatusEvent { + project: "p".to_string(), + event: StoredEvent::StageTransition { + story_id: format!("{n}_story"), + story_name: String::new(), + from_stage: "2_current".to_string(), + to_stage: "3_qa".to_string(), + timestamp_ms: n, + }, + }; + tx.send(transition(1)).unwrap(); + tx.send(transition(2)).unwrap(); + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + let messages = calls.lock().unwrap(); + assert_eq!( + messages.len(), + 2, + "Non-disk events must not be deduped against each other" + ); +} + // ── BotConfig tests ───────────────────────────────────────────────── #[test] diff --git a/server/src/gateway_relay.rs b/server/src/gateway_relay.rs index d0c109be..3e2e0dca 100644 --- a/server/src/gateway_relay.rs +++ b/server/src/gateway_relay.rs @@ -168,6 +168,28 @@ fn status_to_stored(event: StatusEvent) -> Option { reason, timestamp_ms: now_ms, }), + StatusEvent::DiskSpaceWarning { + level, + free_bytes, + target_bytes, + worktrees_bytes, + host_id, + } => Some(StoredEvent::DiskSpaceWarning { + level, + free_bytes, + target_bytes, + worktrees_bytes, + host_id, + timestamp_ms: now_ms, + }), + StatusEvent::DiskSpaceRecovered { + free_bytes, + host_id, + } => Some(StoredEvent::DiskSpaceRecovered { + free_bytes, + host_id, + timestamp_ms: now_ms, + }), // Rate-limit events have no StoredEvent equivalent — skip them. StatusEvent::RateLimitWarning { .. } | StatusEvent::RateLimitHardBlock { .. } => None, } @@ -252,6 +274,36 @@ mod tests { assert!(status_to_stored(ev).is_none()); } + #[test] + fn status_to_stored_disk_space_warning() { + let ev = StatusEvent::DiskSpaceWarning { + level: "warn".into(), + free_bytes: 45_000_000_000, + target_bytes: 10_000_000_000, + worktrees_bytes: 5_000_000_000, + host_id: "sled-a".into(), + }; + let stored = status_to_stored(ev).unwrap(); + assert!(matches!( + stored, + StoredEvent::DiskSpaceWarning { ref level, ref host_id, .. } + if level == "warn" && host_id == "sled-a" + )); + } + + #[test] + fn status_to_stored_disk_space_recovered() { + let ev = StatusEvent::DiskSpaceRecovered { + free_bytes: 60_000_000_000, + host_id: "sled-a".into(), + }; + let stored = status_to_stored(ev).unwrap(); + assert!(matches!( + stored, + StoredEvent::DiskSpaceRecovered { ref host_id, .. } if host_id == "sled-a" + )); + } + #[test] fn status_to_stored_rate_limit_hard_block_is_none() { let ev = StatusEvent::RateLimitHardBlock { diff --git a/server/src/io/watcher/events.rs b/server/src/io/watcher/events.rs index 3d526a2d..574ecfd8 100644 --- a/server/src/io/watcher/events.rs +++ b/server/src/io/watcher/events.rs @@ -111,4 +111,28 @@ pub enum WatcherEvent { /// Total auto-retry budget shared with the auto-block threshold. budget: u32, }, + /// Free disk space on the `/workspace` filesystem crossed a warn/critical + /// threshold (story 1200). + /// Triggers a rate-limited warning notification to configured chat rooms. + DiskSpaceWarning { + /// Severity level: `"warn"` or `"critical"`. + level: String, + /// Free space in bytes at the time of the check. + free_bytes: u64, + /// Size of the `target/` directory in bytes. + target_bytes: u64, + /// Size of the `.huskies/worktrees/` directory in bytes. + worktrees_bytes: u64, + /// Identifier of the sled that observed the reading. + host_id: String, + }, + /// Free disk space recovered above the warn threshold plus its recovery + /// margin, after a warn/critical warning had been sent (story 1200). + /// Triggers a single recovery notification to configured chat rooms. + DiskSpaceRecovered { + /// Free space in bytes at the time of recovery. + free_bytes: u64, + /// Identifier of the sled that observed the recovery. + host_id: String, + }, } diff --git a/server/src/service/disk_watch/io.rs b/server/src/service/disk_watch/io.rs new file mode 100644 index 00000000..10fbee47 --- /dev/null +++ b/server/src/service/disk_watch/io.rs @@ -0,0 +1,144 @@ +//! Side effects for the disk-space watchdog: statvfs free-space reads, +//! directory-size walks, and the periodic check-and-notify entry point. + +use std::ffi::CString; +use std::os::unix::ffi::OsStrExt as _; +use std::path::Path; +use std::time::Instant; + +use tokio::sync::broadcast; + +use crate::config::DiskWatchConfig; +use crate::io::watcher::WatcherEvent; +use crate::service::status::{StatusBroadcaster, StatusEvent}; + +use super::{DiskAction, DiskWatchState, decide_action}; + +/// Read free bytes available on the filesystem containing `path`. +pub fn free_space_bytes(path: &Path) -> std::io::Result { + let c_path = CString::new(path.as_os_str().as_bytes()) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidInput, e))?; + // SAFETY: `c_path` is a valid NUL-terminated string and `stat` is a + // zero-initialized `libc::statvfs` passed by mutable reference, matching + // the `statvfs(2)` contract. + let mut stat: libc::statvfs = unsafe { std::mem::zeroed() }; + let ret = unsafe { libc::statvfs(c_path.as_ptr(), &mut stat) }; + if ret != 0 { + return Err(std::io::Error::last_os_error()); + } + Ok(stat.f_bavail as u64 * stat.f_frsize as u64) +} + +/// Recursively sum file sizes under `path`. Returns 0 if `path` does not exist. +pub fn dir_size_bytes(path: &Path) -> u64 { + walkdir::WalkDir::new(path) + .into_iter() + .filter_map(Result::ok) + .filter(|e| e.file_type().is_file()) + .filter_map(|e| e.metadata().ok()) + .map(|m| m.len()) + .sum() +} + +/// Run one disk-space check against `workspace_root`, updating `state` and +/// broadcasting a [`WatcherEvent`] (local chat notifications) and a +/// [`StatusEvent`] (gateway relay, for cross-sled dedupe — story 1200 AC4) +/// when a warning or recovery notification is due. +/// +/// `host_id` identifies the sled that observed the reading (surfaced in the +/// gateway dedupe key and chat message). +pub fn check_and_notify( + workspace_root: &Path, + config: &DiskWatchConfig, + state: &mut DiskWatchState, + watcher_tx: &broadcast::Sender, + status: &StatusBroadcaster, + host_id: &str, +) { + let Ok(free_bytes) = free_space_bytes(workspace_root) else { + return; + }; + let now = Instant::now(); + match decide_action(free_bytes, config, state, now) { + DiskAction::None => {} + DiskAction::Warning(level) => { + let target_bytes = dir_size_bytes(&workspace_root.join("target")); + let worktrees_bytes = dir_size_bytes(&workspace_root.join(".huskies/worktrees")); + let level_str = level.as_str().to_string(); + let _ = watcher_tx.send(WatcherEvent::DiskSpaceWarning { + level: level_str.clone(), + free_bytes, + target_bytes, + worktrees_bytes, + host_id: host_id.to_string(), + }); + status.publish(StatusEvent::DiskSpaceWarning { + level: level_str, + free_bytes, + target_bytes, + worktrees_bytes, + host_id: host_id.to_string(), + }); + } + DiskAction::Recovery => { + let _ = watcher_tx.send(WatcherEvent::DiskSpaceRecovered { + free_bytes, + host_id: host_id.to_string(), + }); + status.publish(StatusEvent::DiskSpaceRecovered { + free_bytes, + host_id: host_id.to_string(), + }); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn free_space_bytes_reads_a_real_path() { + // The current directory always exists in test runs; just assert the + // call succeeds and returns a plausible non-zero value. + let free = free_space_bytes(Path::new(".")).expect("statvfs should succeed on cwd"); + assert!(free > 0); + } + + #[test] + fn dir_size_bytes_sums_files_in_a_temp_dir() { + let tmp = tempfile::tempdir().unwrap(); + std::fs::write(tmp.path().join("a.txt"), b"hello").unwrap(); + std::fs::write(tmp.path().join("b.txt"), b"world!").unwrap(); + assert_eq!(dir_size_bytes(tmp.path()), 11); + } + + #[test] + fn dir_size_bytes_returns_zero_for_missing_path() { + assert_eq!(dir_size_bytes(Path::new("/no/such/path/1200")), 0); + } + + #[tokio::test] + async fn check_and_notify_sends_warning_event_on_low_space() { + let tmp = tempfile::tempdir().unwrap(); + let (tx, mut rx) = broadcast::channel::(4); + let status = StatusBroadcaster::new(); + let mut status_sub = status.subscribe(); + let config = DiskWatchConfig { + warn_gb: u64::MAX / 1_000_000_000, // force Warn on any real filesystem + critical_gb: 0, + rate_limit_secs: 21_600, + recovery_margin_pct: 10, + }; + let mut state = DiskWatchState::default(); + check_and_notify(tmp.path(), &config, &mut state, &tx, &status, "sled-a"); + let event = rx.try_recv().expect("should broadcast a WatcherEvent"); + assert!(matches!(event, WatcherEvent::DiskSpaceWarning { .. })); + let status_event = + tokio::time::timeout(std::time::Duration::from_millis(100), status_sub.recv()) + .await + .expect("should not time out") + .expect("should publish a StatusEvent"); + assert!(matches!(status_event, StatusEvent::DiskSpaceWarning { .. })); + } +} diff --git a/server/src/service/disk_watch/mod.rs b/server/src/service/disk_watch/mod.rs new file mode 100644 index 00000000..f9c7fdf4 --- /dev/null +++ b/server/src/service/disk_watch/mod.rs @@ -0,0 +1,330 @@ +//! Low-disk-space watchdog (story 1200) — checks free space on the +//! `/workspace` filesystem each tick and decides whether to emit a +//! rate-limited warn/critical warning or a one-shot recovery notice. +//! +//! Follows service-module conventions: this file holds pure classification +//! and rate-limiting logic (no I/O, no `Instant::now()` calls — callers +//! supply `now`); `io.rs` is the only place performing side effects +//! (statvfs reads, directory-size walks, broadcasting `WatcherEvent`s). + +/// Side effects: free-space reads, directory-size walks, and the periodic +/// check-and-notify entry point. +pub mod io; + +use std::time::{Duration, Instant}; + +use crate::config::DiskWatchConfig; + +/// Bytes per gigabyte, using the same decimal (GB, not GiB) convention as the +/// `warn_gb` / `critical_gb` config fields. +const BYTES_PER_GB: u64 = 1_000_000_000; + +/// Disk-space severity level relative to configured thresholds. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DiskLevel { + /// Free space is at or above the warn threshold. + Ok, + /// Free space is below `warn_gb` but at/above `critical_gb`. + Warn, + /// Free space is below `critical_gb`. + Critical, +} + +impl DiskLevel { + /// Wire/display name used in notifications and gateway dedupe keys. + pub fn as_str(self) -> &'static str { + match self { + DiskLevel::Ok => "ok", + DiskLevel::Warn => "warn", + DiskLevel::Critical => "critical", + } + } +} + +/// Classify `free_bytes` against the configured warn/critical thresholds. +pub fn classify_level(free_bytes: u64, config: &DiskWatchConfig) -> DiskLevel { + if free_bytes < config.critical_gb * BYTES_PER_GB { + DiskLevel::Critical + } else if free_bytes < config.warn_gb * BYTES_PER_GB { + DiskLevel::Warn + } else { + DiskLevel::Ok + } +} + +/// Returns `true` if free space has recovered enough to send a recovery +/// notice: at or above `warn_gb * (1 + recovery_margin_pct / 100)`. +pub fn is_recovered(free_bytes: u64, config: &DiskWatchConfig) -> bool { + let recovery_threshold = + config.warn_gb * BYTES_PER_GB * (100 + config.recovery_margin_pct) / 100; + free_bytes >= recovery_threshold +} + +/// Returns `true` if a repeat notification for the same level should be +/// sent, given when it was last sent (`None` if never) and the configured +/// rate limit. +pub fn should_send(last_sent: Option, now: Instant, rate_limit_secs: u64) -> bool { + match last_sent { + None => true, + Some(last) => now.duration_since(last) >= Duration::from_secs(rate_limit_secs), + } +} + +/// Per-sled in-memory state tracked across tick iterations: the last level +/// observed, and when a notification was last sent for each level. +#[derive(Debug, Default)] +pub struct DiskWatchState { + /// The disk level observed on the previous check, `None` before the first check. + pub last_level: Option, + /// When a "warn" notification was last sent. + pub warn_last_sent: Option, + /// When a "critical" notification was last sent. + pub critical_last_sent: Option, +} + +/// The next action to take based on a fresh disk-level observation. +#[derive(Debug, PartialEq)] +pub enum DiskAction { + /// No notification needed. + None, + /// Send a warn/critical warning notification. + Warning(DiskLevel), + /// Send a single recovery notice. + Recovery, +} + +/// Decide what action to take given a fresh free-space reading, updating +/// `state` in place. Pure aside from taking `now` as an argument (no direct +/// I/O or wall-clock reads). +pub fn decide_action( + free_bytes: u64, + config: &DiskWatchConfig, + state: &mut DiskWatchState, + now: Instant, +) -> DiskAction { + let level = classify_level(free_bytes, config); + let was_persisting = matches!( + state.last_level, + Some(DiskLevel::Warn) | Some(DiskLevel::Critical) + ); + state.last_level = Some(level); + + match level { + DiskLevel::Ok => { + if was_persisting && is_recovered(free_bytes, config) { + state.warn_last_sent = None; + state.critical_last_sent = None; + DiskAction::Recovery + } else { + DiskAction::None + } + } + DiskLevel::Warn => { + if should_send(state.warn_last_sent, now, config.rate_limit_secs) { + state.warn_last_sent = Some(now); + DiskAction::Warning(DiskLevel::Warn) + } else { + DiskAction::None + } + } + DiskLevel::Critical => { + if should_send(state.critical_last_sent, now, config.rate_limit_secs) { + state.critical_last_sent = Some(now); + DiskAction::Warning(DiskLevel::Critical) + } else { + DiskAction::None + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn config() -> DiskWatchConfig { + DiskWatchConfig { + warn_gb: 50, + critical_gb: 20, + rate_limit_secs: 6 * 60 * 60, + recovery_margin_pct: 10, + } + } + + const GB: u64 = 1_000_000_000; + + // ── classify_level (threshold crossing) ─────────────────────────────── + + #[test] + fn classify_above_warn_is_ok() { + assert_eq!(classify_level(60 * GB, &config()), DiskLevel::Ok); + } + + #[test] + fn classify_at_warn_boundary_is_warn() { + // Below warn_gb (50) counts as Warn; exactly at warn_gb is still Ok + // (the threshold is "below", not "at or below"). + assert_eq!(classify_level(50 * GB, &config()), DiskLevel::Ok); + assert_eq!(classify_level(49 * GB, &config()), DiskLevel::Warn); + } + + #[test] + fn classify_between_warn_and_critical_is_warn() { + assert_eq!(classify_level(30 * GB, &config()), DiskLevel::Warn); + } + + #[test] + fn classify_at_critical_boundary() { + assert_eq!(classify_level(20 * GB, &config()), DiskLevel::Warn); + assert_eq!(classify_level(19 * GB, &config()), DiskLevel::Critical); + } + + #[test] + fn classify_below_critical_is_critical() { + assert_eq!(classify_level(5 * GB, &config()), DiskLevel::Critical); + } + + // ── is_recovered ─────────────────────────────────────────────────────── + + #[test] + fn is_recovered_requires_margin_above_warn() { + let cfg = config(); + // warn_gb=50, margin=10% => recovery threshold = 55 GB. + assert!(!is_recovered(54 * GB, &cfg)); + assert!(is_recovered(55 * GB, &cfg)); + assert!(is_recovered(60 * GB, &cfg)); + } + + // ── should_send (rate limiting) ──────────────────────────────────────── + + #[test] + fn should_send_when_never_sent() { + assert!(should_send(None, Instant::now(), 21_600)); + } + + #[test] + fn should_not_send_within_rate_limit_window() { + let now = Instant::now(); + let last = now - Duration::from_secs(100); + assert!(!should_send(Some(last), now, 21_600)); + } + + #[test] + fn should_send_after_rate_limit_window_expires() { + let now = Instant::now(); + let last = now - Duration::from_secs(21_601); + assert!(should_send(Some(last), now, 21_600)); + } + + // ── decide_action (threshold crossing + rate limiting + recovery) ────── + + #[test] + fn decide_action_first_warn_reading_sends_warning() { + let cfg = config(); + let mut state = DiskWatchState::default(); + let action = decide_action(30 * GB, &cfg, &mut state, Instant::now()); + assert_eq!(action, DiskAction::Warning(DiskLevel::Warn)); + } + + #[test] + fn decide_action_repeat_warn_within_window_is_suppressed() { + let cfg = config(); + let mut state = DiskWatchState::default(); + let now = Instant::now(); + assert_eq!( + decide_action(30 * GB, &cfg, &mut state, now), + DiskAction::Warning(DiskLevel::Warn) + ); + // Second check 1 minute later — same level, still within 6h window. + let later = now + Duration::from_secs(60); + assert_eq!( + decide_action(30 * GB, &cfg, &mut state, later), + DiskAction::None + ); + } + + #[test] + fn decide_action_repeat_warn_after_window_sends_again() { + let cfg = config(); + let mut state = DiskWatchState::default(); + let now = Instant::now(); + decide_action(30 * GB, &cfg, &mut state, now); + let later = now + Duration::from_secs(6 * 60 * 60 + 1); + assert_eq!( + decide_action(30 * GB, &cfg, &mut state, later), + DiskAction::Warning(DiskLevel::Warn) + ); + } + + #[test] + fn decide_action_warn_and_critical_have_independent_rate_limits() { + let cfg = config(); + let mut state = DiskWatchState::default(); + let now = Instant::now(); + assert_eq!( + decide_action(30 * GB, &cfg, &mut state, now), + DiskAction::Warning(DiskLevel::Warn) + ); + // Dropping straight to critical moments later must still notify — + // it has its own rate-limit bucket, separate from warn's. + let later = now + Duration::from_secs(5); + assert_eq!( + decide_action(10 * GB, &cfg, &mut state, later), + DiskAction::Warning(DiskLevel::Critical) + ); + } + + #[test] + fn decide_action_recovery_after_persisting_warn() { + let cfg = config(); + let mut state = DiskWatchState::default(); + let now = Instant::now(); + decide_action(30 * GB, &cfg, &mut state, now); // Warn + // Free space climbs back above the 55 GB recovery threshold. + assert_eq!( + decide_action(60 * GB, &cfg, &mut state, now), + DiskAction::Recovery + ); + } + + #[test] + fn decide_action_no_recovery_notice_without_prior_warning() { + let cfg = config(); + let mut state = DiskWatchState::default(); + // First-ever reading is healthy — must not fire a spurious recovery notice. + assert_eq!( + decide_action(60 * GB, &cfg, &mut state, Instant::now()), + DiskAction::None + ); + } + + #[test] + fn decide_action_recovery_is_sent_only_once() { + let cfg = config(); + let mut state = DiskWatchState::default(); + let now = Instant::now(); + decide_action(30 * GB, &cfg, &mut state, now); // Warn + assert_eq!( + decide_action(60 * GB, &cfg, &mut state, now), + DiskAction::Recovery + ); + // Still healthy on the next tick — no repeat recovery notice. + assert_eq!( + decide_action(60 * GB, &cfg, &mut state, now), + DiskAction::None + ); + } + + #[test] + fn decide_action_back_above_warn_but_below_margin_is_not_recovery() { + let cfg = config(); + let mut state = DiskWatchState::default(); + let now = Instant::now(); + decide_action(30 * GB, &cfg, &mut state, now); // Warn + // 52 GB is above warn_gb (50) but below the 55 GB recovery margin. + assert_eq!( + decide_action(52 * GB, &cfg, &mut state, now), + DiskAction::None + ); + } +} diff --git a/server/src/service/events/buffer.rs b/server/src/service/events/buffer.rs index 934d8f38..14b66006 100644 --- a/server/src/service/events/buffer.rs +++ b/server/src/service/events/buffer.rs @@ -45,6 +45,30 @@ pub enum StoredEvent { /// Unix timestamp in milliseconds when this event was recorded. timestamp_ms: u64, }, + /// Free disk space on a sled crossed a warn/critical threshold (story 1200). + DiskSpaceWarning { + /// Severity level: `"warn"` or `"critical"`. + level: String, + /// Free space in bytes at the time of the check. + free_bytes: u64, + /// Size of the `target/` directory in bytes. + target_bytes: u64, + /// Size of the `.huskies/worktrees/` directory in bytes. + worktrees_bytes: u64, + /// Identifier of the sled that observed the reading. + host_id: String, + /// Unix timestamp in milliseconds when this event was recorded. + timestamp_ms: u64, + }, + /// Free disk space recovered on a sled after a warn/critical warning (story 1200). + DiskSpaceRecovered { + /// Free space in bytes at the time of recovery. + free_bytes: u64, + /// Identifier of the sled that observed the recovery. + host_id: String, + /// Unix timestamp in milliseconds when this event was recorded. + timestamp_ms: u64, + }, } impl StoredEvent { @@ -54,6 +78,8 @@ impl StoredEvent { StoredEvent::StageTransition { timestamp_ms, .. } => *timestamp_ms, StoredEvent::MergeFailure { timestamp_ms, .. } => *timestamp_ms, StoredEvent::StoryBlocked { timestamp_ms, .. } => *timestamp_ms, + StoredEvent::DiskSpaceWarning { timestamp_ms, .. } => *timestamp_ms, + StoredEvent::DiskSpaceRecovered { timestamp_ms, .. } => *timestamp_ms, } } } diff --git a/server/src/service/gateway/io.rs b/server/src/service/gateway/io.rs index 15c885c0..0a89608f 100644 --- a/server/src/service/gateway/io.rs +++ b/server/src/service/gateway/io.rs @@ -507,15 +507,51 @@ pub fn init_wizard_state(path: &Path) { /// /// The task exits cleanly when the broadcast channel is closed (i.e. when /// `GatewayState` is dropped). +/// Window within which identical disk-space warnings from different sleds are +/// deduped into a single forwarded chat message (story 1200 AC4). Matches the +/// sled-side default `rate_limit_secs`. +const DISK_DEDUPE_WINDOW: std::time::Duration = std::time::Duration::from_secs(6 * 60 * 60); + +/// Returns the dedupe key for a [`crate::service::events::StoredEvent`], or +/// `None` if the event type is never deduped (only disk-space events dedupe +/// across sleds — story 1200 AC4). +fn dedupe_key(event: &crate::service::events::StoredEvent) -> Option { + match event { + crate::service::events::StoredEvent::DiskSpaceWarning { level, .. } => { + Some(format!("disk_warning:{level}")) + } + crate::service::events::StoredEvent::DiskSpaceRecovered { .. } => { + Some("disk_recovered".to_string()) + } + _ => None, + } +} + pub fn spawn_gateway_broadcaster_forwarder( transport: std::sync::Arc, room_ids: Vec, mut rx: tokio::sync::broadcast::Receiver, ) { tokio::spawn(async move { + let mut last_forwarded: std::collections::HashMap = + std::collections::HashMap::new(); loop { match rx.recv().await { Ok(event) => { + if let Some(key) = dedupe_key(&event.event) { + let now = std::time::Instant::now(); + let should_forward = match last_forwarded.get(&key) { + None => true, + Some(last) => now.duration_since(*last) >= DISK_DEDUPE_WINDOW, + }; + if !should_forward { + crate::slog!( + "[gateway-forwarder] Deduping repeated {key} within window" + ); + continue; + } + last_forwarded.insert(key, now); + } let (plain, html) = super::polling::format_gateway_event(&event.project, &event.event); for room_id in &room_ids { @@ -839,4 +875,57 @@ mod tests { let dir = tempfile::tempdir().unwrap(); assert!(read_installed_manifest(dir.path(), "huskies-linux-arm64").is_none()); } + + // ── dedupe_key (story 1200 AC4) ────────────────────────────────────────── + + #[test] + fn dedupe_key_disk_warning_keys_by_level() { + let warn = crate::service::events::StoredEvent::DiskSpaceWarning { + level: "warn".to_string(), + free_bytes: 1, + target_bytes: 0, + worktrees_bytes: 0, + host_id: "sled-a".to_string(), + timestamp_ms: 0, + }; + let critical = crate::service::events::StoredEvent::DiskSpaceWarning { + level: "critical".to_string(), + free_bytes: 1, + target_bytes: 0, + worktrees_bytes: 0, + host_id: "sled-b".to_string(), + timestamp_ms: 0, + }; + // Different sleds at the same level share a key (so they dedupe)... + assert_eq!(dedupe_key(&warn), dedupe_key(&warn)); + // ...but different levels do not. + assert_ne!(dedupe_key(&warn), dedupe_key(&critical)); + } + + #[test] + fn dedupe_key_disk_recovered_has_a_single_shared_key() { + let a = crate::service::events::StoredEvent::DiskSpaceRecovered { + free_bytes: 1, + host_id: "sled-a".to_string(), + timestamp_ms: 0, + }; + let b = crate::service::events::StoredEvent::DiskSpaceRecovered { + free_bytes: 2, + host_id: "sled-b".to_string(), + timestamp_ms: 1, + }; + assert_eq!(dedupe_key(&a), dedupe_key(&b)); + } + + #[test] + fn dedupe_key_non_disk_events_are_not_deduped() { + let transition = crate::service::events::StoredEvent::StageTransition { + story_id: "1_story".to_string(), + story_name: String::new(), + from_stage: "2_current".to_string(), + to_stage: "3_qa".to_string(), + timestamp_ms: 0, + }; + assert_eq!(dedupe_key(&transition), None); + } } diff --git a/server/src/service/gateway/polling.rs b/server/src/service/gateway/polling.rs index a963b06a..84b73787 100644 --- a/server/src/service/gateway/polling.rs +++ b/server/src/service/gateway/polling.rs @@ -7,7 +7,8 @@ use crate::pipeline_state::Stage; use crate::service::events::StoredEvent; use crate::service::notifications::{ - format_blocked_notification, format_error_notification, format_stage_notification, + format_blocked_notification, format_disk_recovery_notification, + format_disk_warning_notification, format_error_notification, format_stage_notification, }; /// Format a [`StoredEvent`] from a project into a gateway notification. @@ -49,6 +50,31 @@ pub fn format_gateway_event(project_name: &str, event: &StoredEvent) -> (String, let (plain, html) = format_blocked_notification(story_id, story_name, reason); (format!("{prefix}{plain}"), format!("{prefix}{html}")) } + StoredEvent::DiskSpaceWarning { + level, + free_bytes, + target_bytes, + worktrees_bytes, + host_id, + .. + } => { + let (plain, html) = format_disk_warning_notification( + level, + host_id, + *free_bytes, + *target_bytes, + *worktrees_bytes, + ); + (format!("{prefix}{plain}"), format!("{prefix}{html}")) + } + StoredEvent::DiskSpaceRecovered { + free_bytes, + host_id, + .. + } => { + let (plain, html) = format_disk_recovery_notification(host_id, *free_bytes); + (format!("{prefix}{plain}"), format!("{prefix}{html}")) + } } } @@ -91,6 +117,14 @@ pub fn format_gateway_audit_line(project: &str, event: &StoredEvent) -> String { "audit ts={ts} project={project} id={story_id} event=story_blocked reason={reason}" ) } + StoredEvent::DiskSpaceWarning { level, host_id, .. } => { + format!( + "audit ts={ts} project={project} event=disk_space_warning level={level} host={host_id}" + ) + } + StoredEvent::DiskSpaceRecovered { host_id, .. } => { + format!("audit ts={ts} project={project} event=disk_space_recovered host={host_id}") + } } } diff --git a/server/src/service/mod.rs b/server/src/service/mod.rs index b0263384..60a139e2 100644 --- a/server/src/service/mod.rs +++ b/server/src/service/mod.rs @@ -15,6 +15,8 @@ pub mod bot_command; pub mod common; /// Diagnostics — server logs, CRDT dump, and permission management. pub mod diagnostics; +/// Low-disk-space watchdog — free-space threshold checks and notifications. +pub mod disk_watch; /// Event-based pipeline triggers: register, list, cancel, and execute on TransitionFired events. pub mod event_triggers; /// Pipeline event buffer for SSE streaming. diff --git a/server/src/service/notifications/events.rs b/server/src/service/notifications/events.rs index c9d131cc..e93ecd03 100644 --- a/server/src/service/notifications/events.rs +++ b/server/src/service/notifications/events.rs @@ -30,6 +30,13 @@ pub enum EventAction { NewItemCreated, /// Post a merge-auto-retry notification naming the attempt and budget. MergeAutoRetry, + /// Post a low-disk-space warning notification (story 1200). + DiskWarning { + /// Severity level: `"warn"` or `"critical"`. + level: String, + }, + /// Post a disk-space recovery notification (story 1200). + DiskRecovery, /// Log server-side only; do not post to chat (e.g. hard rate-limit blocks). LogOnly, /// Reload the project configuration. @@ -57,6 +64,10 @@ pub fn classify(event: &WatcherEvent) -> EventAction { } WatcherEvent::NewItemCreated { .. } => EventAction::NewItemCreated, WatcherEvent::MergeAutoRetry { .. } => EventAction::MergeAutoRetry, + WatcherEvent::DiskSpaceWarning { level, .. } => EventAction::DiskWarning { + level: level.clone(), + }, + WatcherEvent::DiskSpaceRecovered { .. } => EventAction::DiskRecovery, _ => EventAction::Skip, } } diff --git a/server/src/service/notifications/format.rs b/server/src/service/notifications/format.rs index 69b4b1c6..f93ab074 100644 --- a/server/src/service/notifications/format.rs +++ b/server/src/service/notifications/format.rs @@ -272,6 +272,57 @@ pub fn format_merge_auto_retry_notification( (plain, html) } +/// Format a low-disk-space warning notification message (story 1200 AC3). +/// +/// Includes free space, `target/` and `.huskies/worktrees/` directory sizes, +/// and names the `gc` tool as the first remediation step. +/// Returns `(plain_text, html)` suitable for `ChatTransport::send_message`. +pub fn format_disk_warning_notification( + level: &str, + host_id: &str, + free_bytes: u64, + target_bytes: u64, + worktrees_bytes: u64, +) -> (String, String) { + let emoji = if level == "critical" { + "\u{1f6a8}" // 🚨 + } else { + "\u{26a0}\u{fe0f}" // ⚠️ + }; + let free_gb = bytes_to_gb(free_bytes); + let target_gb = bytes_to_gb(target_bytes); + let worktrees_gb = bytes_to_gb(worktrees_bytes); + let plain = format!( + "{emoji} Low disk space on {host_id} ({level}): {free_gb:.1}GB free \ + (target/ {target_gb:.1}GB, worktrees/ {worktrees_gb:.1}GB) \ + \u{2014} first response: run the `gc` tool to reclaim space" + ); + let html = format!( + "{emoji} Low disk space on {host_id} ({level}): {free_gb:.1}GB free \ + (target/ {target_gb:.1}GB, worktrees/ {worktrees_gb:.1}GB) \ + \u{2014} first response: run the gc tool to reclaim space" + ); + (plain, html) +} + +/// Format a disk-space-recovered notification message (story 1200 AC2). +/// +/// Sent once when free space climbs back above the configured recovery +/// margin after a warn/critical warning. +/// Returns `(plain_text, html)` suitable for `ChatTransport::send_message`. +pub fn format_disk_recovery_notification(host_id: &str, free_bytes: u64) -> (String, String) { + let free_gb = bytes_to_gb(free_bytes); + let plain = format!("\u{2705} Disk space recovered on {host_id}: {free_gb:.1}GB free"); + let html = + format!("\u{2705} Disk space recovered on {host_id}: {free_gb:.1}GB free"); + (plain, html) +} + +/// Convert a byte count to gigabytes for display (story 1200). +fn bytes_to_gb(bytes: u64) -> f64 { + bytes as f64 / 1_000_000_000.0 +} + /// Maximum number of trailing gate-output lines included in a merge-failure /// chat notification. /// @@ -813,4 +864,50 @@ mod tests { format_agent_completed_notification("42_story_foo", "", "coder-1", false); assert_eq!(plain, "\u{274C} #42 \u{2014} coder-1 failed"); } + + // ── format_disk_warning_notification ────────────────────────────────────── + + #[test] + fn format_disk_warning_notification_warn_level_includes_content() { + let (plain, html) = format_disk_warning_notification( + "warn", + "sled-a", + 45_000_000_000, + 30_000_000_000, + 10_000_000_000, + ); + assert!(plain.contains("sled-a")); + assert!(plain.contains("45.0GB free")); + assert!(plain.contains("target/ 30.0GB")); + assert!(plain.contains("worktrees/ 10.0GB")); + assert!(plain.contains("`gc` tool")); + assert!(html.contains("gc tool")); + } + + #[test] + fn format_disk_warning_notification_critical_uses_distinct_emoji() { + let (plain, _html) = + format_disk_warning_notification("critical", "sled-b", 5_000_000_000, 0, 0); + assert!(plain.starts_with("\u{1f6a8}")); + } + + #[test] + fn format_disk_warning_notification_warn_uses_warning_emoji() { + let (plain, _html) = + format_disk_warning_notification("warn", "sled-b", 45_000_000_000, 0, 0); + assert!(plain.starts_with("\u{26a0}\u{fe0f}")); + } + + // ── format_disk_recovery_notification ───────────────────────────────────── + + #[test] + fn format_disk_recovery_notification_includes_host_and_free_space() { + let (plain, html) = format_disk_recovery_notification("sled-a", 60_000_000_000); + assert_eq!( + plain, + "\u{2705} Disk space recovered on sled-a: 60.0GB free" + ); + assert!(html.contains("sled-a")); + assert!(html.contains("60.0GB free")); + } } diff --git a/server/src/service/notifications/io/listener.rs b/server/src/service/notifications/io/listener.rs index 7468b65a..dcc87a4e 100644 --- a/server/src/service/notifications/io/listener.rs +++ b/server/src/service/notifications/io/listener.rs @@ -15,7 +15,8 @@ use super::super::events::classify; use super::super::filter::{AGENT_EVENT_DEBOUNCE, should_send_rate_limit}; use super::super::format::{ MERGE_FAILURE_TAIL_LINES, format_agent_completed_notification, - format_agent_started_notification, format_blocked_notification, format_error_notification, + format_agent_started_notification, format_blocked_notification, + format_disk_recovery_notification, format_disk_warning_notification, format_error_notification, format_merge_auto_retry_notification, format_new_item_notification, format_oauth_account_swapped, format_oauth_accounts_exhausted, format_rate_limit_notification, truncate_gate_output, @@ -325,6 +326,59 @@ pub fn spawn_notification_listener( } } } + EventAction::DiskWarning { .. } => { + if !config.status_push_enabled { + continue; + } + let WatcherEvent::DiskSpaceWarning { + ref level, + free_bytes, + target_bytes, + worktrees_bytes, + ref host_id, + } = event + else { + continue; + }; + let (plain, html) = format_disk_warning_notification( + level, + host_id, + free_bytes, + target_bytes, + worktrees_bytes, + ); + slog!("[bot] Sending disk-space warning notification: {plain}"); + for room_id in &rooms_for_notification(&get_room_ids) { + if let Err(e) = transport.send_message(room_id, &plain, &html).await { + slog!( + "[bot] Failed to send disk-space warning notification \ + to {room_id}: {e}" + ); + } + } + } + EventAction::DiskRecovery => { + if !config.status_push_enabled { + continue; + } + let WatcherEvent::DiskSpaceRecovered { + free_bytes, + ref host_id, + } = event + else { + continue; + }; + let (plain, html) = format_disk_recovery_notification(host_id, free_bytes); + slog!("[bot] Sending disk-space recovery notification: {plain}"); + for room_id in &rooms_for_notification(&get_room_ids) { + if let Err(e) = transport.send_message(room_id, &plain, &html).await { + slog!( + "[bot] Failed to send disk-space recovery notification \ + to {room_id}: {e}" + ); + } + } + } EventAction::LogOnly => { // Hard-block: log server-side for debugging; do NOT post to chat. // Hard-block auto-resume is normal operation — the status command diff --git a/server/src/service/notifications/mod.rs b/server/src/service/notifications/mod.rs index d0dced20..f3cd17ee 100644 --- a/server/src/service/notifications/mod.rs +++ b/server/src/service/notifications/mod.rs @@ -19,7 +19,8 @@ pub(super) mod io; pub(super) mod route; pub use format::{ - format_blocked_notification, format_error_notification, format_stage_notification, + format_blocked_notification, format_disk_recovery_notification, + format_disk_warning_notification, format_error_notification, format_stage_notification, }; pub use io::spawn_notification_listener; pub use io::spawn_stage_notification_subscriber; diff --git a/server/src/service/status/format.rs b/server/src/service/status/format.rs index 8eddda20..71b2a5ab 100644 --- a/server/src/service/status/format.rs +++ b/server/src/service/status/format.rs @@ -97,6 +97,22 @@ pub fn format_status_event(event: &StatusEvent) -> String { "\u{26d4} #{number} {name} \u{2014} {agent_name} hard rate-limited until {reset}" ) } + StatusEvent::DiskSpaceWarning { + level, + free_bytes, + host_id, + .. + } => { + let free_gb = *free_bytes as f64 / 1_000_000_000.0; + format!("\u{26a0}\u{fe0f} Low disk space on {host_id} ({level}): {free_gb:.1}GB free") + } + StatusEvent::DiskSpaceRecovered { + free_bytes, + host_id, + } => { + let free_gb = *free_bytes as f64 / 1_000_000_000.0; + format!("\u{2705} Disk space recovered on {host_id}: {free_gb:.1}GB free") + } } } diff --git a/server/src/service/status/mod.rs b/server/src/service/status/mod.rs index 1273c3a9..f1dc9b6a 100644 --- a/server/src/service/status/mod.rs +++ b/server/src/service/status/mod.rs @@ -100,6 +100,26 @@ pub enum StatusEvent { /// UTC instant at which the rate limit resets. reset_at: DateTime, }, + /// Free disk space on a sled crossed a warn/critical threshold (story 1200). + DiskSpaceWarning { + /// Severity level: `"warn"` or `"critical"`. + level: String, + /// Free space in bytes at the time of the check. + free_bytes: u64, + /// Size of the `target/` directory in bytes. + target_bytes: u64, + /// Size of the `.huskies/worktrees/` directory in bytes. + worktrees_bytes: u64, + /// Identifier of the sled that observed the reading. + host_id: String, + }, + /// Free disk space recovered on a sled after a warn/critical warning (story 1200). + DiskSpaceRecovered { + /// Free space in bytes at the time of recovery. + free_bytes: u64, + /// Identifier of the sled that observed the recovery. + host_id: String, + }, } // ── Subscription ────────────────────────────────────────────────────────────── diff --git a/server/src/service/ws/message/convert.rs b/server/src/service/ws/message/convert.rs index 96c79216..41290309 100644 --- a/server/src/service/ws/message/convert.rs +++ b/server/src/service/ws/message/convert.rs @@ -41,6 +41,9 @@ pub fn watcher_event_to_response(e: WatcherEvent) -> Option { WatcherEvent::NewItemCreated { .. } => None, // Merge-auto-retry notifications are forwarded to chat transports only; no WebSocket message. WatcherEvent::MergeAutoRetry { .. } => None, + // Disk-space events are forwarded to chat transports only; no WebSocket message (story 1200). + WatcherEvent::DiskSpaceWarning { .. } => None, + WatcherEvent::DiskSpaceRecovered { .. } => None, } } diff --git a/server/src/startup/tick_loop.rs b/server/src/startup/tick_loop.rs index 693c418e..f4ed28ac 100644 --- a/server/src/startup/tick_loop.rs +++ b/server/src/startup/tick_loop.rs @@ -183,9 +183,15 @@ pub(crate) fn spawn_tick_loop( }) .unwrap_or((30, std::time::Duration::from_secs(4 * 3600))); + let disk_watch_host_id = + std::env::var("HOSTNAME").unwrap_or_else(|_| "unknown-host".to_string()); + let disk_watch_watcher_tx = agents.watcher_tx(); + let disk_watch_status = agents.status_broadcaster(); + tokio::spawn(async move { let mut interval = tokio::time::interval(std::time::Duration::from_secs(1)); let mut tick_count: u64 = 0; + let mut disk_watch_state = service::disk_watch::DiskWatchState::default(); loop { interval.tick().await; tick_count = tick_count.wrapping_add(1); @@ -218,6 +224,25 @@ pub(crate) fn spawn_tick_loop( agents.reap_stale_merge_jobs(); } + // Low-disk-space watchdog (story 1200): check free space on the + // project workspace filesystem every 30 ticks (~30s) and emit a + // rate-limited warn/critical notification, or a recovery notice. + if tick_count.is_multiple_of(30) + && let Some(ref r) = root + { + let disk_config = config::ProjectConfig::load(r) + .map(|c| c.disk_watch) + .unwrap_or_default(); + service::disk_watch::io::check_and_notify( + r, + &disk_config, + &mut disk_watch_state, + &disk_watch_watcher_tx, + &disk_watch_status, + &disk_watch_host_id, + ); + } + // Periodic reconciler: converge subscriber side effects so that // Lagged broadcast events never leave state permanently diverged. if tick_count.is_multiple_of(reconcile_interval) diff --git a/server/src/worktree/cleanup.rs b/server/src/worktree/cleanup.rs index d36c2225..8b743ea3 100644 --- a/server/src/worktree/cleanup.rs +++ b/server/src/worktree/cleanup.rs @@ -194,6 +194,7 @@ mod tests { component: vec![], agent: vec![], watcher: WatcherConfig::default(), + disk_watch: Default::default(), default_qa: "server".to_string(), default_coder_model: None, max_coders: None, diff --git a/server/src/worktree/create.rs b/server/src/worktree/create.rs index ecd3b8f8..4ccf4d94 100644 --- a/server/src/worktree/create.rs +++ b/server/src/worktree/create.rs @@ -233,6 +233,7 @@ mod tests { component: vec![], agent: vec![], watcher: WatcherConfig::default(), + disk_watch: Default::default(), default_qa: "server".to_string(), default_coder_model: None, max_coders: None, diff --git a/server/src/worktree/remove.rs b/server/src/worktree/remove.rs index 26d12b27..a5e053bf 100644 --- a/server/src/worktree/remove.rs +++ b/server/src/worktree/remove.rs @@ -69,6 +69,7 @@ mod tests { component: vec![], agent: vec![], watcher: WatcherConfig::default(), + disk_watch: Default::default(), default_qa: "server".to_string(), default_coder_model: None, max_coders: None, diff --git a/server/src/worktree/sweep.rs b/server/src/worktree/sweep.rs index 8304ea73..08151dfa 100644 --- a/server/src/worktree/sweep.rs +++ b/server/src/worktree/sweep.rs @@ -115,6 +115,7 @@ mod tests { component: vec![], agent: vec![], watcher: WatcherConfig::default(), + disk_watch: Default::default(), default_qa: "server".to_string(), default_coder_model: None, max_coders: None,