zesdex/src/app/workflow/engine.rs

//! Workflow engine: interprets `ScriptPrimitive` values (agent, parallel,
//! pipeline, phase) by spawning subagents, collecting results, and
//! managing concurrency.
//!
//! Key design points:
//! - `Parallel` branches run concurrently (capped by semaphore) — this is
//!   the main advantage over single-turn chat.
//! - `Pipeline` branches run sequentially so each stage sees findings from
//!   the previous one.
//! - `run_workflow_tracked` accepts a `LiveState` callback that receives
//!   real-time agent status updates for the TUI panel.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use serde::{Deserialize, Serialize};
use super::script::{ScriptPrimitive, WorkflowScript};

static FINDINGS: Mutex<Vec<String>> = Mutex::new(Vec::new());

/// The lifecycle state of an agent within a workflow run.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum AgentState {
    Idle,
    Running,
    Completed,
    Failed,
}

/// Timestamped status of one workflow agent.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AgentStatus {
    pub state: AgentState,
    pub started_at: Option<i64>,
    pub completed_at: Option<i64>,
    pub error: Option<String>,
}

/// A single agent tracked within a workflow run.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowAgent {
    pub id: String,
    pub name: String,
    pub status: AgentStatus,
}

/// Orchestrator for running workflow scripts: holds agent roster and a
/// shared finding accumulator visible to all pipeline stages.
#[derive(Debug, Clone)]
pub struct WorkflowEngine {
    pub agents: Vec<WorkflowAgent>,
    pub findings: Vec<String>,
}

impl WorkflowEngine {
    /// Create an empty workflow engine with no agents or findings.
    pub fn new() -> Self {
        WorkflowEngine {
            agents: Vec::new(),
            findings: Vec::new(),
        }
    }
}

/// Shared live state used by `run_workflow_tracked` to push real-time
/// agent status updates into the TUI's `WorkflowEngine`.
///
/// The closure receives `(agent_id, new_status)` and should update the
/// corresponding agent in `AppStateRest::workflow_engine`.
pub type LiveStateFn = Arc<dyn Fn(String, AgentStatus) + Send + Sync>;

/// Spawn a single synchronous subagent with the given prompt, passing it
/// any findings from earlier sibling agents. Updates live state before and
/// after to reflect Running → Completed/Failed transitions.
///
/// Flow: push agent as `Running` → build SubagentContext with prompt +
/// findings preamble → call `run_subagent` (draining the event channel into
/// a throwaway consumer so events are not blocked) → push `Completed` or
/// `Failed`.
///
/// Return: the agent's text output, or an error on failure.
fn spawn_single_agent(
    agent_id: &str,
    agent_name: &str,
    prompt: &str,
    findings_snapshot: Vec<String>,
    live: Option<&LiveStateFn>,
) -> anyhow::Result<String> {
    use crate::app::subagent::context::build_subagent_context;
    use crate::app::subagent::engine::run_subagent;
    use crate::app::subagent::spawn::AgentDefinition;

    let started_at = chrono::Utc::now().timestamp_millis();

    // Notify UI: this agent is now running
    if let Some(f) = live {
        f(agent_id.to_string(), AgentStatus {
            state: AgentState::Running,
            started_at: Some(started_at),
            completed_at: None,
            error: None,
        });
    }

    let def = AgentDefinition::new(agent_name.to_string(), "coder".to_string())
        .with_max_steps(50);
    let mut ctx = build_subagent_context(def);

    let findings_section = if findings_snapshot.is_empty() {
        String::new()
    } else {
        format!(
            "\n\nFindings from sibling agents in this workflow run:\n{}",
            findings_snapshot
                .iter()
                .enumerate()
                .map(|(i, f)| format!("{}. {}", i + 1, f))
                .collect::<Vec<_>>()
                .join("\n")
        )
    };

    ctx.system_prompt = format!("{}{}", prompt, findings_section);

    // Create an mpsc channel and drain events in a background thread so
    // run_subagent's blocking_send never blocks (previously the _rx was
    // dropped immediately, which would cause blocking_send to panic/fail
    // on a closed channel).
    let (tx, rx) = tokio::sync::mpsc::channel(64);
    let _drain_thread = std::thread::spawn(move || {
        // Drain all events; we don't surface them individually to the UI
        // (the live state callbacks handle coarse-grained status).
        let mut rx = rx;
        while rx.blocking_recv().is_some() {}
    });

    let result = run_subagent(ctx, tx);

    let completed_at = chrono::Utc::now().timestamp_millis();

    // Notify UI: agent completed or failed
    if let Some(f) = live {
        match &result {
            Ok(_) => f(agent_id.to_string(), AgentStatus {
                state: AgentState::Completed,
                started_at: Some(started_at),
                completed_at: Some(completed_at),
                error: None,
            }),
            Err(e) => f(agent_id.to_string(), AgentStatus {
                state: AgentState::Failed,
                started_at: Some(started_at),
                completed_at: Some(completed_at),
                error: Some(e.to_string()),
            }),
        }
    }

    result
}

