TeamAAS-2.5/TeamAAS.Base/TeamAAS.FlowEngine/Execution/FlowExecutor.cs

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using System.Windows;
using System.Xml.Linq;
using TeamAAS.FlowEditor.Models;
using TeamAAS.FlowEditor.Plugins;
using TeamAAS.FlowEngine;

namespace TeamAAS.FlowEditor.Execution
{
    /// <summary>
    /// 流程执行引擎 - 并行任务流模型
    /// 1. 无输入连线的节点是任务起点，每个作为独立线程启动
    /// 2. 每个节点等待所有前驱节点完成后再运行（IF判断除外）
    /// 3. 节点完成后，启动后继节点为独立线程（fire-and-forget），自身立即返回
    /// 4. Decision 节点只启动匹配分支的后继
    /// 5. 主线程通过计数器等待所有节点完成
    /// </summary>
    /// <summary>
    /// 执行一个 <see cref="FlowGraph"/>。
    ///
    /// 执行器只负责调度节点、维护节点状态和发布结果；具体业务动作由流程节点插件实现。
    /// 这样可以在不改变流程图数据结构的情况下扩展新的硬件或业务节点。
    /// </summary>
    public class FlowExecutor
    {
        private readonly FlowGraph _graph;
        private CancellationToken _token;

        /// <summary>已启动的节点ID集合（防止多前驱重复启动同一后继）</summary>
        private readonly HashSet<string> _startedNodes = new HashSet<string>();
        private readonly object _startLock = new object();

        /// <summary>正在运行的节点计数（主线程等待此计数归零）</summary>
        private int _pendingCount = 0;
        private readonly object _pendingLock = new object();

        /// <summary>所有已启动的节点任务（StartNode / TryStartExceptionBranchQueue / SkipNode 的 Task.Run 收集于此，供 ExecuteCoreAsync 等待）</summary>
        private readonly System.Collections.Concurrent.ConcurrentBag<Task> _nodeTasks = new System.Collections.Concurrent.ConcurrentBag<Task>();

        /// <summary>单节点模式：不启动后继节点</summary>
        private bool _singleNodeMode = false;

        /// <summary>跳过前驱等待的节点（用于从指定节点开始运行）</summary>
        private readonly HashSet<string> _skipPredecessorNodes = new HashSet<string>();

        public bool IsRunning { get; private set; }

        public event Action<FlowNode> NodeStatusChanged;
        public event Action<bool> ExecutionCompleted;
        public event Action EndNodeEncountered;

        public bool SkipReset { get; set; }

        public FlowExecutor(FlowGraph graph)
        {
            _graph = graph;
            // 反序列化（BinaryFormatter 跳过字段初始化）可能导致 Registry 为 null，兜底绑定 Debug 注册表
            if (_graph.Registry == null) _graph.Registry = ResultRegistry.Debug;
        }

        /// <summary>
        /// 异步执行整个流程图
        /// </summary>
        public async Task ExecuteAsync(CancellationToken token = default)
        {
            _singleNodeMode = false;

            // 空流程：没有任何节点可执行，直接完成（不进入等待循环，避免后续循环运行模式 CPU 满载）
            if (_graph.Nodes.Count == 0)
            {
                IsRunning = true;
                IsRunning = false;
                Application.Current?.Dispatcher.Invoke(() =>
                {
                    ExecutionCompleted?.Invoke(true);
                });
                return;
            }

            await ExecuteCoreAsync(() =>
            {
                var rootNodes = GetRootNodes();
                if (rootNodes.Count == 0 && _graph.Nodes.Count > 0)
                    rootNodes = _graph.Nodes.ToList();
                foreach (var root in rootNodes)
                    StartNode(root);
            }, token);
        }

        /// <summary>
        /// 从指定节点开始执行（向下流转）
        /// </summary>
        public async Task ExecuteFromNodeAsync(FlowNode startNode, CancellationToken token = default)
        {
            _singleNodeMode = false;
            _skipPredecessorNodes.Clear();
            _skipPredecessorNodes.Add(startNode.NodeId);
            await ExecuteCoreAsync(() => StartNode(startNode), token);
        }

