//! Aragorn: Master Orchestrator Agent
//!
//! Aragorn coordinates the parallel execution of SCROLL tasks. It loads tasks
//! from SQLite, builds a dependency graph (DAG), performs topological sorting
//! via Kahn's algorithm to identify execution levels, then executes tasks in
//! parallel within each level using bounded concurrency (tokio + Semaphore).
//!
//! Named after the king who rallies many forces -- Aragorn orchestrates the
//! entire execution pipeline.

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;
use std::time::Instant;

use anyhow::{bail, Context, Result};
use chrono::Utc;
use serde::{Deserialize, Serialize};
use tokio::sync::{Mutex as TokioMutex, Semaphore};
use tracing::{debug, error, info, warn};

use crate::agents::gandalf::Gandalf;
use crate::agents::gimli::Gimli;
use crate::config::AgentConfig;
use crate::db::SqliteDb;
use crate::types::{ScrollEventListener, ScrollStatus, Task, TaskStatus};

// ---------------------------------------------------------------------------
// Public data structures
// ---------------------------------------------------------------------------

/// A fully resolved execution plan for a scroll, organized into dependency
/// levels. Tasks within the same level have no inter-dependencies and can
/// run in parallel.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionPlan {
    /// The scroll this plan belongs to.
    pub scroll_id: String,

    /// Ordered sequence of execution levels.
    pub levels: Vec<ExecutionLevel>,

    /// Total number of tasks in the plan.
    pub total_tasks: usize,

    /// Unix timestamp (seconds) when the plan was created.
    pub created_at: i64,
}

/// A single execution level -- all tasks in a level can run concurrently.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionLevel {
    /// Zero-based level index.
    pub level: usize,

    /// Task IDs in this level.
    pub task_ids: Vec<String>,

    /// Whether there are multiple tasks (i.e. parallelism is meaningful).
    pub can_parallelize: bool,
}

/// The final result of executing a scroll.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionResult {
    /// The scroll that was executed.
    pub scroll_id: String,

    /// Overall status of the execution.
    pub status: ExecutionStatus,

    /// IDs of tasks that completed successfully.
    pub completed_tasks: Vec<String>,

    /// Details of tasks that failed.
    pub failed_tasks: Vec<TaskFailure>,

    /// Wall-clock duration of the execution in seconds.
    pub duration_secs: u64,

    /// Unix timestamp when execution started.
    pub started_at: i64,

    /// Unix timestamp when execution finished (None if still running).
    pub completed_at: Option<i64>,

    /// The DAG execution levels that were computed for this scroll.
    /// Each level contains tasks that can run in parallel.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub execution_plan: Option<ExecutionPlan>,
}

/// Overall status of a scroll execution.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum ExecutionStatus {
    /// All tasks completed successfully.
    Complete,
    /// Some tasks succeeded, some failed.
    PartialFailure,
    /// All tasks failed or execution could not proceed.
    Failed,
    /// Execution was cancelled (e.g. due to upstream failure).
    Cancelled,
}

/// Information about a failed task.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TaskFailure {
    /// The ID of the task that failed.
    pub task_id: String,
    /// Human-readable error description.
    pub error: String,
    /// How many times the task was retried before giving up.
    pub retry_count: usize,
}

/// Snapshot of live execution state for progress tracking.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExecutionState {
    /// The scroll being executed.
    pub scroll_id: String,
    /// Current level being executed (0-indexed).
    pub current_level: usize,
    /// Total number of levels.
    pub total_levels: usize,
    /// IDs of tasks that completed.
    pub completed_tasks: Vec<String>,
    /// IDs of tasks that failed.
    pub failed_tasks: Vec<String>,
    /// IDs of tasks currently running.
    pub running_tasks: Vec<String>,
    /// IDs of tasks that were skipped (due to failed dependencies).
    pub skipped_tasks: Vec<String>,
    /// Progress as a percentage (0.0 - 100.0).
    pub progress_percent: f64,
    /// Whether execution is still in progress.
    pub is_running: bool,
}

// ---------------------------------------------------------------------------
// Internal graph structures
// ---------------------------------------------------------------------------

/// Directed acyclic graph of task dependencies.
#[derive(Debug, Clone)]
pub(crate) struct DependencyGraph {
    /// Node data keyed by task ID.
    nodes: HashMap<String, GraphNode>,
    /// Directed edges: (from, to) meaning `from` must complete before `to`.
    edges: Vec<(String, String)>,
}

/// A node in the dependency graph.
#[derive(Debug, Clone)]
struct GraphNode {
    /// IDs of tasks this node depends on (must finish first).
    dependencies: Vec<String>,
    /// IDs of tasks that depend on this node.
    dependents: Vec<String>,
    /// Number of unresolved incoming edges (used by Kahn's algorithm).
    in_degree: usize,
}

// ---------------------------------------------------------------------------
// Aragorn
// ---------------------------------------------------------------------------

/// The Aragorn orchestrator agent.
pub struct Aragorn {
    /// SQLite database for persisting scroll/task state.
    sqlite: Arc<SqliteDb>,

    /// Gandalf: provides constraint context for code generation tasks.
    gandalf: Option<Arc<Gandalf>>,

    /// Gimli: code generation agent.
    gimli: Option<Arc<Gimli>>,

    /// Elrond: lifecycle guardian for execution health checks.
    elrond: Option<Arc<crate::agents::elrond::Elrond>>,

    /// Maximum number of tasks to run concurrently within a level.
    max_concurrent: usize,

    /// Maximum number of retries for a failed task before giving up.
    max_retries: usize,

    /// Live execution state, keyed by scroll ID.
    execution_state: Arc<TokioMutex<HashMap<String, ExecutionState>>>,

    /// Event listeners notified on scroll status changes (e.g., SyncBridge).
    event_listeners: Vec<Arc<dyn ScrollEventListener>>,
}

impl Aragorn {
    /// Create a new Aragorn instance.
    pub fn new(sqlite: Arc<SqliteDb>, config: &AgentConfig) -> Self {
        info!(
            max_concurrent = config.max_concurrent_tasks,
            max_retries = config.max_retries,
            "Aragorn initialized"
        );
        Self {
            sqlite,
            gandalf: None,
            gimli: None,
            elrond: None,
            max_concurrent: config.max_concurrent_tasks,
            max_retries: config.max_retries,
            execution_state: Arc::new(TokioMutex::new(HashMap::new())),
            event_listeners: Vec::new(),
        }
    }

    /// Set the Gandalf and Gimli agents for code generation delegation.
    pub fn with_agents(mut self, gandalf: Arc<Gandalf>, gimli: Arc<Gimli>) -> Self {
        self.gandalf = Some(gandalf);
        self.gimli = Some(gimli);
        self
    }

    /// Set the Elrond agent for execution health checks between levels.
    pub fn with_elrond(mut self, elrond: Arc<crate::agents::elrond::Elrond>) -> Self {
        self.elrond = Some(elrond);
        self
    }

    /// Add an event listener that will be notified on scroll status changes.
    pub fn with_event_listener(mut self, listener: Arc<dyn ScrollEventListener>) -> Self {
        self.event_listeners.push(listener);
        self
    }

    /// Create an Aragorn with explicit parameters (useful for testing).
    pub fn with_params(sqlite: Arc<SqliteDb>, max_concurrent: usize, max_retries: usize) -> Self {
        info!(max_concurrent, max_retries, "Aragorn initialized (custom params)");
        Self {
            sqlite,
            gandalf: None,
            gimli: None,
            elrond: None,
            max_concurrent,
            max_retries,
            execution_state: Arc::new(TokioMutex::new(HashMap::new())),
            event_listeners: Vec::new(),
        }
    }

