zesdex/apps/infrastructure/src/workflow/hive_mind/cycle.rs

//! Hive-mind cycle execution — run one cycle of parallel nodes.
//!
//! Flow: load settings → resolve LLM credentials → run all directives in the
//! cycle concurrently via a BOUNDED buffer (`buffer_unordered(MAX)`) → collect
//! `Vec<NodeOutput>`. 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::stream::StreamExt;
use tracing::{info, warn};

use zesdex_domain::cms::{AppConfigRepository, SettingsRepository};
use zesdex_domain::core::Store;

use crate::persistence::{JsonAppConfigRepository, JsonSettingsRepository};
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, 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 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<Vec<NodeOutput>> {
    info!(
        "Executing cycle {} with {} directives",
        cycle.index,
        cycle.directives.len()
    );

    let store = Store::new();
    let settings = JsonSettingsRepository::new()
        .load(&store.base_dir)
        .unwrap_or_default();
    let app_config = JsonAppConfigRepository::new()
        .load(&store.base_dir)
        .unwrap_or_default();

    let (provider, model) =
        crate::subagent::provider::resolve_subagent_provider(&settings, &app_config);

    let base_url = app_config
        .providers
        .get(&provider)
        .map(|p| p.api_base.clone())
        .unwrap_or_else(|| zesdex_domain::agent::defaults::DEFAULT_API_BASE.to_string());

    let api_key = crate::llm::provider::resolve_api_key(&settings, &app_config);

    let cycle_index = cycle.index;

    // 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()
        .map(|(i, node_dir): (usize, &NodeDirective)| {
            let dir = node_dir.directive.clone();
            let access_tier = node_dir.access_tier.clone();
            let ctx = SubagentContext::new(
                dir.clone(),
                tool_ctx.clone(),
                access_tier.clone(),
                base_url.clone(),
                api_key.clone(),
                model.clone(),
            );
            let tc = tool_ctx.clone();

            let access = match access_tier.as_str() {
                "write" => AccessTier::Write,
                "full" => AccessTier::Full,
                _ => AccessTier::Read,
            };

            let node_id = format!("Node-{}-{}", cycle_index, i);
            async move {
                match run_agent(ctx, &dir, access, tc).await {
                    Ok(output) => Ok::<NodeOutput, anyhow::Error>(NodeOutput {
                        id: node_id.clone(),
                        directive: dir,
                        output,
                    }),
                    Err(e) => {
                        warn!(node = %node_id, error = %e, "hive-mind node failed (isolated)");
                        Ok::<NodeOutput, anyhow::Error>(NodeOutput {
                            id: node_id,
                            directive: dir,
                            output: format!("[ERROR] {e}"),
                        })
                    }
                }
            }
        })
        .collect();

    // 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)
}