        /// <summary>
        /// 仅执行单个节点（不启动后继）
        /// </summary>
        public async Task ExecuteSingleNodeAsync(FlowNode node, CancellationToken token = default)
        {
            _singleNodeMode = true;
            _skipPredecessorNodes.Clear();
            _skipPredecessorNodes.Add(node.NodeId);
            await ExecuteCoreAsync(() => StartNode(node), token);
        }

        private async Task ExecuteCoreAsync(Action startAction, CancellationToken token)
        {
            _token = token;
            _startedNodes.Clear();
            _pendingCount = 0;
            IsRunning = true;

            bool success = true;

            await Task.Run(() =>
            {
                try
                {
                    if (!SkipReset) ResetAllNodes();
                    startAction();

                    lock (_pendingLock)
                    {
                        while (_pendingCount > 0)
                        {
                            if (_token.IsCancellationRequested) break;
                            Monitor.Wait(_pendingLock, 100);
                        }
                    }
                }
                catch (OperationCanceledException)
                {
                    success = false;
                }
                catch (Exception)
                {
                    success = false;
                }
            });

            IsRunning = false;

            // 兜底：流程结束时强制收尾残留的 Running 状态。
            // 即使 RunNodeAsync 的 try/catch 漏网（例如 Thread.Abort、SEH 异常等），
            // 也不会让 UI 永久卡在 Running 同时 ExecutionCompleted 已触发。
            Application.Current?.Dispatcher.Invoke(() =>
            {
                foreach (var node in _graph.Nodes)
                {
                    if (node.Status == NodeRunStatus.Running)
                    {
                        node.Status = NodeRunStatus.Failed;
                        node.NotifyStatusChanged();
                    }
                }
                ExecutionCompleted?.Invoke(success);
            });
        }

        /// <summary>
        /// 启动一个节点为独立线程（fire-and-forget）
        /// </summary>
        private void StartNode(FlowNode node)
        {
            // 禁用节点：跳过执行，级联启动后继
            if (!node.IsEnabled)
            {
                SkipNode(node);
                return;
            }

            // 异常分支节点不走常规去重+前驱等待路径，改由独立路径触发（支持排队重入）
            if (node.Category == NodeCategory.ExceptionBranch)
            {
                StartExceptionBranchNode(node);
                return;
            }

            // 去重：多前驱可能同时尝试启动同一后继
            lock (_startLock)
            {
                if (_startedNodes.Contains(node.NodeId)) return;
                _startedNodes.Add(node.NodeId);
            }

            // 计数+1
            lock (_pendingLock) { _pendingCount++; }

            // 每个节点独立线程运行，不等待；Task 收集到 _nodeTasks 供 ExecuteCoreAsync 统一等待
            var nodeTask = Task.Run(async () =>
            {
                try
                {
                    await RunNodeAsync(node);
                }
                catch (Exception ex) { AppLogger.Error($"节点任务异常：{node.NodeName} @ 流程 {_graph.GraphName}", ex, nameof(FlowExecutor)); }
                finally
                {
                    // 计数-1，通知主线程
                    lock (_pendingLock)
                    {
                        _pendingCount--;
                        Monitor.Pulse(_pendingLock);
                    }
                }
            });
            _nodeTasks.Add(nodeTask);
        }