    // -----------------------------------------------------------------------
    // Event notification
    // -----------------------------------------------------------------------

    /// Notify all registered event listeners of a scroll status change.
    async fn notify_status_change(&self, scroll_id: &str, status: &ScrollStatus) {
        for listener in &self.event_listeners {
            if tokio::time::timeout(
                std::time::Duration::from_secs(5),
                listener.on_scroll_status_changed(scroll_id, status),
            )
            .await
            .is_err()
            {
                warn!(scroll_id, ?status, "Event listener timed out after 5s, skipping");
            }
        }
    }

    // -----------------------------------------------------------------------
    // Public API
    // -----------------------------------------------------------------------

    /// Execute all tasks for a scroll.
    ///
    /// This is the main entry point. It:
    /// 1. Loads tasks from SQLite.
    /// 2. Builds the dependency graph.
    /// 3. Checks for circular dependencies (fails fast).
    /// 4. Performs topological sort to produce execution levels.
    /// 5. Executes each level in order, with bounded parallelism.
    /// 6. Tracks progress and returns the final result.
    pub async fn execute_scroll(&self, scroll_id: &str) -> Result<ExecutionResult> {
        let start = Instant::now();
        let started_at = Utc::now().timestamp();

        info!(scroll_id, "Starting scroll execution");

        // 1. Load tasks from SQLite.
        let tasks = self
            .sqlite
            .get_tasks_for_scroll(scroll_id)
            .await
            .context("Failed to load tasks for scroll")?;

        if tasks.is_empty() {
            warn!(scroll_id, "No tasks found for scroll");
            return Ok(ExecutionResult {
                scroll_id: scroll_id.to_string(),
                status: ExecutionStatus::Complete,
                completed_tasks: vec![],
                failed_tasks: vec![],
                duration_secs: 0,
                started_at,
                completed_at: Some(Utc::now().timestamp()),
                execution_plan: None,
            });
        }

        info!(scroll_id, task_count = tasks.len(), "Tasks loaded");

        // Mark the scroll as InProgress.
        if let Err(e) = self
            .sqlite
            .update_scroll_status(scroll_id, &ScrollStatus::InProgress)
            .await
        {
            warn!(scroll_id, error = %e, "Failed to update scroll status to InProgress");
        }
        self.notify_status_change(scroll_id, &ScrollStatus::InProgress).await;

        // 2. Build the dependency graph.
        let graph = Self::build_dependency_graph(&tasks);
        info!(
            scroll_id,
            nodes = graph.nodes.len(),
            edges = graph.edges.len(),
            "Dependency graph built"
        );

        // 3. Check for circular dependencies.
        if let Some(cycle) = Self::find_cycle(&graph) {
            error!(scroll_id, cycle = ?cycle, "Circular dependency detected");
            if let Err(e) = self
                .sqlite
                .update_scroll_status(scroll_id, &ScrollStatus::Failed)
                .await
            {
                warn!(scroll_id, error = %e, "Failed to update scroll status to Failed");
            }
            self.notify_status_change(scroll_id, &ScrollStatus::Failed).await;
            bail!(
                "Circular dependency detected in scroll {}: {:?}",
                scroll_id,
                cycle
            );
        }

        // 4. Topological sort into execution levels.
        let plan = Self::topological_sort(scroll_id, &graph)?;
        info!(
            scroll_id,
            levels = plan.levels.len(),
            total_tasks = plan.total_tasks,
            "Execution plan created"
        );

        // 5. Initialize execution state.
        {
            let mut state_map = self.execution_state.lock().await;
            state_map.insert(
                scroll_id.to_string(),
                ExecutionState {
                    scroll_id: scroll_id.to_string(),
                    current_level: 0,
                    total_levels: plan.levels.len(),
                    completed_tasks: vec![],
                    failed_tasks: vec![],
                    running_tasks: vec![],
                    skipped_tasks: vec![],
                    progress_percent: 0.0,
                    is_running: true,
                },
            );
        }

        // 6. Execute level by level.
        let mut all_completed: Vec<String> = Vec::new();
        let mut all_failed: Vec<TaskFailure> = Vec::new();
        let cancelled = false;

        // Track which tasks have failed so we can skip dependents.
        let failed_set: Arc<TokioMutex<HashSet<String>>> =
            Arc::new(TokioMutex::new(HashSet::new()));

        for level in &plan.levels {
            if cancelled {
                // Mark remaining tasks as skipped.
                let mut state_map = self.execution_state.lock().await;
                if let Some(state) = state_map.get_mut(scroll_id) {
                    for tid in &level.task_ids {
                        state.skipped_tasks.push(tid.clone());
                    }
                }
                for tid in &level.task_ids {
                    if let Err(e) = self
                        .sqlite
                        .update_task_status(tid, &TaskStatus::Skipped)
                        .await
                    {
                        warn!(task_id = tid.as_str(), error = %e, "Failed to mark task as Skipped");
                    }
                }
                continue;
            }

            // Update current level in state.
            {
                let mut state_map = self.execution_state.lock().await;
                if let Some(state) = state_map.get_mut(scroll_id) {
                    state.current_level = level.level;
                }
            }

            info!(
                scroll_id,
                level = level.level,
                tasks = level.task_ids.len(),
                parallel = level.can_parallelize,
                "Executing level"
            );

            // Determine which tasks in this level should be skipped due to
            // failed dependencies.
            let current_failed = failed_set.lock().await;
            let mut runnable_ids = Vec::new();
            let mut skip_ids = Vec::new();

            for tid in &level.task_ids {
                if let Some(node) = graph.nodes.get(tid) {
                    let has_failed_dep = node
                        .dependencies
                        .iter()
                        .any(|dep| current_failed.contains(dep));
                    if has_failed_dep {
                        skip_ids.push(tid.clone());
                    } else {
                        runnable_ids.push(tid.clone());
                    }
                } else {
                    runnable_ids.push(tid.clone());
                }
            }
            drop(current_failed);

            // Skip tasks with failed dependencies.
            for tid in &skip_ids {
                warn!(task_id = tid.as_str(), "Skipping task due to failed dependency");
                if let Err(e) = self
                    .sqlite
                    .update_task_status(tid, &TaskStatus::Skipped)
                    .await
                {
                    warn!(task_id = tid.as_str(), error = %e, "Failed to mark task as Skipped");
                }
                failed_set.lock().await.insert(tid.clone());
                let mut state_map = self.execution_state.lock().await;
                if let Some(state) = state_map.get_mut(scroll_id) {
                    state.skipped_tasks.push(tid.clone());
                }
            }

            // Batch pre-generation: generate code for all tasks needing it in one LLM call.
            // This reduces N sampling calls to 1 per level, minimizing user approval prompts.
            if let (Some(gandalf), Some(gimli)) = (&self.gandalf, &self.gimli) {
                let mut batch_tasks = Vec::new();
                let mut batch_contexts = Vec::new();

                for tid in &runnable_ids {
                    if let Ok(task) = self.sqlite.get_task_by_id(tid).await {
                        if task.generated_code.is_some()
                            || task.task_type == crate::types::TaskType::Delete
                        {
                            continue;
                        }
                        let context = gandalf.build_constraint_context(&task).await.unwrap_or(
                            crate::agents::gandalf::ConstraintContext {
                                direct_constraints: vec![],
                                inherited_constraints: vec![],
                                pattern_examples: vec![],
                                related_types: vec![],
                                dependency_context: vec![],
                                summary: String::new(),
                                recipe_context: None,
                            },
                        );
                        batch_tasks.push(task);
                        batch_contexts.push(context);
                    }
                }

                if batch_tasks.len() > 1 {
                    info!(
                        scroll_id,
                        level = level.level,
                        batch_size = batch_tasks.len(),
                        "Batch code generation for level"
                    );
                    match gimli.generate_code_batch(&batch_tasks, &batch_contexts).await {
                        Ok(results) => {
                            for gen in &results {
                                debug!(
                                    task_id = %gen.task_id,
                                    code_len = gen.code.len(),
                                    "Batch-generated code stored"
                                );
                            }
                        }
                        Err(e) => {
                            warn!(
                                scroll_id,
                                error = %e,
                                "Batch code generation failed, individual tasks will generate their own"
                            );
                        }
                    }
                }
            }

            // Execute runnable tasks in parallel.
            let (level_completed, level_failed) = self
                .execute_level_parallel(scroll_id, &runnable_ids, &plan)
                .await;

            // Record failures into the failed set.
            for failure in &level_failed {
                failed_set.lock().await.insert(failure.task_id.clone());
            }

            all_completed.extend(level_completed);
            all_failed.extend(level_failed);

            // Update progress.
            {
                let mut state_map = self.execution_state.lock().await;
                if let Some(state) = state_map.get_mut(scroll_id) {
                    state.completed_tasks = all_completed.clone();
                    state.failed_tasks = all_failed.iter().map(|f| f.task_id.clone()).collect();
                    let done_count = state.completed_tasks.len()
                        + state.failed_tasks.len()
                        + state.skipped_tasks.len();
                    state.progress_percent =
                        (done_count as f64 / plan.total_tasks as f64) * 100.0;
                }
            }

            // Elrond health check between execution levels.
            if let Some(ref elrond) = self.elrond {
                let scroll = match self.sqlite.get_scroll(scroll_id).await {
                    Ok(s) => s,
                    Err(_) => continue,
                };

                // Gather completed and remaining tasks for health check.
                let completed_tasks: Vec<_> = tasks
                    .iter()
                    .filter(|t| all_completed.contains(&t.id))
                    .cloned()
                    .collect();
                let remaining_tasks: Vec<_> = tasks
                    .iter()
                    .filter(|t| !all_completed.contains(&t.id) && !all_failed.iter().any(|f| f.task_id == t.id))
                    .cloned()
                    .collect();
                let failed_pairs: Vec<(String, String)> = all_failed
                    .iter()
                    .map(|f| (f.task_id.clone(), f.error.clone()))
                    .collect();

                let health_result = tokio::time::timeout(
                    std::time::Duration::from_secs(15),
                    elrond.verify_execution_health(&scroll, &completed_tasks, &remaining_tasks, &failed_pairs),
                ).await;
                match health_result {
                    Err(_) => {
                        warn!(scroll_id, "Elrond health check timed out after 15s, continuing");
                    }
                    Ok(Err(e)) => {
                        debug!(scroll_id, error = %e, "Elrond health check failed, continuing");
                    }
                    Ok(Ok(health)) => {
                        if health.should_halt {
                            warn!(
                                scroll_id,
                                reason = %health.reason,
                                "Elrond halted execution"
                            );
                            break;
                        }
                        if !health.healthy {
                            warn!(
                                scroll_id,
                                reason = %health.reason,
                                recommendations = ?health.recommendations,
                                "Elrond flagged unhealthy execution, continuing"
                            );
                        }
                    }
                }
            }
        }

        // 7. Determine final status.
        let status = if all_failed.is_empty() && !cancelled {
            ExecutionStatus::Complete
        } else if all_completed.is_empty() && !all_failed.is_empty() {
            ExecutionStatus::Failed
        } else if cancelled {
            ExecutionStatus::Cancelled
        } else {
            ExecutionStatus::PartialFailure
        };

        let duration_secs = start.elapsed().as_secs();
        let completed_at = Utc::now().timestamp();

        // Update scroll status.
        let scroll_status = match &status {
            ExecutionStatus::Complete => ScrollStatus::Completed,
            _ => ScrollStatus::Failed,
        };
        if let Err(e) = self
            .sqlite
            .update_scroll_status(scroll_id, &scroll_status)
            .await
        {
            warn!(scroll_id, error = %e, "Failed to update final scroll status");
        }
        self.notify_status_change(scroll_id, &scroll_status).await;

        // Mark execution as finished.
        {
            let mut state_map = self.execution_state.lock().await;
            if let Some(state) = state_map.get_mut(scroll_id) {
                state.is_running = false;
                state.progress_percent = 100.0;
                state.running_tasks.clear();
            }
        }

        info!(
            scroll_id,
            status = ?status,
            completed = all_completed.len(),
            failed = all_failed.len(),
            duration_secs,
            "Scroll execution finished"
        );

        Ok(ExecutionResult {
            scroll_id: scroll_id.to_string(),
            status,
            completed_tasks: all_completed,
            failed_tasks: all_failed,
            duration_secs,
            started_at,
            completed_at: Some(completed_at),
            execution_plan: Some(plan.clone()),
        })
    }

