animus-cli: bound plugin/daemon tokio worker threads to cgroup CPU quota (not host cpu count)

AGENTS.md

@@ -41,6 +41,7 @@ crates/
 ├── animus-mcp-oauth/            # OAuth authorization-code + PKCE helpers and proxy bridge for protected MCP servers
 ├── animus-plugin-protocol/      # In-tree stdio plugin protocol types
 ├── animus-plugin-runtime/       # Runtime helpers for plugin implementations
+├── animus-runtime-utils/        # Dependency-light cgroup/runtime sizing helpers
 ├── orchestrator-cli/            # Main `animus` binary
 ├── orchestrator-config/         # Workflow, pack, and template config loading
 ├── orchestrator-core/           # Domain services, subject_adapter, store, bootstrap, state mutation APIs

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
   "crates/animus-plugin-runtime",
+  "crates/animus-runtime-utils",
   "crates/orchestrator-daemon-runtime",
   "crates/orchestrator-logging",
   "crates/orchestrator-plugin-host",
@@ -27,6 +28,7 @@ categories = ["command-line-utilities", "development-tools"]
 # Protocol/wire-type crates live in launchapp-dev/animus-protocol, never in
 # this repo. Every dependency is pinned to the same release so Cargo resolves
 # one coherent wire identity with no workstation paths or legacy source split.
+animus-runtime-utils = { path = "crates/animus-runtime-utils" }
 animus-actor = { git = "https://github.com/launchapp-dev/animus-protocol", tag = "v0.7.0-rc.14" }
 animus-application-protocol = { git = "https://github.com/launchapp-dev/animus-protocol", tag = "v0.7.0-rc.14" }
 animus-config-protocol = { git = "https://github.com/launchapp-dev/animus-protocol", tag = "v0.7.0-rc.14" }

crates/animus-runtime-shared/Cargo.toml

@@ -11,6 +11,7 @@ name = "animus_runtime_shared"
 path = "src/lib.rs"
 
 [dependencies]
+animus-runtime-utils = { workspace = true }
 anyhow = "1.0"
 async-trait = "0.1"
 chrono = { version = "0.4", features = ["serde"] }

crates/animus-runtime-shared/src/lib.rs

@@ -13,6 +13,7 @@
 //! `direct_exec`) is plugin-private and intentionally NOT here.
 
 pub mod actor_env;
+pub use animus_runtime_utils::cgroup_threads;
 pub mod agent_state;
 pub mod config_context;
 pub mod ensure_execution_cwd;

crates/animus-runtime-utils/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "animus-runtime-utils"
+version = "0.1.0"
+edition = "2021"
+license = "Elastic-2.0"
+description = "Dependency-light runtime helpers shared by Animus binaries"
+repository = "https://github.com/launchapp-dev/animus-cli"

crates/animus-runtime-utils/src/cgroup_threads.rs

