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"
dependencies = [
"hive-sh4re",
"libc",
"serde",
"serde_json",
"tempfile",
"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
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>
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
[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"

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

@ -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`. Three things live here:
server declared in an agent's `agent.nix`. Two things live here:
## 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
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.

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`
//! (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};
@ -30,7 +27,6 @@ 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.

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"
);
let cgroups = hive_agent_sock::run_cgroup::RunCgroups::init("tasks");
// Start the background runner loop — scans for pending tasks and
// 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.
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 hive_agent_sock::Request as TodoReq;
use hive_agent_sock::run_cgroup::{self, RunCgroup, RunCgroups};
use tokio::io::AsyncWriteExt as _;
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
/// the harness's in-agent todo socket (`HIVE_AGENT_SOCKET`), where task
/// transitions are pushed as todos. Call once at daemon startup.
pub fn spawn_loop(socket: PathBuf, cgroups: Option<RunCgroups>) {
let cgroups = Arc::new(cgroups);
pub fn spawn_loop(socket: PathBuf) {
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()) {
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()));
loop {
poll_once(&socket, &claimed, &cgroups);
poll_once(&socket, &claimed);
tokio::time::sleep(POLL_INTERVAL).await;
}
}
@ -543,11 +541,7 @@ async fn mark_interrupted(socket: &Path) {
}
}
fn poll_once(
socket: &Path,
claimed: &Arc<Mutex<HashSet<String>>>,
cgroups: &Arc<Option<RunCgroups>>,
) {
fn poll_once(socket: &Path, claimed: &Arc<Mutex<HashSet<String>>>) {
let Ok(rd) = std::fs::read_dir(paths::tasks_dir()) else {
return;
};
@ -573,9 +567,8 @@ fn poll_once(
claimed.lock().unwrap().insert(id.clone());
let socket = socket.to_path_buf();
let claimed = claimed.clone();
let cgroups = Arc::clone(cgroups);
tokio::spawn(async move {
run_task(task, &socket, cgroups.as_ref().as_ref()).await;
run_task(task, &socket).await;
claimed.lock().unwrap().remove(&id);
});
}
@ -585,7 +578,7 @@ fn poll_once(
// 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();
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(
&task.cmd,
&out_path,
@ -631,10 +623,8 @@ async fn run_task(mut task: TaskFile, socket: &Path, cgroups: Option<&RunCgroups
task.timeout_secs,
&cancel,
&force,
cgroup.as_ref().map(|(_, procs)| procs),
)
.await;
drop(cgroup);
// Deregister before writing terminal state — the task is no longer killable.
running().lock().unwrap().remove(&id);
@ -703,20 +693,6 @@ async fn run_task(mut task: TaskFile, socket: &Path, cgroups: Option<&RunCgroups
.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.
/// `timeout_secs = None` means no timeout — run until natural exit.
///
@ -745,7 +721,6 @@ async fn exec_cmd(
timeout_secs: Option<u64>,
cancel: &Notify,
force: &AtomicBool,
cgroup_procs: Option<&std::fs::File>,
) -> Result<ExecOutcome> {
use tokio::process::Command;
// `/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}"))
.stdout(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`/`write` are async-signal-safe and only touch
// the calling (child) process before exec — no allocations, no locks.
// SAFETY: `setpgid`/`nice` are async-signal-safe and only touch the
// calling (child) process before exec — no allocations, no locks.
// `setpgid(0, 0)` makes the child its own process-group leader so the
// 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 {
cmd_builder.pre_exec(move || {
cmd_builder.pre_exec(|| {
libc::setpgid(0, 0);
libc::nice(10);
if let Some(procs) = cgroup_procs {
run_cgroup::enter_in_pre_exec(procs)?;
}
Ok(())
});
}

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

@ -22,15 +22,8 @@ struct Cli {
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]
async fn serve() -> Result<()> {
async fn main() -> Result<()> {
tracing_subscriber::fmt()
.with_env_filter(
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)
.on_runtime(runtime)
.publishing_to(hive_subagent_mcp::swarm_term::Publisher::spawn())
.reading_models_from(hive_subagent_mcp::swarm_models::Reader::spawn())
.entering_runs(hive_agent_sock::run_cgroup::RunCgroups::init("runs")),
.reading_models_from(hive_subagent_mcp::swarm_models::Reader::spawn()),
);
// 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::os::unix::process::ExitStatusExt as _;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant};
use hive_agent_sock::run_cgroup::{self, RunCgroup, RunCgroups};
use hive_claude::{Attach, Claude, Config, SessionStore};
use hive_runtime::{
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:
/// [`available_models`] outside tests.
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
@ -470,9 +460,6 @@ impl State {
term: None,
models: None,
local_models: available_models,
runs_entered: false,
run_cgroups: None,
run_seq: AtomicU64::new(0),
}
}
@ -491,55 +478,6 @@ impl State {
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
/// `check_model`). The swarm key is read only for a named model.
async fn allow_model(&self, model: Option<&str>) -> anyhow::Result<()> {
@ -1456,12 +1394,7 @@ fn start_on_runtime(
trigger: String,
) -> anyhow::Result<String> {
if let SubagentRuntime::Acp { command, sessions } = &state.runtime {
let cgroup = state.run_cgroup(name);
let runtime = acp_runtime(
state.entered_command(command.clone(), cgroup.as_ref()),
sessions,
name,
);
let runtime = acp_runtime(command.clone(), sessions, name);
if let Some(archived) = runtime
.archive()
.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,
..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)?;
if store.find_by_title(name).is_some() {
@ -1675,17 +1608,11 @@ fn continue_on_runtime(
);
}
state.begin_continue(name, dir);
let cgroup = state.run_cgroup(name);
let runtime = acp_runtime(
state.entered_command(command.clone(), cgroup.as_ref()),
sessions,
name,
);
let runtime = acp_runtime(command.clone(), sessions, name);
return spawn_and_track_acp(
state,
name,
runtime,
cgroup,
config,
prompt,
Some(Arc::clone(verdict)),
@ -1809,10 +1736,6 @@ fn spawn_and_track(
prompt: String,
verdict: Option<VerdictTx>,
) -> 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)
.map_err(|e| anyhow::anyhow!("starting the subagent process failed: {e}"))?;
state.track(name, running.cancel_handle().into());
@ -1838,7 +1761,6 @@ fn spawn_and_track(
let state = Arc::clone(state);
let task_name = name.to_owned();
tokio::spawn(async move {
let _cgroup = cgroup;
let mut running = running;
let mut prompt = prompt;
loop {
@ -1931,12 +1853,11 @@ async fn after_turn(
/// `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
/// 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(
state: &Arc<State>,
name: &str,
runtime: AcpRuntime<hive_claude::NeverCompact>,
cgroup: Option<RunCgroup>,
config: Config,
prompt: String,
verdict: Option<VerdictTx>,
@ -1963,14 +1884,12 @@ fn spawn_and_track_acp(
Err(e) => TurnEnd::Failed(format!("ACP error: {e}")),
};
let Some(next) = after_turn(&state, &task_name, end, verdict.as_ref()).await else {
break;
return;
};
state.track(&task_name, canceller.clone().into());
state.note_event(&task_name);
prompt = next;
}
drop(runtime);
drop(cgroup);
});
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}"))
}
/// 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
/// [`acp_session_file`] and answering permission requests per
/// [`acp_permits`].
@ -5258,55 +5172,4 @@ echo '{"type":"result","subtype":"success","is_error":false}'
};
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
# negative value puts the harness — and the agent's own `claude`,
# which it spawns as a child and which therefore inherits the
# value — last in line under memory pressure, behind each subagent
# run (+1000) and bash task (0) (see ./mcp.nix). Losing a
# value — last in line under memory pressure, behind
# `hive-subagent-daemon`'s `+500` (see ./mcp.nix). Losing a
# subagent costs one restartable task; losing this costs the
# session that was supervising it. Not `-1000`, which would
# 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
# loss to the process the kernel chose.
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;
Group = userName;
};
@ -441,15 +436,18 @@ in
# `start`/`continue`/`status`/`interrupt` with no stdio fallback.
Restart = "always";
RestartSec = 3;
# The daemon runs each subagent run in its own `runs/<name>-<n>/`
# cgroup with `memory.oom.group=1` and `oom_score_adj` +1000, so an
# OOM kill takes one whole run, ahead of a bash task (0) and
# `hive-agent` (-500). No `OOMScoreAdjust=` here: the daemon stays
# at 0, and a unit value would also be the floor no run could go
# below. The subgroup name is `DAEMON_SUBGROUP` in hive-agent-sock's
# `run_cgroup`.
Delegate = "memory pids";
DelegateSubgroup = "daemon";
# A HIGHER OOMScoreAdjust means MORE likely to be killed: the
# kernel adds it to the badness score it derives from the
# process's memory footprint, then kills the highest scorer. So
# the positive value here and the negative one on `hive-agent`
# put this unit first in line — a subagent turn can be retried,
# the agent's own turn is the thing everything else hangs off.
# `+500` against `hive-agent`'s `-500` is a full half of the
# score range apart, enough that a fat subagent outranks a fatter
# 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
# soon as the kernel kills *any* process in its cgroup — so one
# 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;
}
{
# A HIGHER OOMScoreAdjust is a MORE likely victim. The subagent daemon
# puts each run at +1000 itself and bash tasks stay at 0, so the harness
# must sit below 0 for the agent's own session to go last. Neither
# daemon unit carries a value: it would make the idle daemon a likelier
# victim than a bash task, and a floor no run under it could go below.
name = "the OOM killer takes the agent's own session last";
# The sign is the whole property, and it is easy to write backwards:
# a HIGHER OOMScoreAdjust is a MORE likely victim. So the subagent
# daemon must be strictly above zero and the harness strictly below
# it — swap the two and the kernel takes the agent's own turn first,
# which is worse than setting nothing at all. Both nested `claude`
# 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 =
(agentHarness agentDefault).serviceConfig.OOMScoreAdjust < 0
&& !((agentSubagentDaemon agentDefault).serviceConfig ? OOMScoreAdjust)
&& !((agentBashDaemon agentDefault).serviceConfig ? OOMScoreAdjust);
let
sub = (agentSubagentDaemon 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
# delegation they cannot, and without the `daemon` leaf (the name
# hive-agent-sock's `run_cgroup` looks for) they refuse to, so runs
# would share the daemon's cgroup again.
name = "the subagent and bash daemons are delegated their cgroup";
ok =
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.
# The pair the case above only makes sense with: preferring this unit
# as the victim is an improvement only if losing one subagent 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, which is the blast radius that made the preference a bad
# trade in the first place.
name = "one subagent losing the OOM draw does not stop the daemon";
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.
name = "one bash task losing the OOM draw does not stop the daemon";
ok = (agentBashDaemon agentDefault).serviceConfig.OOMPolicy or null == "continue";