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.
//! - Findings (inter-agent notes) are scoped per invocation via an
//!   `Arc<Mutex<Vec<String>>>` threaded through `execute_primitive` and
//!   `spawn_single_agent` rather than a global static, preventing data
//!   leaks between concurrent workflow runs.

use std::collections::HashMap;
use std::sync::{Arc, Mutex, atomic::{AtomicBool, Ordering}};
use std::time::Duration;
use serde::{Deserialize, Serialize};
use super::script::{ScriptPrimitive, WorkflowScript};

/// 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>,
    /// Human-readable progress message (e.g. "editing src/main.rs",
    /// "running cargo test"). Shown in the TUI panel alongside the state.
    pub progress: 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, agent_name, new_status)`:
/// - `agent_id`: unique identifier (UUID) for upserting the agent.
/// - `agent_name`: human-readable display name for the TUI panel.
/// - `status`: the agent's lifecycle state and timing.
///
/// Callers should use `agent_id` as the stable key and `agent_name` for
/// display purposes (e.g. the division name in the company pipeline).
pub type LiveStateFn = Arc<dyn Fn(String, 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, linking the `workflow_findings` Arc so the subagent's
/// `note_finding` tool pushes into the same vec → call `run_subagent`
/// (draining the event channel into a consumer so events are not blocked)
/// → push `Completed` or `Failed`.
///
/// Why: the `workflow_findings` Arc is shared by all agents within the same
/// `execute_primitive` scope, so pipeline stages can pass data between each
/// other while different workflow invocations remain isolated.
///
/// When `timeout_ms` is `Some`, the subagent is killed (abandoned on a
/// separate thread) if it does not complete within the deadline, preventing
/// a stuck stage from blocking the entire pipeline forever.
///
/// Return: the agent's text output, or an error on failure.
#[allow(clippy::too_many_lines, clippy::too_many_arguments)]
fn spawn_single_agent(
    agent_id: &str,
    agent_name: &str,
    prompt: &str,
    findings_snapshot: &[String],
    findings: &Arc<Mutex<Vec<String>>>,
    abort_flag: &Option<Arc<AtomicBool>>,
    live: Option<&LiveStateFn>,
    session_dir: &std::path::Path,
    workspaces: &[std::path::PathBuf],
    timeout_ms: Option<u64>,
) -> 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.
    // Pass both the unique agent_id (UUID for stable key) and agent_name
    // (human-readable display name, e.g. division name).
    if let Some(f) = live {
        f(
            agent_id.to_string(),
            agent_name.to_string(),
            AgentStatus {
                state: AgentState::Running,
                started_at: Some(started_at),
                completed_at: None,
                error: None,
                progress: None,
            },
        );
    }

    let mut role = "coder".to_string();
    let mut allowed_tools = None;

    if agent_name.contains("Strategy") {
        let div_def = crate::app::subagent::division::strategy_division();
        role = div_def.role;
        allowed_tools = div_def.allowed_tools;
    } else if agent_name.contains("Engineering") {
        let div_def = crate::app::subagent::division::engineering_division();
        role = div_def.role;
        allowed_tools = div_def.allowed_tools;
    } else if agent_name.contains("Quality") {
        let div_def = crate::app::subagent::division::quality_division();
        role = div_def.role;
        allowed_tools = div_def.allowed_tools;
    } else if agent_name.contains("Security") {
        let div_def = crate::app::subagent::division::security_division();
        role = div_def.role;
        allowed_tools = div_def.allowed_tools;
    } else if agent_name.contains("Documentation") {
        let div_def = crate::app::subagent::division::documentation_division();
        role = div_def.role;
        allowed_tools = div_def.allowed_tools;
    }

    let mut def = AgentDefinition::new(agent_name.to_string(), role);
    if let Some(tools) = allowed_tools {
        def = def.with_allowed_tools(tools);
    }
    let mut ctx = build_subagent_context(&def);
    ctx.session_dir = session_dir.to_path_buf();
    ctx.workspaces = workspaces.to_vec();

    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}");
    // Link the shared findings Arc so note_finding calls within this
    // subagent write into the same vec visible to sibling agents.
    ctx.workflow_findings = Some(findings.clone());
    ctx.abort_flag = abort_flag.clone();

    // Create an mpsc channel and drain events in a background thread.
    // The drain thread also pushes intra-division progress updates to the
    // live callback (current tool being executed), so the TUI panel shows
    // real-time "editing X" or "running build" instead of just "Running…".
    let (tx, rx) = tokio::sync::mpsc::channel(64);
    let drain_agent_id = agent_id.to_string();
    let drain_agent_name = agent_name.to_string();
    let drain_live = live.cloned();
    let drain_started_at = started_at;
    let _drain_thread = std::thread::spawn(move || {
        use crate::app::subagent::event::SubagentEvent;
        let mut rx = rx;
        while let Some(event) = rx.blocking_recv() {
            match &event {
                SubagentEvent::ToolCall { tool, .. } => {
                    tracing::debug!("[subagent] tool call: {}", tool);
                    // Push intra-division progress: which tool is running
                    if let Some(ref f) = drain_live {
                        f(
                            drain_agent_id.clone(),
                            drain_agent_name.clone(),
                            AgentStatus {
                                state: AgentState::Running,
                                started_at: Some(drain_started_at),
                                completed_at: None,
                                error: None,
                                progress: Some(format!("tool: {tool}")),
                            },
                        );
                    }
                }
                SubagentEvent::ToolResult { tool, .. } => {
                    tracing::debug!("[subagent] tool result: {}", tool);
                    if let Some(ref f) = drain_live {
                        f(
                            drain_agent_id.clone(),
                            drain_agent_name.clone(),
                            AgentStatus {
                                state: AgentState::Running,
                                started_at: Some(drain_started_at),
                                completed_at: None,
                                error: None,
                                progress: Some(format!("done: {tool}")),
                            },
                        );
                    }
                }
                SubagentEvent::StepCompleted { .. } => {
                    tracing::trace!("[subagent] step completed");
                }
                SubagentEvent::StepFailed { step, error } => {
                    tracing::warn!("[subagent] step {} failed: {}", step, error);
                }
                SubagentEvent::Completed { .. } => {
                    tracing::debug!("[subagent] completed");
                }
            }
        }
    });

    // Check abort before even starting the subagent.
    if abort_flag.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
        anyhow::bail!("subagent '{agent_name}' aborted before start");
    }

    // Run subagent on a separate thread so the abort flag can be polled.
    // If abort is requested while the subagent is running, we abandon the
    // thread (Rust threads cannot be forcibly killed) and return early.
    let (done_tx, done_rx) = std::sync::mpsc::channel::<anyhow::Result<String>>();
    let bg_ctx = ctx;
    let bg_tx = tx;
    let bg_name = agent_name.to_string();
    let bg_abort = abort_flag.clone();
    std::thread::spawn(move || {
        let _ = done_tx.send(run_subagent(&bg_ctx, &bg_tx));
    });

    let poll_interval = Duration::from_millis(200);
    let result = if let Some(timeout) = timeout_ms {
        let deadline = Duration::from_millis(timeout);
        let mut elapsed = Duration::ZERO;
        loop {
            match done_rx.recv_timeout(poll_interval) {
                Ok(r) => break r,
                Err(_) => {
                    elapsed += poll_interval;
                    if elapsed >= deadline {
                        break Err(anyhow::anyhow!(
                            "subagent '{bg_name}' timed out after {timeout}ms",
                        ));
                    }
                    if bg_abort.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
                        break Err(anyhow::anyhow!(
                            "subagent '{bg_name}' aborted by user",
                        ));
                    }
                }
            }
        }
    } else {
        loop {
            match done_rx.recv_timeout(poll_interval) {
                Ok(r) => break r,
                Err(_) => {
                    if bg_abort.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
                        break Err(anyhow::anyhow!(
                            "subagent '{bg_name}' aborted by user",
                        ));
                    }
                }
            }
        }
    };

    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(),
                agent_name.to_string(),
                AgentStatus {
                    state: AgentState::Completed,
                    started_at: Some(started_at),
                    completed_at: Some(completed_at),
                    error: None,
                    progress: None,
                },
            ),
            Err(e) => f(
                agent_id.to_string(),
                agent_name.to_string(),
                AgentStatus {
                    state: AgentState::Failed,
                    started_at: Some(started_at),
                    completed_at: Some(completed_at),
                    error: Some(e.to_string()),
                    progress: None,
                },
            ),
        }
    }

    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. Findings are scoped to an
/// `Arc<Mutex<Vec<String>>>` rather than a global static, so concurrent
/// workflow runs are isolated from each other.
///
/// `timeout_ms` propagates to individual agents so that no single agent
/// can block the entire workflow beyond the configured deadline.
///
/// Return: a `Vec<String>` of all agent outputs (or error strings) in
/// the order they were submitted.
#[allow(clippy::too_many_arguments)]
pub fn execute_primitive(
    primitive: &ScriptPrimitive,
    args: &HashMap<String, String>,
    concurrency_cap: usize,
    continue_on_error: bool,
    abort_flag: &Option<Arc<AtomicBool>>,
    live: Option<&LiveStateFn>,
    session_dir: &std::path::Path,
    workspaces: &[std::path::PathBuf],
    findings: &Arc<Mutex<Vec<String>>>,
    timeout_ms: Option<u64>,
) -> anyhow::Result<Vec<String>> {
    match primitive {
        ScriptPrimitive::Agent(prompt) => {
            let mut resolved_args = args.clone();
            let findings_snapshot = findings.lock().map(|f| f.clone()).unwrap_or_default();
            if !resolved_args.contains_key("findings") {
                let formatted_findings = if findings_snapshot.is_empty() {
                    "None".to_string()
                } else {
                    findings_snapshot
                        .iter()
                        .enumerate()
                        .map(|(i, f)| format!("{}. {}", i + 1, f))
                        .collect::<Vec<_>>()
                        .join("\n")
                };
                resolved_args.insert("findings".to_string(), formatted_findings);
            }
            let resolved = resolve_template(prompt, &resolved_args);
            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, findings, abort_flag, live, session_dir, workspaces, timeout_ms) {
                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.
            // Each branch shares the same `findings` Arc so note_finding
            // calls within any branch are visible to all other branches.
            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 abort = abort_flag.clone();
                    let live_clone = live.cloned();
                    let session_dir = session_dir.to_path_buf();
                    let workspaces = workspaces.to_vec();
                    let findings = Arc::clone(findings);
                    let to = timeout_ms;

                    std::thread::spawn(move || {
                        let _permit = sem.acquire();
                        let result = execute_primitive(
                            &script, &args, cap, continue_on_error,
                            &abort,
                            live_clone.as_ref(),
                            &session_dir,
                            &workspaces,
                            &findings,
                            to,
                        );
                        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.
            //
            // Abort is checked between stages so the user can cancel the
            // pipeline immediately when moving to the next division, rather
            // than having to wait for the current subagent to finish.
            //
            // 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 through the shared
            // `findings` Arc (same isolation scope as parent).
            let mut all = Vec::new();
            for (idx, script) in scripts.iter().enumerate() {
                // Check abort before each pipeline stage so we don't
                // launch the next division after the user cancelled.
                if abort_flag.as_ref().is_some_and(|f| f.load(Ordering::SeqCst)) {
                    if continue_on_error {
                        all.push(format!("pipeline aborted at stage {idx}"));
                        break;
                    }
                    anyhow::bail!("pipeline aborted by user at stage {idx}");
                }
                match execute_primitive(script, args, concurrency_cap, continue_on_error, abort_flag, live, session_dir, workspaces, findings, timeout_ms) {
                    Ok(outputs) => all.extend(outputs),
                    Err(e) => {
                        if continue_on_error {
                            all.push(format!("pipeline stage {idx} error: {e}"));
                        } else {
                            return Err(e);
                        }
                    }
                }
            }
            Ok(all)
        }

        ScriptPrimitive::Phase { name: _name, script } => {
            execute_primitive(script, args, concurrency_cap, continue_on_error, abort_flag, live, session_dir, workspaces, findings, timeout_ms)
        }
    }
}

/// 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>,
    session_dir: &std::path::Path,
    workspaces: &[std::path::PathBuf],
) -> anyhow::Result<String> {
    run_workflow_tracked(script, args, &None, None, session_dir, workspaces)
}

/// Run a `WorkflowScript` with real-time live-state callbacks so the TUI
/// panel updates as each agent transitions between Idle/Running/Done/Failed.
///
/// Flow: create an empty findings Arc (scoped to this invocation) → cap
/// concurrency to 8 → call `execute_primitive` with the live callback and
/// findings → format results.
///
/// Why: findings are scoped to an `Arc<Mutex<Vec<String>>>` rather than a
/// global static, so concurrent `run_workflow_tracked` calls from different
/// `spawn_agents` invocations remain fully isolated.
///
/// Return: a human-readable summary string.
pub fn run_workflow_tracked(
    script: &WorkflowScript,
    args: &HashMap<String, String>,
    abort_flag: &Option<Arc<AtomicBool>>,
    live: Option<&LiveStateFn>,
    session_dir: &std::path::Path,
    workspaces: &[std::path::PathBuf],
) -> anyhow::Result<String> {
    let concurrency_cap = if script.options.max_concurrency > 0 {
        script.options.max_concurrency.min(10) // allow up to 10 parallel agents
    } else {
        10
    };

    let findings = Arc::new(Mutex::new(Vec::new()));
    let results = execute_primitive(
        &script.script, args, concurrency_cap,
        script.options.continue_on_error, abort_flag, live,
        session_dir, workspaces, &findings,
        script.options.timeout_ms,
    )?;

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

/// 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.
///
/// Panic-safety: if a thread panics while holding a permit, the Mutex
/// becomes poisoned. Both `acquire` and the `Drop` implementation recover
/// from poisoned mutexes by discarding the poison, ensuring the semaphore
/// remains usable after a thread panic.
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_or_else(|e| {
            tracing::warn!("[semaphore] mutex poisoned in acquire, recovering");
            e.into_inner()
        });
        while *count == 0 {
            count = self.condvar.wait(count).unwrap_or_else(|e| {
                tracing::warn!("[semaphore] mutex poisoned in wait, recovering");
                e.into_inner()
            });
        }
        *count -= 1;
        SemaphoreGuard { sem: self }
    }
}

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

impl Drop for SemaphoreGuard<'_> {
    fn drop(&mut self) {
        let mut count = self.sem.count.lock().unwrap_or_else(|e| {
            tracing::warn!("[semaphore] mutex poisoned in drop, recovering");
            e.into_inner()
        });
        *count += 1;
        self.sem.condvar.notify_one();
    }
}