// src/core/executor.rs
#![cfg(not(target_arch = "wasm32"))]

use anyhow::{anyhow, Result};
use futures::future::join_all;
use lazy_static::lazy_static;
use petgraph::graph::NodeIndex;
use petgraph::visit::EdgeRef;
use petgraph::Direction;
use regex::Regex;
use serde_json::{json, Value};
use std::collections::{HashMap, HashSet, VecDeque};
use std::future::Future;
use std::pin::Pin;
use std::sync::{Arc, Mutex};
use tracing::{debug, error, info, warn};

use crate::builtins::BuiltinRegistry;
use crate::core::context::{WorkflowContext, WorkflowEvent};
use crate::core::expr_eval::{self, ExprEvaluator};
use crate::core::graph::{self, NodeType, WorkflowGraph};
use crate::core::instance_arena::{MethodScope, TypedSlot};
use crate::core::parser::GraphParser;
use crate::runtime::python::PythonRuntime;
use crate::services::config::{DebugConfig, JuglansConfig};
use crate::services::local_runtime::LocalRuntime;
use crate::services::prompt_loader::PromptRegistry;
use crate::services::tool_registry::ToolRegistry;

lazy_static! {
    static ref FUNC_CALL_RE: Regex =
        Regex::new(r"(?s)^([a-zA-Z0-9_.]+)\((.*)\)(\.[a-zA-Z0-9_]+)?$").unwrap();
}

/// Context for evaluating outgoing edges from a completed node
struct EdgeEvalContext<'a> {
    executor: &'a Arc<WorkflowExecutor>,
    node_idx: NodeIndex,
    node_succeeded: bool,
    switch_result: &'a Option<String>,
    workflow: &'a Arc<WorkflowGraph>,
    context: &'a WorkflowContext,
    in_degrees: &'a Arc<Mutex<HashMap<NodeIndex, usize>>>,
    ready_queue: &'a Arc<Mutex<VecDeque<NodeIndex>>>,
}

pub struct WorkflowExecutor {
    builtin_registry: Arc<BuiltinRegistry>,
    tool_registry: Arc<ToolRegistry>,
    expr_eval: ExprEvaluator,
    debug_config: DebugConfig,
    /// Configurable runtime limits
    max_loop_iterations: usize,
    /// Python runtime for executing external Python calls
    python_runtime: Option<Arc<Mutex<PythonRuntime>>>,
    /// Imported Python modules (from workflow python: [...] declaration)
    python_imports: Vec<String>,
}

impl WorkflowExecutor {
    pub async fn new(prompt_registry: Arc<PromptRegistry>, runtime: Arc<LocalRuntime>) -> Self {
        Self::new_with_debug(prompt_registry, runtime, DebugConfig::default()).await
    }

    pub async fn new_with_debug(
        prompt_registry: Arc<PromptRegistry>,
        runtime: Arc<LocalRuntime>,
        debug_config: DebugConfig,
    ) -> Self {
        let registry_arc = BuiltinRegistry::new(prompt_registry, runtime);

        // Auto-register devtools schema to ToolRegistry (slug: "devtools")
        let mut tool_registry = ToolRegistry::new();
        registry_arc.register_devtools_to_registry(&mut tool_registry);

        Self {
            builtin_registry: registry_arc,
            tool_registry: Arc::new(tool_registry),
            expr_eval: ExprEvaluator::new(),
            debug_config,
            max_loop_iterations: 100,
            python_runtime: None,
            python_imports: Vec::new(),
        }
    }

    /// Apply runtime limits from configuration
    pub fn apply_limits(&mut self, limits: &crate::services::config::RuntimeLimits) {
        self.max_loop_iterations = limits.max_loop_iterations;
    }

    /// Get a reference to the builtin registry (for injecting into executor)
    pub fn get_registry(&self) -> &Arc<BuiltinRegistry> {
        &self.builtin_registry
    }

    /// Load tool definitions from workflow patterns
    pub async fn load_tools(&mut self, workflow: &WorkflowGraph) {
        use crate::core::tool_loader::ToolLoader;
        use std::path::Path;

        if workflow.tool_patterns.is_empty() {
            return;
        }

        info!(
            "📦 Loading tool definitions from {} pattern(s)...",
            workflow.tool_patterns.len()
        );

        let workflow_base_dir = Path::new("."); // Can be derived from workflow file path
        let mut loaded_count = 0;

        for pattern in &workflow.tool_patterns {
            match ToolLoader::load_from_glob(pattern, workflow_base_dir) {
                Ok(tools) => {
                    loaded_count += tools.len();
                    // Need a mutable reference, so use Arc::get_mut or Mutex
                    // For now, create a new registry and replace it
                    let mut registry = (*self.tool_registry).clone();
                    registry.register_all(tools);
                    self.tool_registry = Arc::new(registry);
                }
                Err(e) => {
                    warn!("Failed to load tools from pattern '{}': {}", pattern, e);
                }
            }
        }

        if loaded_count > 0 {
            info!(
                "  ✅ Loaded {} tool resource(s) with {} total tools",
                self.tool_registry.count(),
                loaded_count
            );
        }
    }

    /// Get a reference to the tool registry
    pub fn get_tool_registry(&self) -> &Arc<ToolRegistry> {
        &self.tool_registry
    }

    /// Replace the tool registry
    pub fn set_tool_registry(&mut self, registry: Arc<ToolRegistry>) {
        self.tool_registry = registry;
    }

    /// Initialize Python runtime if the workflow has python imports
    pub fn init_python_runtime(
        &mut self,
        workflow: &WorkflowGraph,
        worker_count: usize,
    ) -> Result<()> {
        if workflow.python_imports.is_empty() {
            return Ok(());
        }

        info!(
            "🐍 Initializing Python runtime with {} import(s)...",
            workflow.python_imports.len()
        );

        // Store the imports for later resolution
        self.python_imports = workflow.python_imports.clone();

        let mut runtime = PythonRuntime::new(worker_count)?;
        runtime.set_imports(workflow.python_imports.clone());
        self.python_runtime = Some(Arc::new(Mutex::new(runtime)));

        info!(
            "  ✅ Python runtime initialized with imports: {:?}",
            workflow.python_imports
        );
        Ok(())
    }

    /// Check if a tool name is a Python module call
    fn is_python_call(&self, name: &str) -> bool {
        if self.python_imports.is_empty() {
            return false;
        }

        // Check if the tool name starts with any imported module
        for import in &self.python_imports {
            // Handle file imports (e.g., "./utils.py" -> "utils")
            let module_name = if import.ends_with(".py") {
                import
                    .rsplit('/')
                    .next()
                    .map(|f| f.trim_end_matches(".py"))
                    .unwrap_or(import)
            } else {
                import.as_str()
            };

            if name == module_name || name.starts_with(&format!("{}.", module_name)) {
                return true;
            }
        }
        false
    }

    /// Execute a Python function call
    fn execute_python_call(
        &self,
        name: &str,
        params: &HashMap<String, String>,
    ) -> Result<Option<Value>> {
        let runtime = self
            .python_runtime
            .as_ref()
            .ok_or_else(|| anyhow!("Python runtime not initialized"))?;

        let rt = runtime
            .lock()
            .map_err(|e| anyhow!("Failed to lock Python runtime: {}", e))?;

        // Convert params to kwargs
        let mut kwargs: HashMap<String, Value> = HashMap::new();
        for (k, v) in params {
            // Try to parse as JSON, fall back to string
            let val = serde_json::from_str(v).unwrap_or(Value::String(v.clone()));
            kwargs.insert(k.clone(), val);
        }

        let result = rt.call(name, Vec::new(), kwargs)?;
        Ok(Some(result))
    }