@@ -0,0 +1,197 @@
+//! Compute the Tokio worker-thread count from the cgroup CPU quota so
+//! binaries don't default to the host CPU count when running inside a
+//! resource-capped container.
+//!
+//! Priority:
+//!   1. `TOKIO_WORKER_THREADS` env var (explicit operator override)
+//!   2. cgroup v2 `cpu.max` quota (`/sys/fs/cgroup/cpu.max`)
+//!   3. cgroup v1 quota (`/sys/fs/cgroup/cpu/cpu.cfs_quota_us`)
+//!   4. `std::thread::available_parallelism()` (host-visible CPU count)
+//!
+//! The result is always >= 1.
+//!
+//! This module lives in the dependency-light `animus-runtime-utils` crate so
+//! every in-workspace runtime entry point can use it without dependency cycles.
+
+/// Return the number of Tokio worker threads appropriate for this process,
+/// honouring the cgroup CPU quota when present.
+///
+/// Pass the return value to `tokio::runtime::Builder::worker_threads` instead
+/// of relying on `#[tokio::main]`'s default (which uses the host CPU count,
+/// not the container quota).
+pub fn tokio_worker_threads() -> usize {
+    read_env_override().or_else(cgroup_v2_threads).or_else(cgroup_v1_threads).unwrap_or_else(system_threads).max(1)
+}
+
+fn read_env_override() -> Option<usize> {
+    let s = std::env::var("TOKIO_WORKER_THREADS").ok()?;
+    parse_env_override(s.trim())
+}
+
+/// Parse a `TOKIO_WORKER_THREADS` value; returns `None` for zero or non-numeric.
+pub(crate) fn parse_env_override(s: &str) -> Option<usize> {
+    s.parse::<usize>().ok().filter(|&n| n > 0)
+}
+
+fn cgroup_v2_threads() -> Option<usize> {
+    let content = std::fs::read_to_string("/sys/fs/cgroup/cpu.max").ok()?;
+    parse_cgroup_v2(&content)
+}
+
+/// Parse a cgroup v2 `cpu.max` file content (`<quota_us> <period_us>`).
+///
+/// Returns `None` when the quota is unlimited (`"max"`) or the file is malformed.
+pub(crate) fn parse_cgroup_v2(content: &str) -> Option<usize> {
+    let mut parts = content.split_whitespace();
+    let quota_str = parts.next()?;
+    let period_str = parts.next()?;
+    if parts.next().is_some() || quota_str == "max" {
+        return None;
+    }
+    let quota: u64 = quota_str.parse().ok()?;
+    let period: u64 = period_str.parse().ok()?;
+    if quota == 0 || period == 0 {
+        return None;
+    }
+    Some(quota.div_ceil(period).try_into().unwrap_or(usize::MAX).max(1))
+}
+
+fn cgroup_v1_threads() -> Option<usize> {
+    // The CPU controller may be mounted directly at the cgroup root or in a
+    // named controller directory, depending on the container runtime.
+    const CONTROLLER_PATHS: &[&str] = &["/sys/fs/cgroup", "/sys/fs/cgroup/cpu", "/sys/fs/cgroup/cpu,cpuacct"];
+
+    CONTROLLER_PATHS.iter().find_map(|controller_path| {
+        let quota_path = format!("{controller_path}/cpu.cfs_quota_us");
+        let period_path = format!("{controller_path}/cpu.cfs_period_us");
+        let quota_str = std::fs::read_to_string(quota_path).ok()?;
+        let period_str = std::fs::read_to_string(period_path).ok()?;
+        parse_cgroup_v1(quota_str.trim(), period_str.trim())
+    })
+}
+
+/// Parse cgroup v1 quota/period strings from `cpu.cfs_quota_us` / `cpu.cfs_period_us`.
+///
+/// Returns `None` when the quota is unlimited (`-1`) or either value is malformed.
+pub(crate) fn parse_cgroup_v1(quota_str: &str, period_str: &str) -> Option<usize> {
+    let quota: i64 = quota_str.parse().ok()?;
+    if quota < 0 {
+        return None; // -1 = unlimited
+    }
+    let period: u64 = period_str.parse().ok()?;
+    if quota == 0 || period == 0 {
+        return None;
+    }
+    Some((quota as u64).div_ceil(period).try_into().unwrap_or(usize::MAX).max(1))
+}
+
+fn system_threads() -> usize {
+    std::thread::available_parallelism().map_or(1, std::num::NonZeroUsize::get)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn system_threads_is_at_least_one() {
+        assert!(system_threads() >= 1);
+    }
+
+    #[test]
+    fn result_is_at_least_one() {
+        assert!(tokio_worker_threads() >= 1);
+    }
+
+    #[test]
+    fn env_override_zero_is_rejected() {
+        assert!(parse_env_override("0").is_none());
+    }
+
+    #[test]
+    fn env_override_positive_is_accepted() {
+        assert_eq!(parse_env_override("4"), Some(4));
+    }
+
+    #[test]
+    fn env_override_overflow_is_rejected() {
+        assert!(parse_env_override("999999999999999999999999999999999999999").is_none());
+    }
+
+    #[test]
+    fn env_override_invalid_is_rejected() {
+        assert!(parse_env_override("abc").is_none());
+    }
+
+    #[test]
+    fn cgroup_v2_unlimited_returns_none() {
+        assert!(parse_cgroup_v2("max 100000").is_none());
+    }
+
+    #[test]
+    fn cgroup_v2_two_cpus() {
+        // 200_000 quota / 100_000 period = 2 CPUs
+        assert_eq!(parse_cgroup_v2("200000 100000"), Some(2));
+    }
+
+    #[test]
+    fn cgroup_v2_sub_cpu_quota_has_one_worker() {
+        // 50_000 / 100_000 = 0.5 CPUs → clamped to 1
+        assert_eq!(parse_cgroup_v2("50000 100000"), Some(1));
+    }
+
+    #[test]
+    fn cgroup_v2_fractional_cpu_rounds_up() {
+        assert_eq!(parse_cgroup_v2("150000 100000"), Some(2));
+    }
+
+    #[test]
+    fn cgroup_v2_zero_period_returns_none() {
+        assert!(parse_cgroup_v2("200000 0").is_none());
+    }
+
+    #[test]
+    fn cgroup_v2_zero_quota_returns_none() {
+        assert!(parse_cgroup_v2("0 100000").is_none());
+    }
+
+    #[test]
+    fn cgroup_v2_extra_fields_returns_none() {
+        assert!(parse_cgroup_v2("200000 100000 unexpected").is_none());
+    }
+
+    #[test]
+    fn cgroup_v2_missing_period_returns_none() {
+        assert!(parse_cgroup_v2("200000").is_none());
+    }
+
+    #[test]
+    fn cgroup_v1_unlimited_returns_none() {
+        assert!(parse_cgroup_v1("-1", "100000").is_none());
+    }
+
+    #[test]
+    fn cgroup_v1_two_cpus() {
+        assert_eq!(parse_cgroup_v1("200000", "100000"), Some(2));
+    }
+
+    #[test]
+    fn cgroup_v1_fractional_cpu_rounds_up() {
+        assert_eq!(parse_cgroup_v1("150000", "100000"), Some(2));
+    }
+
+    #[test]
+    fn cgroup_v1_zero_period_returns_none() {
+        assert!(parse_cgroup_v1("200000", "0").is_none());
+    }
+
+    #[test]
+    fn cgroup_v1_zero_quota_returns_none() {
+        assert!(parse_cgroup_v1("0", "100000").is_none());
+    }
+
+    #[test]
+    fn cgroup_v1_malformed_quota_returns_none() {
+        assert!(parse_cgroup_v1("not-a-number", "100000").is_none());
+    }
+}