    /// Build an execution plan for a scroll without running it.
    ///
    /// Useful for dry-run / preview of what would happen.
    pub async fn plan_scroll(&self, scroll_id: &str) -> Result<ExecutionPlan> {
        let tasks = self
            .sqlite
            .get_tasks_for_scroll(scroll_id)
            .await
            .context("Failed to load tasks for scroll")?;

        let graph = Self::build_dependency_graph(&tasks);

        if let Some(cycle) = Self::find_cycle(&graph) {
            bail!(
                "Circular dependency detected in scroll {}: {:?}",
                scroll_id,
                cycle
            );
        }

        Self::topological_sort(scroll_id, &graph)
    }

    /// Get the current execution state for a scroll.
    ///
    /// Returns `None` if the scroll has never been executed or if its state
    /// has been cleaned up.
    pub async fn get_execution_status(&self, scroll_id: &str) -> Option<ExecutionState> {
        let state_map = self.execution_state.lock().await;
        state_map.get(scroll_id).cloned()
    }

    // -----------------------------------------------------------------------
    // Graph construction
    // -----------------------------------------------------------------------

    /// Build a dependency graph from a list of tasks.
    ///
    /// Each task becomes a node. Each entry in `task.dependencies` creates a
    /// directed edge from the dependency to the task. Dependencies referencing
    /// task IDs not in the list are silently dropped with a warning.
    pub(crate) fn build_dependency_graph(tasks: &[Task]) -> DependencyGraph {
        let task_ids: HashSet<String> = tasks.iter().map(|t| t.id.clone()).collect();
        let mut nodes: HashMap<String, GraphNode> = HashMap::new();
        let mut edges: Vec<(String, String)> = Vec::new();

        // Initialize all nodes.
        for task in tasks {
            nodes.insert(
                task.id.clone(),
                GraphNode {
                    dependencies: Vec::new(),
                    dependents: Vec::new(),
                    in_degree: 0,
                },
            );
        }

        // Build edges from dependency declarations.
        for task in tasks {
            for dep_id in &task.dependencies {
                if !task_ids.contains(dep_id) {
                    warn!(
                        task_id = task.id.as_str(),
                        dependency = dep_id.as_str(),
                        "Dependency references unknown task, skipping"
                    );
                    continue;
                }

                // Edge: dep_id -> task.id (dep must complete before task).
                edges.push((dep_id.clone(), task.id.clone()));

                if let Some(node) = nodes.get_mut(&task.id) {
                    node.dependencies.push(dep_id.clone());
                    node.in_degree += 1;
                }

                if let Some(dep_node) = nodes.get_mut(dep_id) {
                    dep_node.dependents.push(task.id.clone());
                }
            }
        }

        DependencyGraph { nodes, edges }
    }