    async fn process_parameter(&self, param_str: &str, context: &WorkflowContext) -> Result<Value> {
        let clean_param = param_str.trim();

        if let Some(caps) = FUNC_CALL_RE.captures(clean_param) {
            let tool_name = &caps[1];

            // Only dispatch as tool call if it's a registered tool
            // (not an expression function like len(), str(), filter(), etc.)
            let is_registered_tool =
                self.builtin_registry.get(tool_name).is_some() || self.is_python_call(tool_name);

            if is_registered_tool {
                let args_str = &caps[2];
                let field_access = caps.get(3).map(|m| m.as_str().trim_start_matches('.'));

                let raw_args = GraphParser::parse_arguments_str(args_str);
                let mut resolved_args = HashMap::new();

                for (k, v) in raw_args {
                    let resolved_val = Box::pin(self.process_parameter(&v, context)).await?;
                    let val_str = match resolved_val {
                        Value::String(s) => s,
                        Value::Null => "".to_string(),
                        other => other.to_string(),
                    };
                    resolved_args.insert(k, val_str);
                }

                let result_val = self
                    .execute_tool_internal(tool_name, &resolved_args, context)
                    .await?;

                if let Some(field) = field_access {
                    if let Some(obj) = result_val.as_ref().and_then(|v| v.as_object()) {
                        let field_val = obj.get(field).cloned().unwrap_or(Value::Null);
                        return Ok(field_val);
                    }
                }
                return Ok(result_val.unwrap_or(Value::Null));
            }
            // else: fall through to expression evaluator below
        }

        // Node reference syntax: [identifier] → preserve as literal string.
        // Used by on_tool=[handler], on_token=[handler], on_result=[handler].
        // Without this, [stream_chunk] would evaluate as array [null] via expr evaluator.
        if clean_param.starts_with('[') && clean_param.ends_with(']') && clean_param.len() > 2 {
            let inner = clean_param[1..clean_param.len() - 1].trim();
            let is_node_ref = !inner.is_empty()
                && inner.starts_with(|c: char| c.is_alphabetic() || c == '_')
                && inner.chars().all(|c: char| c.is_alphanumeric() || c == '_')
                && !matches!(inner, "true" | "false" | "null");
            if is_node_ref {
                return Ok(Value::String(clean_param.to_string()));
            }
        }

        // Quoted -> string literal, return directly without going through expression evaluator.
        // Prevents "display_only" from being resolved as a variable to Null, "trade-extractor" being parsed as trade - extractor = 0
        if clean_param.starts_with('"') && clean_param.ends_with('"') {
            let inner = &clean_param[1..clean_param.len() - 1];
            // No inner quotes -> pure string literal, return directly
            // Has inner quotes -> "str" + expr + "str" concatenation expression, pass to evaluator
            if !inner.contains('"') {
                return Ok(Value::String(inner.to_string()));
            }
        }

        // Parse and evaluate using expression evaluator (replaces Rhai).
        // Resolver maps variable paths (e.g. "ctx.intent" -> "intent") to WorkflowContext.
        let context_ref = context;
        let show_variables = self.debug_config.show_variables;

        // Phase C-2: prefer direct-index evaluation inside method body (zero HashMap, zero locks)
        // parse_cached → optimize_for_method → eval_method_expr
        let method_result = context_ref
            .with_method_scope(|scope| {
                let ast = self.expr_eval.parse_cached(clean_param).ok()?;
                let optimized = expr_eval::ExprEvaluator::optimize_for_method(
                    &ast,
                    &scope.class_def,
                    &scope.method_params,
                );
                self.expr_eval
                    .eval_method_expr(&optimized, &scope.field_cache, &scope.param_values)
                    .ok()
            })
            .flatten();
        if let Some(slot) = method_result {
            return Ok(slot.to_value());
        }

        // General path: TypedSlot fast path + Value fallback
        let typed_resolver = |path: &str| -> Option<TypedSlot> {
            let clean = path.strip_prefix("ctx.").unwrap_or(path);
            if let Some(slot) = context_ref.resolve_path_typed(clean) {
                return Some(slot);
            }
            let resolved = context_ref.resolve_path(clean).ok().flatten();
            if show_variables {
                tracing::info!("🔍 [Debug] Resolve: ${} → {:?}", path, resolved);
            }
            resolved.map(TypedSlot::from_value)
        };

        match self.expr_eval.eval_typed(clean_param, &typed_resolver) {
            Ok(slot) => Ok(slot.to_value()),
            Err(e) => {
                // On evaluation failure, return as string literal
                debug!("  [Expr] Failed to evaluate '{}': {}", clean_param, e);
                Ok(json!(clean_param))
            }
        }
    }

    pub async fn execute_tool_internal(
        &self,
        name: &str,
        params: &HashMap<String, String>,
        context: &WorkflowContext,
    ) -> Result<Option<Value>> {
        // 1. Check built-in tools first
        if let Some(tool) = self.builtin_registry.get(name) {
            return tool.execute(params, context).await;
        }

        // 2. Check Python imports
        if self.is_python_call(name) {
            debug!("🐍 Executing Python call: {}", name);
            return self.execute_python_call(name, params);
        }

        Err(anyhow!("Function/Tool '{}' not found", name))
    }

    async fn evaluate_condition_async(
        &self,
        script: &str,
        context: &WorkflowContext,
    ) -> Result<bool> {
        let val = self.process_parameter(script, context).await?;
        // Python-like truthiness replaces Rhai's as_bool
        let result = expr_eval::is_truthy(&val);
        if self.debug_config.show_conditions {
            info!(
                "🔀 [Debug] Condition '{}' → {} (raw: {:?})",
                script, result, val
            );
        }
        Ok(result)
    }