        /// <summary>
        /// 运行单个节点：等待前驱 → 执行 → 启动后继
        /// 关键修复：try/finally 保证任何异常路径都把 Status 从 Running 切到最终态，
        /// 避免外层 Task.Run 的 catch 吞掉异常后 Status 永久卡在 Running、_pendingCount 归零触发 ExecutionCompleted(true)。
        /// </summary>
        private async Task RunNodeAsync(FlowNode node)
        {
            // 跟踪本节点是否已置为 Running；异常时据此决定是否需要兜底收尾
            bool wasMarkedRunning = false;

            try
            {
                if (_token.IsCancellationRequested)
                {
                    UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                    return;
                }

                var predecessors = GetPredecessors(node);
                bool skipWait = _skipPredecessorNodes.Contains(node.NodeId);

                // 等待所有前驱节点完成（从指定节点开始运行时跳过等待）
                if (predecessors.Count > 0 && !skipWait)
                {
                    while (true)
                    {
                        if (_token.IsCancellationRequested)
                        {
                            UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                            return;
                        }

                        // 前驱有失败 → 跳过本节点，但仍启动后继
                        if (predecessors.Any(p => p.Status == NodeRunStatus.Failed))
                        {
                            UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                            StartSuccessors(node, null);
                            return;
                        }

                        // 所有前驱完成（Success或Skipped）
                        if (predecessors.All(p => p.Status == NodeRunStatus.Success || p.Status == NodeRunStatus.Skipped))
                        {
                            // 所有前驱都是Skipped（无Success）→ 级联跳过
                            if (!predecessors.Any(p => p.Status == NodeRunStatus.Success))
                            {
                                UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                                var cascadeResults = new Dictionary<string, object>
                                {
                                    ["状态"] = NodeRunStatus.Skipped.ToString(),
                                    ["耗时"] = 0
                                };
                                _graph.Registry.SetResult(_graph.GraphId, node.NodeId, node.NodeName, cascadeResults);
                                Application.Current?.Dispatcher.BeginInvoke(
                                    System.Windows.Threading.DispatcherPriority.Normal,
                                    new Action(() =>
                                    {
                                        node.SetResults(_graph.GraphId, cascadeResults);
                                    }));
                                if (!_singleNodeMode)
                                    StartSuccessors(node, null);
                                return;
                            }
                            break;
                        }

                        try { await Task.Delay(50, _token); }
                        catch (OperationCanceledException)
                        {
                            UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                            return;
                        }
                    }
                }

                // 禁用的节点直接跳过，但仍启动后继
                if (!node.IsEnabled)
                {
                    UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                    StartSuccessors(node, null);
                    return;
                }

                // 标记运行中
                UpdateNodeStatus(node, NodeRunStatus.Running, 0);
                wasMarkedRunning = true;
                await Task.Delay(50, _token); // 短暂延迟让UI能看到Running状态（取消时按外层异常路径处理）

                // 执行节点
                var sw = Stopwatch.StartNew();
                NodeRunStatus result;
                List<string> branchTargets = null;
                Dictionary<string, object> execResults = null;
                try
                {
                    // 新插件系统：通过 PluginLoader 创建实例
                    var pluginDisplayName = node.ToolName ?? node.PluginId;
                    var plugin = PluginLoader.Instance.CreateInstance(pluginDisplayName);
                    if (plugin != null)
                    {

                        // 恢复模型数据
                        if (node.PluginModel is Plugins.IFlowNodePlugin savedModel)
                            plugin.GetModel = savedModel.GetModel;
                        // 新实例需重新绑定所属流程的结果注册表
                        plugin.Registry = _graph.Registry;
                        #region 运行前休眠
                        // 从节点读取运行前/后休眠时间
                        int preDelayMs = node.PreDelayMs;
                        int postDelayMs = node.PostDelayMs;

                        // 运行前休眠
                        if (preDelayMs > 0)
                        {
                            try { await Task.Delay(preDelayMs, _token); }
                            catch (OperationCanceledException)
                            {
                                UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                                return;
                            }
                        }
                        #endregion
                        var status = await RunPluginAsync(plugin, node, node.TimeoutMs);
                        result = status;
                        execResults = plugin.LastResults ?? new Dictionary<string, object>();
                        execResults["状态"] = status.ToString();
                        execResults["结果"] = (status == NodeRunStatus.Success || status == NodeRunStatus.Skipped);

                        execResults["模型"] = plugin;
                        if (node.Category == NodeCategory.Decision)
                            branchTargets = plugin.BranchTargets;
                        #region 运行后休眠
                        // 运行后休眠（在启动后继之前）
                        if (postDelayMs > 0)
                        {
                            try { await Task.Delay(postDelayMs, _token); }
                            catch (OperationCanceledException) { return; }
                        }
                        #endregion
                    }
                    else
                    {
                        AppLogger.Error($"未找到插件，节点判为失败：{pluginDisplayName}（节点 {node.NodeName} @ 流程 {_graph.GraphName}）", nameof(FlowExecutor));
                        result = NodeRunStatus.Failed;
                    }
                }
                catch (Exception ex)
                {
                    AppLogger.Error($"节点执行失败：{node.NodeName} @ 流程 {_graph.GraphName}", ex, nameof(FlowExecutor));
                    result = NodeRunStatus.Failed;
                }

                sw.Stop();
                UpdateNodeStatus(node, result, (int)sw.ElapsedMilliseconds);
                node.CostTime = (int)sw.ElapsedMilliseconds;
                execResults["耗时"] = node.CostTime;
                // 存储执行结果到节点（UI线程更新）
                if (execResults != null && execResults.Count > 0)
                {
                    _graph.Registry.SetResult(_graph.GraphId, node.NodeId, node.NodeName, execResults);
                    Application.Current?.Dispatcher.BeginInvoke(
                        System.Windows.Threading.DispatcherPriority.Normal,
                        new Action(() =>
                        {
                            node.SetResults(_graph.GraphId, execResults);
                        }));
                }



                // 执行失败：不启动常规后继，但显式触发 ExceptionBranch 类型的直接后继
                if (result == NodeRunStatus.Failed)
                {
                    StartExceptionBranchSuccessors(node);
                    return;
                }
                // 警告返回时 如果可跳过警告，则可继续后续节点  否则暂停
                if (result == NodeRunStatus.Warning && !node.IsSkipWarning) return;

                // End节点：不启动后继，触发结束事件
                if (node.Category == NodeCategory.End
                    && execResults != null
                    && execResults.TryGetValue("操作", out var opText)
                    && opText?.ToString() == "结束整个流程")
                {
                    EndNodeEncountered?.Invoke();
                    return;
                }

                // 启动后继节点（fire-and-forget，不等待）
                if (!_singleNodeMode)
                    StartSuccessors(node, branchTargets);
            }
            catch (Exception)
            {
                // 异常路径兜底：如果已置 Running，强制收尾为 Failed，防止 UI 永久卡 Running
                if (wasMarkedRunning && node.Status == NodeRunStatus.Running)
                {
                    try { UpdateNodeStatus(node, NodeRunStatus.Failed, node.CostTime); } catch { }
                }
                throw;  // 继续向上抛，由 Task.Run 的 catch 吞掉，_pendingCount 在 finally 减
            }
        }