type ParallelResult = (usize, anyhow::Result<Vec<String>>);

/// Recursively execute a `ScriptPrimitive` tree, respecting an overall
/// concurrency cap for parallel branches.
///
/// Flow: match the primitive →
///   `Agent` → `spawn_single_agent`
///   `Parallel` → spawn threads up to `concurrency_cap` (semaphore-gated),
///               collect results in submission order
///   `Pipeline` → execute stages sequentially; findings flow between stages
///   `Phase` → recurse (pass-through wrapper)
///
/// Why: `Parallel` uses OS threads + a semaphore so the main async event
/// loop remains responsive. `Pipeline` is sequential so each stage sees
/// findings deposited by the previous one.
///
/// Return: a `Vec<String>` of all agent outputs (or error strings) in
/// the order they were submitted.
pub fn execute_primitive(
    primitive: &ScriptPrimitive,
    args: &HashMap<String, String>,
    concurrency_cap: usize,
    continue_on_error: bool,
    live: Option<&LiveStateFn>,
) -> anyhow::Result<Vec<String>> {
    match primitive {
        ScriptPrimitive::Agent(prompt) => {
            let resolved = resolve_template(prompt, args);
            let findings_snapshot = FINDINGS.lock().map(|f| f.clone()).unwrap_or_default();
            let agent_id = uuid::Uuid::new_v4().to_string();
            let agent_name = resolved.chars().take(40).collect::<String>();
            match spawn_single_agent(&agent_id, &agent_name, &resolved, findings_snapshot, live) {
                Ok(text) => Ok(vec![text]),
                Err(e) => {
                    if continue_on_error {
                        Ok(vec![format!("agent error: {}", e)])
                    } else {
                        Err(e)
                    }
                }
            }
        }

        ScriptPrimitive::Parallel(scripts) => {
            // All branches run concurrently, capped by semaphore.
            // This is the primary advantage over single-turn chat: multiple
            // independent subagents work simultaneously.
            let semaphore = Arc::new(Semaphore::new(concurrency_cap.max(1)));
            let results: Arc<Mutex<Vec<ParallelResult>>> =
                Arc::new(Mutex::new(Vec::new()));

            let handles: Vec<_> = scripts
                .iter()
                .enumerate()
                .map(|(idx, script)| {
                    let script = script.clone();
                    let args = args.clone();
                    let sem = Arc::clone(&semaphore);
                    let results = Arc::clone(&results);
                    let cap = concurrency_cap;
                    let live_clone = live.cloned();

                    std::thread::spawn(move || {
                        let _permit = sem.acquire();
                        let result = execute_primitive(
                            &script, &args, cap, continue_on_error,
                            live_clone.as_ref(),
                        );
                        if let Ok(mut locked) = results.lock() {
                            locked.push((idx, result));
                        }
                    })
                })
                .collect();

            for handle in handles {
                let _ = handle.join();
            }

            let mut locked = results.lock().map_err(|_| anyhow::anyhow!("parallel results lock poisoned"))?;
            locked.sort_by_key(|(idx, _)| *idx);
            let mut all = Vec::new();
            for (_, res) in locked.drain(..) {
                match res {
                    Ok(outputs) => all.extend(outputs),
                    Err(e) => all.push(format!("agent error: {}", e)),
                }
            }
            Ok(all)
        }

        ScriptPrimitive::Pipeline(scripts) => {
            // Sequential: each stage runs only after the previous completes.
            //
            // Why: parallel execution defeats the purpose of a pipeline whose
            // stages are supposed to build on each other's output. Findings
            // written by stage N are visible to stage N+1 because we share
            // the global FINDINGS mutex.
            let mut all = Vec::new();
            for (idx, script) in scripts.iter().enumerate() {
                match execute_primitive(script, args, concurrency_cap, continue_on_error, live) {
                    Ok(outputs) => all.extend(outputs),
                    Err(e) => {
                        if continue_on_error {
                            all.push(format!("pipeline stage {} error: {}", idx, e));
                        } else {
                            return Err(e);
                        }
                    }
                }
            }
            Ok(all)
        }

        ScriptPrimitive::Phase { name: _name, script } => {
            execute_primitive(script, args, concurrency_cap, continue_on_error, live)
        }
    }
}