    async fn run_single_node(
        self: Arc<Self>,
        node_idx: NodeIndex,
        workflow: &Arc<WorkflowGraph>,
        context: &WorkflowContext,
    ) -> Result<Option<Value>> {
        let node = &workflow.graph[node_idx];
        let status_suffix = if let Ok(Some(s)) = context.resolve_path("reply.status") {
            format!(" | Status: {}", s.as_str().unwrap_or(""))
        } else {
            "".to_string()
        };

        debug!("│ → [{}]{}", node.id, status_suffix);

        if self.debug_config.show_nodes {
            info!("📦 [Debug] Node [{}]: {:?}", node.id, node.node_type);
        }

        // Check if this node is mocked (injected output from mock() builtin)
        if let Ok(Some(mock_val)) = context.resolve_path(&format!("_mocks.{}", node.id)) {
            info!("  🎭 Mock [{}]", node.id);
            return Ok(Some(mock_val));
        }

        match &node.node_type {
            NodeType::Literal(val) => {
                debug!("│   Literal value assigned");
                Ok(Some(val.clone()))
            }
            NodeType::Task(action) => {
                // Check if this is struct/class instantiation (function call style: Name(field=value))
                if workflow.classes.contains_key(&action.name) {
                    let class_def = workflow.classes.get(&action.name).unwrap();
                    let mut fields_vec = Vec::with_capacity(class_def.fields.len());
                    for field in &class_def.fields {
                        let value = if let Some(arg_expr) = action.params.get(&field.name) {
                            self.process_parameter(arg_expr, context).await?
                        } else if let Some(default_expr) = &field.default {
                            self.process_parameter(default_expr, context).await?
                        } else {
                            Value::Null
                        };
                        fields_vec.push(value);
                    }
                    let class_def_arc = Arc::clone(class_def);
                    let id = context.alloc_instance(
                        node.id.clone(),
                        action.name.clone(),
                        class_def_arc,
                        fields_vec,
                    )?;
                    // Deferred materialization: return arena proxy, resolve_path lazy-loads on demand
                    return Ok(Some(json!({"__arena_ref__": id.0})));
                }

                // Check if this is a function node call
                // Look in current workflow first, then fallback to root workflow
                let func_workflow = if workflow.functions.contains_key(&action.name) {
                    Some(workflow.clone())
                } else {
                    context
                        .get_root_workflow()
                        .filter(|rw| rw.functions.contains_key(&action.name))
                };
                if let Some(func_wf) = func_workflow {
                    // Resolve all function args concurrently
                    let param_futures: Vec<_> = action
                        .params
                        .iter()
                        .map(|(key, val_template)| {
                            let key = key.clone();
                            let val_template = val_template.clone();
                            let executor = self.clone();
                            let ctx = context.clone();
                            async move {
                                let val = executor.process_parameter(&val_template, &ctx).await?;
                                Ok::<(String, Value), anyhow::Error>((key, val))
                            }
                        })
                        .collect();
                    let resolved = futures::future::join_all(param_futures).await;
                    let mut args = HashMap::new();
                    for result in resolved {
                        let (key, val) = result?;
                        args.insert(key, val);
                    }
                    // emit tool_start (convert Value args to String for display)
                    let display_params: HashMap<String, String> = args
                        .iter()
                        .map(|(k, v)| {
                            (
                                k.clone(),
                                match v {
                                    Value::String(s) => s.clone(),
                                    other => other.to_string(),
                                },
                            )
                        })
                        .collect();
                    context.emit_node_start(&node.id, &action.name, &display_params);
                    let result = self
                        .execute_function(action.name.clone(), args, func_wf, context)
                        .await;
                    context.emit_node_complete(&node.id, &action.name, &result);
                    return result;
                }

                // Check if this is a method call on an instance (instance.method)
                if let Some((instance_path, method_name)) = action.name.rsplit_once('.') {
                    if context.lookup_instance(instance_path).is_ok() {
                        let instance_id = context.lookup_instance(instance_path)?;
                        let class_def =
                            context.arena().class_def(instance_id).ok_or_else(|| {
                                anyhow!("Instance '{}' class_def not found in arena", instance_path)
                            })?;
                        let class_name =
                            context.arena().class_name(instance_id).unwrap_or_default();
                        let method_def = class_def.methods.get(method_name).ok_or_else(|| {
                            anyhow!("Class '{}' has no method '{}'", class_name, method_name)
                        })?;

                        // Resolve params
                        let param_futures: Vec<_> = action
                            .params
                            .iter()
                            .map(|(key, val_template)| {
                                let key = key.clone();
                                let val_template = val_template.clone();
                                let executor = self.clone();
                                let ctx = context.clone();
                                async move {
                                    let val =
                                        executor.process_parameter(&val_template, &ctx).await?;
                                    Ok::<(String, Value), anyhow::Error>((key, val))
                                }
                            })
                            .collect();
                        let resolved = futures::future::join_all(param_futures).await;
                        let mut args = HashMap::new();
                        for result in resolved {
                            let (key, val) = result?;
                            args.insert(key, val);
                        }

                        // Push method scope
                        use crate::core::instance_arena::{MethodScope, TypedSlot};
                        let field_cache = context
                            .arena()
                            .snapshot_fields(instance_id)
                            .unwrap_or_default();
                        let scope = MethodScope {
                            instance_id,
                            class_def: Arc::clone(&class_def),
                            instance_path: instance_path.to_string(),
                            dirty: HashMap::new(),
                            field_cache,
                            method_params: method_def.params.clone(),
                            param_values: Vec::new(),
                        };
                        context.push_method_scope(scope)?;

                        // Bind method parameters (supports both named and positional argN)
                        let mut param_slots = Vec::with_capacity(method_def.params.len());
                        let mut pos_idx = 0usize;
                        for param_name in &method_def.params {
                            if param_name == "self" {
                                param_slots.push(TypedSlot::Null);
                                continue;
                            }
                            let val = if let Some(v) = args.get(param_name) {
                                Some(v.clone())
                            } else {
                                // Fallback: try positional argN
                                let key = format!("arg{}", pos_idx);
                                args.get(&key).cloned()
                            };
                            pos_idx += 1;
                            if let Some(v) = val {
                                param_slots.push(TypedSlot::from_value(v.clone()));
                                context.set(param_name.clone(), v)?;
                            } else {
                                param_slots.push(TypedSlot::Null);
                            }
                        }
                        context.set_method_param_values(param_slots)?;

                        // Execute method body
                        let body_arc = Arc::clone(&method_def.body);
                        self.clone().execute_graph(body_arc, context).await?;

                        // Pop scope, flush dirty fields
                        if let Some(scope) = context.pop_method_scope()? {
                            context.flush_dirty_to_arena(&scope);
                        }

                        return Ok(Some(json!({"__arena_ref__": instance_id.0})));
                    }
                }

                // Resolve all task params concurrently
                let param_futures: Vec<_> = action
                    .params
                    .iter()
                    .map(|(key, val_template)| {
                        let key = key.clone();
                        let val_template = val_template.clone();
                        let executor = self.clone();
                        let ctx = context.clone();
                        async move {
                            let processed_val =
                                executor.process_parameter(&val_template, &ctx).await?;
                            let val_str = match processed_val {
                                Value::String(s) => s,
                                other => other.to_string(),
                            };
                            Ok::<(String, String), anyhow::Error>((key, val_str))
                        }
                    })
                    .collect();
                let resolved = futures::future::join_all(param_futures).await;
                let mut rendered_params = HashMap::new();
                for result in resolved {
                    let (key, val_str) = result?;
                    rendered_params.insert(key, val_str);
                }
                debug!("  Arguments: {:?}", rendered_params);
                context.emit_node_start(&node.id, &action.name, &rendered_params);
                let result = self
                    .execute_tool_internal(&action.name, &rendered_params, context)
                    .await;
                context.emit_node_complete(&node.id, &action.name, &result);
                result
            }
            NodeType::Assert(expr_str) => {
                debug!("│   Assert: {}", expr_str);
                context.emit_node_start(&node.id, "assert", &HashMap::new());
                let resolver = |path: &str| -> Option<Value> {
                    let clean = path.strip_prefix("ctx.").unwrap_or(path);
                    match clean {
                        "_tools_called" => {
                            let tools: Vec<Value> = context
                                .trace_entries()
                                .iter()
                                .map(|e| json!(e.tool.clone()))
                                .collect();
                            Some(json!(tools))
                        }
                        "_trace" => {
                            let entries: Vec<Value> = context
                                .trace_entries()
                                .iter()
                                .map(|e| {
                                    json!({
                                        "tool": e.tool,
                                        "params": e.params,
                                        "status": format!("{:?}", e.status),
                                    })
                                })
                                .collect();
                            Some(json!(entries))
                        }
                        _ => context.resolve_path(clean).ok().flatten(),
                    }
                };
                let result = match self.expr_eval.eval(expr_str, &resolver) {
                    Ok(val) => {
                        if expr_eval::is_truthy(&val) {
                            Ok(Some(json!({"assert": true, "expr": expr_str})))
                        } else {
                            Err(anyhow!("Assertion failed: `{}` → {:?}", expr_str, val))
                        }
                    }
                    Err(e) => Err(anyhow!("Assertion error: `{}` → {}", expr_str, e)),
                };
                context.emit_node_complete(&node.id, "assert", &result);
                result
            }
            NodeType::ReturnErr(err_obj) => {
                debug!("│   ReturnErr: {:?}", err_obj);
                context.emit_node_start(&node.id, "return_err", &HashMap::new());
                // Render template variables in string fields that reference $variables
                let rendered = if let Value::Object(map) = err_obj {
                    let mut rendered_map = serde_json::Map::new();
                    for (k, v) in map {
                        if let Value::String(s) = v {
                            if s.contains('$') || s.contains('{') {
                                // Contains variable references — resolve
                                let resolved = self.process_parameter(s, context).await.ok();
                                rendered_map.insert(
                                    k.clone(),
                                    resolved.unwrap_or_else(|| Value::String(s.clone())),
                                );
                            } else {
                                // Plain string — keep as-is
                                rendered_map.insert(k.clone(), Value::String(s.clone()));
                            }
                        } else {
                            rendered_map.insert(k.clone(), v.clone());
                        }
                    }
                    Value::Object(rendered_map)
                } else {
                    err_obj.clone()
                };
                let kind = rendered
                    .get("kind")
                    .and_then(|v| v.as_str())
                    .unwrap_or("custom");
                let message = rendered
                    .get("message")
                    .and_then(|v| v.as_str())
                    .unwrap_or("error");
                let result: Result<Option<Value>> = Err(anyhow!("[{}] {}", kind, message));
                context.emit_node_complete(&node.id, "return_err", &result);
                result
            }
            NodeType::Yield(expr) => {
                debug!("│   Yield: {}", expr);
                let value = self.process_parameter(expr, context).await?;
                context.emit(WorkflowEvent::Yield(value.clone()));
                Ok(Some(value))
            }
            NodeType::AssignCall { var, action } => {
                debug!("│   AssignCall: {} = {}(...)", var, action.name);
                // Resolve params (same as Task)
                let param_futures: Vec<_> = action
                    .params
                    .iter()
                    .map(|(key, val_template)| {
                        let key = key.clone();
                        let val_template = val_template.clone();
                        let executor = self.clone();
                        let ctx = context.clone();
                        async move {
                            let processed_val =
                                executor.process_parameter(&val_template, &ctx).await?;
                            let val_str = match processed_val {
                                Value::String(s) => s,
                                other => other.to_string(),
                            };
                            Ok::<(String, String), anyhow::Error>((key, val_str))
                        }
                    })
                    .collect();
                let resolved = futures::future::join_all(param_futures).await;
                let mut rendered_params = HashMap::new();
                for result in resolved {
                    let (key, val_str) = result?;
                    rendered_params.insert(key, val_str);
                }
                context.emit_node_start(&node.id, &action.name, &rendered_params);
                let result = self
                    .execute_tool_internal(&action.name, &rendered_params, context)
                    .await;
                // Store result in variable
                if let Ok(Some(ref val)) = result {
                    context.set(var.clone(), val.clone())?;
                }
                context.emit_node_complete(&node.id, &action.name, &result);
                result
            }
            NodeType::Foreach {
                item,
                list,
                body,
                parallel,
            } => {
                let clean_path = list.strip_prefix("ctx.").unwrap_or(list);
                debug!(
                    "│   Foreach{}: {} in {} ({})",
                    if *parallel { " parallel" } else { "" },
                    item,
                    list,
                    clean_path
                );
                let list_val = context
                    .resolve_path(clean_path)?
                    .ok_or_else(|| anyhow!("Foreach list variable '{}' not found", clean_path))?;
                let array = list_val
                    .as_array()
                    .ok_or_else(|| anyhow!("Variable '{}' is not an array.", list))?;

                if *parallel {
                    // Parallel foreach: spawn all iterations concurrently
                    info!("│   Foreach parallel: {} iterations", array.len());
                    let mut tasks = vec![];
                    for (i, val) in array.iter().enumerate() {
                        let ctx_clone = context.fork();
                        ctx_clone.set(item.clone(), val.clone())?;
                        ctx_clone.set("_index".to_string(), serde_json::json!(i))?;
                        let body_arc = Arc::new(*body.clone());
                        let executor = self.clone();
                        tasks.push(tokio::spawn(async move {
                            executor.execute_graph(body_arc, &ctx_clone).await?;
                            // Extract output from this iteration's context
                            let output = ctx_clone
                                .resolve_path("output")?
                                .unwrap_or(serde_json::Value::Null);
                            Ok::<serde_json::Value, anyhow::Error>(output)
                        }));
                    }
                    let results = futures::future::join_all(tasks).await;
                    let mut outputs = vec![];
                    for (i, result) in results.into_iter().enumerate() {
                        match result {
                            Ok(Ok(output)) => outputs.push(output),
                            Ok(Err(e)) => {
                                return Err(anyhow!("foreach parallel error at index {}: {}", i, e))
                            }
                            Err(e) => {
                                return Err(anyhow!(
                                    "foreach parallel task panic at index {}: {}",
                                    i,
                                    e
                                ))
                            }
                        }
                    }
                    let collected = serde_json::json!(outputs);
                    context.set("output".to_string(), collected.clone())?;
                    Ok(Some(collected))
                } else {
                    // Sequential foreach: original behavior
                    for (i, val) in array.iter().enumerate() {
                        debug!("│   ├─ Iteration {}/{}", i + 1, array.len());
                        context.set(item.clone(), val.clone())?;
                        let body_arc = Arc::new(*body.clone());
                        if let Err(e) = self.clone().execute_graph(body_arc, context).await {
                            return Err(anyhow!("Error inside foreach body at index {}: {}", i, e));
                        }
                    }
                    Ok(None)
                }
            }
            NodeType::Loop { condition, body } => {
                info!(
                    "  [Control] Entering while loop with condition: '{}'",
                    condition
                );
                let mut loop_count = 0;
                loop {
                    if loop_count > self.max_loop_iterations {
                        return Err(anyhow!(
                            "Loop limit exceeded (max: {}).",
                            self.max_loop_iterations
                        ));
                    }
                    if !self.evaluate_condition_async(condition, context).await? {
                        info!("  [Control] Loop condition is false, exiting loop.");
                        break;
                    }
                    let body_arc = Arc::new(*body.clone());
                    if let Err(e) = self.clone().execute_graph(body_arc, context).await {
                        return Err(anyhow!("Error inside loop body: {}", e));
                    }
                    loop_count += 1;
                }
                Ok(None)
            }
            NodeType::_ExternalCall {
                call_path,
                args,
                kwargs,
            } => {
                debug!(
                    "│   ExternalCall: {} args={:?} kwargs={:?}",
                    call_path, args, kwargs
                );

                // Resolve all kwargs and args concurrently
                use std::future::Future;
                use std::pin::Pin;
                type ParamFut =
                    Pin<Box<dyn Future<Output = anyhow::Result<(String, String)>> + Send>>;
                let mut param_futures: Vec<ParamFut> = Vec::new();
                for (k, v) in kwargs {
                    let key = k.clone();
                    let val_template = v.clone();
                    let executor = self.clone();
                    let ctx = context.clone();
                    param_futures.push(Box::pin(async move {
                        let resolved = executor.process_parameter(&val_template, &ctx).await?;
                        let val_str = match resolved {
                            Value::String(s) => s,
                            other => other.to_string(),
                        };
                        Ok((key, val_str))
                    }));
                }
                // Add positional args as numbered kwargs
                for (i, arg) in args.iter().enumerate() {
                    let key = format!("__arg{}", i);
                    let val_template = arg.clone();
                    let executor = self.clone();
                    let ctx = context.clone();
                    param_futures.push(Box::pin(async move {
                        let resolved = executor.process_parameter(&val_template, &ctx).await?;
                        let val_str = match resolved {
                            Value::String(s) => s,
                            other => other.to_string(),
                        };
                        Ok((key, val_str))
                    }));
                }
                let resolved = futures::future::join_all(param_futures).await;
                let mut resolved_kwargs: HashMap<String, String> = HashMap::new();
                for result in resolved {
                    let (key, val_str) = result?;
                    resolved_kwargs.insert(key, val_str);
                }

                self.execute_python_call(call_path, &resolved_kwargs)
            }
            NodeType::NewInstance { class_name, args } => {
                debug!("│   NewInstance: {} args={:?}", class_name, args);
                let class_def = workflow
                    .classes
                    .get(class_name)
                    .ok_or_else(|| anyhow!("Class '{}' not found", class_name))?;

                let mut fields_vec = Vec::with_capacity(class_def.fields.len());
                for field in &class_def.fields {
                    let value = if let Some(arg_expr) = args.get(&field.name) {
                        self.process_parameter(arg_expr, context).await?
                    } else if let Some(default_expr) = &field.default {
                        self.process_parameter(default_expr, context).await?
                    } else {
                        return Err(anyhow!(
                            "Class '{}' field '{}' has no default and was not provided",
                            class_name,
                            field.name
                        ));
                    };
                    fields_vec.push(value);
                }

                let class_def_arc = Arc::clone(class_def);
                let id = context.alloc_instance(
                    node.id.clone(),
                    class_name.clone(),
                    class_def_arc,
                    fields_vec,
                )?;
                // Deferred materialization: return arena proxy
                Ok(Some(json!({"__arena_ref__": id.0})))
            }
            NodeType::MethodCall {
                instance_path,
                method_name,
                args,
            } => {
                debug!(
                    "│   MethodCall: ${}.{}({:?})",
                    instance_path, method_name, args
                );

                // 1. Arena lookup (no clone)
                let instance_id = context.lookup_instance(instance_path)?;
                let class_def = context.arena().class_def(instance_id).ok_or_else(|| {
                    anyhow!("Instance '{}' class_def not found in arena", instance_path)
                })?;
                let class_name = context.arena().class_name(instance_id).unwrap_or_default();

                let method_def = class_def.methods.get(method_name).ok_or_else(|| {
                    anyhow!("Class '{}' has no method '{}'", class_name, method_name)
                })?;

                // 2. Snapshot fields + push method scope (single lock, then zero-lock reads)
                let field_cache = context
                    .arena()
                    .snapshot_fields(instance_id)
                    .unwrap_or_default();
                let scope = MethodScope {
                    instance_id,
                    class_def: Arc::clone(&class_def),
                    instance_path: instance_path.to_string(),
                    dirty: HashMap::new(),
                    field_cache,
                    method_params: method_def.params.clone(),
                    param_values: Vec::new(),
                };
                context.push_method_scope(scope)?;

                // 3. Bind method parameters + store TypedSlot values for C-2 fast path
                let mut param_slots = Vec::with_capacity(method_def.params.len());
                for param_name in &method_def.params {
                    if let Some(arg_expr) = args.get(param_name) {
                        let val = self.process_parameter(arg_expr, context).await?;
                        param_slots.push(TypedSlot::from_value(val.clone()));
                        context.set(param_name.clone(), val)?;
                    } else {
                        param_slots.push(TypedSlot::Null);
                    }
                }
                context.set_method_param_values(param_slots.clone())?;

                // 4. Execute method body
                let body_arc = Arc::clone(&method_def.body);
                self.clone().execute_graph(body_arc, context).await?;

                // 6. Pop scope, flush dirty to arena (minimal clones)
                if let Some(scope) = context.pop_method_scope()? {
                    context.flush_dirty_to_arena(&scope);
                }

                // 7. Deferred materialization: return arena proxy, resolve_path lazy-loads on demand
                Ok(Some(json!({"__arena_ref__": instance_id.0})))
            }
        }
    }