    // -----------------------------------------------------------------------
    // Cycle detection
    // -----------------------------------------------------------------------

    /// Detect cycles in the dependency graph using DFS with a recursion stack.
    ///
    /// Returns `Some(cycle_path)` if a cycle is found, or `None` if the graph
    /// is acyclic.
    pub(crate) fn find_cycle(graph: &DependencyGraph) -> Option<Vec<String>> {
        let mut visited: HashSet<String> = HashSet::new();
        let mut rec_stack: HashSet<String> = HashSet::new();
        let mut path: Vec<String> = Vec::new();

        for node_id in graph.nodes.keys() {
            if !visited.contains(node_id) {
                if let Some(cycle) =
                    Self::dfs_cycle(node_id, graph, &mut visited, &mut rec_stack, &mut path)
                {
                    return Some(cycle);
                }
            }
        }

        None
    }

    /// Recursive DFS helper for cycle detection.
    fn dfs_cycle(
        node_id: &str,
        graph: &DependencyGraph,
        visited: &mut HashSet<String>,
        rec_stack: &mut HashSet<String>,
        path: &mut Vec<String>,
    ) -> Option<Vec<String>> {
        visited.insert(node_id.to_string());
        rec_stack.insert(node_id.to_string());
        path.push(node_id.to_string());

        if let Some(node) = graph.nodes.get(node_id) {
            for dependent in &node.dependents {
                if !visited.contains(dependent.as_str()) {
                    if let Some(cycle) =
                        Self::dfs_cycle(dependent, graph, visited, rec_stack, path)
                    {
                        return Some(cycle);
                    }
                } else if rec_stack.contains(dependent.as_str()) {
                    // Found a cycle. Extract the cycle path from where the
                    // repeated node first appears.
                    let mut cycle_path: Vec<String> = Vec::new();
                    let mut found_start = false;
                    for p in path.iter() {
                        if p == dependent {
                            found_start = true;
                        }
                        if found_start {
                            cycle_path.push(p.clone());
                        }
                    }
                    cycle_path.push(dependent.clone()); // close the cycle
                    return Some(cycle_path);
                }
            }
        }

        path.pop();
        rec_stack.remove(node_id);
        None
    }

    // -----------------------------------------------------------------------
    // Topological sort (Kahn's algorithm)
    // -----------------------------------------------------------------------

    /// Perform a topological sort using Kahn's algorithm, producing execution
    /// levels. Each level contains tasks whose dependencies are all satisfied
    /// by previous levels.
    pub(crate) fn topological_sort(
        scroll_id: &str,
        graph: &DependencyGraph,
    ) -> Result<ExecutionPlan> {
        // Clone in-degrees so we can mutate them.
        let mut in_degree: HashMap<String, usize> = graph
            .nodes
            .iter()
            .map(|(id, node)| (id.clone(), node.in_degree))
            .collect();

        // Seed the queue with all zero-in-degree nodes.
        let mut queue: VecDeque<String> = in_degree
            .iter()
            .filter(|(_, &deg)| deg == 0)
            .map(|(id, _)| id.clone())
            .collect();

        let mut levels: Vec<ExecutionLevel> = Vec::new();
        let mut processed_count = 0;

        while !queue.is_empty() {
            // All nodes currently in the queue form one execution level.
            let mut current_level_ids: Vec<String> = Vec::new();

            let level_size = queue.len();
            for _ in 0..level_size {
                let node_id = queue.pop_front().unwrap();
                current_level_ids.push(node_id);
            }

            // For each node in this level, decrement in-degrees of dependents.
            for node_id in &current_level_ids {
                if let Some(node) = graph.nodes.get(node_id) {
                    for dependent in &node.dependents {
                        if let Some(deg) = in_degree.get_mut(dependent) {
                            *deg -= 1;
                            if *deg == 0 {
                                queue.push_back(dependent.clone());
                            }
                        }
                    }
                }
            }

            let can_parallelize = current_level_ids.len() > 1;
            processed_count += current_level_ids.len();

            // Sort IDs within the level for deterministic ordering.
            current_level_ids.sort();

            levels.push(ExecutionLevel {
                level: levels.len(),
                task_ids: current_level_ids,
                can_parallelize,
            });
        }

        // If we did not process all nodes, there is a cycle (should not happen
        // if find_cycle was called first, but guard anyway).
        if processed_count != graph.nodes.len() {
            bail!(
                "Topological sort failed for scroll {}: {} of {} nodes processed (cycle likely)",
                scroll_id,
                processed_count,
                graph.nodes.len()
            );
        }

        Ok(ExecutionPlan {
            scroll_id: scroll_id.to_string(),
            levels,
            total_tasks: processed_count,
            created_at: Utc::now().timestamp(),
        })
    }

    // -----------------------------------------------------------------------
    // Parallel execution
    // -----------------------------------------------------------------------

    /// Execute all tasks in a single level with bounded concurrency.
    ///
    /// Returns (completed_task_ids, failed_tasks).
    async fn execute_level_parallel(
        &self,
        scroll_id: &str,
        task_ids: &[String],
        _plan: &ExecutionPlan,
    ) -> (Vec<String>, Vec<TaskFailure>) {
        if task_ids.is_empty() {
            return (vec![], vec![]);
        }

        let semaphore = Arc::new(Semaphore::new(self.max_concurrent));
        let completed: Arc<TokioMutex<Vec<String>>> = Arc::new(TokioMutex::new(Vec::new()));
        let failed: Arc<TokioMutex<Vec<TaskFailure>>> = Arc::new(TokioMutex::new(Vec::new()));

        let mut handles = Vec::new();

        for task_id in task_ids {
            let sem = Arc::clone(&semaphore);
            let sqlite = Arc::clone(&self.sqlite);
            let gandalf = self.gandalf.clone();
            let gimli = self.gimli.clone();
            let completed = Arc::clone(&completed);
            let failed = Arc::clone(&failed);
            let exec_state = Arc::clone(&self.execution_state);
            let max_retries = self.max_retries;
            let tid = task_id.clone();
            let sid = scroll_id.to_string();

            let handle = tokio::spawn(async move {
                // Acquire semaphore permit.
                let _permit = match sem.acquire().await {
                    Ok(permit) => permit,
                    Err(_) => {
                        error!(task_id = tid.as_str(), "Semaphore closed unexpectedly");
                        failed.lock().await.push(TaskFailure {
                            task_id: tid,
                            error: "Semaphore closed unexpectedly".into(),
                            retry_count: 0,
                        });
                        return;
                    }
                };

                // Mark as running in execution state.
                {
                    let mut state_map = exec_state.lock().await;
                    if let Some(state) = state_map.get_mut(&sid) {
                        state.running_tasks.push(tid.clone());
                    }
                }

                debug!(task_id = tid.as_str(), "Task execution starting");

                // Execute with retries (120s timeout per task).
                let result = match tokio::time::timeout(
                    std::time::Duration::from_secs(120),
                    Self::execute_single_task_with_retry(
                        &sqlite,
                        gandalf.as_ref(),
                        gimli.as_ref(),
                        &tid,
                        max_retries,
                    ),
                ).await {
                    Ok(r) => r,
                    Err(_) => {
                        warn!(task_id = tid.as_str(), "Task execution timed out after 120s");
                        Err(("Task execution timed out after 120s".into(), max_retries))
                    }
                };

                // Remove from running in execution state.
                {
                    let mut state_map = exec_state.lock().await;
                    if let Some(state) = state_map.get_mut(&sid) {
                        state.running_tasks.retain(|id| id != &tid);
                    }
                }

                match result {
                    Ok(()) => {
                        info!(task_id = tid.as_str(), "Task completed successfully");
                        completed.lock().await.push(tid);
                    }
                    Err((err_msg, retry_count)) => {
                        error!(
                            task_id = tid.as_str(),
                            error = err_msg.as_str(),
                            retries = retry_count,
                            "Task failed after retries"
                        );
                        failed.lock().await.push(TaskFailure {
                            task_id: tid,
                            error: err_msg,
                            retry_count,
                        });
                    }
                }
            });

            handles.push(handle);
        }

        // Wait for all tasks in this level to finish.
        for handle in handles {
            if let Err(e) = handle.await {
                error!(error = %e, "Task join handle error");
            }
        }

        let completed = completed.lock().await.clone();
        let failed = failed.lock().await.clone();

        (completed, failed)
    }