        /// <summary>
        /// 运行插件并施加超时看门狗。timeoutMs<=0 时直接同步运行（保持原行为）。
        /// 超时后判为 Failed 并继续后续流程；阻塞式硬件调用无法强杀，仅保证流程不再无限等待。
        /// </summary>
        private async Task<NodeRunStatus> RunPluginAsync(IFlowNodePlugin plugin, FlowNode node, int timeoutMs)
        {
            if (timeoutMs <= 0)
                return plugin.Run(_token);

            var runTask = Task.Run(() => plugin.Run(_token));
            var timeoutTask = Task.Delay(timeoutMs, _token);
            // 观察两者潜在异常，避免未观察任务异常
            _ = runTask.ContinueWith(t => { var _e = t.Exception; }, TaskContinuationOptions.OnlyOnFaulted);
            _ = timeoutTask.ContinueWith(t => { var _e = t.Exception; }, TaskContinuationOptions.OnlyOnFaulted);

            var completed = await Task.WhenAny(runTask, timeoutTask);
            if (completed == runTask)
                return await runTask;   // 正常完成；插件抛异常则向上抛，由调用方 catch 记录

            if (_token.IsCancellationRequested)
                throw new OperationCanceledException(_token);

            AppLogger.Error($"节点执行超时（{timeoutMs}ms）：{node.NodeName} @ 流程 {_graph.GraphName}；阻塞式硬件调用可能仍在后台线程运行，无法强制中断", nameof(FlowExecutor));
            return NodeRunStatus.Failed;
        }