    pub async fn _run(
        self: Arc<Self>,
        workflow: Arc<WorkflowGraph>,
        config: &JuglansConfig,
    ) -> Result<WorkflowContext> {
        self.run_with_input(workflow, config, None).await
    }

    pub async fn run_with_input(
        self: Arc<Self>,
        workflow: Arc<WorkflowGraph>,
        config: &JuglansConfig,
        input: Option<Value>,
    ) -> Result<WorkflowContext> {
        info!(
            "🚀 Starting Execution: {} (v{})",
            workflow.name, workflow.version
        );
        debug!("👤 User: {}", config.account.name);
        let context = WorkflowContext::new();

        // Inject juglans.toml configuration into $config
        if let Ok(config_value) = serde_json::to_value(config) {
            context.set("config".to_string(), config_value)?;
        }

        // Set input data to ctx.input
        if let Some(input_val) = input {
            if let Some(obj) = input_val.as_object() {
                for (key, val) in obj {
                    context.set(format!("input.{}", key), val.clone())?;
                }
            }
            // Also set the complete input object
            context.set("input".to_string(), input_val)?;
        }

        info!("\n--- Execution Log ---");
        self.execute_graph(workflow, &context).await?;
        info!("🎉 Workflow finished successfully.");
        Ok(context)
    }