    /// Execute a single task with retry logic.
    ///
    /// Returns `Ok(())` on success, or `Err((error_message, retry_count))` on
    /// final failure.
    async fn execute_single_task_with_retry(
        sqlite: &Arc<SqliteDb>,
        gandalf: Option<&Arc<Gandalf>>,
        gimli: Option<&Arc<Gimli>>,
        task_id: &str,
        max_retries: usize,
    ) -> std::result::Result<(), (String, usize)> {
        let mut last_error = String::new();

        for attempt in 0..=max_retries {
            if attempt > 0 {
                debug!(
                    task_id,
                    attempt,
                    max_retries,
                    "Retrying task"
                );
                // Brief backoff before retry.
                tokio::time::sleep(std::time::Duration::from_millis(100 * attempt as u64)).await;
            }

            match Self::execute_single_task(sqlite, gandalf, gimli, task_id).await {
                Ok(()) => return Ok(()),
                Err(e) => {
                    last_error = format!("{:#}", e);
                    warn!(
                        task_id,
                        attempt,
                        error = last_error.as_str(),
                        "Task attempt failed"
                    );
                }
            }
        }

        // All retries exhausted -- mark as Failed.
        if let Err(e) = sqlite.update_task_status(task_id, &TaskStatus::Failed).await {
            warn!(task_id, error = %e, "Failed to mark task as Failed in DB");
        }

        Err((last_error, max_retries))
    }

    /// Execute a single task.
    ///
    /// Loads the task from SQLite, marks it InProgress, delegates to Gandalf
    /// for constraint context and Gimli for code generation, stores the result,
    /// then marks it Completed. Records agent activity for each agent invocation.
    async fn execute_single_task(
        sqlite: &Arc<SqliteDb>,
        gandalf: Option<&Arc<Gandalf>>,
        gimli: Option<&Arc<Gimli>>,
        task_id: &str,
    ) -> Result<()> {
        use crate::types::{AgentActivity, AgentActivityStatus};

        // Mark task as InProgress.
        sqlite
            .update_task_status(task_id, &TaskStatus::InProgress)
            .await
            .context("Failed to mark task InProgress")?;

        debug!(task_id, "Task marked InProgress");

        // Load the task from the database.
        let task = sqlite.get_task_by_id(task_id).await
            .context("Task not found in database")?;

        let scroll_id = task.scroll_id.clone();

        // If code was pre-generated (e.g. via add_tasks), skip Gandalf+Gimli.
        if task.generated_code.is_some() {
            let activity_id = format!("act-aragorn-{}-{}", task_id, Utc::now().timestamp_millis());
            let code_len = task.generated_code.as_ref().map(|c| c.len()).unwrap_or(0);
            let _ = sqlite.record_agent_activity(&AgentActivity {
                id: activity_id.clone(),
                scroll_id: scroll_id.clone(),
                task_id: task_id.to_string(),
                agent_name: "Aragorn".into(),
                action: "code_pre_generated".into(),
                detail: format!("Pre-generated code: {} bytes for {}", code_len, task.target_file),
                started_at: Utc::now(),
                completed_at: Some(Utc::now()),
                status: AgentActivityStatus::Completed,
            }).await;

            info!(task_id, target_file = %task.target_file, code_len, "Using pre-generated code, skipping Gandalf+Gimli");

            sqlite
                .update_task_status(task_id, &TaskStatus::Completed)
                .await
                .context("Failed to mark pre-generated task Completed")?;

            debug!(task_id, "Pre-generated task marked Completed");
            return Ok(());
        }

        // Delegate to Gimli for code generation if agents are available.
        if let (Some(gandalf), Some(gimli)) = (gandalf, gimli) {
            // 1. Build constraint context from Gandalf's knowledge graph.
            let gandalf_activity_id = format!("act-gandalf-{}-{}", task_id, Utc::now().timestamp_millis());
            let gandalf_started = Utc::now();
            let _ = sqlite.record_agent_activity(&AgentActivity {
                id: gandalf_activity_id.clone(),
                scroll_id: scroll_id.clone(),
                task_id: task_id.to_string(),
                agent_name: "Gandalf".into(),
                action: "build_constraint_context".into(),
                detail: String::new(),
                started_at: gandalf_started,
                completed_at: None,
                status: AgentActivityStatus::Running,
            }).await;

            let context = match gandalf.build_constraint_context(&task).await {
                Ok(ctx) => {
                    let detail = format!(
                        "Extracted {} direct + {} inherited constraints, {} patterns",
                        ctx.direct_constraints.len(),
                        ctx.inherited_constraints.len(),
                        ctx.pattern_examples.len(),
                    );
                    let _ = sqlite.complete_agent_activity(
                        &gandalf_activity_id, &AgentActivityStatus::Completed, &detail,
                    ).await;
                    ctx
                }
                Err(e) => {
                    let _ = sqlite.complete_agent_activity(
                        &gandalf_activity_id, &AgentActivityStatus::Failed, &format!("{}", e),
                    ).await;
                    warn!(task_id, error = %e, "Failed to build constraint context, using empty");
                    crate::agents::gandalf::ConstraintContext {
                        direct_constraints: vec![],
                        inherited_constraints: vec![],
                        pattern_examples: vec![],
                        related_types: vec![],
                        dependency_context: vec![],
                        summary: String::new(),
                        recipe_context: None,
                    }
                }
            };

            // 2. Generate code via Gimli.
            let gimli_activity_id = format!("act-gimli-{}-{}", task_id, Utc::now().timestamp_millis());
            let _ = sqlite.record_agent_activity(&AgentActivity {
                id: gimli_activity_id.clone(),
                scroll_id: scroll_id.clone(),
                task_id: task_id.to_string(),
                agent_name: "Gimli".into(),
                action: "generate_code".into(),
                detail: String::new(),
                started_at: Utc::now(),
                completed_at: None,
                status: AgentActivityStatus::Running,
            }).await;

            match gimli.generate_code(&task, &context).await {
                Ok(generated) => {
                    let detail = format!(
                        "Generated {} bytes for {}",
                        generated.code.len(),
                        generated.target_file,
                    );
                    let _ = sqlite.complete_agent_activity(
                        &gimli_activity_id, &AgentActivityStatus::Completed, &detail,
                    ).await;

                    info!(
                        task_id,
                        target_file = %generated.target_file,
                        code_len = generated.code.len(),
                        "Code generated"
                    );

                    // Note: Gimli already persists the code to SQLite inside
                    // generate_code(), so no need for a redundant write here.
                }
                Err(e) => {
                    let _ = sqlite.complete_agent_activity(
                        &gimli_activity_id, &AgentActivityStatus::Failed, &format!("{}", e),
                    ).await;
                    return Err(e).context("Gimli code generation failed");
                }
            }
        } else {
            debug!(task_id, "No code generation agents configured, completing task as lifecycle-only");
        }

        // Mark task as Completed.
        sqlite
            .update_task_status(task_id, &TaskStatus::Completed)
            .await
            .context("Failed to mark task Completed")?;

        debug!(task_id, "Task marked Completed");

        Ok(())
    }

