feat(agent): raise a todo when the agent's disk gets tight

Closes 2718.

The operator has been going through agent dirs by hand with ncdu,
deleting 20+GB target dirs. Agents had no way to know they were the
ones sitting on the space.

New `disk_watch` module in the harness: every 15 minutes it statvfs's
the filesystem backing the agent's state dir and, past 80%, raises a
keyed `disk` todo telling the agent to free space — with the operator's
rules inline: only delete things that are actually big, build output
first, and never delete something still needed, ask for more space
instead.

Over threshold it also walks the agent's own tree (`/agents/<label>`
plus `$HOME`) and names the directories worth looking at, so the todo
says where the bytes actually went rather than just that the disk is
full. The walk is bounded on every axis — entry budget, recursion cap,
report depth — pinned to the state dir's device so it can't wander into
`/nix` or the shared bind mounts, and it does not traverse symlinks. It
reports the deepest oversized directory on each branch, so the agent
gets pointed at `<workspace>/target` rather than at `/agents/<label>`.

Anti-nag is the whole design constraint. The todo is keyed, and the
summary is deliberately stable: the percentage is bucketed to 5 points
and no raw byte counts appear anywhere in it. An unchanged situation
re-upserts as `changed == false` and never fires the wake, so a disk
that has been steady at 89% for a week sits quietly in the loose-ends
list; crossing into a new bucket speaks up once. Dropping back under
the threshold clears the row.

Harness-local by construction, per the operator's call that this gets
no core wiring: hive-c0re cannot push a todo at all (the store and its
wake live inside the container), and running in-process means this
skips even the in-agent socket and calls `Todos::upsert` directly.

Worth recording, since it shaped the scope: btrfs does NOT fold qgroup
limits into statfs. Measured with quota counting enabled and a 20G
limit set on a real subvolume, statvfs returns byte-identical whole-FS
numbers for that subvolume, an ordinary agent dir, and the root. So
this watches host-FS pressure, which is valid before and after the
planned subvolume migration; per-agent quota awareness would need the
limit handed to the agent explicitly.

Cargo.lock