    /// Clean up unreachable nodes: check and skip nodes whose predecessors have all completed
    /// but still have non-zero in-degree. These nodes can never execute (predecessors didn't
    /// activate edges leading to them).
    fn cleanup_unreachable_nodes(
        workflow: &Arc<WorkflowGraph>,
        in_degrees: &Arc<Mutex<HashMap<NodeIndex, usize>>>,
        ready_queue: &Arc<Mutex<VecDeque<NodeIndex>>>,
        completed_nodes: &Arc<Mutex<HashSet<NodeIndex>>>,
    ) {
        let unreachable_nodes = Arc::new(Mutex::new(HashSet::new()));
        let completed = completed_nodes.lock().unwrap().clone();
        let degrees = in_degrees.lock().unwrap();

        // Find all unreachable nodes (skip test_* nodes)
        let unreachable: Vec<NodeIndex> = degrees
            .iter()
            .filter(|(idx, &degree)| {
                if degree == 0 || completed.contains(idx) {
                    return false;
                }

                // Skip test ROOT nodes — they are intentionally not executed
                // (test nodes with incoming edges are part of the main flow)
                if graph::is_test_node_id(&workflow.graph[**idx].id)
                    && workflow
                        .graph
                        .edges_directed(**idx, Direction::Incoming)
                        .count()
                        == 0
                {
                    return false;
                }

                // Check if all predecessors have completed
                let all_predecessors_done = workflow
                    .graph
                    .edges_directed(**idx, Direction::Incoming)
                    .all(|e| completed.contains(&e.source()));

                all_predecessors_done
            })
            .map(|(idx, _)| *idx)
            .collect();

        drop(degrees);
        drop(completed);

        // Recursively process unreachable nodes
        for node_idx in unreachable {
            Self::mark_unreachable_recursive(
                node_idx,
                workflow,
                in_degrees,
                ready_queue,
                completed_nodes,
                &unreachable_nodes,
            );
        }
    }

