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

subagent: keep runs at +1000 when their cgroup is unavailable

Browse source 
Without a run cgroup (no delegation, setup failed, or one create
failed) a subagent run was started plain at oom_score_adj 0, below the
+500 the unit gave it before. Every run of the daemon now goes through
the entry wrapper; it sets +1000 and enters the cgroup only when there
is one. run_cgroup::entry takes the cgroup as an Option.
This commit is contained in:
atlas 2026-10-12 23:46:18 +02:00 • committed by mara
commit afe6026d44
5 changed files with 159 additions and 64 deletions
Download patch file Download diff file Expand all files Collapse all files

3
docs/scheduler/coordinator.md
View file

@ -491,7 +491,8 @@ 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.
up logs one warning at start and runs everything in its own cgroup; subagent
runs still get +1000.
<details><summary>Implementation: swarm caps</summary>

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

@ -14,9 +14,11 @@
//! 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:
//! - [`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;
@ -149,30 +151,34 @@ impl RunCgroup {
.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)>) {
/// 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![
(
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(),
),
],
)
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 {
@ -218,8 +224,8 @@ pub fn enter_in_pre_exec(procs: RawFd) -> io::Result<()> {
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
/// 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() {
@ -336,7 +342,7 @@ mod tests {
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);
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"));
@ -344,6 +350,16 @@ mod tests {
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");

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

@ -65,7 +65,7 @@ async fn serve() -> Result<()> {
.on_runtime(runtime)
.publishing_to(hive_subagent_mcp::swarm_term::Publisher::spawn())
.reading_models_from(hive_subagent_mcp::swarm_models::Reader::spawn())
.in_run_cgroups(hive_agent_sock::run_cgroup::RunCgroups::init("runs")),
.entering_runs(hive_agent_sock::run_cgroup::RunCgroups::init("runs")),
);
// Serve the MCP tools over streamable-http forever. No background poll

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

@ -93,7 +93,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, PoisonError};
use std::time::{Duration, Instant};
use hive_agent_sock::run_cgroup::{RunCgroup, RunCgroups};
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,8 +407,11 @@ 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 ([`State::in_run_cgroups`]).
/// one.
run_cgroups: Option<RunCgroups>,
/// The `<n>` of the next run cgroup `<name>-<n>`.
run_seq: AtomicU64,
@ -467,6 +470,7 @@ impl State {
term: None,
models: None,
local_models: available_models,
runs_entered: false,
run_cgroups: None,
run_seq: AtomicU64::new(0),
}
@ -487,10 +491,12 @@ impl State {
self
}
/// Run each subagent run in its own cgroup under `cgroups`, at
/// `oom_score_adj` +1000.
/// 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 in_run_cgroups(mut self, cgroups: Option<RunCgroups>) -> Self {
pub fn entering_runs(mut self, cgroups: Option<RunCgroups>) -> Self {
self.runs_entered = true;
self.run_cgroups = cgroups;
self
}
@ -507,6 +513,33 @@ impl State {
.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<()> {
@ -1424,7 +1457,11 @@ fn start_on_runtime(
) -> anyhow::Result<String> {
if let SubagentRuntime::Acp { command, sessions } = &state.runtime {
let cgroup = state.run_cgroup(name);
let runtime = acp_runtime(command.clone(), sessions, name, cgroup.as_ref());
let runtime = acp_runtime(
state.entered_command(command.clone(), cgroup.as_ref()),
sessions,
name,
);
if let Some(archived) = runtime
.archive()
.map_err(|e| anyhow::anyhow!("archiving the prior `{name}` session failed: {e}"))?
@ -1639,7 +1676,11 @@ fn continue_on_runtime(
}
state.begin_continue(name, dir);
let cgroup = state.run_cgroup(name);
let runtime = acp_runtime(command.clone(), sessions, name, cgroup.as_ref());
let runtime = acp_runtime(
state.entered_command(command.clone(), cgroup.as_ref()),
sessions,
name,
);
return spawn_and_track_acp(
state,
name,
@ -1771,7 +1812,7 @@ fn spawn_and_track(
// 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 = &entered_config(config, cgroup.as_ref());
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());
@ -1946,46 +1987,19 @@ fn acp_session_file(sessions: &Path, name: &str) -> PathBuf {
/// own session (-500, `nix/agent-modules/agent-service.nix`).
const RUN_OOM_SCORE_ADJ: i32 = 1000;
/// `config` with its program started inside `cgroup`, if there is one.
/// `hive-claude` builds the `Command` itself and offers no `pre_exec`, so
/// the program becomes this daemon's binary as a [`RunCgroup::entry`]
/// wrapper.
fn entered_config(config: &Config, cgroup: Option<&RunCgroup>) -> Config {
let mut config = config.clone();
if let Some(cgroup) = cgroup {
// `hive-claude`'s own default program.
let program = config.program.take().unwrap_or_else(|| "claude".into());
let (entry, env) = cgroup.entry(&program, RUN_OOM_SCORE_ADJ);
config.program = Some(entry);
config.env.extend(env);
}
config
}
/// [`entered_config`] for an ACP agent's command.
fn entered_command(mut command: AcpCommand, cgroup: Option<&RunCgroup>) -> AcpCommand {
if let Some(cgroup) = cgroup {
let (entry, env) = cgroup.entry(Path::new(&command.command), RUN_OOM_SCORE_ADJ);
command.command = entry.to_string_lossy().into_owned();
command.env.extend(env);
}
command
}
/// The ACP runtime `name`'s run drives, keeping its session id in
/// [`acp_session_file`] and answering permission requests per
/// [`acp_permits`]. Its agent process starts inside `cgroup`.
/// [`acp_permits`].
fn acp_runtime(
command: AcpCommand,
sessions: &Path,
name: &str,
cgroup: Option<&RunCgroup>,
) -> AcpRuntime<hive_claude::NeverCompact> {
let groups = hive_sh4re::permissions::effective_tool_groups();
let permit: PermissionPolicy =
Arc::new(move |ask: &PermissionAsk<'_>| acp_permits(ask, &groups));
AcpRuntime::new(
entered_command(command, cgroup),
command,
acp_session_file(sessions, name),
permit,
hive_claude::NeverCompact,
@ -5244,4 +5258,55 @@ 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());
}
}

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

@ -26,3 +26,16 @@ fn the_entry_enters_the_cgroup_raises_the_score_and_execs_the_program() {
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");
}