@@ -1530,6 +1530,7 @@ dependencies = [
  "http-body-util",
  "hyper",
  "hyper-util",
+ "libc",
  "reqwest 0.13.1",
  "rmcp",
  "rusqlite",

docs/turn-loop.md

@@ -101,6 +101,15 @@ installs claude plugins, spawns `forge_notify::run` + `web_ui::serve`,
 and either drops into `serve_loop` directly (`Online`) or parks on
 the login flow first (`NeedsLogin`).
 
+`spawn_todo_socket` opens the todos store and, alongside
+`todo_server::run` (the socket the out-of-process producers dial),
+spawns `disk_watch::run` — an *in-process* todo producer, so it shares
+the store + wake `Notify` directly rather than dialling its own socket.
+It raises a keyed `disk` todo when the filesystem backing the agent's
+state gets tight, naming the agent's own biggest directories; the
+summary is bucketed and carries no raw byte counts, so an unchanged
+situation re-upserts as `changed == false` and never re-wakes.
+
 Plugin install failures are not fatal: each entry comes back as a
 human-readable failure string that gets routed via
 `Surface::send_to_parent` to the agent's topology parent (the

hive-agent/Cargo.toml

@@ -23,6 +23,7 @@ hive-claude.workspace = true
 hive-agent-sock.workspace = true
 hive-core-agent-sock.workspace = true
 hive-sh4re.workspace = true
+libc.workspace = true
 rmcp.workspace = true
 rusqlite.workspace = true
 schemars.workspace = true

hive-agent/src/disk_watch.rs

@@ -0,0 +1,369 @@
+//! Disk-pressure watch: raises a loose-ends-v2 todo when the filesystem
+//! backing this agent's state gets tight, pointing at the agent's own
+//! biggest directories so the todo says where the bytes actually went.
+//!
+//! Harness-local by design. hive-c0re cannot push a todo — the todo store
+//! and its wake `Notify` live inside the container, reachable only over
+//! `HIVE_AGENT_SOCKET` — and the operator explicitly ruled out core
+//! wiring for this. Running in-process also means we skip the socket
+//! entirely and call [`crate::todos::Todos`] directly.
+//!
+//! Anti-nag: the todo is keyed (`disk`/[`TODO_KEY`]), and the summary is
+//! deliberately *stable* — the percentage is bucketed and no raw byte
+//! counts appear in it. An unchanged summary makes `upsert` report
+//! `changed == false`, so a steadily-full disk sits quiet in the
+//! loose-ends list instead of waking the agent every tick. Only crossing
+//! into a new bucket (or a change in which directories are big enough to
+//! list) speaks up again. Dropping back under the threshold clears the
+//! todo.
+
+use std::ffi::CString;
+use std::fmt::Write as _;
+use std::os::unix::ffi::OsStrExt as _;
+use std::os::unix::fs::MetadataExt as _;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::Notify;
+
+use crate::todos::Todos;
+
+/// How often the filesystem is probed. The `statvfs` is free; the
+/// directory scan behind it only runs once we're already over threshold.
+const CHECK_INTERVAL: Duration = Duration::from_mins(15);
+/// Percent-used past which the agent gets a todo. Matches the operator's
+/// ask ("more than 80%").
+const WARN_PCT: u64 = 80;
+/// Percentages are reported in buckets this wide. This is the anti-nag
+/// knob: a disk drifting 89% → 91% keeps the same summary and stays
+/// silent; a jump to 95% earns a fresh wake.
+const BUCKET_PCT: u64 = 5;
+/// Producing subsystem marker for the todo row.
+const SUBSYSTEM: &str = "disk";
+/// Fixed dedup key — there is only ever one disk todo per agent.
+const TODO_KEY: &str = "usage";
+/// Only directories at least this large are worth naming. Coarse on
+/// purpose: the listed set shouldn't churn while a build runs.
+const BIG_DIR_BYTES: u64 = 5 << 30;
+/// At most this many directories are named in the todo.
+const MAX_LISTED: usize = 5;
+/// Directories deeper than this (relative to a scan root) are summed into
+/// their parent but never listed by name — a todo pointing at
+/// `…/target/debug/build/foo-1a2b/out` helps nobody.
+const REPORT_DEPTH: usize = 4;
+/// Hard recursion cap, independent of [`REPORT_DEPTH`]: bounds stack
+/// depth on a pathological tree.
+const MAX_DEPTH: usize = 24;
+/// Directory entries the scan will look at before giving up. Bounds the
+/// worst case on a huge tree; a truncated scan just names fewer dirs.
+const SCAN_BUDGET: u64 = 400_000;
+
+/// Background loop: periodically reconcile the disk todo. Detached task —
+/// runs for the harness's lifetime; errors are logged, never fatal.
+pub async fn run(todos: Arc<Todos>, wake: Arc<Notify>) {
+    loop {
+        // Sleep first: boot is the busiest the container gets, and a disk
+        // that's been full for a week can wait one interval.
+        tokio::time::sleep(CHECK_INTERVAL).await;
+        // The scan walks the agent's tree — blocking work, off the runtime.
+        let summary = match tokio::task::spawn_blocking(probe).await {
+            Ok(s) => s,
+            Err(e) => {
+                tracing::warn!(error = %e, "disk watch probe panicked");
+                continue;
+            }
+        };
+        reconcile(&todos, &wake, summary.as_deref());
+    }
+}
+
+/// Apply one probe result to the todo store: raise/refresh while over
+/// threshold, clear once back under. Split from [`run`] so the store
+/// interaction is testable without a timer.
+fn reconcile(todos: &Todos, wake: &Notify, summary: Option<&str>) {
+    match summary {
+        Some(summary) => match todos.upsert(SUBSYSTEM, Some(TODO_KEY), summary, None) {
+            // Only a genuine change wakes the agent — an identical summary
+            // means the situation is unchanged and already in its list.
+            Ok((_, true)) => wake.notify_one(),
+            Ok((_, false)) => {}
+            Err(e) => tracing::warn!(error = ?e, "disk todo upsert failed"),
+        },
+        None => {
+            if let Err(e) = todos.clear(SUBSYSTEM, Some(TODO_KEY)) {
+                tracing::warn!(error = ?e, "disk todo clear failed");
+            }
+        }
+    }
+}
+
+/// One probe: measure the filesystem backing the state dir and, when it's
+/// over threshold, scan the agent's own tree for the big directories worth
+/// naming. `None` means "nothing to report" — under threshold, or the
+/// syscall failed.
+fn probe() -> Option<String> {
+    let state = crate::paths::state_dir();
+    let pct = used_pct(&state)?;
+    if pct < WARN_PCT {
+        return None;
+    }
+    summary_for(pct, &big_dirs())
+}
+
+/// Pure rendering half of [`probe`]: the todo text for a given usage
+/// percentage + set of oversized directories. Separated so the threshold
+/// and — more importantly — the *stability* of the summary are unit-tested
+/// without a real filesystem.
+fn summary_for(pct: u64, dirs: &[PathBuf]) -> Option<String> {
+    if pct < WARN_PCT {
+        return None;
+    }
+    let bucket = pct / BUCKET_PCT * BUCKET_PCT;
+    let mut out = format!(
+        "disk over {bucket}% full on the filesystem holding your state — free up space if you safely can.\n\
+         Only delete things that are actually big (tens of GiB); a few MB of notes won't move the needle.\n\
+         First candidates: regenerable build output in your workspace (`target/`, `node_modules/`, `dist/`).\n\
+         Never delete anything still needed. If nothing is safe to drop, tell the operator you need more space \
+         rather than forcing it — be cautious with deletions in general."
+    );
+    if !dirs.is_empty() {
+        out.push_str(
+            "\nBiggest directories under your own tree (`du -sh` them before removing anything):",
+        );
+        for dir in dirs {
+            // Infallible: writing into a String.
+            let _ = write!(out, "\n  - {}", dir.display());
+        }
+    }
+    Some(out)
+}
+
+/// Percent of the filesystem containing `path` that is in use, as `df`
+/// reports it (`used / (used + available)`, so root-reserved blocks count
+/// as used). `None` if the `statvfs` syscall fails.
+fn used_pct(path: &Path) -> Option<u64> {
+    let c_path = CString::new(path.as_os_str().as_bytes()).ok()?;
+    // SAFETY: `statvfs` reads only through the valid NUL-terminated
+    // `c_path` pointer and writes into the zeroed `stat` we own. The
+    // return code is checked before any field is read.
+    let mut stat: libc::statvfs = unsafe { std::mem::zeroed() };
+    let rc = unsafe { libc::statvfs(c_path.as_ptr(), &raw mut stat) };
+    if rc != 0 {
+        return None;
+    }
+    let used = stat.f_blocks.checked_sub(stat.f_bfree)?;
+    let capacity = used.checked_add(stat.f_bavail)?;
+    if capacity == 0 {
+        return None;
+    }
+    // Integer math on block counts — no float, so no rounding to explain.
+    Some(used.saturating_mul(100) / capacity)
+}
+
+/// Scan roots: everything this agent owns and could plausibly free. The
+/// `/agents/<label>` parent covers `state/` (workspaces live there) and
+/// `harness/`; `$HOME` covers caches like `~/.cargo`. Roots nested inside
+/// another root are dropped so nothing is counted twice.
+fn roots() -> Vec<PathBuf> {
+    let state = crate::paths::state_dir();
+    let mut found = vec![
+        state
+            .parent()
+            .map_or_else(|| state.clone(), Path::to_path_buf),
+    ];
+    if let Some(home) = std::env::var_os("HOME") {
+        found.push(PathBuf::from(home));
+    }
+    found.retain(|p| p.is_dir());
+    found.sort();
+    found.dedup();
+    let mut kept: Vec<PathBuf> = Vec::new();
+    for root in found {
+        // Sorted, so a parent is always seen before anything nested in it.
+        if !kept.iter().any(|k| root.starts_with(k)) {
+            kept.push(root);
+        }
+    }
+    kept
+}
+
+/// Walk the agent's own tree and return the directories worth naming:
+/// at least [`BIG_DIR_BYTES`], deepest-first (a `target/` dir, not the
+/// `/agents/<label>` that contains it), capped at [`MAX_LISTED`].
+fn big_dirs() -> Vec<PathBuf> {
+    let mut budget = SCAN_BUDGET;
+    let mut big = Vec::new();
+    for root in roots() {
+        // Stay on the filesystem we measured: bind mounts (`/nix`,
+        // `/shared`, `/knowledge`) aren't this agent's to clean up.
+        let Ok(md) = std::fs::metadata(&root) else {
+            continue;
+        };
+        dir_bytes(&root, md.dev(), 0, &mut budget, &mut big);
+    }
+    let mut out = leafmost(big);
+    out.truncate(MAX_LISTED);
+    out
+}
+
+/// Recursive size accumulation. Pushes every directory that is both
+/// shallow enough to name and at least [`BIG_DIR_BYTES`] into `big`;
+/// returns the total bytes under `path`.
+fn dir_bytes(path: &Path, dev: u64, depth: usize, budget: &mut u64, big: &mut Vec<PathBuf>) -> u64 {
+    if depth >= MAX_DEPTH {
+        return 0;
+    }
+    let Ok(entries) = std::fs::read_dir(path) else {
+        return 0;
+    };
+    let mut total: u64 = 0;
+    for entry in entries.flatten() {
+        if *budget == 0 {
+            break;
+        }
+        *budget -= 1;
+        // `DirEntry::metadata` does not traverse symlinks, so a link into
+        // the nix store is counted as the link itself, not its target.
+        let Ok(md) = entry.metadata() else {
+            continue;
+        };
+        if md.is_symlink() || md.dev() != dev {
+            continue;
+        }
+        if md.is_dir() {
+            let child = entry.path();
+            let bytes = dir_bytes(&child, dev, depth + 1, budget, big);
+            if depth < REPORT_DEPTH && bytes >= BIG_DIR_BYTES {
+                big.push(child);
+            }
+            total = total.saturating_add(bytes);
+        } else {
+            total = total.saturating_add(md.len());
+        }
+    }
+    total
+}
+
+/// Keep only the deepest oversized directory on each branch: if both a
+/// workspace and the `target/` inside it are over threshold, name the
+/// `target/`. Sorted output, so the todo summary is order-stable across
+/// ticks.
+fn leafmost(mut dirs: Vec<PathBuf>) -> Vec<PathBuf> {
+    dirs.sort();
+    let mut out: Vec<PathBuf> = Vec::new();
+    for dir in dirs {
+        // Lexicographic order puts a parent immediately before its
+        // descendants, so the ancestors to drop are always on the tail.
+        while out.last().is_some_and(|last| dir.starts_with(last)) {
+            out.pop();
+        }
+        out.push(dir);
+    }
+    out
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn p(s: &str) -> PathBuf {
+        PathBuf::from(s)
+    }
+
+    #[test]
+    fn under_threshold_is_silent() {
+        assert!(summary_for(WARN_PCT - 1, &[]).is_none());
+    }
+
+    #[test]
+    fn at_threshold_speaks_up() {
+        let s = summary_for(WARN_PCT, &[]).expect("todo");
+        assert!(s.contains("over 80% full"));
+        assert!(s.contains("tens of GiB"));
+        assert!(s.contains("tell the operator"));
+    }
+
+    /// The anti-nag property: usage drifting inside one bucket must render
+    /// byte-identically, so `Todos::upsert` reports `changed == false` and
+    /// nothing wakes the agent.
+    #[test]
+    fn drift_inside_a_bucket_is_identical() {
+        let a = summary_for(85, &[]).expect("todo");
+        let b = summary_for(89, &[]).expect("todo");
+        assert_eq!(a, b);
+    }
+
+    #[test]
+    fn crossing_a_bucket_changes_the_summary() {
+        let a = summary_for(89, &[]).expect("todo");
+        let b = summary_for(90, &[]).expect("todo");
+        assert_ne!(a, b);
+        assert!(b.contains("over 90% full"));
+    }
+
+    #[test]
+    fn big_dirs_are_named() {
+        let s = summary_for(90, &[p("/agents/x/state/wt/target")]).expect("todo");
+        assert!(s.contains("/agents/x/state/wt/target"));
+        assert!(s.contains("du -sh"));
+    }
+
+    #[test]
+    fn leafmost_drops_ancestors() {
+        let got = leafmost(vec![
+            p("/a/b"),
+            p("/a"),
+            p("/a/b/target"),
+            p("/a/c"),
+            p("/a/b-other"),
+        ]);
+        // `Path`'s `Ord` compares component-wise, so `/a/b/target` sorts
+        // before `/a/b-other` (`b` < `b-other`) — which is exactly the
+        // property `leafmost` leans on: a parent always lands immediately
+        // before its own descendants, never separated by a sibling that
+        // merely shares a name prefix.
+        assert_eq!(
+            got,
+            vec![p("/a/b/target"), p("/a/b-other"), p("/a/c")],
+            "only the deepest dir on each branch survives, and a sibling \
+             sharing a name prefix is not treated as a descendant"
+        );
+    }
+
+    #[test]
+    fn leafmost_keeps_sibling_branches() {
+        let got = leafmost(vec![p("/a"), p("/a/x/target"), p("/a/y/target")]);
+        assert_eq!(got, vec![p("/a/x/target"), p("/a/y/target")]);
+    }
+
+    #[test]
+    fn roots_are_disjoint() {
+        for (i, a) in roots().iter().enumerate() {
+            for (j, b) in roots().iter().enumerate() {
+                assert!(i == j || !a.starts_with(b), "{a:?} nested in {b:?}");
+            }
+        }
+    }
+
+    /// The store round-trip: first raise wakes, an identical re-probe does
+    /// not, and dropping back under threshold clears the row.
+    #[test]
+    fn reconcile_wakes_once_then_clears() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let todos = Todos::open(&dir.path().join("state.sqlite")).expect("open");
+        let wake = Notify::new();
+        let summary = summary_for(90, &[]).expect("todo");
+
+        reconcile(&todos, &wake, Some(&summary));
+        assert_eq!(todos.list(Some(SUBSYSTEM)).expect("list").len(), 1);
+
+        // Same summary again — still exactly one row, and (by
+        // `upsert`'s contract) no wake was armed.
+        reconcile(&todos, &wake, Some(&summary));
+        assert_eq!(todos.list(Some(SUBSYSTEM)).expect("list").len(), 1);
+
+        reconcile(&todos, &wake, None);
+        assert!(todos.list(Some(SUBSYSTEM)).expect("list").is_empty());
+    }
+}

hive-agent/src/main.rs

@@ -11,6 +11,7 @@
 
 mod client;
 mod db_migrate;
+mod disk_watch;
 mod events;
 mod forge_notify;
 mod harness_state;
@@ -500,6 +501,10 @@ fn spawn_todo_socket(
                     tracing::error!(error = %e, "in-agent todo socket exited with error");
                 }
             });
+            // Disk-pressure watch: an in-process todo producer, so it
+            // shares this store + wake directly instead of dialling the
+            // socket the out-of-process producers use.
+            tokio::spawn(disk_watch::run(store.clone(), todo_wake.clone()));
             Some(store)
         }
         Err(e) => {