crates/animus-runtime-utils/src/lib.rs

@@ -0,0 +1,3 @@
+//! Dependency-light helpers shared by Animus runtime crates.
+
+pub mod cgroup_threads;

crates/orchestrator-cli/src/bin/animus_mcp_proxy.rs

@@ -71,8 +71,17 @@ impl animus_mcp_oauth::proxy::BearerTokenSource for BrokerBearerSource {
     }
 }
 
-#[tokio::main]
-async fn main() -> Result<()> {
+fn main() -> Result<()> {
+    let worker_threads = animus_runtime_shared::cgroup_threads::tokio_worker_threads();
+    tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(worker_threads)
+        .enable_all()
+        .build()
+        .expect("failed to build tokio runtime")
+        .block_on(async_main())
+}
+
+async fn async_main() -> Result<()> {
     let args = Args::parse();
     let project_root = args
         .project_root

crates/orchestrator-cli/src/main.rs

@@ -12,8 +12,21 @@ mod shared;
 pub(crate) use cli_types::*;
 pub(crate) use shared::*;
 
-#[tokio::main]
-async fn main() {
+fn main() {
+    // Use a cgroup-aware worker thread count so the daemon doesn't default to
+    // the host CPU count (e.g. 48 on Railway) inside a resource-capped
+    // container, exhausting the cgroup PID limit with ~480 Tokio threads.
+    // `TOKIO_WORKER_THREADS` still overrides (operator escape hatch).
+    let worker_threads = animus_runtime_shared::cgroup_threads::tokio_worker_threads();
+    tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(worker_threads)
+        .enable_all()
+        .build()
+        .expect("failed to build tokio runtime")
+        .block_on(async_main());
+}
+
+async fn async_main() {
     // Pre-scan argv for `--json` so that clap argparse failures (unknown
     // subcommands, bad flag values) still emit the `animus.cli.v1` error
     // envelope when the caller asked for machine-readable output. `Cli::parse`

crates/orchestrator-cli/src/services/operations/ops_plugin.rs

@@ -5127,6 +5127,7 @@ struct InstallProvenance {
 fn probe_manifest(binary_path: &Path) -> Result<PluginManifest> {
     let output = std::process::Command::new(binary_path)
         .arg("--manifest")
+        .env("TOKIO_WORKER_THREADS", animus_runtime_shared::cgroup_threads::tokio_worker_threads().to_string())
         .stdout(Stdio::piped())
         .stderr(Stdio::piped())
         .output()
@@ -7130,6 +7131,31 @@ fn compute_lock_verify(args: PluginLockVerifyArgs, project_root: &str) -> Result
 mod tests {
     use super::*;
 
+    #[cfg(unix)]
+    #[test]
+    fn probe_manifest_propagates_tokio_worker_thread_bound() {
+        use std::os::unix::fs::PermissionsExt;
+
+        let temp = tempfile::tempdir().expect("tempdir");
+        let plugin = temp.path().join("animus-plugin-worker-env");
+        let expected = animus_runtime_shared::cgroup_threads::tokio_worker_threads();
+        let script = format!(
+            "#!/bin/sh\n\
+             if [ \"$TOKIO_WORKER_THREADS\" != \"{expected}\" ]; then\n\
+               echo \"unexpected TOKIO_WORKER_THREADS=$TOKIO_WORKER_THREADS\" >&2\n\
+               exit 17\n\
+             fi\n\
+             printf '{{\"name\":\"worker-env\",\"version\":\"0.1.0\",\"plugin_kind\":\"custom\",\"description\":\"t\",\"protocol_version\":\"1.0.0\",\"capabilities\":[]}}\\n'\n"
+        );
+        std::fs::write(&plugin, script).expect("write plugin");
+        let mut perms = std::fs::metadata(&plugin).expect("metadata").permissions();
+        perms.set_mode(0o755);
+        std::fs::set_permissions(&plugin, perms).expect("chmod");
+
+        let manifest = probe_manifest(&plugin).expect("manifest probe must receive the tokio worker bound");
+        assert_eq!(manifest.name, "worker-env");
+    }
+
     /// Build a single-target integrity map for the CURRENT build target whose
     /// archive + installed-binary shas both equal `sha`. Mirrors how an install
     /// on this platform records its own target; lets the existing lock tests

crates/orchestrator-daemon-runtime/src/dispatch/process_manager.rs

@@ -426,9 +426,11 @@ impl ProcessManager {
         // Bound the workflow-runner's tokio pool for the same reason we cap plugins
         // (orchestrator-plugin-host host.rs): a bare `#[tokio::main]` otherwise sizes
         // the worker pool to all CPU cores, compounding the PID/thread pressure. This
-        // path inherits the daemon env, so only impose the default when unset.
+        // path inherits the daemon env, so only impose the default when unset, using
+        // the cgroup-aware count rather than a hard-coded value.
         if std::env::var_os("TOKIO_WORKER_THREADS").is_none() {
-            command.env("TOKIO_WORKER_THREADS", "2");
+            command
+                .env("TOKIO_WORKER_THREADS", animus_runtime_shared::cgroup_threads::tokio_worker_threads().to_string());
         }
         // Phase skills pass-down: resolve the union of phase-level `skills:`
         // and the executing agent profile's `skills:` daemon-side (scoped

crates/orchestrator-plugin-host/Cargo.toml

@@ -17,6 +17,7 @@ anyhow = "1.0"
 animus-plugin-protocol = { workspace = true }
 animus-actor = { workspace = true }
 animus-subject-protocol-wire = { workspace = true }
+animus-runtime-utils = { workspace = true }
 animus-session-backend = { workspace = true }
 async-trait = "0.1"
 chrono = { version = "0.4", features = ["serde"] }