/// Run a `WorkflowScript` with the given template arguments and produce a
/// summary string. Uses no live-state callback.
///
/// Return: a human-readable summary string.
pub fn run_workflow(script: &WorkflowScript, args: &HashMap<String, String>) -> anyhow::Result<String> {
    run_workflow_tracked(script, args, None)
}

/// Run a `WorkflowScript` with real-time live-state callbacks so the TUI
/// panel updates as each agent transitions between Idle/Running/Done/Failed.
///
/// Flow: clear the global finding store → cap concurrency to 8 → call
/// `execute_primitive` with the live callback → format results.
///
/// Return: a human-readable summary string.
pub fn run_workflow_tracked(
    script: &WorkflowScript,
    args: &HashMap<String, String>,
    live: Option<LiveStateFn>,
) -> anyhow::Result<String> {
    if let Ok(mut findings) = FINDINGS.lock() {
        findings.clear();
    }

    let concurrency_cap = if script.options.max_concurrency > 0 {
        script.options.max_concurrency.min(8) // allow up to 8 parallel agents
    } else {
        4
    };

    let live_ref = live.as_ref();
    let results = execute_primitive(
        &script.script, args, concurrency_cap,
        script.options.continue_on_error, live_ref,
    )?;

    let summary = if results.is_empty() {
        "workflow completed with no output".to_string()
    } else {
        format!(
            "workflow '{}' completed. {} agent result(s):\n{}",
            script.name,
            results.len(),
            results
                .iter()
                .enumerate()
                .map(|(i, r)| format!("[{}] {}", i + 1, r.lines().next().unwrap_or(r)))
                .collect::<Vec<_>>()
                .join("\n")
        )
    };

    Ok(summary)
}

/// Add a finding text to the global workflow findings list, making it
/// visible to sibling agents spawned later in the same run.
pub fn note_finding(text: &str) {
    if let Ok(mut findings) = FINDINGS.lock() {
        findings.push(text.to_string());
    }
}

/// Simple template engine: replace `{{key}}` placeholders with values
/// from `args`.
///
/// Why: a structured template engine is unnecessary for the limited
/// use-case; this is intentionally simple and safe.
fn resolve_template(template: &str, args: &HashMap<String, String>) -> String {
    let mut result = template.to_string();
    for (key, value) in args {
        result = result.replace(&format!("{{{{{}}}}}", key), value);
    }
    result
}

/// A counting semaphore built from a `Mutex` + `Condvar`.
///
/// Used by `execute_primitive` to cap concurrent parallel branches.
struct Semaphore {
    count: Mutex<usize>,
    condvar: std::sync::Condvar,
}

impl Semaphore {
    fn new(count: usize) -> Self {
        Semaphore {
            count: Mutex::new(count),
            condvar: std::sync::Condvar::new(),
        }
    }

    fn acquire(&self) -> SemaphoreGuard<'_> {
        let mut count = self.count.lock().unwrap();
        while *count == 0 {
            count = self.condvar.wait(count).unwrap();
        }
        *count -= 1;
        SemaphoreGuard { sem: self }
    }
}

struct SemaphoreGuard<'a> {
    sem: &'a Semaphore,
}

impl<'a> Drop for SemaphoreGuard<'a> {
    fn drop(&mut self) {
        let mut count = self.sem.count.lock().unwrap();
        *count += 1;
        self.sem.condvar.notify_one();
    }
}