subagent, bash: run each subagent run and bash task in its own cgroup

hive-subagent-daemon and hive-bash-daemon are now delegated their cgroup
(Delegate=memory pids, DelegateSubgroup=daemon). At start each daemon
enables memory+pids below its unit and creates runs/ (subagent) or
tasks/ (bash). Each subagent run gets runs/<name>-<n>/ and each bash
task tasks/<id>/, both with memory.oom.group=1, and the daemon removes
the directory when the run ends. An OOM kill now takes one whole run
instead of one process out of the daemon's shared cgroup, and never the
daemon itself.

Subagent runs get oom_score_adj +1000; bash tasks and both daemons stay
at 0; the harness stays at -500. The unit-level OOMScoreAdjust=500 on
the subagent daemon goes: it was both the daemon's own score and a floor
no run below it could be lowered past.

Bash tasks enter their cgroup from the existing pre_exec hook, writing
to a cgroup.procs fd opened by the parent. hive-claude builds its
Command itself and exposes neither a pre_exec hook nor the child's pid,
so a subagent's claude (and an ACP agent's process) is started as the
daemon's own binary with the run's cgroup in its environment; main()
enters the cgroup, raises oom_score_adj and execs the real program, the
trampoline swarm-review-classifier already uses. Both enter before the
program executes, so nothing it forks starts outside the run's cgroup.

Without a delegated `daemon` leaf (tests, a dev shell) each daemon logs
one warning and runs everything in its own cgroup, as before.

The shared helper is hive-agent-sock's run_cgroup module, the crate
both daemons already use for what in-container plugins share.

Closes #5007

hive-agent-sock/Cargo.toml

@@ -8,7 +8,12 @@ readme = "README.md"
 workspace = true
 
 [dependencies]
+# `run_cgroup`: `write(2)` between fork and exec.
+libc.workspace = true
 serde.workspace = true
 serde_json.workspace = true
 tracing.workspace = true
 hive-sh4re.workspace = true
+
+[dev-dependencies]
+tempfile = "3"

hive-agent-sock/README.md

@@ -2,7 +2,7 @@
 
 What's common across the in-container "agent plugins" — the matrix / bash /
 subagent MCP daemons and `forge_notify` today, plus any user-configured MCP
-server declared in an agent's `agent.nix`. Two things live here:
+server declared in an agent's `agent.nix`. Three things live here:
 
 ## Wire types for the in-agent socket
 
@@ -38,3 +38,12 @@ entry into the shape claude expects in a
 `availableToSubagents`) — both already depend on this crate for the socket
 types above, so a new plugin-facing shared type belongs here rather than in
 a crate of its own.
+
+## `run_cgroup`: one cgroup per run
+
+`hive-bash-mcp` puts each bash task in `<unit>/tasks/<id>/` and
+`hive-subagent-mcp` each subagent run in `<unit>/runs/<name>-<n>/`, both with
+`memory.oom.group=1`, so the OOM killer takes one whole run and leaves the
+daemon alone. Both daemons are delegated their cgroup with `Delegate=memory
+pids` and `DelegateSubgroup=daemon` (`nix/agent-modules/mcp.nix`); outside such
+a unit the module logs one warning and runs stay in the daemon's cgroup.

hive-agent-sock/src/lib.rs

@@ -20,6 +20,9 @@
 //! `hive-agent` (the main session's `--mcp-config`) and `hive-subagent-mcp`
 //! (the subagent-eligible subset) — not a socket protocol at all, but this
 //! crate is the natural home since both consumers already depend on it.
+//!
+//! **`run_cgroup`**: the per-run cgroups `hive-bash-mcp` and
+//! `hive-subagent-mcp` put their tasks and subagent runs in.
 
 use serde::{Deserialize, Serialize};
 
@@ -27,6 +30,7 @@ use hive_sh4re::inbox::LooseEnd;
 
 pub mod extra_mcp;
 pub mod paths;
+pub mod run_cgroup;
 
 /// In-container path of the harness-served in-agent socket. The harness
 /// binds it on boot; the in-container producers dial it for todo ops.