    /// Recursively mark a node and its successors as unreachable
    fn mark_unreachable_recursive(
        node_idx: NodeIndex,
        workflow: &Arc<WorkflowGraph>,
        in_degrees: &Arc<Mutex<HashMap<NodeIndex, usize>>>,
        ready_queue: &Arc<Mutex<VecDeque<NodeIndex>>>,
        completed_nodes: &Arc<Mutex<HashSet<NodeIndex>>>,
        unreachable_nodes: &Arc<Mutex<HashSet<NodeIndex>>>,
    ) {
        // Check if already processed
        if completed_nodes.lock().unwrap().contains(&node_idx) {
            return;
        }

        info!(
            "  -> Node [{}] is unreachable (skipping)",
            workflow.graph[node_idx].id
        );

        // Mark as completed (even though not executed)
        completed_nodes.lock().unwrap().insert(node_idx);
        // Also mark as unreachable
        unreachable_nodes.lock().unwrap().insert(node_idx);

        // Process all successor nodes
        for edge in workflow.graph.edges(node_idx) {
            let successor_idx = edge.target();

            let mut degrees = in_degrees.lock().unwrap();
            if let Some(degree) = degrees.get_mut(&successor_idx) {
                *degree -= 1;
                let new_degree = *degree;
                drop(degrees);

                if new_degree == 0 {
                    // Check if all predecessors of this successor are unreachable
                    let unreachable = unreachable_nodes.lock().unwrap();
                    let all_preds_unreachable = workflow
                        .graph
                        .edges_directed(successor_idx, Direction::Incoming)
                        .all(|e| unreachable.contains(&e.source()));
                    drop(unreachable);

                    if all_preds_unreachable {
                        // All predecessors unreachable, continue recursive marking
                        Self::mark_unreachable_recursive(
                            successor_idx,
                            workflow,
                            in_degrees,
                            ready_queue,
                            completed_nodes,
                            unreachable_nodes,
                        );
                    } else {
                        // Has normally-executed predecessors, add to queue
                        info!(
                            "Node [{}] is now ready to run.",
                            workflow.graph[successor_idx].id
                        );
                        ready_queue.lock().unwrap().push_back(successor_idx);
                    }
                }
            }
        }
    }

    /// Store node execution result into context.
    ///
    /// Result model (Rust-inspired):
    /// - Ok: `$output` is the raw value directly (zero overhead)
    /// - Err: `$output = { err: { kind, message, node } }` (structured error wrapper)
    fn store_node_result(node_id: &str, result: &Result<Option<Value>>, context: &WorkflowContext) {
        match result {
            Ok(Some(output)) => {
                debug!(
                    "  [Output] Result for [{}]: {}",
                    node_id,
                    serde_json::to_string(output).unwrap_or_default()
                );
                info!("  ✅ Success");
                // `.ok()` instead of `.unwrap()` — the set can fail when
                // a node is named "output" and ctx["output"] is already a
                // non-object value from a prior node. Crashing here would
                // kill the whole event stream, so we tolerate the miss.
                let _ = context.set(format!("{}.output", node_id), output.clone());
                let _ = context.set("output".to_string(), output.clone());
            }
            Ok(None) => {
                debug!("  [Output] No primary output for [{}].", node_id);
                info!("  ✅ Success");
                let _ = context.set(format!("{}.output", node_id), Value::Null);
                let _ = context.set("output".to_string(), Value::Null);
            }
            Err(e) => {
                warn!("  [Output] Error for [{}]: {}", node_id, e);
                error!("  ❌ Failed: {}", e);
                let kind = classify_error(e);
                let err_value = json!({
                    "err": {
                        "kind": kind,
                        "message": e.to_string(),
                        "node": node_id,
                    }
                });
                let _ = context.set(format!("{}.output", node_id), err_value.clone());
                let _ = context.set("output".to_string(), err_value);
                let _ = context.set(format!("{}.error", node_id), json!(e.to_string()));
                let _ = context.set(
                    "error".to_string(),
                    json!({
                        "node": node_id,
                        "message": e.to_string()
                    }),
                );
            }
        }
    }

    /// Resolve switch subject value for a node (if it has a switch route)
    fn resolve_switch_subject(
        node_id: &str,
        workflow: &WorkflowGraph,
        context: &WorkflowContext,
    ) -> Option<String> {
        let switch_route = workflow.switch_routes.get(node_id)?;

        let subject_value = if switch_route.subject.is_empty() {
            context.resolve_path("output").ok().flatten()
        } else {
            let clean_subject = switch_route.subject.trim_start_matches('$');
            let clean_subject = clean_subject.strip_prefix("ctx.").unwrap_or(clean_subject);
            context.resolve_path(clean_subject).ok().flatten()
        };

        let subject_str = match &subject_value {
            Some(Value::String(s)) => s.clone(),
            Some(Value::Number(n)) => n.to_string(),
            Some(Value::Bool(b)) => b.to_string(),
            Some(Value::Null) => String::new(),
            Some(v) => v.to_string(),
            None => String::new(),
        };

        info!(
            "  🔀 Switch on '{}' = '{}'",
            switch_route.subject, subject_str
        );
        Some(subject_str)
    }

    /// Activate a successor node by decrementing its in-degree and enqueueing if ready
    fn activate_successor(
        successor_idx: NodeIndex,
        workflow: &WorkflowGraph,
        in_degrees: &Arc<Mutex<HashMap<NodeIndex, usize>>>,
        ready_queue: &Arc<Mutex<VecDeque<NodeIndex>>>,
    ) {
        let mut degrees_guard = in_degrees.lock().unwrap();
        if let Some(degree) = degrees_guard.get_mut(&successor_idx) {
            *degree -= 1;
            if *degree == 0 {
                info!(
                    "Node [{}] is now ready to run.",
                    workflow.graph[successor_idx].id
                );
                ready_queue.lock().unwrap().push_back(successor_idx);
            }
        }
    }

