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

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 orchestrator_plugin_host::cgroup_threads;
 pub mod agent_state;
 pub mod config_context;
 pub mod ensure_execution_cwd;

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

@@ -63,8 +63,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-daemon-runtime/src/dispatch/process_manager.rs

@@ -324,9 +324,13 @@ 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/src/cgroup_threads.rs

@@ -0,0 +1,154 @@
+//! 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()` (OS default)
+//!
+//! The result is always >= 1.
+//!
+//! This module lives in `orchestrator-plugin-host` (a leaf crate) so that
+//! both `orchestrator-plugin-host` itself and `animus-runtime-shared` (which
+//! depends on this crate) can use it without creating a dependency cycle.
+
+/// 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 quota_str == "max" {
+        return None;
+    }
+    let quota: u64 = quota_str.parse().ok()?;
+    let period: u64 = period_str.parse().ok()?;
+    if period == 0 {
+        return None;
+    }
+    Some(((quota / period) as usize).max(1))
+}
+
+fn cgroup_v1_threads() -> Option<usize> {
+    let quota_str = std::fs::read_to_string("/sys/fs/cgroup/cpu/cpu.cfs_quota_us").ok()?;
+    let period_str = std::fs::read_to_string("/sys/fs/cgroup/cpu/cpu.cfs_period_us").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 period == 0 {
+        return None;
+    }
+    Some(((quota as u64 / period) as usize).max(1))
+}
+
+fn system_threads() -> usize {
+    std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1)
+}
+
+#[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_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_fractional_rounds_down_to_one() {
+        // 50_000 / 100_000 = 0.5 CPUs → clamped to 1
+        assert_eq!(parse_cgroup_v2("50000 100000"), Some(1));
+    }
+
+    #[test]
+    fn cgroup_v2_zero_period_returns_none() {
+        assert!(parse_cgroup_v2("200000 0").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_zero_period_returns_none() {
+        assert!(parse_cgroup_v1("200000", "0").is_none());
+    }
+}

crates/orchestrator-plugin-host/src/host.rs

@@ -762,19 +762,16 @@ impl PluginHost {
         }
 
         // Bound each plugin's tokio runtime. A bare `#[tokio::main]` (every Animus
-        // stdio plugin) sizes its multi-thread worker pool to
-        // `available_parallelism()` — all CPU cores. With v0.6's resident-plugin
-        // fleet (config_source + subject backends + queue + workflow_runner +
-        // providers + transport) that is hundreds of threads on a many-core host,
-        // exhausting the PID/thread budget so new forks — including the provider CLI
-        // an agent phase spawns — fail with EAGAIN and the run hangs. Plugins are
-        // I/O-bound stdio RPC servers, so a tiny pool is sufficient. `env_clear()`
-        // above dropped any inherited value (TOKIO_WORKER_THREADS is not in the base
-        // allowlist), so set it explicitly here, honoring an operator override on the
-        // daemon env so a deploy can still tune it up or down.
+        // stdio plugin) sizes its multi-thread worker pool to the host CPU count.
+        // With v0.6's resident-plugin fleet that is hundreds of threads on a many-core
+        // host, exhausting the PID/thread budget. Use the cgroup CPU quota instead.
+        // `env_clear()` above dropped any inherited value (TOKIO_WORKER_THREADS is
+        // not in the base allowlist), so set it explicitly here, honoring an operator
+        // override on the daemon env so a deploy can still tune it up or down.
         command.env(
             "TOKIO_WORKER_THREADS",
-            std::env::var_os("TOKIO_WORKER_THREADS").unwrap_or_else(|| std::ffi::OsString::from("2")),
+            std::env::var_os("TOKIO_WORKER_THREADS")
+                .unwrap_or_else(|| std::ffi::OsString::from(crate::cgroup_threads::tokio_worker_threads().to_string())),
         );
 
         for missing in &options.missing_required_env {

crates/orchestrator-plugin-host/src/lib.rs

@@ -2,6 +2,7 @@
 
 mod discovery;
 mod host;
+pub mod cgroup_threads;
 pub mod lockfile;
 pub mod manifest_cache;
 mod registry;