        /// <summary>
        /// 启动后继节点（每个后继作为独立线程，不等待）
        /// </summary>
        private void StartSuccessors(FlowNode node, List<string> branchTargets)
        {
            var outgoing = _graph.Connections
                .Where(c => c.SourceNodeId == node.NodeId)
                .ToList();

            // 过滤掉 ExceptionBranch 后继：仅由前驱 Failed 路径显式触发，
            // 避免 Success/Skipped/Decision 分支误启动异常分支节点
            outgoing = outgoing
                .Where(c =>
                {
                    var t = _graph.GetNode(c.TargetNodeId);
                    return t == null || t.Category != NodeCategory.ExceptionBranch;
                })
                .ToList();

            // Decision节点：按NodeName筛选后继，非选中分支级联跳过
            if (node.Category == NodeCategory.Decision && branchTargets != null && branchTargets.Count > 0)
            {
                var selectedConns = new List<FlowConnection>();
                var skippedConns = new List<FlowConnection>();

                foreach (var conn in outgoing)
                {
                    var target = _graph.GetNode(conn.TargetNodeId);
                    if (target != null && branchTargets.Contains(target.NodeName))
                        selectedConns.Add(conn);
                    else
                        skippedConns.Add(conn);
                }

                // 非选中分支：标记Skipped并级联启动其后继
                foreach (var conn in skippedConns)
                {
                    var target = _graph.GetNode(conn.TargetNodeId);
                    if (target != null)
                        SkipNode(target);
                }

                outgoing = selectedConns;
            }

            foreach (var conn in outgoing)
            {
                if (_token.IsCancellationRequested) break;
                var target = _graph.GetNode(conn.TargetNodeId);
                if (target != null)
                {
                    StartNode(target);
                }
            }
        }

        /// <summary>
        /// 启动失败节点的异常分支后继（仅 Category==ExceptionBranch 的直接后继）。
        /// 与 StartSuccessors 互斥：常规后继由 StartSuccessors 处理，异常分支后继由此处显式触发。
        /// </summary>
        private void StartExceptionBranchSuccessors(FlowNode failedNode)
        {
            if (_singleNodeMode) return;
            if (_token.IsCancellationRequested) return;

            var exceptionTargets = _graph.Connections
                .Where(c => c.SourceNodeId == failedNode.NodeId)
                .Select(c => _graph.GetNode(c.TargetNodeId))
                .Where(t => t != null && t.Category == NodeCategory.ExceptionBranch)
                .Distinct()
                .ToList();

            foreach (var exNode in exceptionTargets)
            {
                if (_token.IsCancellationRequested) break;
                StartExceptionBranchNode(exNode);
            }
        }

        /// <summary>
        /// 触发异常分支节点：原子 +1 触发计数，并尝试启动串行消费 Task。
        /// 多次触发会排队（计数累积），由运行中的 Task 在下一轮迭代消费。
        /// </summary>
        private void StartExceptionBranchNode(FlowNode node)
        {
            if (!node.IsEnabled)
            {
                SkipNode(node);
                return;
            }
            node.IncrementExceptionTrigger();
            TryStartExceptionBranchQueue(node);
        }

        /// <summary>
        /// 尝试获取运行权并启动队列消费 Task。运行权用 CAS 保证全局唯一。
        /// </summary>
        private void TryStartExceptionBranchQueue(FlowNode node)
        {
            if (!node.TryAcquireExceptionRun()) return;  // 已有 Task 在跑，触发已排队，直接返回

            lock (_pendingLock) { _pendingCount++; }

            var exceptionTask = Task.Run(async () =>
            {
                try
                {
                    await RunExceptionBranchAsync(node);
                }
                catch (Exception ex) { AppLogger.Error($"异常分支任务异常：{node.NodeName} @ 流程 {_graph.GraphName}", ex, nameof(FlowExecutor)); }
                finally
                {
                    node.ReleaseExceptionRun();
                    lock (_pendingLock)
                    {
                        _pendingCount--;
                        Monitor.Pulse(_pendingLock);
                    }
                    // 防漏触发：在 Release 之前到达的新触发不会被本 Task 消费，再尝试启动一次
                    if (node.ExceptionTriggers > 0 && !_token.IsCancellationRequested)
                        TryStartExceptionBranchQueue(node);
                }
            });
            _nodeTasks.Add(exceptionTask);
        }

