diff --git a/Cargo.lock b/Cargo.lock index ca42a32..4c1ac04 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4862,7 +4862,7 @@ dependencies = [ [[package]] name = "zesdex-api" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "argon2", @@ -4885,7 +4885,7 @@ dependencies = [ [[package]] name = "zesdex-application" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "base64", @@ -4902,7 +4902,7 @@ dependencies = [ [[package]] name = "zesdex-bootstrap" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "chrono", @@ -4919,7 +4919,7 @@ dependencies = [ [[package]] name = "zesdex-daemon" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "base64", @@ -4943,7 +4943,7 @@ dependencies = [ [[package]] name = "zesdex-domain" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "base64", @@ -4959,7 +4959,7 @@ dependencies = [ [[package]] name = "zesdex-gateway" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "axum", @@ -4986,7 +4986,7 @@ dependencies = [ [[package]] name = "zesdex-grpc" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "axum", @@ -5003,7 +5003,7 @@ dependencies = [ [[package]] name = "zesdex-infrastructure" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "argon2", @@ -5051,7 +5051,7 @@ dependencies = [ [[package]] name = "zesdex-tui" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "base64", @@ -5077,7 +5077,7 @@ dependencies = [ [[package]] name = "zesdex-web" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "axum", @@ -5097,7 +5097,7 @@ dependencies = [ [[package]] name = "zesdex-ws" -version = "1.18.4" +version = "1.19.0" dependencies = [ "anyhow", "axum", diff --git a/apps/infrastructure/src/best_practice/explore.rs b/apps/infrastructure/src/best_practice/explore.rs index 2d88e20..5703c37 100644 --- a/apps/infrastructure/src/best_practice/explore.rs +++ b/apps/infrastructure/src/best_practice/explore.rs @@ -15,7 +15,7 @@ //! read of up to 3 relevant files). //! 4. Join the result and return a concise bullet summary as a tool message. -use anyhow::{Context, Result}; +use anyhow::Result; use serde_json::{json, Value}; use tracing::{info, warn}; @@ -118,7 +118,7 @@ impl Tool for ExploreCodebase { model, ); - let rt = tokio::runtime::Runtime::new().context("create explore tokio runtime")?; + let rt = crate::runtime::runtime(); let result = rt.block_on(run_agent( subagent_ctx, &directive, diff --git a/apps/infrastructure/src/lib.rs b/apps/infrastructure/src/lib.rs index 029a2c3..41f83fd 100644 --- a/apps/infrastructure/src/lib.rs +++ b/apps/infrastructure/src/lib.rs @@ -34,6 +34,7 @@ pub mod llm; pub mod mcp; pub mod middleware; pub mod persistence; +pub mod runtime; pub mod subagent; pub mod tools; pub mod utils; diff --git a/apps/infrastructure/src/runtime.rs b/apps/infrastructure/src/runtime.rs new file mode 100644 index 0000000..8233ce5 --- /dev/null +++ b/apps/infrastructure/src/runtime.rs @@ -0,0 +1,62 @@ +//! Process-wide shared Tokio runtime for sync → async bridging. +//! +//! Many `Tool::run` implementations are synchronous but need to drive async +//! work (LLM calls, subagent execution). Creating a fresh +//! [`tokio::runtime::Runtime`] on every call is expensive (spawns a thread +//! pool + runtime each time) and can fail randomly under thread pressure. +//! +//! # Flow +//! +//! [`runtime()`] returns a lazily-initialised process-wide runtime created +//! exactly once via [`std::sync::OnceLock`]. Callers use +//! `runtime().block_on(...)` exactly like they would with a local runtime — +//! the only difference is the runtime is shared, so the cost is paid once per +//! process instead of once per tool call. +//! +//! # Safety +//! +//! `block_on` panics if called from within a running Tokio runtime. The +//! tools that use this helper are synchronous (`Tool::run`), so this is safe +//! in practice. Async code should never call `runtime().block_on`. + +use std::sync::OnceLock; + +/// Maximum worker threads for the shared runtime. Kept modest — tools are +/// mostly I/O-bound and rarely need more concurrency than this. +const RUNTIME_WORKER_THREADS: usize = 8; + +static SHARED_RUNTIME: OnceLock = OnceLock::new(); + +/// Return the process-wide shared Tokio runtime, initialising it on first use. +/// +/// The runtime is configured with `worker_threads = 8` and +/// `enable_all()` (time + IO drivers) so streams, timers, and network calls +/// all work. If initialisation fails (extremely rare — resource exhaustion at +/// startup), the process aborts with a clear message rather than returning +/// an error on every subsequent call. +pub fn runtime() -> &'static tokio::runtime::Runtime { + SHARED_RUNTIME.get_or_init(|| { + tokio::runtime::Builder::new_multi_thread() + .worker_threads(RUNTIME_WORKER_THREADS) + .thread_name("zesdex-shared-rt") + .enable_all() + .build() + .expect("failed to create shared tokio runtime") + }) +} + +#[cfg(test)] +mod tests { + use super::runtime; + + #[test] + fn runtime_is_singleton() { + assert!(std::ptr::eq(runtime(), runtime())); + } + + #[test] + fn runtime_blocks_and_resolves() { + let val = runtime().block_on(async { 6 * 7 }); + assert_eq!(val, 42); + } +} diff --git a/apps/infrastructure/src/subagent/engine.rs b/apps/infrastructure/src/subagent/engine.rs index 3eb2f0b..7b98ddc 100644 --- a/apps/infrastructure/src/subagent/engine.rs +++ b/apps/infrastructure/src/subagent/engine.rs @@ -23,6 +23,37 @@ use zesdex_domain::subagent_directive; /// Maximum number of tool-call iterations before the engine gives up. const MAX_ITERATIONS: u32 = 25; +/// A single tool-result message is truncated before entering the subagent's +/// context so it cannot blow the window (matches the main turn service). +const TOOL_OUTPUT_MAX_CHARS: usize = 12_000; + +/// Maximum consecutive identical tool errors before the engine injects a +/// recovery note steering the model to a different approach. +const MAX_CONSECUTIVE_TOOL_ERRORS: usize = 3; + +/// Pick a `max_tokens` budget proportional to the directive's length. +fn adaptive_max_tokens(directive_len: usize) -> u32 { + if directive_len <= 80 { + 800 + } else if directive_len <= 400 { + 1600 + } else { + 4096 + } +} + +fn truncate_tool_output(output: String) -> String { + if output.len() <= TOOL_OUTPUT_MAX_CHARS { + return output; + } + let mut result: String = output.chars().take(TOOL_OUTPUT_MAX_CHARS).collect(); + result.push_str(&format!( + "\n...[truncated {} chars]", + output.len() - TOOL_OUTPUT_MAX_CHARS + )); + result +} + /// Emit an `AgentProgress` event onto the turn-event queue, if one is /// configured in the `ToolCtx`. fn report_progress(tool_ctx: &ToolCtx, progress: AgentProgress) { @@ -85,11 +116,17 @@ pub async fn run_agent( Some(ctx.base_url.clone()), ); + let max_tokens = adaptive_max_tokens(directive.len()); + + // Track repeated tool errors so the agent can recover from a dead end. + let mut consecutive_errors = 0usize; + let mut last_tool = String::new(); + // Limited iteration loop so we don't run forever for iteration in 0..MAX_ITERATIONS { use zesdex_application::ports::ProviderService; let (response_msg, _usage) = client - .chat(&messages, Some(defs.clone()), Some(4096), None) + .chat(&messages, Some(defs.clone()), Some(max_tokens), Some(0.2)) .await?; let content = response_msg.content.clone().unwrap_or_default(); @@ -113,7 +150,7 @@ pub async fn run_agent( &tool_ctx, AgentProgress::running( "subagent", - format!("{}:{}", directive, tool_name), + format!("{}:{tool_name}", directive), Some(tool_name.clone()), ), ); @@ -127,7 +164,29 @@ pub async fn run_agent( format!("Unknown tool: {tool_name}") }; - messages.push(ChatMessage::tool(tc.id.clone(), result)); + // Error-recovery: if the same tool keeps failing, inject a + // system note steering the model to a different approach. + if result.starts_with("Error:") { + if last_tool.as_str() == tool_name.as_str() { + consecutive_errors += 1; + } else { + consecutive_errors = 1; + last_tool = tool_name.to_string(); + } + if consecutive_errors >= MAX_CONSECUTIVE_TOOL_ERRORS { + messages.push(ChatMessage::system( + zesdex_domain::agent::prompt::error_recovery_note(tool_name, &result), + )); + consecutive_errors = 0; + } + } else { + consecutive_errors = 0; + } + + messages.push(ChatMessage::tool( + tc.id.clone(), + truncate_tool_output(result), + )); } // Add assistant response if there was text content diff --git a/apps/infrastructure/src/subagent/spawn.rs b/apps/infrastructure/src/subagent/spawn.rs index 3745092..e03bebc 100644 --- a/apps/infrastructure/src/subagent/spawn.rs +++ b/apps/infrastructure/src/subagent/spawn.rs @@ -32,7 +32,6 @@ pub fn spawn_subagent( ) -> thread::JoinHandle> { info!("Spawning subagent: {directive}"); thread::spawn(move || { - let rt = tokio::runtime::Runtime::new()?; - rt.block_on(run_agent(ctx, &directive, access, tool_ctx)) + crate::runtime::runtime().block_on(run_agent(ctx, &directive, access, tool_ctx)) }) } diff --git a/apps/infrastructure/src/tools/parallel_delegate.rs b/apps/infrastructure/src/tools/parallel_delegate.rs index 461000e..ca17632 100644 --- a/apps/infrastructure/src/tools/parallel_delegate.rs +++ b/apps/infrastructure/src/tools/parallel_delegate.rs @@ -123,7 +123,7 @@ impl Tool for ParallelDelegate { .collect() } else { // Auto-split using LLM - let rt = tokio::runtime::Runtime::new()?; + let rt = crate::runtime::runtime(); let directives = rt.block_on(auto_split_task( &task, max_parallel, @@ -146,49 +146,53 @@ impl Tool for ParallelDelegate { "parallel delegation: starting subagents" ); - // Spawn agents in parallel - let mut handles = Vec::new(); - for (i, (directive, access)) in directives.iter().enumerate() { - let subagent_ctx = SubagentContext::new( - directive.clone(), - ctx.clone(), - format!("{access:?}"), - base_url.clone(), - api_key.clone(), - model.clone(), - ); - - debug!(agent_index = i, access = ?access, "spawning parallel agent"); - let handle = spawn_subagent(subagent_ctx, directive.clone(), *access, ctx.clone()); - handles.push((i, handle)); - } - - // Join all results + // Spawn agents in parallel — bounded: never more than `max_parallel` + // subagent threads in flight at once (Claude Code-style isolation). let mut results: Vec<(usize, String, String)> = Vec::new(); - for (i, handle) in handles { - match handle.join() { - Ok(Ok(output)) => { - info!(agent_index = i, "parallel agent completed"); - results.push((i, directives[i].0.clone(), output)); - } - Ok(Err(e)) => { - warn!(agent_index = i, error = %e, "parallel agent failed"); - results.push((i, directives[i].0.clone(), format!("[ERROR] {e}"))); - } - Err(e) => { - warn!(agent_index = i, error = ?e, "parallel agent panicked"); - results.push(( - i, - directives[i].0.clone(), - "[ERROR] Agent panicked".to_string(), - )); + for batch in directives.chunks(max_parallel) { + let mut handles = Vec::with_capacity(batch.len()); + for (i, (directive, access)) in batch.iter().enumerate() { + let global_idx = results.len() + i; + let subagent_ctx = SubagentContext::new( + directive.clone(), + ctx.clone(), + format!("{access:?}"), + base_url.clone(), + api_key.clone(), + model.clone(), + ); + + debug!(agent_index = global_idx, access = ?access, "spawning parallel agent"); + let handle = spawn_subagent(subagent_ctx, directive.clone(), *access, ctx.clone()); + handles.push((global_idx, handle)); + } + + // Join this batch before spawning the next. + for (i, handle) in handles { + match handle.join() { + Ok(Ok(output)) => { + info!(agent_index = i, "parallel agent completed"); + results.push((i, directives[i].0.clone(), output)); + } + Ok(Err(e)) => { + warn!(agent_index = i, error = %e, "parallel agent failed"); + results.push((i, directives[i].0.clone(), format!("[ERROR] {e}"))); + } + Err(e) => { + warn!(agent_index = i, error = ?e, "parallel agent panicked"); + results.push(( + i, + directives[i].0.clone(), + "[ERROR] Agent panicked".to_string(), + )); + } } } } // Consolidate results if synthesize && results.len() > 1 { - let rt = tokio::runtime::Runtime::new()?; + let rt = crate::runtime::runtime(); let consolidated = rt.block_on(consolidate_results(&results, &base_url, &api_key, &model))?; Ok(format!( diff --git a/apps/infrastructure/src/tools/utility/dir_cache_update.rs b/apps/infrastructure/src/tools/utility/dir_cache_update.rs index 94651c2..66ad4f4 100644 --- a/apps/infrastructure/src/tools/utility/dir_cache_update.rs +++ b/apps/infrastructure/src/tools/utility/dir_cache_update.rs @@ -63,7 +63,7 @@ impl crate::tools::Tool for DirCacheUpdate { // Persist the resolved paths into the shared DirCache so the TUI // and other tools can read the cached listing without re-scanning. let dc = ctx.dir_cache.clone(); - let rt = tokio::runtime::Runtime::new()?; + let rt = crate::runtime::runtime(); rt.block_on(async { dc.write().await.set(resolved).await }); info!(count, "directory cache updated"); diff --git a/apps/infrastructure/src/tools/workflow.rs b/apps/infrastructure/src/tools/workflow.rs index a0addb6..3aca073 100644 --- a/apps/infrastructure/src/tools/workflow.rs +++ b/apps/infrastructure/src/tools/workflow.rs @@ -65,7 +65,7 @@ impl Tool for WorkflowRun { zesdex_domain::agent::defaults::DEFAULT_MODEL.to_string(), None, ); - let rt = tokio::runtime::Runtime::new()?; + let rt = crate::runtime::runtime(); let result: Vec = rt.block_on(async { execute_workflow(&script, ctx, &llm_client).await })?; @@ -229,7 +229,7 @@ impl Tool for HiveMind { .ok_or_else(|| anyhow::anyhow!("missing 'cycles' array"))?; info!("Hive mind starting with {} cycles", cycles_val.len()); - let rt = tokio::runtime::Runtime::new()?; + let rt = crate::runtime::runtime(); let mut all_node_outputs = Vec::new(); for (cycle_idx, cycle_val) in cycles_val.iter().enumerate() { diff --git a/apps/infrastructure/src/workflow/hive_mind/cycle.rs b/apps/infrastructure/src/workflow/hive_mind/cycle.rs index eda2efa..d682691 100644 --- a/apps/infrastructure/src/workflow/hive_mind/cycle.rs +++ b/apps/infrastructure/src/workflow/hive_mind/cycle.rs @@ -1,11 +1,15 @@ //! Hive-mind cycle execution — run one cycle of parallel nodes. //! //! Flow: load settings → resolve LLM credentials → run all directives in the -//! cycle concurrently via try_join_all → collect Vec. +//! cycle concurrently via a BOUNDED buffer (`buffer_unordered(MAX)`) → collect +//! `Vec`. Unlike `try_join_all`, a single failing node does NOT +//! fail the whole cycle — failed nodes are logged and replaced with an +//! `[ERROR]` output so the remaining results are preserved (like Claude +//! Code's isolated subagents). use anyhow::Result; -use futures_util::future::try_join_all; -use tracing::info; +use futures_util::stream::StreamExt; +use tracing::{info, warn}; use zesdex_domain::cms::{AppConfigRepository, SettingsRepository}; use zesdex_domain::core::Store; @@ -15,16 +19,21 @@ use crate::subagent::context::SubagentContext; use crate::subagent::division::AccessTier; use crate::subagent::engine::run_agent; use crate::tools::ToolCtx; -use zesdex_domain::workflow::{CognitiveCycle, NodeOutput}; +use zesdex_domain::workflow::{CognitiveCycle, NodeDirective, NodeOutput}; + +/// Maximum number of hive-mind nodes running concurrently per cycle. +/// Keeps thread/runtime pressure bounded (Claude Code-style). +const MAX_CONCURRENT_NODES: usize = 8; /// Execute one cycle: run each node directive and collect outputs. /// /// Flow: /// 1. Load `Settings` and `AppConfig` from the store directory. /// 2. Resolve provider, model, base_url, and api_key. -/// 3. Spawn all directives concurrently — each builds a `SubagentContext` -/// and calls `run_agent` (Full access). -/// 4. `try_join_all` waits for all to complete, then collect `NodeOutput`s. +/// 3. Spawn directives with bounded concurrency — each builds a +/// `SubagentContext` and calls `run_agent`. +/// 4. Collect `NodeOutput`s; failed nodes are logged and replaced with an +/// `[ERROR]` placeholder so the cycle still completes. pub async fn execute_cycle(cycle: &CognitiveCycle, tool_ctx: &ToolCtx) -> Result> { info!( "Executing cycle {} with {} directives", @@ -53,10 +62,9 @@ pub async fn execute_cycle(cycle: &CognitiveCycle, tool_ctx: &ToolCtx) -> Result let cycle_index = cycle.index; - use zesdex_domain::workflow::NodeDirective; - - // Run all directives in this cycle concurrently. - let handles: Vec<_> = cycle + // Run all directives with bounded concurrency. Each node is its own + // future; failures are collected, not propagated (isolated errors). + let tasks: Vec<_> = cycle .directives .iter() .enumerate() @@ -79,18 +87,33 @@ pub async fn execute_cycle(cycle: &CognitiveCycle, tool_ctx: &ToolCtx) -> Result _ => AccessTier::Read, }; + let node_id = format!("Node-{}-{}", cycle_index, i); async move { - let result = run_agent(ctx, &dir, access, tc).await?; - Ok::(NodeOutput { - id: format!("Node-{}-{}", cycle_index, i), - directive: dir, - output: result, - }) + match run_agent(ctx, &dir, access, tc).await { + Ok(output) => Ok::(NodeOutput { + id: node_id.clone(), + directive: dir, + output, + }), + Err(e) => { + warn!(node = %node_id, error = %e, "hive-mind node failed (isolated)"); + Ok::(NodeOutput { + id: node_id, + directive: dir, + output: format!("[ERROR] {e}"), + }) + } + } } }) .collect(); - let results = try_join_all(handles).await?; + // Bounded concurrency: run at most MAX_CONCURRENT_NODES futures at once. + let mut stream = futures_util::stream::iter(tasks).buffer_unordered(MAX_CONCURRENT_NODES); + let mut results = Vec::with_capacity(cycle.directives.len()); + while let Some(node) = stream.next().await { + results.push(node?); + } Ok(results) } diff --git a/apps/interfaces/daemon/src/handler.rs b/apps/interfaces/daemon/src/handler.rs index 3f1d29d..1450c06 100644 --- a/apps/interfaces/daemon/src/handler.rs +++ b/apps/interfaces/daemon/src/handler.rs @@ -471,8 +471,7 @@ fn handle_compact(state: &mut AppStateRest) { let client = zesdex_infrastructure::llm::provider::LlmClient::new(api_key, model, api_base); if let Some(ref mut rt) = state.session_runtime { - let tokio_rt = - tokio::runtime::Runtime::new().expect("create tokio runtime for AI compaction"); + let tokio_rt = zesdex_infrastructure::runtime::runtime(); if let Ok(()) = tokio_rt.block_on( zesdex_application::agent::turn_service::compact_messages_with_ai( &mut rt.messages,