From afe6026d4484f1b49e86c06a944bd73dc4d796c1 Mon Sep 17 00:00:00 2001 From: atlas Date: Mon, 12 Oct 2026 23:46:18 +0200 Subject: [PATCH] subagent: keep runs at +1000 when their cgroup is unavailable 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. --- docs/scheduler/coordinator.md | 3 +- hive-agent-sock/src/run_cgroup.rs | 66 ++++++++----- hive-subagent-mcp/src/main.rs | 2 +- hive-subagent-mcp/src/session.rs | 139 ++++++++++++++++++++------- hive-subagent-mcp/tests/run_entry.rs | 13 +++ 5 files changed, 159 insertions(+), 64 deletions(-) diff --git a/docs/scheduler/coordinator.md b/docs/scheduler/coordinator.md index c40fc929..e7fb554e 100644 --- a/docs/scheduler/coordinator.md +++ b/docs/scheduler/coordinator.md @@ -491,7 +491,8 @@ outside it. Inside the container these are `/sys/fs/cgroup/system.slice/hive-subagent-daemon.service/runs/-/` and `/sys/fs/cgroup/system.slice/hive-bash-daemon.service/tasks//`; 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.
Implementation: swarm caps diff --git a/hive-agent-sock/src/run_cgroup.rs b/hive-agent-sock/src/run_cgroup.rs index eca80225..509818a6 100644 --- a/hive-agent-sock/src/run_cgroup.rs +++ b/hive-agent-sock/src/run_cgroup.rs @@ -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"); diff --git a/hive-subagent-mcp/src/main.rs b/hive-subagent-mcp/src/main.rs index a8ffc4df..f3be23fb 100644 --- a/hive-subagent-mcp/src/main.rs +++ b/hive-subagent-mcp/src/main.rs @@ -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 diff --git a/hive-subagent-mcp/src/session.rs b/hive-subagent-mcp/src/session.rs index 907ae83e..43ca24b6 100644 --- a/hive-subagent-mcp/src/session.rs +++ b/hive-subagent-mcp/src/session.rs @@ -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, + /// 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, /// The `` of the next run cgroup `-`. 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) -> Self { + pub fn entering_runs(mut self, cgroups: Option) -> 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 { 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 { 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()); + } } diff --git a/hive-subagent-mcp/tests/run_entry.rs b/hive-subagent-mcp/tests/run_entry.rs index b194276f..dba0d9fe 100644 --- a/hive-subagent-mcp/tests/run_entry.rs +++ b/hive-subagent-mcp/tests/run_entry.rs @@ -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"); +}