        /// <summary>
        /// 异常分支节点执行循环：串行消费触发队列，每次执行完检查是否还有排队触发。
        /// 不等待前驱（前驱已 Failed 才会到达此处）。
        /// </summary>
        private async Task RunExceptionBranchAsync(FlowNode node)
        {
            while (node.ExceptionTriggers > 0)
            {
                node.DecrementExceptionTrigger();
                if (_token.IsCancellationRequested)
                {
                    UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                    break;
                }
                await RunExceptionBranchOnceAsync(node);
            }
        }

        /// <summary>
        /// 异常分支节点单次执行：跳过前驱等待，直接运行插件并启动常规后继。
        /// 逻辑与 RunNodeAsync 的"执行+结果+后继"段对齐，但不进入前驱等待循环。
        /// </summary>
        private async Task RunExceptionBranchOnceAsync(FlowNode node)
        {
            bool wasMarkedRunning = false;
            try
            {
                if (_token.IsCancellationRequested)
                {
                    UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                    return;
                }

                UpdateNodeStatus(node, NodeRunStatus.Running, 0);
                wasMarkedRunning = true;
                Thread.Sleep(50);

                var sw = Stopwatch.StartNew();
                NodeRunStatus result;
                Dictionary<string, object> execResults = null;

                int preDelayMs = node.PreDelayMs;
                int postDelayMs = node.PostDelayMs;

                if (preDelayMs > 0)
                {
                    try { await Task.Delay(preDelayMs, _token); }
                    catch (OperationCanceledException)
                    {
                        UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);
                        return;
                    }
                }

                try
                {
                    var pluginDisplayName = node.ToolName ?? node.PluginId;
                    var plugin = PluginLoader.Instance.CreateInstance(pluginDisplayName);
                    if (plugin != null)
                    {

                        if (node.PluginModel is IFlowNodePlugin savedModel)
                            plugin.GetModel = savedModel.GetModel;
                        plugin.Registry = _graph.Registry;
                        var status = await RunPluginAsync(plugin, node, node.TimeoutMs);
                        result = status;
                        execResults = plugin.LastResults ?? new Dictionary<string, object>();
                        execResults["状态"] = status.ToString();
                        execResults["结果"] = (status == NodeRunStatus.Success || status == NodeRunStatus.Skipped);
                        execResults["耗时"] = plugin.CostTime;
                        execResults["模型"] = plugin;
                    }
                    else
                    {
                        AppLogger.Error($"未找到插件，异常分支节点判为失败：{pluginDisplayName}（节点 {node.NodeName} @ 流程 {_graph.GraphName}）", nameof(FlowExecutor));
                        result = NodeRunStatus.Failed;
                    }
                }
                catch (Exception ex)
                {
                    AppLogger.Error($"异常分支节点执行失败：{node.NodeName} @ 流程 {_graph.GraphName}", ex, nameof(FlowExecutor));
                    result = NodeRunStatus.Failed;
                }
                if (postDelayMs > 0)
                {
                    try { await Task.Delay(postDelayMs, _token); }
                    catch (OperationCanceledException) { return; }
                }
                sw.Stop();
                UpdateNodeStatus(node, result, (int)sw.ElapsedMilliseconds);
                node.CostTime = (int)sw.ElapsedMilliseconds;
                execResults["耗时"] = node.CostTime;
                if (execResults != null && execResults.Count > 0)
                {
                    _graph.Registry.SetResult(_graph.GraphId, node.NodeId, node.NodeName, execResults);
                    Application.Current?.Dispatcher.BeginInvoke(
                        System.Windows.Threading.DispatcherPriority.Normal,
                        new Action(() =>
                        {
                            node.SetResults(_graph.GraphId, execResults);
                        }));
                }
                if (!_singleNodeMode)
                    StartSuccessors(node, null);

                if (result == NodeRunStatus.Failed)
                    StartExceptionBranchSuccessors(node);
            }
            catch (Exception)
            {
                if (wasMarkedRunning && node.Status == NodeRunStatus.Running)
                {
                    try { UpdateNodeStatus(node, NodeRunStatus.Failed, node.CostTime); } catch { }
                }
                throw;
            }
        }

