fix(agents): bound how many subagents one session runs at once

Port of upstream grok-build 1.0.1 (landed upstream 2026-08-06). The
coordinator fire-and-forgot every spawn, so a wide fan-out (graph
waves, swarm bursts, model-driven Task storms) could exhaust file
descriptors.

- kigi-tools task/admission: SubagentLimits — 32 concurrent per session
  by default (KIGI_MAX_CONCURRENT_SUBAGENTS), queue|fail behavior
  (KIGI_SUBAGENT_LIMIT_BEHAVIOR), zero clamped to 1, read once at the
  composition root; parse_positive_env helper.
- Each parent session gets one spawn semaphore (map entries live for
  the process so a resumed session shares its predecessor's limit);
  handle_subagent_request holds an owned permit for the whole child
  run — queue awaits FIFO with a unified-log notice at saturation,
  fail routes through send_pre_spawn_failure. After a queued wait, a
  closed parent command channel (teardown) fails the spawn instead of
  running against a dead parent.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KSy9S31pZNWXtRZNVJ1sDf

crates/codegen/kigi-shell/src/agent/mvp_agent/agent_ops.rs

@@ -979,6 +979,9 @@ impl MvpAgent {
             activity,
             loading_sessions: RefCell::new(HashMap::new()),
             dispatch_locks: RefCell::new(HashMap::new()),
+            subagent_limits:
+                kigi_tools::implementations::kigi::task::admission::SubagentLimits::from_env(),
+            subagent_spawn_permits: RefCell::new(HashMap::new()),
             session_threads: RefCell::new(HashMap::new()),
             resident_roster_titles: RefCell::new(HashMap::new()),
             initialize_request: OnceLock::new(),

crates/codegen/kigi-shell/src/agent/mvp_agent/mod.rs

@@ -474,6 +474,18 @@ pub struct MvpAgent {
     dispatch_locks: RefCell<
         HashMap<acp::SessionId, std::rc::Rc<tokio::sync::Mutex<()>>>,
     >,
+    /// Session-scoped subagent spawn limits, read once from env at
+    /// construction (composition root).
+    subagent_limits: kigi_tools::implementations::kigi::task::admission::SubagentLimits,
+    /// Per-parent-session spawn semaphores sized to `subagent_limits`; a
+    /// wide fan-out queues on the semaphore instead of exhausting file
+    /// descriptors. Keyed by session id string (the coordinator's
+    /// `parent_session_id`). Entries deliberately live for the whole
+    /// process: a removed-then-resumed session keeps its id, so sharing one
+    /// semaphore across generations keeps the limit correct while old
+    /// children still run. Growth is bounded by distinct sessions per
+    /// process and each entry is tiny. LEADER-SAFE(per-session).
+    subagent_spawn_permits: RefCell<HashMap<String, std::sync::Arc<tokio::sync::Semaphore>>>,
     /// LEADER-SAFE(per-session): keyed by SessionId. Mirrors `sessions` lifecycle.
     session_threads: RefCell<HashMap<acp::SessionId, SessionThread>>,
     /// Title per resident session id, refreshed each `build_roster`. Lets the

crates/codegen/kigi-shell/src/agent/mvp_agent/subagent_coordinator.rs

@@ -272,6 +272,19 @@ impl MvpAgent {
         self.try_build_subagent_spawn_context(parent_session_id)
             .expect("parent session must exist when spawning subagents")
     }
+    /// Get-or-create the parent session's spawn semaphore (see
+    /// [`Self::subagent_spawn_permits`]). Cheap clone of the shared `Arc`.
+    fn subagent_spawn_permit(&self, parent_session_id: &str) -> std::sync::Arc<tokio::sync::Semaphore> {
+        self.subagent_spawn_permits
+            .borrow_mut()
+            .entry(parent_session_id.to_string())
+            .or_insert_with(|| {
+                std::sync::Arc::new(tokio::sync::Semaphore::new(
+                    self.subagent_limits.effective_max_concurrent(),
+                ))
+            })
+            .clone()
+    }
     /// Fallible variant of [`Self::build_subagent_spawn_context`]: returns
     /// `None` when the parent `SessionHandle` is absent (evicted / torn down)
     /// instead of panicking, so read-only paths that can race a teardown can
@@ -387,6 +400,8 @@ impl MvpAgent {
         };
         Some(crate::agent::subagent::SubagentSpawnContext {
             lsp: parent_lsp,
+            spawn_limits: self.subagent_limits,
+            spawn_permits: self.subagent_spawn_permit(parent_session_id),
             gateway: self.gateway.clone(),
             client_hooks: Default::default(),
             sampling_config: self.sampling_config.borrow().clone(),

crates/codegen/kigi-shell/src/agent/subagent/handle_request.rs

@@ -81,6 +81,64 @@ pub(crate) async fn handle_subagent_request(
         }
         _ => {}
     }
+    // Session-scoped concurrency admission: hold one permit for the whole
+    // child run so a wide fan-out queues (default) or fails (configured)
+    // instead of exhausting file descriptors. Queued spawns are not yet
+    // visible in the coordinator; they admit in FIFO order.
+    use kigi_tools::implementations::kigi::task::admission::LimitBehavior;
+    let _spawn_permit = match ctx.spawn_limits.behavior {
+        LimitBehavior::Fail => match ctx.spawn_permits.clone().try_acquire_owned() {
+            Ok(permit) => permit,
+            Err(_) => {
+                let limit = ctx.spawn_limits.effective_max_concurrent();
+                kigi_log::unified_log::warn(
+                    "subagent.spawn.rejected_at_concurrent_limit",
+                    Some(&ctx.parent_session_id),
+                    Some(serde_json::json!({ "limit": limit })),
+                );
+                let msg = format!(
+                    "Session subagent limit reached ({limit} running); \
+                     retry after one finishes, or raise KIGI_MAX_CONCURRENT_SUBAGENTS"
+                );
+                send_pre_spawn_failure(request, &msg, coordinator, &ctx, gateway);
+                return;
+            }
+        },
+        LimitBehavior::Queue => {
+            if ctx.spawn_permits.available_permits() == 0 {
+                kigi_log::unified_log::warn(
+                    "subagent.spawn.queued_at_concurrent_limit",
+                    Some(&ctx.parent_session_id),
+                    Some(serde_json::json!({
+                        "limit": ctx.spawn_limits.effective_max_concurrent(),
+                        "subagent_id": request.id,
+                    })),
+                );
+            }
+            ctx.spawn_permits
+                .clone()
+                .acquire_owned()
+                .await
+                .expect("subagent spawn semaphore is never closed")
+        }
+    };
+    // The parent can end while a spawn is queued on the semaphore (a queued
+    // request is not yet in the coordinator, so the teardown cancel sweep
+    // cannot see it). A closed parent command channel is the teardown
+    // signal. The channel closes when the actor finishes shutting down, so a
+    // permit granted during that brief drain can still slip through — the
+    // same in-flight window an ordinary spawn racing teardown already has;
+    // closing it fully would require queueing inside the coordinator.
+    if ctx.parent_cmd_tx.as_ref().is_some_and(|tx| tx.is_closed()) {
+        send_pre_spawn_failure(
+            request,
+            "Parent session ended while the spawn was queued at the subagent limit.",
+            coordinator,
+            &ctx,
+            gateway,
+        );
+        return;
+    }
     let run_in_background = request.run_in_background
         || definition.background.unwrap_or(false);
     let cancel_token = CancellationToken::new();

crates/codegen/kigi-shell/src/agent/subagent/mod.rs

@@ -133,6 +133,13 @@ impl AutoCompactThresholdTiers {
 pub(crate) struct SubagentSpawnContext {
     /// Parent's LSP runtime — inherited via ToolContext, same as fs/terminal.
     pub lsp: Option<std::sync::Arc<dyn kigi_tools::implementations::lsp::LspBackend>>,
+    /// Session-scoped spawn limits (env-resolved once at the composition root).
+    pub spawn_limits: kigi_tools::implementations::kigi::task::admission::SubagentLimits,
+    /// Parent session's spawn semaphore, sized to
+    /// `spawn_limits.effective_max_concurrent()`. A permit is held for the
+    /// whole child run, so a wide fan-out queues (or fails, per behavior)
+    /// instead of exhausting file descriptors.
+    pub spawn_permits: std::sync::Arc<tokio::sync::Semaphore>,
     #[expect(
         dead_code,
         reason = "unused in production; remove expect when wired or delete the item"

crates/codegen/kigi-shell/src/test_support/lsp_runtime.rs

@@ -33,6 +33,12 @@ pub(crate) fn ctx_with_toggle(toggle: HashMap<String, bool>) -> SubagentSpawnCon
     let (tx, _rx) = mpsc::unbounded_channel();
     SubagentSpawnContext {
         lsp: None,
+        spawn_limits:
+            kigi_tools::implementations::kigi::task::admission::SubagentLimits::default(),
+        spawn_permits: std::sync::Arc::new(tokio::sync::Semaphore::new(
+            kigi_tools::implementations::kigi::task::admission::SubagentLimits::default()
+                .effective_max_concurrent(),
+        )),
         parent_max_turns: None,
         gateway: test_gateway(),
         client_hooks: Default::default(),

crates/codegen/kigi-tools/src/implementations/kigi/task/admission.rs

@@ -0,0 +1,117 @@
+//! Session-scoped subagent spawn limits, enforced by the shell's spawn path.
+
+use crate::util::env::parse_positive_env;
+
+pub const DEFAULT_MAX_CONCURRENT: usize = 32;
+
+/// What happens to a spawn that arrives at the concurrent limit.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub enum LimitBehavior {
+    #[default]
+    Queue,
+    Fail,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct SubagentLimits {
+    /// Subagents one session may run at once. Zero does not disable the
+    /// limit: [`Self::effective_max_concurrent`] clamps it to 1.
+    pub max_concurrent: usize,
+    pub behavior: LimitBehavior,
+}
+
+impl Default for SubagentLimits {
+    fn default() -> Self {
+        Self {
+            max_concurrent: DEFAULT_MAX_CONCURRENT,
+            behavior: LimitBehavior::Queue,
+        }
+    }
+}
+
+impl SubagentLimits {
+    /// Read once at the composition root; inject everywhere else.
+    pub fn from_env() -> Self {
+        Self::from_lookup(|var| std::env::var(var).ok())
+    }
+
+    /// A limit can be adjusted but never disabled.
+    fn from_lookup(lookup: impl Fn(&str) -> Option<String>) -> Self {
+        let default = Self::default();
+        let max_concurrent = parse_positive_env(
+            "KIGI_MAX_CONCURRENT_SUBAGENTS",
+            lookup("KIGI_MAX_CONCURRENT_SUBAGENTS"),
+        )
+        .unwrap_or(default.max_concurrent);
+        let behavior = match lookup("KIGI_SUBAGENT_LIMIT_BEHAVIOR") {
+            None => LimitBehavior::Queue,
+            Some(value) if value.eq_ignore_ascii_case("fail") => LimitBehavior::Fail,
+            Some(value) if value.eq_ignore_ascii_case("queue") => LimitBehavior::Queue,
+            Some(value) => {
+                tracing::warn!(
+                    %value,
+                    "KIGI_SUBAGENT_LIMIT_BEHAVIOR is neither `queue` nor `fail`; keeping `queue`"
+                );
+                LimitBehavior::Queue
+            }
+        };
+        Self {
+            max_concurrent,
+            behavior,
+        }
+    }
+
+    /// The semaphore size: a zero limit must still admit one child, or
+    /// queued spawns starve.
+    pub fn effective_max_concurrent(&self) -> usize {
+        self.max_concurrent.max(1)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{LimitBehavior, SubagentLimits};
+
+    #[test]
+    fn limits_accept_plain_positive_digits_and_ignore_everything_else() {
+        let from = |concurrent: Option<&str>, behavior: Option<&str>| {
+            SubagentLimits::from_lookup(|var| {
+                match var {
+                    "KIGI_MAX_CONCURRENT_SUBAGENTS" => concurrent,
+                    "KIGI_SUBAGENT_LIMIT_BEHAVIOR" => behavior,
+                    other => panic!("unexpected lookup: {other}"),
+                }
+                .map(str::to_owned)
+            })
+        };
+
+        assert_eq!(from(None, None), SubagentLimits::default());
+        assert_eq!(
+            from(Some("5"), Some("FAIL")),
+            SubagentLimits {
+                max_concurrent: 5,
+                behavior: LimitBehavior::Fail,
+            }
+        );
+        // Adjustable but never disabled: zero, negatives, non-digits,
+        // scientific notation, digit separators, and out-of-range values all
+        // keep defaults.
+        for ignored in ["0", "-1", "abc", "1e3", "20_000", "18446744073709551616"] {
+            assert_eq!(
+                from(Some(ignored), Some("nonsense")),
+                SubagentLimits::default(),
+                "value {ignored:?} should fall back to the defaults"
+            );
+        }
+    }
+
+    #[test]
+    fn a_zero_limit_is_clamped_to_one_not_disabled() {
+        let limits = SubagentLimits {
+            max_concurrent: 0,
+            behavior: LimitBehavior::Queue,
+        };
+        assert_eq!(limits.effective_max_concurrent(), 1);
+        assert_eq!(SubagentLimits::default().effective_max_concurrent(), 32);
+    }
+}

crates/codegen/kigi-tools/src/implementations/kigi/task/mod.rs

@@ -12,6 +12,7 @@
 //! - `SessionIdResource` — current session ID for parent scoping (optional)
 //! - `TaskModelValidator` — validates explicit model slugs before spawn
 
+pub mod admission;
 pub mod backend;
 pub mod types;
 

crates/codegen/kigi-tools/src/util/env.rs

@@ -6,6 +6,27 @@
 
 pub use kigi_tty_utils::{detach_from_tty, pager_env};
 
+/// Parse a positive whole number in plain digits (no sign, no separators,
+/// no scientific notation). `None` for zero or anything non-canonical.
+pub fn parse_positive(value: &str) -> Option<u64> {
+    value.parse::<u64>().ok().filter(|&parsed| parsed > 0)
+}
+
+/// Parse a positive whole-number env value into a count. A set-but-invalid
+/// value warns and reads as unset, so the caller's default applies.
+pub fn parse_positive_env(var: &str, value: Option<String>) -> Option<usize> {
+    let value = value?;
+    let parsed = parse_positive(&value).and_then(|parsed| usize::try_from(parsed).ok());
+    if parsed.is_none() {
+        tracing::warn!(
+            var,
+            %value,
+            "env value is not a positive whole number in plain digits; using the default"
+        );
+    }
+    parsed
+}
+
 /// Env var set on agent-spawned terminal processes so host tools (e.g. `x ban`)
 /// can distinguish agent invocations from human interactive shells.
 /// Note: the CLI also uses `KIGI_AGENT` as an