vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

Compare commits

base: vinzenz:afe6026d4484f1b49e86c06a944bd73dc4d796c1
Branches Tags
vinzenz:main
..
compare: vinzenz:baf282131fc25d851262d80f7f873a394310f317
Branches Tags
vinzenz:main
14 changed files with 54 additions and 693 deletions
Download patch file Download diff file Expand all files Collapse all files

2
Cargo.lock generated
View file

@ -1756,10 +1756,8 @@ name = "hive-agent-sock"
version = "0.1.0" version = "0.1.0"
dependencies = [ dependencies = [
"hive-sh4re", "hive-sh4re",
"libc",
"serde", "serde",
"serde_json", "serde_json",
"tempfile",
"tracing", "tracing",
] ]

15
docs/scheduler/coordinator.md
View file

@ -479,21 +479,6 @@ changed cap reaches them at the next container start without a rebuild.
`hyperhive-memory-cap-check.service` logs a warning naming the path in the `hyperhive-memory-cap-check.service` logs a warning naming the path in the
container's journal. The container's `MemoryMax=` still applies. container's journal. The container's `MemoryMax=` still applies.
When the container reaches its cap, the kernel kills in this order:
1. a subagent run (`oom_score_adj` +1000);
2. a bash task (0);
3. the harness and the agent's own `claude` (-500).
Each subagent run and each bash task has its own cgroup with
`memory.oom.group=1`, so the kernel kills all of a run or task and nothing
outside it. Inside the container these are
`/sys/fs/cgroup/system.slice/hive-subagent-daemon.service/runs/<name>-<n>/`
and `/sys/fs/cgroup/system.slice/hive-bash-daemon.service/tasks/<id>/`; the
daemon itself sits in `daemon/` beside them. A daemon that can't set these
up logs one warning at start and runs everything in its own cgroup; subagent
runs still get +1000.
<details><summary>Implementation: swarm caps</summary> <details><summary>Implementation: swarm caps</summary>
swarm-controller keeps the caps in the `agent-limits` KV bucket, one key per swarm-controller keeps the caps in the `agent-limits` KV bucket, one key per

5
hive-agent-sock/Cargo.toml
View file

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

11
hive-agent-sock/README.md
View file

@ -2,7 +2,7 @@
What's common across the in-container "agent plugins" — the matrix / bash / What's common across the in-container "agent plugins" — the matrix / bash /
subagent MCP daemons and `forge_notify` today, plus any user-configured MCP subagent MCP daemons and `forge_notify` today, plus any user-configured MCP
server declared in an agent's `agent.nix`. Three things live here: server declared in an agent's `agent.nix`. Two things live here:
## Wire types for the in-agent socket ## Wire types for the in-agent socket
@ -38,12 +38,3 @@ entry into the shape claude expects in a
`availableToSubagents`) — both already depend on this crate for the socket `availableToSubagents`) — both already depend on this crate for the socket
types above, so a new plugin-facing shared type belongs here rather than in types above, so a new plugin-facing shared type belongs here rather than in
a crate of its own. 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.

4
hive-agent-sock/src/lib.rs
View file

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

372
hive-agent-sock/src/run_cgroup.rs
View file

