fix(startup): thread-exhaustion resilience — cap workers, never panic

Port of upstream grok-build 0.2.113/0.2.114 fixes, extended to Kigi's
startup order.

- kigi-tty-utils runtime module: worker-cap policy (8 max) — tokio's
  per-core default pins 100+ thread slots on many-core shared hosts
  against RLIMIT_NPROC / pids.max, and later spawns die with EAGAIN.
- kigi-bin builds the main runtime with capped workers and reports an
  out-of-threads host with a diagnostic + exit(1) instead of a panic.
- The pre-runtime optimization threads (HTTP warmup, early model
  prefetch) spawn recoverably and degrade — their panic would have
  preempted the runtime's own diagnostic.
- The per-session actor thread and its runtime report spawn/build
  failure through the existing acp::Error / AgentBuildError boundaries
  (new RuntimeBuild variant) instead of panicking the agent host.
- fsnotify's watcher thread spawn maps into FsNotifyError::WatcherStart
  so callers degrade instead of crashing.

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

crates/codegen/kigi-agent/src/error.rs

@@ -17,6 +17,11 @@ pub enum AgentBuildError {
     #[error("IO error during agent construction: {0}")]
     IoError(#[from] std::io::Error),
 
+    /// The per-session runtime could not be built — fd or thread
+    /// exhaustion (EMFILE/EAGAIN), not a definition problem.
+    #[error("failed to build the session runtime: {0}")]
+    RuntimeBuild(std::io::Error),
+
     /// Carries template line numbers and surrounding context.
     #[error("template rendering error: {0}")]
     MiniJinjaError(#[from] minijinja::Error),

crates/codegen/kigi-bin/src/main.rs

@@ -1221,10 +1221,26 @@ fn main() {
             "Found crashed sessions from a previous run"
         );
     }
-    let runtime = tokio::runtime::Builder::new_multi_thread()
+    // Capped workers: tokio's per-core default pins 100+ thread slots on
+    // many-core shared hosts and later spawns die with EAGAIN.
+    let runtime = match tokio::runtime::Builder::new_multi_thread()
+        .worker_threads(kigi_tty_utils::runtime::capped_worker_threads().get())
         .enable_all()
         .build()
-        .unwrap_or_else(|e| panic!("failed to start tokio runtime: {e}"));
+    {
+        Ok(runtime) => runtime,
+        Err(e) => {
+            // A host out of threads (RLIMIT_NPROC / pids.max) is an
+            // environment problem, not a bug — report it, don't panic.
+            kigi_tty_utils::restore_native_stderr();
+            eprintln!("Error: failed to start the async runtime: {e}");
+            eprintln!(
+                "The host may be out of threads (RLIMIT_NPROC or the cgroup pids ceiling); \
+                 close other processes or raise the limit, then retry."
+            );
+            std::process::exit(1);
+        }
+    };
     let result = run_and_shutdown(runtime, async_main(), RUNTIME_SHUTDOWN_GRACE);
     kigi_log::debug_log::flush();
     if let Err(e) = result {

crates/codegen/kigi-fsnotify/src/watcher.rs

@@ -971,7 +971,7 @@ pub(crate) fn start_with_timeout(
 
     let progress_for_thread = progress.clone();
 
-    let thread = std::thread::spawn(move || {
+    let watcher_loop = move || {
         let update_stage = |stage: &'static str| {
             if let Ok(mut p) = progress_for_thread.lock() {
                 p.set_stage(stage);
@@ -1351,7 +1351,13 @@ pub(crate) fn start_with_timeout(
                 let _ = ready_tx.send(Err(Box::new(e)));
             }
         }
-    });
+    };
+    // Recoverable spawn: a host out of threads must surface as an error the
+    // caller can degrade on, not a `std::thread::spawn` panic.
+    let thread = std::thread::Builder::new()
+        .name("fsnotify-watcher".into())
+        .spawn(watcher_loop)
+        .map_err(|e| crate::FsNotifyError::WatcherStart(Box::new(e)))?;
 
     // Wait for watcher to be ready (with timeout)
     if let Ok(mut p) = progress.lock() {

crates/codegen/kigi-shell/src/agent/models.rs

@@ -1938,7 +1938,7 @@ fn resolve_prefetch_env(kimi_code_config: Option<KimiCodeConfig>) -> Option<Pref
 /// credentials from disk.
 pub fn start_early_prefetch_with_auth(auth: Option<KimiAuth>) -> Option<EarlyPrefetchHandle> {
     let env = resolve_prefetch_env_with_auth(auth)?;
-    Some(spawn_prefetch_thread(env))
+    spawn_prefetch_thread(env)
 }
 
 /// Start the model-catalog prefetch on a background thread.
@@ -1949,11 +1949,16 @@ pub fn start_early_prefetch(
     kimi_code_config: Option<KimiCodeConfig>,
 ) -> Option<EarlyPrefetchHandle> {
     let env = resolve_prefetch_env(kimi_code_config)?;
-    Some(spawn_prefetch_thread(env))
+    spawn_prefetch_thread(env)
 }
 
-fn spawn_prefetch_thread(env: PrefetchEnv) -> EarlyPrefetchHandle {
-    std::thread::spawn(move || {
+/// `None` when the OS refuses a thread: the prefetch is an optimization and
+/// runs before the runtime builds, so a spawn panic here would preempt the
+/// runtime's own out-of-threads diagnostic.
+fn spawn_prefetch_thread(env: PrefetchEnv) -> Option<EarlyPrefetchHandle> {
+    let spawned = std::thread::Builder::new()
+        .name("early-prefetch".into())
+        .spawn(move || {
         let mut timer = crate::instrumentation_timer!("startup.early_prefetch");
         let proxy_endpoint = env.endpoints.proxy_url();
         timer.with_field("endpoint", proxy_endpoint.as_str());
@@ -1993,7 +1998,14 @@ fn spawn_prefetch_thread(env: PrefetchEnv) -> EarlyPrefetchHandle {
         }
 
         EarlyPrefetchResult { models }
-    })
+    });
+    match spawned {
+        Ok(handle) => Some(handle),
+        Err(e) => {
+            tracing::error!(error = %e, "early prefetch thread spawn failed; skipping prefetch");
+            None
+        }
+    }
 }
 
 /// Map a model id (catalog key or routing slug) to its catalog key.

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

@@ -726,10 +726,18 @@ pub struct MvpAgent {
 /// sampling client is discarded, but the TLS root certificates it loads
 /// are cached at the process level by `rustls-native-certs`.
 pub fn warm_async_http_client() {
-    std::thread::spawn(|| {
-        let _timer = crate::instrumentation_timer!("startup.async_http_warmup");
-        let _ = crate::http::shared_client();
-    });
+    // Recoverable spawn: the warmup is an optimization that runs before the
+    // runtime builds — an out-of-threads host must reach the runtime's own
+    // diagnostic, not panic here.
+    let spawned = std::thread::Builder::new()
+        .name("http-warmup".into())
+        .spawn(|| {
+            let _timer = crate::instrumentation_timer!("startup.async_http_warmup");
+            let _ = crate::http::shared_client();
+        });
+    if let Err(e) = spawned {
+        tracing::warn!(error = %e, "async HTTP warmup thread spawn failed; first request warms instead");
+    }
 }
 pub(crate) fn resolve_required_agent_type(
     model_agent_type: Option<&str>,

crates/codegen/kigi-shell/src/session/acp_session_impl/spawn.rs

@@ -1639,10 +1639,23 @@ pub(crate) async fn spawn_session_on_thread(
                 };
                 (initial_last_compaction, initial_prompt_texts)
             };
-            let rt = tokio::runtime::Builder::new_current_thread()
+            // Runtime construction acquires fds (epoll/kqueue, waker) and
+            // fails with EMFILE/EAGAIN under resource pressure — report
+            // through the init channel instead of panicking the thread.
+            let rt = match tokio::runtime::Builder::new_current_thread()
                 .enable_all()
                 .build()
-                .expect("session runtime");
+            {
+                Ok(rt) => rt,
+                Err(e) => {
+                    tracing::error!(
+                        error = %e,
+                        "failed to build session runtime (resource exhaustion?)"
+                    );
+                    let _ = init_tx.send(Err(kigi_agent::AgentBuildError::RuntimeBuild(e)));
+                    return;
+                }
+            };
             let local = tokio::task::LocalSet::new();
             local.block_on(&rt, async move {
                 let _trace_span = parent_traceparent.as_ref().map(|tp| {
@@ -1763,8 +1776,19 @@ pub(crate) async fn spawn_session_on_thread(
                 }));
                 let _ = session_done_rx.await;
             });
-        })
-        .expect("spawn session thread");
+        });
+    let join_handle = match join_handle {
+        Ok(handle) => handle,
+        Err(e) => {
+            tracing::error!(
+                error = %e,
+                "failed to spawn session thread (thread/PID limit or memory pressure?)"
+            );
+            return Err(
+                acp::Error::internal_error().data(format!("failed to spawn session thread: {e}"))
+            );
+        }
+    };
     let init = init_rx
         .await
         .map_err(|_| {

crates/codegen/kigi-tty-utils/src/lib.rs

@@ -40,6 +40,7 @@ use std::collections::HashMap;
 use std::io;
 
 mod process_scope;
+pub mod runtime;
 pub use process_scope::{ProcessScope, global_process_scope};
 
 // TTY detach — pre_exec building block

crates/codegen/kigi-tty-utils/src/runtime.rs

@@ -0,0 +1,55 @@
+//! Worker-thread policy for multi-thread tokio runtimes.
+//!
+//! Tokio defaults to one worker per core. On many-core shared hosts (100+-core
+//! HPC login nodes) that pins 100+ thread slots per kigi process against
+//! per-user ceilings — systemd user-slice `pids.max` (commonly 1000) or
+//! `RLIMIT_NPROC` — and later thread spawns die with EAGAIN. Kigi's runtimes
+//! are I/O-bound, so throughput does not scale with workers past a small
+//! count.
+//!
+//! This is the single home for the cap policy; every multi-thread runtime in
+//! the workspace derives its worker count from here so the policy cannot
+//! drift across crates.
+
+use std::num::NonZeroUsize;
+
+/// Maximum runtime worker threads for any kigi process.
+pub const MAX_WORKER_THREADS: NonZeroUsize = NonZeroUsize::new(8).unwrap();
+
+/// Pure, testable: `min(cores, MAX_WORKER_THREADS)`.
+pub fn cap_worker_threads(cores: NonZeroUsize) -> NonZeroUsize {
+    cores.min(MAX_WORKER_THREADS)
+}
+
+/// Reads the host: `min(available_parallelism, MAX_WORKER_THREADS)`.
+pub fn capped_worker_threads() -> NonZeroUsize {
+    cap_worker_threads(std::thread::available_parallelism().unwrap_or(NonZeroUsize::MIN))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn nz(n: usize) -> NonZeroUsize {
+        NonZeroUsize::new(n).unwrap()
+    }
+
+    #[test]
+    fn cap_is_identity_at_or_below_max() {
+        assert_eq!(cap_worker_threads(nz(1)), nz(1));
+        assert_eq!(cap_worker_threads(nz(4)), nz(4));
+        assert_eq!(cap_worker_threads(nz(8)), nz(8));
+    }
+
+    #[test]
+    fn cap_clamps_many_core_hosts() {
+        assert_eq!(cap_worker_threads(nz(9)), nz(8));
+        assert_eq!(cap_worker_threads(nz(360)), nz(8));
+    }
+
+    #[test]
+    fn capped_worker_threads_stays_in_bounds() {
+        let n = capped_worker_threads();
+        assert!(n <= MAX_WORKER_THREADS, "got {n}");
+    }
+}