        /// <summary>
        /// 跳过节点：标记Skipped + 设置结果 + 启动后继（防止下游死等）
        /// </summary>
        private void SkipNode(FlowNode node)
        {
            lock (_startLock)
            {
                if (_startedNodes.Contains(node.NodeId)) return;
                _startedNodes.Add(node.NodeId);
            }

            lock (_pendingLock) { _pendingCount++; }

            var skipTask = Task.Run(() =>
            {
                try
                {
                    UpdateNodeStatus(node, NodeRunStatus.Skipped, 0);

                    var skipResults = new Dictionary<string, object>
                    {
                        ["状态"] = NodeRunStatus.Skipped.ToString(),
                        ["耗时"] = 0
                    };
                    _graph.Registry.SetResult(_graph.GraphId, node.NodeId, node.NodeName, skipResults);
                    Application.Current?.Dispatcher.BeginInvoke(
                        System.Windows.Threading.DispatcherPriority.Normal,
                        new Action(() =>
                        {
                            node.SetResults(_graph.GraphId, skipResults);
                        }));

                    if (!_singleNodeMode)
                        StartSuccessors(node, null);
                }
                finally
                {
                    lock (_pendingLock)
                    {
                        _pendingCount--;
                        Monitor.Pulse(_pendingLock);
                    }
                }
            });
            _nodeTasks.Add(skipTask);
        }

        /// <summary>
        /// 获取前驱节点
        /// </summary>
        private List<FlowNode> GetPredecessors(FlowNode node)
        {
            return _graph.Connections
                .Where(c => c.TargetNodeId == node.NodeId)
                .Select(c => _graph.GetNode(c.SourceNodeId))
                .Where(n => n != null)
                .ToList();
        }

        /// <summary>
        /// 获取根节点（没有输入连线的节点）
        /// </summary>
        private List<FlowNode> GetRootNodes()
        {
            var targetIds = new HashSet<string>(
                _graph.Connections.Select(c => c.TargetNodeId)
            );
            return _graph.Nodes
                .Where(n => !targetIds.Contains(n.NodeId)
                            && n.Category != NodeCategory.ExceptionBranch)
                .ToList();
        }

        /// <summary>
        /// 更新节点状态。所有赋值与 UI 通知都切到 UI 线程，避免后台线程触发 PropertyChanged 引发 WPF 绑定异常。
        /// 用 Dispatcher.BeginInvoke（异步、非阻塞）而非 Invoke（同步阻塞）：
        /// 后台每个节点线程都会并发更新状态，若用同步 Invoke，UI 线程繁忙（模态窗体泵消息、画布/结果表重绘）时
        /// 会把大量更新挤在一起——前后的 Running 与完成态被同一渲染周期"合并"，导致看不到闪灯/耗时跳动，
        /// 且最后的 ExecutionCompleted（IsRunning 复位）也会被拖后，表现为"要过段时间才恢复"。
        /// </summary>
        private void UpdateNodeStatus(FlowNode node, NodeRunStatus status, int costTime)
        {
            Application.Current?.Dispatcher.BeginInvoke(
                System.Windows.Threading.DispatcherPriority.Normal,
                new Action(() =>
                {
                    node.Status = status;
                    node.CostTime = costTime;
                    node.NotifyStatusChanged();
                    NodeStatusChanged?.Invoke(node);
                }));
        }

        /// <summary>
        /// 重置所有节点状态
        /// </summary>
        private void ResetAllNodes()
        {
            Application.Current?.Dispatcher.Invoke(() =>
            {
                foreach (var node in _graph.Nodes)
                {
                    node.Status = NodeRunStatus.NotStarted;
                    node.CostTime = 0;
                    node.ResultItems?.Clear();
                    node.NotifyStatusChanged();
                    if (node.Category == NodeCategory.ExceptionBranch)
                        node.ResetExceptionState();
                }
            });

            // 清空当前结果（历史记录保留，跨多次执行累积）
            _graph.Registry.ClearFlow(_graph.GraphId);
        }
    }
}