@ -1,372 +0,0 @@
//! 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;
//! - [`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.
//! It also sets the run's `oom_score_adj`, and does that alone when the
//! run has no cgroup.
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` at `oom_score_adj`,
/// inside `cgroup` when there is one: this daemon's own binary, whose `main`
/// starts with [`exec_entry`].
#[must_use]
pub fn entry(
program: &Path,
oom_score_adj: i32,
cgroup: Option<&RunCgroup>,
) -> (PathBuf, Vec<(String, String)>) {
let mut env = vec![
(
ENTRY_PROGRAM_ENV.to_owned(),
program.to_string_lossy().into_owned(),
),
(
ENTRY_OOM_SCORE_ADJ_ENV.to_owned(),
oom_score_adj.to_string(),
),
];
if let Some(cgroup) = cgroup {
env.push((
ENTRY_CGROUP_ENV.to_owned(),
cgroup.dir.to_string_lossy().into_owned(),
));
}
(PathBuf::from("/proc/self/exe"), env)
}
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 [`entry`]: enter the run's
/// cgroup if it has one, 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) = entry(Path::new("claude"), 1000, Some(&run));
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 entry_without_a_cgroup_still_sets_the_score() {
let (program, env) = entry(Path::new("claude"), 1000, None);
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_OOM_SCORE_ADJ_ENV), Some("1000"));
assert_eq!(get(ENTRY_CGROUP_ENV), None);
}
#[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");
}
}

4
hive-bash-mcp/src/main.rs
View file

@ -39,11 +39,9 @@ async fn main() -> Result<()> {
"hive-bash-daemon starting" "hive-bash-daemon starting"
); );
let cgroups = hive_agent_sock::run_cgroup::RunCgroups::init("tasks");
// Start the background runner loop — scans for pending tasks and // Start the background runner loop — scans for pending tasks and
// spawns them, pushing todos to the harness on task transitions. // spawns them, pushing todos to the harness on task transitions.
hive_bash_mcp::runner::spawn_loop(todo_socket, cgroups); hive_bash_mcp::runner::spawn_loop(todo_socket);
// Serve the MCP tools over streamable-http forever. // Serve the MCP tools over streamable-http forever.
hive_bash_mcp::mcp::serve_http(cli.http).await hive_bash_mcp::mcp::serve_http(cli.http).await

51
hive-bash-mcp/src/runner.rs
View file

@ -34,7 +34,6 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
use anyhow::{Result, bail}; use anyhow::{Result, bail};
use hive_agent_sock::Request as TodoReq; use hive_agent_sock::Request as TodoReq;
use hive_agent_sock::run_cgroup::{self, RunCgroup, RunCgroups};
use tokio::io::AsyncWriteExt as _; use tokio::io::AsyncWriteExt as _;
use tokio::sync::Notify; use tokio::sync::Notify;
@ -432,14 +431,13 @@ pub fn kill_task(id: &str, force: bool) -> (bool, bool) {
/// Spawn the background runner loop as a detached tokio task. `socket` is /// Spawn the background runner loop as a detached tokio task. `socket` is
/// the harness's in-agent todo socket (`HIVE_AGENT_SOCKET`), where task /// the harness's in-agent todo socket (`HIVE_AGENT_SOCKET`), where task
/// transitions are pushed as todos. Call once at daemon startup. /// transitions are pushed as todos. Call once at daemon startup.
pub fn spawn_loop(socket: PathBuf, cgroups: Option<RunCgroups>) { pub fn spawn_loop(socket: PathBuf) {
let cgroups = Arc::new(cgroups);
tokio::spawn(async move { tokio::spawn(async move {
run_loop(socket, cgroups).await; run_loop(socket).await;
}); });
} }
async fn run_loop(socket: PathBuf, cgroups: Arc<Option<RunCgroups>>) { async fn run_loop(socket: PathBuf) {
if let Err(e) = std::fs::create_dir_all(paths::tasks_dir()) { if let Err(e) = std::fs::create_dir_all(paths::tasks_dir()) {
tracing::warn!(error = ?e, "bash_runner: create tasks dir failed"); tracing::warn!(error = ?e, "bash_runner: create tasks dir failed");
} }
@ -452,7 +450,7 @@ async fn run_loop(socket: PathBuf, cgroups: Arc<Option<RunCgroups>>) {
let claimed: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new())); let claimed: Arc<Mutex<HashSet<String>>> = Arc::new(Mutex::new(HashSet::new()));
loop { loop {
poll_once(&socket, &claimed, &cgroups); poll_once(&socket, &claimed);
tokio::time::sleep(POLL_INTERVAL).await; tokio::time::sleep(POLL_INTERVAL).await;
} }
} }
@ -543,11 +541,7 @@ async fn mark_interrupted(socket: &Path) {
} }
} }
fn poll_once( fn poll_once(socket: &Path, claimed: &Arc<Mutex<HashSet<String>>>) {
socket: &Path,
claimed: &Arc<Mutex<HashSet<String>>>,
cgroups: &Arc<Option<RunCgroups>>,
) {
let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else { let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else {
return; return;
}; };
@ -573,9 +567,8 @@ fn poll_once(
claimed.lock().unwrap().insert(id.clone()); claimed.lock().unwrap().insert(id.clone());
let socket = socket.to_path_buf(); let socket = socket.to_path_buf();
let claimed = claimed.clone(); let claimed = claimed.clone();
let cgroups = Arc::clone(cgroups);
tokio::spawn(async move { tokio::spawn(async move {
run_task(task, &socket, cgroups.as_ref().as_ref()).await; run_task(task, &socket).await;
claimed.lock().unwrap().remove(&id); claimed.lock().unwrap().remove(&id);
}); });
} }
@ -585,7 +578,7 @@ fn poll_once(
// Task execution // Task execution
// --------------------------------------------------------------------------- // ---------------------------------------------------------------------------
async fn run_task(mut task: TaskFile, socket: &Path, cgroups: Option<&RunCgroups>) { async fn run_task(mut task: TaskFile, socket: &Path) {
let id = task.id.clone(); let id = task.id.clone();
tracing::info!(id = %id, cmd = %task.cmd, "bash_runner: starting task"); tracing::info!(id = %id, cmd = %task.cmd, "bash_runner: starting task");
@ -623,7 +616,6 @@ async fn run_task(mut task: TaskFile, socket: &Path, cgroups: Option<&RunCgroups
}, },
); );
let cgroup = cgroups.and_then(|cgroups| task_cgroup(cgroups, &id));
let outcome = exec_cmd( let outcome = exec_cmd(
&task.cmd, &task.cmd,
&out_path, &out_path,
@ -631,10 +623,8 @@ async fn run_task(mut task: TaskFile, socket: &Path, cgroups: Option<&RunCgroups
task.timeout_secs, task.timeout_secs,
&cancel, &cancel,
&force, &force,
cgroup.as_ref().map(|(_, procs)| procs),
) )
.await; .await;
drop(cgroup);
// Deregister before writing terminal state — the task is no longer killable. // Deregister before writing terminal state — the task is no longer killable.
running().lock().unwrap().remove(&id); running().lock().unwrap().remove(&id);
@ -703,20 +693,6 @@ async fn run_task(mut task: TaskFile, socket: &Path, cgroups: Option<&RunCgroups
.await; .await;
} }
/// `id`'s cgroup and its open `cgroup.procs`, or `None` (logged) when either
/// fails, in which case the task runs in the daemon's cgroup.
fn task_cgroup(cgroups: &RunCgroups, id: &str) -> Option<(RunCgroup, std::fs::File)> {
let opened = cgroups.create(id).and_then(|cgroup| {
let procs = cgroup.procs()?;
Ok((cgroup, procs))
});
opened
.inspect_err(|e| {
tracing::warn!(id = %id, error = %e, "bash_runner: task cgroup unavailable");
})
.ok()
}
/// Run `bash -c cmd` in its own process group, streaming output to files. /// Run `bash -c cmd` in its own process group, streaming output to files.
/// `timeout_secs = None` means no timeout — run until natural exit. /// `timeout_secs = None` means no timeout — run until natural exit.
/// ///
@ -745,7 +721,6 @@ async fn exec_cmd(
timeout_secs: Option<u64>, timeout_secs: Option<u64>,
cancel: &Notify, cancel: &Notify,
force: &AtomicBool, force: &AtomicBool,
cgroup_procs: Option<&std::fs::File>,
) -> Result<ExecOutcome> { ) -> Result<ExecOutcome> {
use tokio::process::Command; use tokio::process::Command;
// `/usr/bin/env bash` rather than a bare `bash`: `/usr/bin/env` is at a // `/usr/bin/env bash` rather than a bare `bash`: `/usr/bin/env` is at a
@ -759,20 +734,14 @@ async fn exec_cmd(
.arg(format!("set -o pipefail; {cmd}")) .arg(format!("set -o pipefail; {cmd}"))
.stdout(std::process::Stdio::piped()) .stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped()); .stderr(std::process::Stdio::piped());
let cgroup_procs = cgroup_procs.map(std::os::fd::AsRawFd::as_raw_fd); // SAFETY: `setpgid`/`nice` are async-signal-safe and only touch the
// SAFETY: `setpgid`/`nice`/`write` are async-signal-safe and only touch // calling (child) process before exec — no allocations, no locks.
// the calling (child) process before exec — no allocations, no locks.
// `setpgid(0, 0)` makes the child its own process-group leader so the // `setpgid(0, 0)` makes the child its own process-group leader so the
// whole subtree can be signalled as a group (`kill(-pgid, …)`). // whole subtree can be signalled as a group (`kill(-pgid, …)`).
// Entering the task's cgroup before exec means nothing the task forks
// starts outside it.
unsafe { unsafe {
cmd_builder.pre_exec(move || { cmd_builder.pre_exec(|| {
libc::setpgid(0, 0); libc::setpgid(0, 0);
libc::nice(10); libc::nice(10);
if let Some(procs) = cgroup_procs {
run_cgroup::enter_in_pre_exec(procs)?;
}
Ok(()) Ok(())
}); });
} }

12
hive-subagent-mcp/src/main.rs
View file

@ -22,15 +22,8 @@ struct Cli {
http: std::net::SocketAddr, http: std::net::SocketAddr,
} }
fn main() -> Result<()> {
// Each subagent run starts as this binary, which enters the run's cgroup
// and execs the run's program (`RunCgroup::entry`).
hive_agent_sock::run_cgroup::exec_entry();
serve()
}
#[tokio::main] #[tokio::main]
async fn serve() -> Result<()> { async fn main() -> Result<()> {
tracing_subscriber::fmt() tracing_subscriber::fmt()
.with_env_filter( .with_env_filter(
tracing_subscriber::EnvFilter::try_from_env("RUST_LOG") tracing_subscriber::EnvFilter::try_from_env("RUST_LOG")
@ -64,8 +57,7 @@ async fn serve() -> Result<()> {
hive_subagent_mcp::session::State::new(todo_socket, signal_base) hive_subagent_mcp::session::State::new(todo_socket, signal_base)
.on_runtime(runtime) .on_runtime(runtime)
.publishing_to(hive_subagent_mcp::swarm_term::Publisher::spawn()) .publishing_to(hive_subagent_mcp::swarm_term::Publisher::spawn())
.reading_models_from(hive_subagent_mcp::swarm_models::Reader::spawn()) .reading_models_from(hive_subagent_mcp::swarm_models::Reader::spawn()),
.entering_runs(hive_agent_sock::run_cgroup::RunCgroups::init("runs")),
); );
// Serve the MCP tools over streamable-http forever. No background poll // Serve the MCP tools over streamable-http forever. No background poll

147
hive-subagent-mcp/src/session.rs
View file

@ -89,11 +89,9 @@
use std::collections::HashMap; use std::collections::HashMap;
use std::os::unix::process::ExitStatusExt as _; use std::os::unix::process::ExitStatusExt as _;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, PoisonError}; use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant}; use std::time::{Duration, Instant};
use hive_agent_sock::run_cgroup::{self, RunCgroup, RunCgroups};
use hive_claude::{Attach, Claude, Config, SessionStore}; use hive_claude::{Attach, Claude, Config, SessionStore};
use hive_runtime::{ use hive_runtime::{
AcpCommand, AcpRuntime, PermissionAsk, PermissionPolicy, Runtime as _, RuntimeSpec, TurnStop, AcpCommand, AcpRuntime, PermissionAsk, PermissionPolicy, Runtime as _, RuntimeSpec, TurnStop,
@ -407,14 +405,6 @@ pub struct State {
/// The agent's own model list, checked when the swarm sets none: /// The agent's own model list, checked when the swarm sets none:
/// [`available_models`] outside tests. /// [`available_models`] outside tests.
local_models: fn() -> Option<String>, local_models: fn() -> Option<String>,
/// Whether runs start through this binary's entry wrapper
/// ([`State::entering_runs`]).
runs_entered: bool,
/// Where each run gets its own cgroup, when this daemon was delegated
/// one.
run_cgroups: Option<RunCgroups>,
/// The `<n>` of the next run cgroup `<name>-<n>`.
run_seq: AtomicU64,
} }
/// Which opaque URL segment belongs to which session — the whole of a /// Which opaque URL segment belongs to which session — the whole of a
@ -470,9 +460,6 @@ impl State {
term: None, term: None,
models: None, models: None,
local_models: available_models, local_models: available_models,
runs_entered: false,
run_cgroups: None,
run_seq: AtomicU64::new(0),
} }
} }
@ -491,55 +478,6 @@ impl State {
self self
} }
/// Start every subagent run through this binary's entry wrapper, whose
/// `main` calls `run_cgroup::exec_entry` first: at `oom_score_adj` +1000
/// always, and in its own cgroup under `cgroups` when there are any.
#[must_use]
pub fn entering_runs(mut self, cgroups: Option<RunCgroups>) -> Self {
self.runs_entered = true;
self.run_cgroups = cgroups;
self
}
/// A new cgroup for one run of `name`; `None` without run cgroups, or
/// (logged) when creating it failed, and the run then stays in the
/// daemon's cgroup.
fn run_cgroup(&self, name: &str) -> Option<RunCgroup> {
let cgroups = self.run_cgroups.as_ref()?;
let n = self.run_seq.fetch_add(1, Ordering::Relaxed);
cgroups
.create(&format!("{name}-{n}"))
.inspect_err(|e| tracing::warn!(name, error = %e, "subagent: run cgroup unavailable"))
.ok()
}
/// `config` with its program started through the entry wrapper, at
/// [`RUN_OOM_SCORE_ADJ`] and inside `cgroup` if there is one; unchanged
/// without [`State::entering_runs`]. `hive-claude` builds the `Command`
/// itself and offers no `pre_exec`, hence the wrapper.
fn entered_config(&self, config: &Config, cgroup: Option<&RunCgroup>) -> Config {
let mut config = config.clone();
if self.runs_entered {
// `hive-claude`'s own default program.
let program = config.program.take().unwrap_or_else(|| "claude".into());
let (entry, env) = run_cgroup::entry(&program, RUN_OOM_SCORE_ADJ, cgroup);
config.program = Some(entry);
config.env.extend(env);
}
config
}
/// [`State::entered_config`] for an ACP agent's command.
fn entered_command(&self, mut command: AcpCommand, cgroup: Option<&RunCgroup>) -> AcpCommand {
if self.runs_entered {
let (entry, env) =
run_cgroup::entry(Path::new(&command.command), RUN_OOM_SCORE_ADJ, cgroup);
command.command = entry.to_string_lossy().into_owned();
command.env.extend(env);
}
command
}
/// Refuse a `model` outside the list in effect for this agent (see /// Refuse a `model` outside the list in effect for this agent (see
/// `check_model`). The swarm key is read only for a named model. /// `check_model`). The swarm key is read only for a named model.
async fn allow_model(&self, model: Option<&str>) -> anyhow::Result<()> { async fn allow_model(&self, model: Option<&str>) -> anyhow::Result<()> {
@ -1456,12 +1394,7 @@ fn start_on_runtime(
trigger: String, trigger: String,
) -> anyhow::Result<String> { ) -> anyhow::Result<String> {
if let SubagentRuntime::Acp { command, sessions } = &state.runtime { if let SubagentRuntime::Acp { command, sessions } = &state.runtime {
let cgroup = state.run_cgroup(name); let runtime = acp_runtime(command.clone(), sessions, name);
let runtime = acp_runtime(
state.entered_command(command.clone(), cgroup.as_ref()),
sessions,
name,
);
if let Some(archived) = runtime if let Some(archived) = runtime
.archive() .archive()
.map_err(|e| anyhow::anyhow!("archiving the prior `{name}` session failed: {e}"))? .map_err(|e| anyhow::anyhow!("archiving the prior `{name}` session failed: {e}"))?
@ -1476,7 +1409,7 @@ fn start_on_runtime(
system_prompt_file: system_prompt, system_prompt_file: system_prompt,
..config ..config
}; };
return spawn_and_track_acp(state, name, runtime, cgroup, config, trigger, None); return spawn_and_track_acp(state, name, runtime, config, trigger, None);
} }
let store = build_store(&config)?; let store = build_store(&config)?;
if store.find_by_title(name).is_some() { if store.find_by_title(name).is_some() {
@ -1675,17 +1608,11 @@ fn continue_on_runtime(
); );
} }
state.begin_continue(name, dir); state.begin_continue(name, dir);
let cgroup = state.run_cgroup(name); let runtime = acp_runtime(command.clone(), sessions, name);
let runtime = acp_runtime(
state.entered_command(command.clone(), cgroup.as_ref()),
sessions,
name,
);
return spawn_and_track_acp( return spawn_and_track_acp(
state, state,
name, name,
runtime, runtime,
cgroup,
config, config,
prompt, prompt,
Some(Arc::clone(verdict)), Some(Arc::clone(verdict)),
@ -1809,10 +1736,6 @@ fn spawn_and_track(
prompt: String, prompt: String,
verdict: Option<VerdictTx>, verdict: Option<VerdictTx>,
) -> anyhow::Result<String> { ) -> anyhow::Result<String> {
// Every turn of the run, and all they start, share one cgroup, released
// when the background task below ends.
let cgroup = state.run_cgroup(name);
let config = &state.entered_config(config, cgroup.as_ref());
let running = Claude::spawn(config, attach) let running = Claude::spawn(config, attach)
.map_err(|e| anyhow::anyhow!("starting the subagent process failed: {e}"))?; .map_err(|e| anyhow::anyhow!("starting the subagent process failed: {e}"))?;
state.track(name, running.cancel_handle().into()); state.track(name, running.cancel_handle().into());
@ -1838,7 +1761,6 @@ fn spawn_and_track(
let state = Arc::clone(state); let state = Arc::clone(state);
let task_name = name.to_owned(); let task_name = name.to_owned();
tokio::spawn(async move { tokio::spawn(async move {
let _cgroup = cgroup;
let mut running = running; let mut running = running;
let mut prompt = prompt; let mut prompt = prompt;
loop { loop {
@ -1931,12 +1853,11 @@ async fn after_turn(
/// `runtime`, whose agent process is spawned by the first turn and killed /// `runtime`, whose agent process is spawned by the first turn and killed
/// when the run ends and the runtime is dropped. Returns before the agent /// when the run ends and the runtime is dropped. Returns before the agent
/// exists, so a spawn failure is the first turn's `Failed` end rather than /// exists, so a spawn failure is the first turn's `Failed` end rather than
/// this call's error. The run's `cgroup` is released after the runtime. /// this call's error.
fn spawn_and_track_acp( fn spawn_and_track_acp(
state: &Arc<State>, state: &Arc<State>,
name: &str, name: &str,
runtime: AcpRuntime<hive_claude::NeverCompact>, runtime: AcpRuntime<hive_claude::NeverCompact>,
cgroup: Option<RunCgroup>,
config: Config, config: Config,
prompt: String, prompt: String,
verdict: Option<VerdictTx>, verdict: Option<VerdictTx>,
@ -1963,14 +1884,12 @@ fn spawn_and_track_acp(
Err(e) => TurnEnd::Failed(format!("ACP error: {e}")), Err(e) => TurnEnd::Failed(format!("ACP error: {e}")),
}; };
let Some(next) = after_turn(&state, &task_name, end, verdict.as_ref()).await else { let Some(next) = after_turn(&state, &task_name, end, verdict.as_ref()).await else {
break; return;
}; };
state.track(&task_name, canceller.clone().into()); state.track(&task_name, canceller.clone().into());
state.note_event(&task_name); state.note_event(&task_name);
prompt = next; prompt = next;
} }
drop(runtime);
drop(cgroup);
}); });
Ok(format!("subagent `{name}` started")) Ok(format!("subagent `{name}` started"))
@ -1982,11 +1901,6 @@ fn acp_session_file(sessions: &Path, name: &str) -> PathBuf {
sessions.join(format!("subagent-acp-session-{name}")) sessions.join(format!("subagent-acp-session-{name}"))
} }
/// A subagent run's `oom_score_adj`: the maximum, so under memory pressure
/// the kernel takes a subagent run before a bash task (0) or the agent's
/// own session (-500, `nix/agent-modules/agent-service.nix`).
const RUN_OOM_SCORE_ADJ: i32 = 1000;
/// The ACP runtime `name`'s run drives, keeping its session id in /// The ACP runtime `name`'s run drives, keeping its session id in
/// [`acp_session_file`] and answering permission requests per /// [`acp_session_file`] and answering permission requests per
/// [`acp_permits`]. /// [`acp_permits`].
@ -5258,55 +5172,4 @@ echo '{"type":"result","subtype":"success","is_error":false}'
}; };
assert!(acp_permits(&mcp, none)); assert!(acp_permits(&mcp, none));
} }
fn env_of<'a>(env: &'a [(String, String)], name: &str) -> Option<&'a str> {
env.iter().find(|(n, _)| n == name).map(|(_, v)| v.as_str())
}
#[test]
fn a_run_without_a_cgroup_still_starts_at_the_subagent_score() {
let state = State::new(PathBuf::from("/dev/null"), signal_url()).entering_runs(None);
let config = state.entered_config(&Config::default(), None);
assert_eq!(config.program.as_deref(), Some(Path::new("/proc/self/exe")));
assert_eq!(
env_of(&config.env, "HIVE_RUN_ENTRY_PROGRAM"),
Some("claude")
);
assert_eq!(
env_of(&config.env, "HIVE_RUN_ENTRY_OOM_SCORE_ADJ"),
Some("1000")
);
assert_eq!(env_of(&config.env, "HIVE_RUN_ENTRY_CGROUP"), None);
let command = AcpCommand {
command: "agent".to_owned(),
args: Vec::new(),
env: std::collections::BTreeMap::new(),
api_key_env: None,
};
let command = state.entered_command(command, None);
assert_eq!(command.command, "/proc/self/exe");
assert_eq!(
command
.env
.get("HIVE_RUN_ENTRY_PROGRAM")
.map(String::as_str),
Some("agent")
);
assert_eq!(
command
.env
.get("HIVE_RUN_ENTRY_OOM_SCORE_ADJ")
.map(String::as_str),
Some("1000")
);
}
#[test]
fn runs_are_started_plain_unless_the_daemon_enters_them() {
let state = State::new(PathBuf::from("/dev/null"), signal_url());
let config = state.entered_config(&Config::default(), None);
assert_eq!(config.program, None);
assert!(config.env.is_empty());
}
} }

41
hive-subagent-mcp/tests/run_entry.rs
View file

@ -1,41 +0,0 @@
//! The daemon binary as a run's entry wrapper (`run_cgroup::RunCgroup::entry`
//! sets these variables; its unit test pins the names).
use std::process::Command;
#[test]
fn the_entry_enters_the_cgroup_raises_the_score_and_execs_the_program() {
let cgroup = std::env::temp_dir().join(format!("hive-run-entry-{}", std::process::id()));
std::fs::create_dir_all(&cgroup).expect("fake cgroup");
let out = Command::new(env!("CARGO_BIN_EXE_hive-subagent-daemon"))
// Stands in for `claude`: prints what it was started with.
.env("HIVE_RUN_ENTRY_PROGRAM", "/bin/sh")
.env("HIVE_RUN_ENTRY_CGROUP", &cgroup)
.env("HIVE_RUN_ENTRY_OOM_SCORE_ADJ", "1000")
.args([
"-c",
r#"printf '%s|%s|%s' "$(cat /proc/self/oom_score_adj)" "${HIVE_RUN_ENTRY_CGROUP-unset}" "$1""#,
"sh",
"an-arg",
])
.output()
.expect("the binary runs");
let procs = std::fs::read_to_string(cgroup.join("cgroup.procs"));
std::fs::remove_dir_all(&cgroup).ok();
assert!(out.status.success(), "{out:?}");
assert_eq!(String::from_utf8_lossy(&out.stdout), "1000|unset|an-arg");
assert_eq!(procs.expect("cgroup.procs written"), "0");
}
#[test]
fn the_entry_without_a_cgroup_still_raises_the_score() {
let out = Command::new(env!("CARGO_BIN_EXE_hive-subagent-daemon"))
.env("HIVE_RUN_ENTRY_PROGRAM", "/bin/sh")
.env_remove("HIVE_RUN_ENTRY_CGROUP")
.env("HIVE_RUN_ENTRY_OOM_SCORE_ADJ", "1000")
.args(["-c", "cat /proc/self/oom_score_adj"])
.output()
.expect("the binary runs");
assert!(out.status.success(), "{out:?}");
assert_eq!(String::from_utf8_lossy(&out.stdout).trim(), "1000");
}

4
nix/agent-modules/agent-service.nix
View file

@ -539,8 +539,8 @@ in
# A LOWER OOMScoreAdjust means LESS likely to be killed, so this # A LOWER OOMScoreAdjust means LESS likely to be killed, so this
# negative value puts the harness — and the agent's own `claude`, # negative value puts the harness — and the agent's own `claude`,
# which it spawns as a child and which therefore inherits the # which it spawns as a child and which therefore inherits the
# value — last in line under memory pressure, behind each subagent # value — last in line under memory pressure, behind
# run (+1000) and bash task (0) (see ./mcp.nix). Losing a # `hive-subagent-daemon`'s `+500` (see ./mcp.nix). Losing a
# subagent costs one restartable task; losing this costs the # subagent costs one restartable task; losing this costs the
# session that was supervising it. Not `-1000`, which would # session that was supervising it. Not `-1000`, which would
# exempt the harness entirely and leave the kernel with nothing # exempt the harness entirely and leave the kernel with nothing

26
nix/agent-modules/mcp.nix
View file

@ -338,11 +338,6 @@ in
# take every running sibling task with it. `continue` scopes the # take every running sibling task with it. `continue` scopes the
# loss to the process the kernel chose. # loss to the process the kernel chose.
OOMPolicy = "continue"; OOMPolicy = "continue";
# The daemon runs each task in its own `tasks/<id>/` cgroup with
# `memory.oom.group=1`, so an OOM kill takes one whole task. The
# subgroup name is `DAEMON_SUBGROUP` in hive-agent-sock's `run_cgroup`.
Delegate = "memory pids";
DelegateSubgroup = "daemon";
User = userName; User = userName;
Group = userName; Group = userName;
}; };
@ -441,15 +436,18 @@ in
# `start`/`continue`/`status`/`interrupt` with no stdio fallback. # `start`/`continue`/`status`/`interrupt` with no stdio fallback.
Restart = "always"; Restart = "always";
RestartSec = 3; RestartSec = 3;
# The daemon runs each subagent run in its own `runs/<name>-<n>/` # A HIGHER OOMScoreAdjust means MORE likely to be killed: the
# cgroup with `memory.oom.group=1` and `oom_score_adj` +1000, so an # kernel adds it to the badness score it derives from the
# OOM kill takes one whole run, ahead of a bash task (0) and # process's memory footprint, then kills the highest scorer. So
# `hive-agent` (-500). No `OOMScoreAdjust=` here: the daemon stays # the positive value here and the negative one on `hive-agent`
# at 0, and a unit value would also be the floor no run could go # put this unit first in line — a subagent turn can be retried,
# below. The subgroup name is `DAEMON_SUBGROUP` in hive-agent-sock's # the agent's own turn is the thing everything else hangs off.
# `run_cgroup`. # `+500` against `hive-agent`'s `-500` is a full half of the
Delegate = "memory pids"; # score range apart, enough that a fat subagent outranks a fatter
DelegateSubgroup = "daemon"; # harness rather than merely tying with it. Every nested `claude`
# inherits the value from the daemon, so the ordering covers the
# whole subtree and not just this process.
OOMScoreAdjust = 500;
# systemd's default is `stop`, which tears the whole unit down as # systemd's default is `stop`, which tears the whole unit down as
# soon as the kernel kills *any* process in its cgroup — so one # soon as the kernel kills *any* process in its cgroup — so one
# over-large subagent takes the daemon and every sibling session # over-large subagent takes the daemon and every sibling session

53
nix/module-eval/agent-memory.nix
View file

@ -55,45 +55,34 @@ let
&& lib.elem "multi-user.target" agentDefault.systemd.services.hyperhive-memory-cap-check.wantedBy; && lib.elem "multi-user.target" agentDefault.systemd.services.hyperhive-memory-cap-check.wantedBy;
} }
{ {
# A HIGHER OOMScoreAdjust is a MORE likely victim. The subagent daemon # The sign is the whole property, and it is easy to write backwards:
# puts each run at +1000 itself and bash tasks stay at 0, so the harness # a HIGHER OOMScoreAdjust is a MORE likely victim. So the subagent
# must sit below 0 for the agent's own session to go last. Neither # daemon must be strictly above zero and the harness strictly below
# daemon unit carries a value: it would make the idle daemon a likelier # it — swap the two and the kernel takes the agent's own turn first,
# victim than a bash task, and a floor no run under it could go below. # which is worse than setting nothing at all. Both nested `claude`
name = "the OOM killer takes the agent's own session last"; # processes inherit their unit's value, so ordering the units orders
# the sessions. Held as an inequality rather than two constants: what
# must not drift is the order, not the magnitudes.
name = "the OOM killer prefers a subagent over the agent's own session";
ok = ok =
(agentHarness agentDefault).serviceConfig.OOMScoreAdjust < 0 let
&& !((agentSubagentDaemon agentDefault).serviceConfig ? OOMScoreAdjust) sub = (agentSubagentDaemon agentDefault).serviceConfig.OOMScoreAdjust;
&& !((agentBashDaemon agentDefault).serviceConfig ? OOMScoreAdjust); own = (agentHarness agentDefault).serviceConfig.OOMScoreAdjust;
in
sub > 0 && own < 0 && sub > own;
} }
{ {
# Both daemons create a cgroup per run under their own; without the # The pair the case above only makes sense with: preferring this unit
# delegation they cannot, and without the `daemon` leaf (the name # as the victim is an improvement only if losing one subagent isn't
# hive-agent-sock's `run_cgroup` looks for) they refuse to, so runs # losing all of them. systemd's default `stop` would take the daemon
# would share the daemon's cgroup again. # and every sibling session down with whichever process the kernel
name = "the subagent and bash daemons are delegated their cgroup"; # chose, which is the blast radius that made the preference a bad
ok = # trade in the first place.
lib.all
(
unit:
unit.serviceConfig.Delegate or null == "memory pids"
&& unit.serviceConfig.DelegateSubgroup or null == "daemon"
)
[
(agentSubagentDaemon agentDefault)
(agentBashDaemon agentDefault)
];
}
{
# Preferring a subagent run as the victim is an improvement only if
# losing one run isn't losing all of them. systemd's default `stop`
# would take the daemon and every sibling session down with whichever
# process the kernel chose.
name = "one subagent losing the OOM draw does not stop the daemon"; name = "one subagent losing the OOM draw does not stop the daemon";
ok = (agentSubagentDaemon agentDefault).serviceConfig.OOMPolicy == "continue"; ok = (agentSubagentDaemon agentDefault).serviceConfig.OOMPolicy == "continue";
} }
{ {
# Every bash task runs under the daemon's unit; under `stop` one # Each bash task is a process in the daemon's cgroup; under `stop` one
# OOM-killed task would end every running sibling task. # OOM-killed task would end every running sibling task.
name = "one bash task losing the OOM draw does not stop the daemon"; name = "one bash task losing the OOM draw does not stop the daemon";
ok = (agentBashDaemon agentDefault).serviceConfig.OOMPolicy or null == "continue"; ok = (agentBashDaemon agentDefault).serviceConfig.OOMPolicy or null == "continue";