    // -----------------------------------------------------------------------
    // Utilities
    // -----------------------------------------------------------------------

    /// Clear persisted execution state for a scroll.
    pub async fn clear_state(&self, scroll_id: &str) {
        let mut state_map = self.execution_state.lock().await;
        state_map.remove(scroll_id);
    }

    /// Check whether a dependency graph has any tasks with missing (external)
    /// dependencies. Returns a list of (task_id, missing_dep_id) pairs.
    pub fn find_missing_dependencies(tasks: &[Task]) -> Vec<(String, String)> {
        let task_ids: HashSet<String> = tasks.iter().map(|t| t.id.clone()).collect();
        let mut missing = Vec::new();

        for task in tasks {
            for dep_id in &task.dependencies {
                if !task_ids.contains(dep_id) {
                    missing.push((task.id.clone(), dep_id.clone()));
                }
            }
        }

        missing
    }
}

// ===========================================================================
// Tests
// ===========================================================================

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::*;
    use chrono::Utc;
    use tempfile::TempDir;

    // -- Helpers ------------------------------------------------------------

    /// Create an in-memory SQLite DB wrapped in Arc.
    async fn setup_db() -> (Arc<SqliteDb>, TempDir) {
        let tmp = TempDir::new().expect("temp dir");
        let db_path = tmp.path().join("test_aragorn.db");
        let sqlite = Arc::new(SqliteDb::new(&db_path).expect("sqlite"));
        (sqlite, tmp)
    }

    /// Convenience builder for a Task with only the fields we care about.
    fn make_task(id: &str, scroll_id: &str, deps: Vec<&str>) -> Task {
        Task {
            id: id.to_string(),
            scroll_id: scroll_id.to_string(),
            description: format!("Task {}", id),
            target_file: format!("src/{}.rs", id),
            task_type: TaskType::Create,
            status: TaskStatus::Pending,
            dependencies: deps.into_iter().map(String::from).collect(),
            generated_code: None,
            diff: None,
            created_at: Utc::now(),
            group_label: None,
        }
    }

    /// Seed a scroll and its tasks into the database.
    async fn seed_scroll(sqlite: &SqliteDb, scroll_id: &str, tasks: &[Task]) {
        let now = Utc::now();
        let scroll = Scroll {
            id: scroll_id.to_string(),
            title: format!("Test Scroll {}", scroll_id),
            intent: "Testing".to_string(),
            constraints: vec![],
            tasks: vec![],
            metadata: ScrollMetadata {
                execution_mode: ExecutionMode::Forge,
                auto_forge_config: None,
                recipe_used: None,
                search_context: vec![],
            },
            status: ScrollStatus::Validated,
            created_at: now,
            updated_at: now,
            cleanup_after: None,
        };
        sqlite.create_scroll(&scroll).await.expect("create scroll");

        for task in tasks {
            sqlite.create_task(task).await.expect("create task");
        }
    }

    // -- Tests --------------------------------------------------------------

    #[test]
    fn test_build_dependency_graph_no_deps() {
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec![]),
            make_task("t3", "s1", vec![]),
        ];

        let graph = Aragorn::build_dependency_graph(&tasks);

        assert_eq!(graph.nodes.len(), 3);
        assert_eq!(graph.edges.len(), 0);

        for node in graph.nodes.values() {
            assert_eq!(node.in_degree, 0, "All tasks should have zero in-degree");
            assert!(node.dependencies.is_empty());
        }
    }

    #[test]
    fn test_build_dependency_graph_linear_chain() {
        // t1 -> t2 -> t3
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1"]),
            make_task("t3", "s1", vec!["t2"]),
        ];

        let graph = Aragorn::build_dependency_graph(&tasks);

        assert_eq!(graph.nodes.len(), 3);
        assert_eq!(graph.edges.len(), 2);

        assert_eq!(graph.nodes["t1"].in_degree, 0);
        assert_eq!(graph.nodes["t2"].in_degree, 1);
        assert_eq!(graph.nodes["t3"].in_degree, 1);

        assert!(graph.nodes["t1"].dependents.contains(&"t2".to_string()));
        assert!(graph.nodes["t2"].dependents.contains(&"t3".to_string()));
    }

    #[test]
    fn test_build_dependency_graph_diamond() {
        //     t1
        //    /  \
        //   t2   t3
        //    \  /
        //     t4
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1"]),
            make_task("t3", "s1", vec!["t1"]),
            make_task("t4", "s1", vec!["t2", "t3"]),
        ];

        let graph = Aragorn::build_dependency_graph(&tasks);

        assert_eq!(graph.nodes.len(), 4);
        assert_eq!(graph.edges.len(), 4); // t1->t2, t1->t3, t2->t4, t3->t4

        assert_eq!(graph.nodes["t1"].in_degree, 0);
        assert_eq!(graph.nodes["t2"].in_degree, 1);
        assert_eq!(graph.nodes["t3"].in_degree, 1);
        assert_eq!(graph.nodes["t4"].in_degree, 2);
    }

    #[test]
    fn test_build_dependency_graph_missing_dep_skipped() {
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1", "t_nonexistent"]),
        ];

        let graph = Aragorn::build_dependency_graph(&tasks);

        // The missing dep should be skipped; only the valid dep creates an edge.
        assert_eq!(graph.edges.len(), 1);
        assert_eq!(graph.nodes["t2"].in_degree, 1);
        assert_eq!(graph.nodes["t2"].dependencies, vec!["t1".to_string()]);
    }

    #[test]
    fn test_find_missing_dependencies() {
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1", "t_missing"]),
            make_task("t3", "s1", vec!["t_also_missing"]),
        ];

        let missing = Aragorn::find_missing_dependencies(&tasks);

        assert_eq!(missing.len(), 2);
        assert!(missing.contains(&("t2".to_string(), "t_missing".to_string())));
        assert!(missing.contains(&("t3".to_string(), "t_also_missing".to_string())));
    }

    #[test]
    fn test_cycle_detection_no_cycle() {
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1"]),
            make_task("t3", "s1", vec!["t2"]),
        ];

        let graph = Aragorn::build_dependency_graph(&tasks);
        let cycle = Aragorn::find_cycle(&graph);

        assert!(cycle.is_none(), "Linear chain should have no cycle");
    }

    #[test]
    fn test_cycle_detection_simple_cycle() {
        // Manually create a graph with a cycle: t1 -> t2 -> t3 -> t1
        // We cannot express this through make_task deps (since build_dependency_graph
        // works on edges from dep -> task), so we build the graph directly.
        let mut nodes = HashMap::new();
        nodes.insert(
            "t1".to_string(),
            GraphNode {
                dependencies: vec!["t3".to_string()],
                dependents: vec!["t2".to_string()],
                in_degree: 1,
            },
        );
        nodes.insert(
            "t2".to_string(),
            GraphNode {
                dependencies: vec!["t1".to_string()],
                dependents: vec!["t3".to_string()],
                in_degree: 1,
            },
        );
        nodes.insert(
            "t3".to_string(),
            GraphNode {
                dependencies: vec!["t2".to_string()],
                dependents: vec!["t1".to_string()],
                in_degree: 1,
            },
        );

        let graph = DependencyGraph {
            nodes,
            edges: vec![
                ("t1".to_string(), "t2".to_string()),
                ("t2".to_string(), "t3".to_string()),
                ("t3".to_string(), "t1".to_string()),
            ],
        };

        let cycle = Aragorn::find_cycle(&graph);
        assert!(cycle.is_some(), "Should detect the cycle");

        let cycle_path = cycle.unwrap();
        // The cycle should contain all three nodes.
        assert!(cycle_path.len() >= 3, "Cycle path should have at least 3 nodes");
    }

    #[test]
    fn test_cycle_detection_self_loop() {
        // A task depends on itself.
        let mut nodes = HashMap::new();
        nodes.insert(
            "t1".to_string(),
            GraphNode {
                dependencies: vec!["t1".to_string()],
                dependents: vec!["t1".to_string()],
                in_degree: 1,
            },
        );

        let graph = DependencyGraph {
            nodes,
            edges: vec![("t1".to_string(), "t1".to_string())],
        };

        let cycle = Aragorn::find_cycle(&graph);
        assert!(cycle.is_some(), "Self-loop should be detected as a cycle");
    }

    #[test]
    fn test_topological_sort_linear() {
        // t1 -> t2 -> t3
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1"]),
            make_task("t3", "s1", vec!["t2"]),
        ];

        let graph = Aragorn::build_dependency_graph(&tasks);
        let plan = Aragorn::topological_sort("s1", &graph).unwrap();

        assert_eq!(plan.levels.len(), 3, "Linear chain should have 3 levels");
        assert_eq!(plan.total_tasks, 3);

        assert_eq!(plan.levels[0].task_ids, vec!["t1"]);
        assert_eq!(plan.levels[1].task_ids, vec!["t2"]);
        assert_eq!(plan.levels[2].task_ids, vec!["t3"]);

        assert!(!plan.levels[0].can_parallelize);
        assert!(!plan.levels[1].can_parallelize);
        assert!(!plan.levels[2].can_parallelize);
    }

    #[test]
    fn test_topological_sort_all_independent() {
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec![]),
            make_task("t3", "s1", vec![]),
        ];

        let graph = Aragorn::build_dependency_graph(&tasks);
        let plan = Aragorn::topological_sort("s1", &graph).unwrap();

        assert_eq!(plan.levels.len(), 1, "All independent tasks should be in one level");
        assert_eq!(plan.levels[0].task_ids.len(), 3);
        assert!(plan.levels[0].can_parallelize);
    }

    #[test]
    fn test_topological_sort_diamond() {
        //     t1
        //    /  \
        //   t2   t3
        //    \  /
        //     t4
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1"]),
            make_task("t3", "s1", vec!["t1"]),
            make_task("t4", "s1", vec!["t2", "t3"]),
        ];

        let graph = Aragorn::build_dependency_graph(&tasks);
        let plan = Aragorn::topological_sort("s1", &graph).unwrap();

        assert_eq!(plan.levels.len(), 3);
        assert_eq!(plan.total_tasks, 4);

        // Level 0: t1
        assert_eq!(plan.levels[0].task_ids, vec!["t1"]);

        // Level 1: t2, t3 (parallel)
        assert_eq!(plan.levels[1].task_ids.len(), 2);
        assert!(plan.levels[1].task_ids.contains(&"t2".to_string()));
        assert!(plan.levels[1].task_ids.contains(&"t3".to_string()));
        assert!(plan.levels[1].can_parallelize);

        // Level 2: t4
        assert_eq!(plan.levels[2].task_ids, vec!["t4"]);
    }

    #[test]
    fn test_topological_sort_complex_graph() {
        // t1 ----> t3 ----> t5
        // t2 --/        \-> t6
        //   \---> t4 ------/
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec![]),
            make_task("t3", "s1", vec!["t1", "t2"]),
            make_task("t4", "s1", vec!["t2"]),
            make_task("t5", "s1", vec!["t3"]),
            make_task("t6", "s1", vec!["t3", "t4"]),
        ];

        let graph = Aragorn::build_dependency_graph(&tasks);
        assert!(Aragorn::find_cycle(&graph).is_none());

        let plan = Aragorn::topological_sort("s1", &graph).unwrap();

        assert_eq!(plan.total_tasks, 6);

        // Level 0: t1, t2 (no deps)
        assert_eq!(plan.levels[0].task_ids.len(), 2);
        assert!(plan.levels[0].task_ids.contains(&"t1".to_string()));
        assert!(plan.levels[0].task_ids.contains(&"t2".to_string()));

        // Verify ordering: t3 and t4 must come after t1 and t2.
        let t3_level = plan
            .levels
            .iter()
            .find(|l| l.task_ids.contains(&"t3".to_string()))
            .unwrap()
            .level;
        let t4_level = plan
            .levels
            .iter()
            .find(|l| l.task_ids.contains(&"t4".to_string()))
            .unwrap()
            .level;

        assert!(t3_level > 0, "t3 should be after level 0");
        assert!(t4_level > 0, "t4 should be after level 0");

        let t5_level = plan
            .levels
            .iter()
            .find(|l| l.task_ids.contains(&"t5".to_string()))
            .unwrap()
            .level;
        let t6_level = plan
            .levels
            .iter()
            .find(|l| l.task_ids.contains(&"t6".to_string()))
            .unwrap()
            .level;

        assert!(t5_level > t3_level, "t5 must come after t3");
        assert!(t6_level > t3_level, "t6 must come after t3");
        assert!(t6_level > t4_level, "t6 must come after t4");
    }

    #[test]
    fn test_topological_sort_empty_graph() {
        let tasks: Vec<Task> = vec![];
        let graph = Aragorn::build_dependency_graph(&tasks);
        let plan = Aragorn::topological_sort("s1", &graph).unwrap();

        assert_eq!(plan.levels.len(), 0);
        assert_eq!(plan.total_tasks, 0);
    }

    #[tokio::test]
    async fn test_execute_scroll_all_independent() {
        let (sqlite, _tmp) = setup_db().await;

        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec![]),
            make_task("t3", "s1", vec![]),
        ];

        seed_scroll(&sqlite, "s1", &tasks).await;

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 2);
        let result = aragorn.execute_scroll("s1").await.unwrap();

        assert_eq!(result.status, ExecutionStatus::Complete);
        assert_eq!(result.completed_tasks.len(), 3);
        assert!(result.failed_tasks.is_empty());
        assert!(result.completed_at.is_some());

        // Verify tasks are marked Completed in DB.
        let db_tasks = sqlite.get_tasks_for_scroll("s1").await.unwrap();
        for task in &db_tasks {
            assert_eq!(task.status, TaskStatus::Completed, "Task {} should be Completed", task.id);
        }
    }

    #[tokio::test]
    async fn test_execute_scroll_with_dependencies() {
        let (sqlite, _tmp) = setup_db().await;

        // t1 -> t2 -> t3
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1"]),
            make_task("t3", "s1", vec!["t2"]),
        ];

        seed_scroll(&sqlite, "s1", &tasks).await;

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 2);
        let result = aragorn.execute_scroll("s1").await.unwrap();

        assert_eq!(result.status, ExecutionStatus::Complete);
        assert_eq!(result.completed_tasks.len(), 3);
        assert!(result.failed_tasks.is_empty());
    }

    #[tokio::test]
    async fn test_execute_scroll_empty() {
        let (sqlite, _tmp) = setup_db().await;

        // Create scroll with no tasks.
        seed_scroll(&sqlite, "s1", &[]).await;

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 2);
        let result = aragorn.execute_scroll("s1").await.unwrap();

        assert_eq!(result.status, ExecutionStatus::Complete);
        assert!(result.completed_tasks.is_empty());
        assert!(result.failed_tasks.is_empty());
        assert_eq!(result.duration_secs, 0);
    }

    #[tokio::test]
    async fn test_execute_scroll_diamond_dependency() {
        let (sqlite, _tmp) = setup_db().await;

        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1"]),
            make_task("t3", "s1", vec!["t1"]),
            make_task("t4", "s1", vec!["t2", "t3"]),
        ];

        seed_scroll(&sqlite, "s1", &tasks).await;

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 2);
        let result = aragorn.execute_scroll("s1").await.unwrap();

        assert_eq!(result.status, ExecutionStatus::Complete);
        assert_eq!(result.completed_tasks.len(), 4);
    }

    #[tokio::test]
    async fn test_get_execution_status_during_and_after() {
        let (sqlite, _tmp) = setup_db().await;

        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec![]),
        ];

        seed_scroll(&sqlite, "s1", &tasks).await;

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 2);

        // Before execution: no state.
        assert!(aragorn.get_execution_status("s1").await.is_none());

        // Execute.
        let _result = aragorn.execute_scroll("s1").await.unwrap();

        // After execution: state should exist and show finished.
        let state = aragorn.get_execution_status("s1").await;
        assert!(state.is_some());

        let state = state.unwrap();
        assert!(!state.is_running, "Execution should be finished");
        assert_eq!(state.progress_percent, 100.0);
        assert!(state.running_tasks.is_empty());
    }

    #[tokio::test]
    async fn test_plan_scroll_without_execution() {
        let (sqlite, _tmp) = setup_db().await;

        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1"]),
            make_task("t3", "s1", vec!["t1"]),
            make_task("t4", "s1", vec!["t2", "t3"]),
        ];

        seed_scroll(&sqlite, "s1", &tasks).await;

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 2);
        let plan = aragorn.plan_scroll("s1").await.unwrap();

        assert_eq!(plan.scroll_id, "s1");
        assert_eq!(plan.total_tasks, 4);
        assert_eq!(plan.levels.len(), 3);

        // Verify tasks are still Pending (plan should not mutate state).
        let db_tasks = sqlite.get_tasks_for_scroll("s1").await.unwrap();
        for task in &db_tasks {
            assert_eq!(task.status, TaskStatus::Pending, "Plan should not change task status");
        }
    }

    #[tokio::test]
    async fn test_clear_state() {
        let (sqlite, _tmp) = setup_db().await;

        let tasks = vec![make_task("t1", "s1", vec![])];
        seed_scroll(&sqlite, "s1", &tasks).await;

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 2);
        let _result = aragorn.execute_scroll("s1").await.unwrap();

        assert!(aragorn.get_execution_status("s1").await.is_some());

        aragorn.clear_state("s1").await;

        assert!(aragorn.get_execution_status("s1").await.is_none());
    }

    #[test]
    fn test_execution_level_can_parallelize() {
        let level_single = ExecutionLevel {
            level: 0,
            task_ids: vec!["t1".to_string()],
            can_parallelize: false,
        };
        assert!(!level_single.can_parallelize);

        let level_multi = ExecutionLevel {
            level: 0,
            task_ids: vec!["t1".to_string(), "t2".to_string()],
            can_parallelize: true,
        };
        assert!(level_multi.can_parallelize);
    }

    #[tokio::test]
    async fn test_execute_scroll_updates_scroll_status() {
        let (sqlite, _tmp) = setup_db().await;

        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec!["t1"]),
        ];

        seed_scroll(&sqlite, "s1", &tasks).await;

        // Verify initial status.
        let scroll = sqlite.get_scroll("s1").await.unwrap();
        assert_eq!(scroll.status, ScrollStatus::Validated);

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 2);
        let result = aragorn.execute_scroll("s1").await.unwrap();

        assert_eq!(result.status, ExecutionStatus::Complete);

        // Verify final scroll status.
        let scroll = sqlite.get_scroll("s1").await.unwrap();
        assert_eq!(scroll.status, ScrollStatus::Completed);
    }

    #[tokio::test]
    async fn test_execute_scroll_concurrency_bounded() {
        // This test verifies that many tasks can run without issue under
        // a low concurrency limit.
        let (sqlite, _tmp) = setup_db().await;

        let mut tasks = Vec::new();
        for i in 0..20 {
            tasks.push(make_task(&format!("t{}", i), "s1", vec![]));
        }

        seed_scroll(&sqlite, "s1", &tasks).await;

        // Set max_concurrent to 2 -- all 20 tasks are independent so they
        // form a single level, but only 2 run at a time.
        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 2, 1);
        let result = aragorn.execute_scroll("s1").await.unwrap();

        assert_eq!(result.status, ExecutionStatus::Complete);
        assert_eq!(result.completed_tasks.len(), 20);
    }

    // -- Tests: execution_plan in result --

    #[tokio::test]
    async fn test_execution_result_includes_plan() {
        let (sqlite, _tmp) = setup_db().await;

        // Create two independent tasks and one dependent.
        let tasks = vec![
            make_task("t1", "s1", vec![]),
            make_task("t2", "s1", vec![]),
            make_task("t3", "s1", vec!["t1".into(), "t2".into()]),
        ];
        seed_scroll(&sqlite, "s1", &tasks).await;

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 1);
        let result = aragorn.execute_scroll("s1").await.unwrap();

        assert_eq!(result.status, ExecutionStatus::Complete);
        assert!(result.execution_plan.is_some());
        let plan = result.execution_plan.unwrap();
        assert_eq!(plan.levels.len(), 2);
        assert_eq!(plan.total_tasks, 3);
        // Level 0: t1 and t2 (independent).
        assert_eq!(plan.levels[0].task_ids.len(), 2);
        assert!(plan.levels[0].can_parallelize);
        // Level 1: t3 (depends on t1 and t2).
        assert_eq!(plan.levels[1].task_ids.len(), 1);
        assert_eq!(plan.levels[1].task_ids[0], "t3");
    }

    #[tokio::test]
    async fn test_execution_result_empty_scroll_has_no_plan() {
        let (sqlite, _tmp) = setup_db().await;
        seed_scroll(&sqlite, "s1", &[]).await;

        let aragorn = Aragorn::with_params(Arc::clone(&sqlite), 4, 1);
        let result = aragorn.execute_scroll("s1").await.unwrap();

        assert_eq!(result.status, ExecutionStatus::Complete);
        assert!(result.execution_plan.is_none());
    }

    #[tokio::test]
    async fn test_execution_result_serialization_includes_plan() {
        let result = ExecutionResult {
            scroll_id: "s1".into(),
            status: ExecutionStatus::Complete,
            completed_tasks: vec!["t1".into()],
            failed_tasks: vec![],
            duration_secs: 5,
            started_at: 1000,
            completed_at: Some(1005),
            execution_plan: Some(ExecutionPlan {
                scroll_id: "s1".into(),
                levels: vec![ExecutionLevel {
                    level: 0,
                    task_ids: vec!["t1".into()],
                    can_parallelize: false,
                }],
                total_tasks: 1,
                created_at: 1000,
            }),
        };

        let json = serde_json::to_value(&result).unwrap();
        assert!(json.get("execution_plan").is_some());
        let plan = &json["execution_plan"];
        assert_eq!(plan["levels"][0]["task_ids"][0], "t1");
    }

    #[tokio::test]
    async fn test_execution_result_serialization_omits_none_plan() {
        let result = ExecutionResult {
            scroll_id: "s1".into(),
            status: ExecutionStatus::Complete,
            completed_tasks: vec![],
            failed_tasks: vec![],
            duration_secs: 0,
            started_at: 1000,
            completed_at: Some(1000),
            execution_plan: None,
        };

        let json = serde_json::to_value(&result).unwrap();
        // skip_serializing_if = "Option::is_none" should omit it.
        assert!(json.get("execution_plan").is_none());
    }
}