    /// Evaluate outgoing edges from a completed node and enqueue ready successors
    async fn evaluate_outgoing_edges(edge_ctx: EdgeEvalContext<'_>) {
        let EdgeEvalContext {
            executor,
            node_idx,
            node_succeeded,
            switch_result,
            workflow,
            context,
            in_degrees,
            ready_queue,
        } = edge_ctx;
        let node_id = &workflow.graph[node_idx].id;
        let mut switch_matched = false;

        // Check if this node has a Result switch route (ok/err cases)
        let is_result_switch = workflow
            .switch_routes
            .get(node_id)
            .map(|r| r.cases.iter().any(|c| c.is_ok || c.is_err))
            .unwrap_or(false);

        // For err "kind" matching, resolve the error kind from context
        let error_kind = if !node_succeeded {
            context.resolve_path("output").ok().flatten().and_then(|v| {
                v.get("err")
                    .and_then(|e| e.get("kind"))
                    .and_then(|k| k.as_str())
                    .map(|s| s.to_string())
            })
        } else {
            None
        };

        // For result switches, use two-pass matching:
        // Pass 1: ok + specific err "kind" cases (highest priority)
        // Pass 2: catch-all err case (only if no specific match found)
        if is_result_switch {
            // Pass 1: ok and specific err "kind"
            for edge in workflow.graph.edges(node_idx) {
                let (edge_info, successor_idx) = (edge.weight(), edge.target());
                if let Some(ref case_value) = edge_info.switch_case {
                    if case_value == "__ok__" && node_succeeded && !switch_matched {
                        switch_matched = true;
                        info!(
                            "  -> Switch ok, taking path to [{}]",
                            workflow.graph[successor_idx].id
                        );
                        Self::activate_successor(successor_idx, workflow, in_degrees, ready_queue);
                    } else if case_value.starts_with("__err_")
                        && case_value.ends_with("__")
                        && case_value != "__err__"
                        && !node_succeeded
                        && !switch_matched
                    {
                        let case_kind = &case_value[6..case_value.len() - 2];
                        if error_kind.as_deref() == Some(case_kind) {
                            switch_matched = true;
                            info!(
                                "  -> Switch err \"{}\", taking path to [{}]",
                                case_kind, workflow.graph[successor_idx].id
                            );
                            Self::activate_successor(
                                successor_idx,
                                workflow,
                                in_degrees,
                                ready_queue,
                            );
                        }
                    }
                }
            }
            // Pass 2: catch-all err (only if no specific match)
            if !switch_matched && !node_succeeded {
                for edge in workflow.graph.edges(node_idx) {
                    let (edge_info, successor_idx) = (edge.weight(), edge.target());
                    if let Some(ref case_value) = edge_info.switch_case {
                        if case_value == "__err__" && !switch_matched {
                            switch_matched = true;
                            info!(
                                "  -> Switch err (catch-all), taking path to [{}]",
                                workflow.graph[successor_idx].id
                            );
                            Self::activate_successor(
                                successor_idx,
                                workflow,
                                in_degrees,
                                ready_queue,
                            );
                        }
                    }
                }
            }
            // If result switch handled the error, clear the global error
            if switch_matched && !node_succeeded {
                let _ = context.set("error".to_string(), serde_json::Value::Null);
            }
        }

        for edge in workflow.graph.edges(node_idx) {
            let (edge_info, successor_idx) = (edge.weight(), edge.target());
            let mut proceed = false;

            if is_result_switch {
                // Already handled above — skip result switch edges in main loop
            } else if edge_info.is_error_path {
                if !node_succeeded {
                    proceed = true;
                    info!(
                        "  -> Taking 'on error' path to [{}]",
                        workflow.graph[successor_idx].id
                    );
                }
            } else if node_succeeded {
                if let Some(ref switch_value) = switch_result {
                    if let Some(ref case_value) = edge_info.switch_case {
                        if case_value == switch_value && !switch_matched {
                            proceed = true;
                            switch_matched = true;
                            info!(
                                "  -> Switch case '{}' matched, taking path to [{}]",
                                case_value, workflow.graph[successor_idx].id
                            );
                        }
                    } else if edge_info.switch_case.is_none()
                        && workflow.switch_routes.contains_key(node_id)
                    {
                        // Default case — handled after all cases
                    } else if let Some(condition) = &edge_info.condition {
                        if executor
                            .evaluate_condition_async(condition, context)
                            .await
                            .unwrap_or(false)
                        {
                            proceed = true;
                            info!(
                                "  -> Condition TRUE, taking path to [{}]",
                                workflow.graph[successor_idx].id
                            );
                        }
                    } else {
                        proceed = true;
                        info!(
                            "  -> Taking unconditional path to [{}]",
                            workflow.graph[successor_idx].id
                        );
                    }
                } else if let Some(condition) = &edge_info.condition {
                    if executor
                        .evaluate_condition_async(condition, context)
                        .await
                        .unwrap_or(false)
                    {
                        proceed = true;
                        info!(
                            "  -> Condition TRUE, taking path to [{}]",
                            workflow.graph[successor_idx].id
                        );
                    } else {
                        info!(
                            "  -> Condition FALSE, skipping path to [{}]",
                            workflow.graph[successor_idx].id
                        );
                    }
                } else {
                    proceed = true;
                    info!(
                        "  -> Taking unconditional path to [{}]",
                        workflow.graph[successor_idx].id
                    );
                }
            }

            if proceed {
                Self::activate_successor(successor_idx, workflow, in_degrees, ready_queue);
            }
        }

        // Handle default switch case if no case matched
        let has_switch = switch_result.is_some() || is_result_switch;
        if has_switch && !switch_matched {
            for edge in workflow.graph.edges(node_idx) {
                let (edge_info, successor_idx) = (edge.weight(), edge.target());
                if edge_info.switch_case.is_none()
                    && !edge_info.is_error_path
                    && edge_info.condition.is_none()
                {
                    info!(
                        "  -> Switch default, taking path to [{}]",
                        workflow.graph[successor_idx].id
                    );
                    Self::activate_successor(successor_idx, workflow, in_degrees, ready_queue);
                    break;
                }
            }
        }
    }

