diff --git a/crates/codegen/kigi-shell/src/agent/mvp_agent/agent_ops.rs b/crates/codegen/kigi-shell/src/agent/mvp_agent/agent_ops.rs index 4057fc7..df2d507 100644 --- a/crates/codegen/kigi-shell/src/agent/mvp_agent/agent_ops.rs +++ b/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(), diff --git a/crates/codegen/kigi-shell/src/agent/mvp_agent/mod.rs b/crates/codegen/kigi-shell/src/agent/mvp_agent/mod.rs index 6e31f61..54ba2a2 100644 --- a/crates/codegen/kigi-shell/src/agent/mvp_agent/mod.rs +++ b/crates/codegen/kigi-shell/src/agent/mvp_agent/mod.rs @@ -474,6 +474,18 @@ pub struct MvpAgent { dispatch_locks: RefCell< HashMap>>, >, + /// 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>>, /// LEADER-SAFE(per-session): keyed by SessionId. Mirrors `sessions` lifecycle. session_threads: RefCell>, /// Title per resident session id, refreshed each `build_roster`. Lets the diff --git a/crates/codegen/kigi-shell/src/agent/mvp_agent/subagent_coordinator.rs b/crates/codegen/kigi-shell/src/agent/mvp_agent/subagent_coordinator.rs index 3d96f8a..422e3d0 100644 --- a/crates/codegen/kigi-shell/src/agent/mvp_agent/subagent_coordinator.rs +++ b/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 { + 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(), diff --git a/crates/codegen/kigi-shell/src/agent/subagent/handle_request.rs b/crates/codegen/kigi-shell/src/agent/subagent/handle_request.rs index 177b226..4bd58c0 100644 --- a/crates/codegen/kigi-shell/src/agent/subagent/handle_request.rs +++ b/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(); diff --git a/crates/codegen/kigi-shell/src/agent/subagent/mod.rs b/crates/codegen/kigi-shell/src/agent/subagent/mod.rs index 1943943..20047b2 100644 --- a/crates/codegen/kigi-shell/src/agent/subagent/mod.rs +++ b/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>, + /// 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, #[expect( dead_code, reason = "unused in production; remove expect when wired or delete the item" diff --git a/crates/codegen/kigi-shell/src/test_support/lsp_runtime.rs b/crates/codegen/kigi-shell/src/test_support/lsp_runtime.rs index 66f9a25..5222106 100644 --- a/crates/codegen/kigi-shell/src/test_support/lsp_runtime.rs +++ b/crates/codegen/kigi-shell/src/test_support/lsp_runtime.rs @@ -33,6 +33,12 @@ pub(crate) fn ctx_with_toggle(toggle: HashMap) -> 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(), diff --git a/crates/codegen/kigi-tools/src/implementations/kigi/task/admission.rs b/crates/codegen/kigi-tools/src/implementations/kigi/task/admission.rs new file mode 100644 index 0000000..26901d2 --- /dev/null +++ b/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) -> 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); + } +} diff --git a/crates/codegen/kigi-tools/src/implementations/kigi/task/mod.rs b/crates/codegen/kigi-tools/src/implementations/kigi/task/mod.rs index dcb9c3f..a041502 100644 --- a/crates/codegen/kigi-tools/src/implementations/kigi/task/mod.rs +++ b/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; diff --git a/crates/codegen/kigi-tools/src/util/env.rs b/crates/codegen/kigi-tools/src/util/env.rs index aa9ba6d..4f0c0f6 100644 --- a/crates/codegen/kigi-tools/src/util/env.rs +++ b/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 { + value.parse::().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) -> Option { + 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