hive-agent-sock/src/run_cgroup.rs

@@ -0,0 +1,356 @@
+//! One cgroup per run, under a daemon unit that systemd delegated with
+//! `Delegate=memory pids` and `DelegateSubgroup=daemon`
+//! (`nix/agent-modules/mcp.nix`). The daemon process lives in the unit's
+//! `daemon/` leaf; each run gets `<unit>/<kind>/<name>/` with
+//! `memory.oom.group=1`, so the OOM killer takes a whole run, never part of
+//! one and never the daemon.
+//!
+//! systemd enables no controller below a service's own cgroup, so
+//! [`RunCgroups::init`] writes `+memory +pids` into the unit's and the
+//! `<kind>/` node's `cgroup.subtree_control` itself. Without that the run
+//! cgroups would have no `memory.oom.group` file.
+//!
+//! Two ways into a run's cgroup, both before the run's program executes, so
+//! everything it later forks is born inside:
+//! - [`enter_in_pre_exec`], from a `Command::pre_exec` hook, for a spawn
+//!   this crate's caller builds itself;
+//! - [`RunCgroup::entry`], for a spawn built by a library with no such hook:
+//!   it runs this daemon's own binary, whose `main` calls [`exec_entry`]
+//!   first thing, which enters the cgroup and then execs the real program.
+
+use std::ffi::OsStr;
+use std::fs::File;
+use std::io;
+use std::os::fd::RawFd;
+use std::os::unix::process::CommandExt;
+use std::path::{Path, PathBuf};
+use std::time::Duration;
+
+/// The leaf systemd puts the daemon process in. Must equal the unit's
+/// `DelegateSubgroup=` in `nix/agent-modules/mcp.nix`.
+const DAEMON_SUBGROUP: &str = "daemon";
+
+const CGROUP_ROOT: &str = "/sys/fs/cgroup";
+const CONTROLLERS: &str = "+memory +pids";
+
+/// The program [`exec_entry`] execs. Its presence is what makes a process
+/// an entry wrapper rather than the daemon.
+const ENTRY_PROGRAM_ENV: &str = "HIVE_RUN_ENTRY_PROGRAM";
+const ENTRY_CGROUP_ENV: &str = "HIVE_RUN_ENTRY_CGROUP";
+const ENTRY_OOM_SCORE_ADJ_ENV: &str = "HIVE_RUN_ENTRY_OOM_SCORE_ADJ";
+
+/// How long a run's cgroup may stay populated after the run ended before
+/// removal is left to the next daemon start.
+const RELEASE_GRACE: Duration = Duration::from_secs(30);
+const RELEASE_POLL: Duration = Duration::from_millis(250);
+
+/// The `<unit>/<kind>/` node run cgroups are created under.
+#[derive(Debug)]
+pub struct RunCgroups {
+    parent: PathBuf,
+}
+
+impl RunCgroups {
+    /// Set up `<unit>/<kind>/` for this process's unit. `None`, with one
+    /// warning, when this process is not in a delegated `daemon/` leaf on
+    /// cgroup v2 (a test, a dev shell, a unit without `Delegate=`): runs then
+    /// stay in the daemon's own cgroup.
+    #[must_use]
+    pub fn init(kind: &str) -> Option<Self> {
+        let result = std::fs::read_to_string("/proc/self/cgroup")
+            .and_then(|own| Self::init_at(Path::new(CGROUP_ROOT), &own, kind));
+        match result {
+            Ok(cgroups) => {
+                tracing::info!(parent = %cgroups.parent.display(), "run cgroups ready");
+                Some(cgroups)
+            }
+            Err(e) => {
+                tracing::warn!(
+                    error = %e,
+                    "run cgroups unavailable; runs share the daemon's cgroup"
+                );
+                None
+            }
+        }
+    }
+
+    /// [`RunCgroups::init`] against cgroup2 mounted at `root`, with
+    /// `proc_self_cgroup` the content of `/proc/self/cgroup`.
+    fn init_at(root: &Path, proc_self_cgroup: &str, kind: &str) -> io::Result<Self> {
+        let own = proc_self_cgroup
+            .lines()
+            .find_map(|line| line.strip_prefix("0::"))
+            .ok_or_else(|| io::Error::other("no cgroup v2 entry in /proc/self/cgroup"))?;
+        let own = root.join(own.trim().trim_start_matches('/'));
+        if own.file_name().and_then(|n| n.to_str()) != Some(DAEMON_SUBGROUP) {
+            return Err(io::Error::other(format!(
+                "{} is not a delegated `{DAEMON_SUBGROUP}` leaf",
+                own.display()
+            )));
+        }
+        let unit = own
+            .parent()
+            .ok_or_else(|| io::Error::other("daemon leaf has no parent"))?;
+        std::fs::write(unit.join("cgroup.subtree_control"), CONTROLLERS)?;
+        let parent = unit.join(kind);
+        match std::fs::create_dir(&parent) {
+            Ok(()) => {}
+            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {}
+            Err(e) => return Err(e),
+        }
+        std::fs::write(parent.join("cgroup.subtree_control"), CONTROLLERS)?;
+        // Runs that outlived the previous daemon's grace leave empty dirs;
+        // a populated one belongs to a process still alive and stays.
+        for entry in std::fs::read_dir(&parent)?.flatten() {
+            if entry.file_type().is_ok_and(|t| t.is_dir()) {
+                let _ = std::fs::remove_dir(entry.path());
+            }
+        }
+        Ok(Self { parent })
+    }
+
+    /// Create the run cgroup `name`, with `memory.oom.group` set.
+    ///
+    /// # Errors
+    ///
+    /// `AlreadyExists` when a populated cgroup of that name is still there,
+    /// or the underlying failure creating it or writing `memory.oom.group`.
+    pub fn create(&self, name: &str) -> io::Result<RunCgroup> {
+        let dir = self.parent.join(name);
+        match std::fs::create_dir(&dir) {
+            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {
+                std::fs::remove_dir(&dir).map_err(|_| e)?;
+                std::fs::create_dir(&dir)?;
+            }
+            other => other?,
+        }
+        let cgroup = RunCgroup { dir };
+        std::fs::write(cgroup.dir.join("memory.oom.group"), "1")?;
+        Ok(cgroup)
+    }
+}
+
+/// One run's cgroup. Dropping it removes the directory once the run's last
+/// process has left, waiting up to 30 seconds on a background thread.
+#[derive(Debug)]
+pub struct RunCgroup {
+    dir: PathBuf,
+}
+
+impl RunCgroup {
+    /// The run's `cgroup.procs`, open for [`enter_in_pre_exec`]. Keep it
+    /// open until the spawn has returned.
+    ///
+    /// # Errors
+    ///
+    /// The failure opening the file.
+    pub fn procs(&self) -> io::Result<File> {
+        File::options()
+            .write(true)
+            .open(self.dir.join("cgroup.procs"))
+    }
+
+    /// The program and extra environment that run `program` inside this
+    /// cgroup with `oom_score_adj`: this daemon's own binary, whose `main`
+    /// starts with [`exec_entry`].
+    #[must_use]
+    pub fn entry(&self, program: &Path, oom_score_adj: i32) -> (PathBuf, Vec<(String, String)>) {
+        (
+            PathBuf::from("/proc/self/exe"),
+            vec![
+                (
+                    ENTRY_PROGRAM_ENV.to_owned(),
+                    program.to_string_lossy().into_owned(),
+                ),
+                (
+                    ENTRY_CGROUP_ENV.to_owned(),
+                    self.dir.to_string_lossy().into_owned(),
+                ),
+                (
+                    ENTRY_OOM_SCORE_ADJ_ENV.to_owned(),
+                    oom_score_adj.to_string(),
+                ),
+            ],
+        )
+    }
+}
+
+impl Drop for RunCgroup {
+    fn drop(&mut self) {
+        match std::fs::remove_dir(&self.dir) {
+            Err(e) if e.raw_os_error() == Some(libc::EBUSY) => {
+                let dir = std::mem::take(&mut self.dir);
+                std::thread::spawn(move || release_when_empty(&dir));
+            }
+            Err(e) if e.kind() != io::ErrorKind::NotFound => {
+                tracing::warn!(dir = %self.dir.display(), error = %e, "removing run cgroup failed");
+            }
+            _ => {}
+        }
+    }
+}
+
+/// Retry removing a still-populated run cgroup until it empties or
+/// [`RELEASE_GRACE`] runs out; past that the next daemon start removes it.
+fn release_when_empty(dir: &Path) {
+    let deadline = std::time::Instant::now() + RELEASE_GRACE;
+    while std::time::Instant::now() < deadline {
+        std::thread::sleep(RELEASE_POLL);
+        match std::fs::remove_dir(dir) {
+            Err(e) if e.raw_os_error() == Some(libc::EBUSY) => {}
+            _ => return,
+        }
+    }
+    tracing::info!(dir = %dir.display(), "run cgroup still populated after the run ended; left in place");
+}
+
+/// Move the calling process into the cgroup whose `cgroup.procs` is `procs`.
+/// Only `write(2)`, so it is safe between fork and exec.
+///
+/// # Errors
+///
+/// The `write(2)` failure.
+pub fn enter_in_pre_exec(procs: RawFd) -> io::Result<()> {
+    // SAFETY: a plain write of a static buffer to a caller-owned fd.
+    if unsafe { libc::write(procs, b"0".as_ptr().cast(), 1) } < 0 {
+        return Err(io::Error::last_os_error());
+    }
+    Ok(())
+}
+
+/// When this process was started by [`RunCgroup::entry`]: enter the run's
+/// cgroup, set its `oom_score_adj`, and exec the run's program with this
+/// process's arguments; on failure print the reason and exit 127. Otherwise
+/// return, doing nothing. Call it first in `main`.
+pub fn exec_entry() {
+    let Some(program) = std::env::var_os(ENTRY_PROGRAM_ENV) else {
+        return;
+    };
+    let err = enter_and_exec(&program);
+    eprintln!("run cgroup entry failed: {err}");
+    std::process::exit(127);
+}
+
+fn enter_and_exec(program: &OsStr) -> io::Error {
+    if let Some(cgroup) = std::env::var_os(ENTRY_CGROUP_ENV)
+        && let Err(e) = std::fs::write(Path::new(&cgroup).join("cgroup.procs"), "0")
+    {
+        return e;
+    }
+    if let Some(adj) = std::env::var_os(ENTRY_OOM_SCORE_ADJ_ENV)
+        && let Err(e) = std::fs::write("/proc/self/oom_score_adj", adj.as_encoded_bytes())
+    {
+        return e;
+    }
+    std::process::Command::new(program)
+        .args(std::env::args_os().skip(1))
+        .env_remove(ENTRY_PROGRAM_ENV)
+        .env_remove(ENTRY_CGROUP_ENV)
+        .env_remove(ENTRY_OOM_SCORE_ADJ_ENV)
+        .exec()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const KIND: &str = "runs";
+
+    fn unit_tree() -> (tempfile::TempDir, PathBuf) {
+        let root = tempfile::tempdir().expect("tempdir");
+        let unit = root.path().join("system.slice/x.service");
+        std::fs::create_dir_all(unit.join(DAEMON_SUBGROUP)).expect("daemon leaf");
+        (root, unit)
+    }
+
+    const OWN: &str = "0::/system.slice/x.service/daemon\n";
+
+    #[test]
+    fn init_enables_controllers_on_the_unit_and_the_kind_node() {
+        let (root, unit) = unit_tree();
+        let cgroups = RunCgroups::init_at(root.path(), OWN, KIND).expect("init");
+        assert_eq!(cgroups.parent, unit.join(KIND));
+        for node in [&unit, &unit.join(KIND)] {
+            let control =
+                std::fs::read_to_string(node.join("cgroup.subtree_control")).expect("written");
+            assert_eq!(control, CONTROLLERS);
+        }
+    }
+
+    #[test]
+    fn init_refuses_a_process_outside_a_daemon_leaf() {
+        let (root, unit) = unit_tree();
+        let err = RunCgroups::init_at(root.path(), "0::/system.slice/x.service\n", KIND)
+            .expect_err("not a daemon leaf");
+        assert!(err.to_string().contains(DAEMON_SUBGROUP), "{err}");
+        assert!(!unit.join("cgroup.subtree_control").exists());
+    }
+
+    #[test]
+    fn init_refuses_cgroup_v1() {
+        let (root, _unit) = unit_tree();
+        assert!(RunCgroups::init_at(root.path(), "1:memory:/x\n", KIND).is_err());
+    }
+
+    #[test]
+    fn init_removes_empty_leftover_runs_and_keeps_populated_ones() {
+        let (root, unit) = unit_tree();
+        let empty = unit.join(KIND).join("old-0");
+        let populated = unit.join(KIND).join("old-1");
+        std::fs::create_dir_all(&empty).expect("empty");
+        std::fs::create_dir_all(&populated).expect("populated");
+        std::fs::write(populated.join("cgroup.procs"), "42").expect("procs");
+        RunCgroups::init_at(root.path(), OWN, KIND).expect("init");
+        assert!(!empty.exists());
+        assert!(populated.exists());
+    }
+
+    #[test]
+    fn create_sets_oom_group_and_drop_removes_the_dir() {
+        let (root, unit) = unit_tree();
+        let cgroups = RunCgroups::init_at(root.path(), OWN, KIND).expect("init");
+        let run = cgroups.create("alpha-0").expect("create");
+        let dir = unit.join(KIND).join("alpha-0");
+        assert_eq!(run.dir, dir);
+        assert_eq!(
+            std::fs::read_to_string(dir.join("memory.oom.group")).expect("written"),
+            "1"
+        );
+        // cgroupfs drops a cgroup's interface files with the directory; a
+        // plain tempdir needs them gone first.
+        std::fs::remove_file(dir.join("memory.oom.group")).expect("rm");
+        drop(run);
+        assert!(!dir.exists());
+    }
+
+    #[test]
+    fn create_replaces_an_empty_leftover_of_the_same_name() {
+        let (root, unit) = unit_tree();
+        let cgroups = RunCgroups::init_at(root.path(), OWN, KIND).expect("init");
+        std::fs::create_dir(unit.join(KIND).join("t1")).expect("leftover");
+        cgroups.create("t1").expect("create over an empty leftover");
+    }
+
+    #[test]
+    fn entry_names_this_binary_and_the_run() {
+        let (root, _unit) = unit_tree();
+        let cgroups = RunCgroups::init_at(root.path(), OWN, KIND).expect("init");
+        let run = cgroups.create("alpha-0").expect("create");
+        let (program, env) = run.entry(Path::new("claude"), 1000);
+        assert_eq!(program, Path::new("/proc/self/exe"));
+        let get = |k: &str| env.iter().find(|(n, _)| n == k).map(|(_, v)| v.as_str());
+        assert_eq!(get(ENTRY_PROGRAM_ENV), Some("claude"));
+        assert_eq!(get(ENTRY_CGROUP_ENV), run.dir.to_str());
+        assert_eq!(get(ENTRY_OOM_SCORE_ADJ_ENV), Some("1000"));
+    }
+
+    #[test]
+    fn enter_in_pre_exec_writes_zero_to_the_fd() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let path = dir.path().join("cgroup.procs");
+        std::fs::write(&path, "").expect("create");
+        let procs = File::options().write(true).open(&path).expect("open");
+        enter_in_pre_exec(std::os::fd::AsRawFd::as_raw_fd(&procs)).expect("write");
+        assert_eq!(std::fs::read_to_string(&path).expect("read"), "0");
+    }
+}