    /// Execute a function node: bind parameters to context, execute sub-graph, return output
    pub fn execute_function<'a>(
        self: Arc<Self>,
        func_name: String,
        args: HashMap<String, Value>,
        workflow: Arc<WorkflowGraph>,
        context: &'a WorkflowContext,
    ) -> Pin<Box<dyn Future<Output = Result<Option<Value>>> + Send + 'a>> {
        Box::pin(async move {
            let func_def = workflow
                .functions
                .get(&func_name)
                .ok_or_else(|| anyhow!("Function '{}' not found", func_name))?;

            info!(
                "│ ⚡ Function [{}]({})",
                func_name,
                func_def
                    .params
                    .iter()
                    .filter_map(|p| args.get(p).map(|v| format!("{}={}", p, v)))
                    .collect::<Vec<_>>()
                    .join(", ")
            );

            // Bind parameters to context
            for param_name in &func_def.params {
                if let Some(val) = args.get(param_name) {
                    context.set(param_name.clone(), val.clone())?;
                }
            }

            // Execute function body sub-graph
            let body_arc = Arc::clone(&func_def.body);
            self.clone().execute_graph(body_arc, context).await?;

            // Return sub-graph output
            Ok(context.resolve_path("output")?.or(Some(Value::Null)))
        })
    }

    /// Find and execute a single node by name (used by on_tool handler)
    pub async fn run_single_node_by_name(
        self: Arc<Self>,
        name: &str,
        workflow: &Arc<WorkflowGraph>,
        context: &WorkflowContext,
    ) -> Result<Option<Value>> {
        let node_idx = *workflow
            .node_map
            .get(name)
            .ok_or_else(|| anyhow!("Node '{}' not found in workflow", name))?;
        self.run_single_node(node_idx, workflow, context).await
    }

    pub fn execute_graph<'a>(
        self: Arc<Self>,
        workflow: Arc<WorkflowGraph>,
        context: &'a WorkflowContext,
    ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>> {
        Box::pin(async move {
            // Inject current workflow into context (used by on_tool=[node] handler)
            context.set_current_workflow(workflow.clone());
            // Pin root workflow on first call (never overwritten by sub-graph execution)
            context.set_root_workflow(workflow.clone());

            // Inject class definition registry into context and expr_eval (for instance field index lookup)
            if !workflow.classes.is_empty() {
                context.set_class_registry(&workflow.classes);
                self.expr_eval
                    .set_class_registry(Arc::new(workflow.classes.clone()));
            }

            // Register simple workflow functions as expression functions (fn_registry bridge)
            if !workflow.functions.is_empty() {
                self.expr_eval.register_expr_functions(&workflow.functions);
            }

            // Exclude test_* ROOT nodes (in_degree=0) from normal execution count.
            // test_* nodes connected to the main flow (in_degree>0) are regular nodes.
            // Also exclude _deco_* nodes (decorator registration nodes handled by serve()).
            let excluded_node_count = workflow
                .graph
                .node_indices()
                .filter(|&idx| {
                    let id = &workflow.graph[idx].id;
                    // test_* ROOT nodes (in_degree=0) are excluded
                    graph::is_test_node_id(id)
                        && workflow
                            .graph
                            .edges_directed(idx, Direction::Incoming)
                            .count()
                            == 0
                })
                .count();
            let total_nodes = workflow.graph.node_count() - excluded_node_count;
            if total_nodes == 0 {
                return Ok(());
            }
            let in_degrees: Arc<Mutex<HashMap<NodeIndex, usize>>> = Arc::new(Mutex::new(
                workflow
                    .graph
                    .node_indices()
                    .map(|idx| {
                        (
                            idx,
                            workflow
                                .graph
                                .edges_directed(idx, Direction::Incoming)
                                .count(),
                        )
                    })
                    .collect(),
            ));
            let completed_nodes = Arc::new(Mutex::new(HashSet::new()));
            let ready_queue: Arc<Mutex<VecDeque<NodeIndex>>> = Arc::new(Mutex::new({
                let guard = in_degrees.lock().unwrap();
                guard
                    .iter()
                    .filter(|(idx, &degree)| {
                        let id = &workflow.graph[**idx].id;
                        degree == 0 && !graph::is_test_node_id(id)
                    })
                    .map(|(&idx, _)| idx)
                    .collect()
            }));

            while completed_nodes.lock().unwrap().len() < total_nodes {
                let mut tasks = vec![];
                let current_batch: Vec<NodeIndex> =
                    { ready_queue.lock().unwrap().drain(..).collect() };
                if current_batch.is_empty() {
                    let completed = completed_nodes.lock().unwrap().len();
                    if completed < total_nodes {
                        // Try to clean up unreachable nodes
                        info!("Detecting unreachable nodes...");
                        Self::cleanup_unreachable_nodes(
                            &workflow,
                            &in_degrees,
                            &ready_queue,
                            &completed_nodes,
                        );

                        // Check if new nodes were added to the queue
                        if ready_queue.lock().unwrap().is_empty() {
                            info!("Workflow graph execution finished early/deadlocked. ({} / {} nodes ran)", completed, total_nodes);
                            break;
                        }
                        // Otherwise continue executing newly added nodes
                        continue;
                    }
                    break;
                }
                info!(
                    "--- Starting execution batch of {} parallel nodes ---",
                    current_batch.len()
                );
                for node_idx in current_batch {
                    let self_clone = self.clone();
                    let workflow_clone = workflow.clone();
                    let context_clone = context.clone();
                    let in_degrees_clone = in_degrees.clone();
                    let ready_queue_clone = ready_queue.clone();
                    let completed_nodes_clone = completed_nodes.clone();

                    tasks.push(tokio::spawn(async move {
                        let node = &workflow_clone.graph[node_idx];
                        let node_result = self_clone
                            .clone()
                            .run_single_node(node_idx, &workflow_clone, &context_clone)
                            .await;
                        let node_succeeded = node_result.is_ok();

                        // Store node result in context
                        // Skip for NewInstance — it already stored the instance at node.id (line 713)
                        if !matches!(node.node_type, NodeType::NewInstance { .. }) {
                            Self::store_node_result(&node.id, &node_result, &context_clone);
                        }

                        if self_clone.debug_config.show_context {
                            if let Ok(ctx_val) = context_clone.get_as_value() {
                                info!(
                                    "📋 [Debug] Context after [{}]: {}",
                                    node.id,
                                    serde_json::to_string_pretty(&ctx_val).unwrap_or_default()
                                );
                            }
                        }

                        completed_nodes_clone.lock().unwrap().insert(node_idx);

                        // Evaluate switch subject if applicable
                        let switch_result =
                            Self::resolve_switch_subject(&node.id, &workflow_clone, &context_clone);

                        // Evaluate outgoing edges and enqueue ready successors
                        Self::evaluate_outgoing_edges(EdgeEvalContext {
                            executor: &self_clone,
                            node_idx,
                            node_succeeded,
                            switch_result: &switch_result,
                            workflow: &workflow_clone,
                            context: &context_clone,
                            in_degrees: &in_degrees_clone,
                            ready_queue: &ready_queue_clone,
                        })
                        .await;
                    }));
                }
                join_all(tasks).await;
            }

            // Check for unhandled node errors (root cause of deadlocks)
            if let Ok(Some(error_val)) = context.resolve_path("error") {
                if !error_val.is_null() {
                    let node = error_val
                        .get("node")
                        .and_then(|v| v.as_str())
                        .unwrap_or("unknown");
                    let msg = error_val
                        .get("message")
                        .and_then(|v| v.as_str())
                        .unwrap_or("unknown error");
                    return Err(anyhow::anyhow!("Node [{}] failed: {}", node, msg));
                }
            }

            Ok(())
        })
    }
}

/// Classify an anyhow error into a kind string for structured error output.
///
/// Checks for `[kind]` prefix pattern (from `return err`), then falls back
/// to heuristic keyword matching on the error message.
fn classify_error(e: &anyhow::Error) -> String {
    let msg = e.to_string();

    // Check for explicit [kind] prefix (from return err / fail)
    if msg.starts_with('[') {
        if let Some(end) = msg.find(']') {
            return msg[1..end].to_string();
        }
    }

    // Heuristic classification based on error message keywords
    let lower = msg.to_lowercase();
    if lower.contains("timeout") || lower.contains("timed out") {
        "timeout".to_string()
    } else if lower.contains("connection refused")
        || lower.contains("network")
        || lower.contains("dns")
        || lower.contains("connect error")
    {
        "network".to_string()
    } else if lower.contains("not found") || lower.contains("no such file") {
        "not_found".to_string()
    } else if lower.contains("parse")
        || lower.contains("invalid json")
        || lower.contains("expected")
    {
        "parse".to_string()
    } else if lower.contains("assertion") || lower.contains("assert") {
        "assertion".to_string()
    } else if lower.contains("python") || lower.contains("traceback") {
        "python".to_string()
    } else if lower.contains("permission")
        || lower.contains("unauthorized")
        || lower.contains("403")
    {
        "auth".to_string()
    } else {
        "unknown".to_string()
    }
}
