package com.agent.management.engine;

import com.agent.management.engine.hermes.ClarifyHooks;
import com.agent.management.external.SseEventBus;
import com.agent.management.model.entity.WorkflowRunNode;
import com.agent.management.model.entity.WorkflowRunVisit;
import com.agent.management.repository.WorkflowPauseRepository;
import com.agent.management.repository.WorkflowRunVisitRepository;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.nio.file.Path;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 按 DAG 拓扑执行节点（Dataflow 模型）。
 *
 * <p>调度规则：节点所有前驱完成后立即被调度，不等同层兄弟节点。
 * 通过 {@link CompletableFuture#whenComplete} 回调链驱动后继节点；
 * 不再有"层"barrier，慢节点不再阻塞快节点的下游链路。</p>
 *
 * <p>线程模型：</p>
 * <ul>
 *   <li>{@link #nodeExecutor}：快节点 worker 线程池，{@link #executeOneNode} 在此执行</li>
 *   <li>{@link #clarifyWaitExecutor}：长阻塞节点专用池，承担 agent / smartAction 等 Hermes 调用，
 *       避免长时间等待用户 clarify 答题时把 nodeExecutor 占满，导致其他快节点无 worker 可用</li>
 *   <li>回调链（onNodeCompleted / tryScheduleSuccessors）在节点所在池中执行</li>
 *   <li>主线程仅在 {@code ctx.done.get(...)} 上等待</li>
 * </ul>
 */
@Slf4j
@Component
public class WorkflowLevelExecutor {

    // 注册 JavaTimeModule：ExecutionLog 含 Instant 字段，默认 ObjectMapper 不支持 Java 8 时间类型，
    // 否则 writeValueAsString 抛 InvalidDefinitionException，logs 字段会被 catch 块吞掉，运行记录页无法回放思考过程。
    private static final ObjectMapper MAPPER = new ObjectMapper()
            .registerModule(new JavaTimeModule());

    /**
     * dataflow 调度主线程等待超时（兜底，避免死锁时永久阻塞）。
     * <p>v1.1：从 30 分钟调到 24 小时。Askfor 场景用户可能几小时后才回来回答 clarify，
     * 工作流线程一直阻塞。真正的"无超时"由前端断线重连机制保证，这里只是最终兜底。</p>
     */
    private static final long EXECUTION_TIMEOUT_MINUTES = 1440L;

    private final Map<String, NodeExecutor> executorMap;
    private final SseEventBus eventBus;
    private final NodeWorkspaceBuilder workspaceBuilder;
    /**
     * v1.1：clarify_request 暂停点持久化（断线重连后恢复 Askfor UI）。
     * 非必需依赖：测试场景或引擎独立测试时允许为 null。
     */
    private final WorkflowPauseRepository pauseRepository;
    /**
     * v2.0：DAG 模式下也持久化 Visit 快照（与 AGENTIC 模式统一）。
     * 非必需依赖：测试场景允许为 null（不持久化，仅 SSE 推送）。
     */
    private final WorkflowRunVisitRepository visitRepository;
    /**
     * 快节点执行的线程池。独立于 WorkflowEngine.executor（工作流主线程池），
     * 避免节点任务与外层 wrapper 争用同一池导致死锁。
     *
     * <p>仅承担 userInput / llm / condition / output / skill / knowledgeRetrieval 等
     * 同步短任务节点；agent / smartAction 等可能长时间阻塞的节点走 {@link #clarifyWaitExecutor}。</p>
     */
    private final ExecutorService nodeExecutor = Executors.newFixedThreadPool(
            Math.max(8, Runtime.getRuntime().availableProcessors() * 2),
            r -> {
                Thread t = new Thread(r, "workflow-node-executor");
                t.setDaemon(true);
                return t;
            }
    );

    /**
     * 长阻塞节点专用线程池（"clarify 等待池"）。
     *
     * <p>背景：Hermes Agent / SmartAction 通过 SSE 长连接串行读取 Bridge 事件，
     * 当 Python 侧触发 {@code clarify_request} 时，整个 SSE 读取会一直阻塞到用户答完题
     * （可能长达数小时）。如果这类节点跑在 {@link #nodeExecutor} 里，多个工作流同时卡 clarify
     * 会快速耗尽池容量，导致后续快节点（userInput / output 等）拿不到 worker。</p>
     *
     * <p>H-8 修复：原 {@code Executors.newFixedThreadPool} 用无界 {@code LinkedBlockingQueue}，
     * 大量并发长任务会无限堆积导致 OOM。改为 {@link ThreadPoolExecutor} + 200 容量有界队列 +
     * {@code CallerRunsPolicy} 拒绝策略：队列满时让提交线程同步执行新任务，形成自然反压，
     * 既不丢任务也避免 OOM；同时通过 corePoolSize/maxPoolSize 弹性伸缩，正常负载下 core 即可消化。</p>
     *
     * <ul>
     *   <li>corePoolSize = {@code max(8, CPU*2)}：常规负载</li>
     *   <li>maxPoolSize = {@code max(32, CPU*4)}：clarify 长等待下扩容</li>
     *   <li>队列容量 200：超过 core 但未达 max 时的缓冲；达 max 后触发 CallerRunsPolicy</li>
     *   <li>独立线程名 {@code "clarify-wait-executor"}，便于线程 dump 定位</li>
     * </ul>
     */
    private final ExecutorService clarifyWaitExecutor = new ThreadPoolExecutor(
            Math.max(8, Runtime.getRuntime().availableProcessors() * 2),
            Math.max(32, Runtime.getRuntime().availableProcessors() * 4),
            60L, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(200),
            r -> {
                Thread t = new Thread(r, "clarify-wait-executor");
                t.setDaemon(true);
                return t;
            },
            new ThreadPoolExecutor.CallerRunsPolicy()
    );

    /**
     * 判断节点是否属于"长阻塞"类型，需提交到 {@link #clarifyWaitExecutor}。
     *
     * <p>当前覆盖 agent / smartAction：在 hermes.enabled=true 时它们走 SSE 阻塞读，
     * 可能数小时等 clarify；在 hermes.enabled=false 时它们仍是较重的 LLM 同步调用，
     * 分到独立池也不会拖慢调度。</p>
     */
    private boolean isLongBlockingNode(String nodeType) {
        return "agent".equals(nodeType) || "smartAction".equals(nodeType);
    }

    /**
     * 按节点类型选择执行池：长阻塞节点 → {@link #clarifyWaitExecutor}，其他 → {@link #nodeExecutor}。
     */
    private ExecutorService selectPoolFor(String nodeType) {
        return isLongBlockingNode(nodeType) ? clarifyWaitExecutor : nodeExecutor;
    }

    @Autowired
    public WorkflowLevelExecutor(Map<String, NodeExecutor> injected, SseEventBus eventBus,
                                  NodeWorkspaceBuilder workspaceBuilder,
                                  @Autowired(required = false) WorkflowPauseRepository pauseRepository,
                                  @Autowired(required = false) WorkflowRunVisitRepository visitRepository) {
        // Spring 注入 Map<String, NodeExecutor> 时 key 是 bean name（如 "userInputExecutor"），
        // 而引擎查找用的是 getType() 返回值（如 "userInput"）。两者命名不一致，
        // 此处按 getType() 重建 map，避免对 bean 命名的依赖。
        Map<String, NodeExecutor> typed = new java.util.HashMap<>();
        for (NodeExecutor executor : injected.values()) {
            typed.put(executor.getType(), executor);
        }
        this.executorMap = typed;
        this.eventBus = eventBus;
        this.workspaceBuilder = workspaceBuilder;
        this.pauseRepository = pauseRepository;
        this.visitRepository = visitRepository;
        log.info("[WorkflowLevelExecutor] 已注册节点执行器: {}", this.executorMap.keySet());
    }

    /** 兼容旧测试用例：无 pauseRepository / visitRepository 的构造器 */
    public WorkflowLevelExecutor(Map<String, NodeExecutor> injected, SseEventBus eventBus,
                                  NodeWorkspaceBuilder workspaceBuilder) {
        this(injected, eventBus, workspaceBuilder, null, null);
    }

    /** 兼容测试用例：仅 pauseRepository，无 visitRepository */
    public WorkflowLevelExecutor(Map<String, NodeExecutor> injected, SseEventBus eventBus,
                                  NodeWorkspaceBuilder workspaceBuilder,
                                  WorkflowPauseRepository pauseRepository) {
        this(injected, eventBus, workspaceBuilder, pauseRepository, null);
    }

    @PreDestroy
    public void shutdown() {
        nodeExecutor.shutdown();
        clarifyWaitExecutor.shutdown();
        try {
            if (!nodeExecutor.awaitTermination(5, TimeUnit.SECONDS)) {
                nodeExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            nodeExecutor.shutdownNow();
        }
        try {
            // clarify 等待池的线程可能卡在用户长时间未答的 clarify 上，
            // 给稍长的优雅关闭时间，超时后强制中断（线程是 daemon，不影响 JVM 退出）
            if (!clarifyWaitExecutor.awaitTermination(8, TimeUnit.SECONDS)) {
                clarifyWaitExecutor.shutdownNow();
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            clarifyWaitExecutor.shutdownNow();
        }
    }

    /**
     * 一次工作流执行的产物：最终输出 + 每节点记录 + 是否失败 + 错误信息
     */
    public static class ExecutionOutcome {
        public final Map<String, Object> finalOutputs;
        public final List<WorkflowRunNode> nodeRecords;
        public final boolean failed;
        public final String errorMsg;

        public ExecutionOutcome(Map<String, Object> finalOutputs, List<WorkflowRunNode> nodeRecords,
                                boolean failed, String errorMsg) {
            this.finalOutputs = finalOutputs;
            this.nodeRecords = nodeRecords;
            this.failed = failed;
            this.errorMsg = errorMsg;
        }
    }

    /**
     * Dataflow 调度的共享上下文。
     *
     * <p>所有并发访问的可变状态集中在此对象，便于审计与生命周期管理。
     * dag / context / runId / runRecordId / emitter 为不可变引用，回调链中通过 ctx.dag / ctx.context 直接访问。</p>
     */
    private static final class DataflowCtx {
        final DagResolver.ResolvedDag dag;
        final WorkflowContext context;
        final String runId;
        final Long runRecordId;
        final SseEmitter emitter;

        /** 活跃节点集合（待执行或执行中）；非活跃节点入度归零后会被标记 SKIPPED */
        final Set<String> activeNodes = ConcurrentHashMap.newKeySet();
        /** 活跃边集合（条件分支按选中 sourceHandle 过滤后实际生效的边） */
        final Set<String> activeEdges = ConcurrentHashMap.newKeySet();
        /** 节点的剩余入度（按 source 去重，与 DagResolver 入度规则一致）；归零则可调度 */
        final Map<String, AtomicInteger> pendingInputs = new ConcurrentHashMap<>();
        /** nodeId → 在飞 future；abort 时用于 cancel */
        final Map<String, CompletableFuture<NodeExecutionResult>> inFlightFutures = new ConcurrentHashMap<>();
        /** 已完成（含 SKIPPED）的节点集合；用于去重 onNodeCompleted 与 triggerAbort 之间的双重处理 */
        final Set<String> completedNodes = ConcurrentHashMap.newKeySet();
        /**
         * nodeId → 本次执行的 visitUuid（executeOneNode 入口写入，onNodeCompleted/handleSkippedNode 中读取后移除）。
         * <p>用于让 sink lambda 在 node_stream 事件中携带正确的 visitUuid；
         * DAG 模式下同一节点不会并发执行（completedNodes 去重），不会有 race。</p>
         */
        final Map<String, String> nodeVisitUuids = new ConcurrentHashMap<>();

        /** 剩余未完成节点数；归零时唤醒主线程（complete done） */
        final AtomicInteger remainingNodes;
        /** abort 标志（CAS 去重，保证 workflow_error 只推一次） */
        final AtomicBoolean abortFlag = new AtomicBoolean(false);
        /** abort 错误信息（首个失败的 abort 节点写入） */
        final AtomicReference<String> abortMsgRef = new AtomicReference<>(null);
        /** 全局递增的 sortOrder（按完成顺序赋值） */
        final AtomicInteger sortOrderSeq = new AtomicInteger(0);

        /** 最终输出（output 节点完成时写入）；持锁访问 */
        final Map<String, Object> finalOutputs = new LinkedHashMap<>();
        /** 节点执行记录；持锁访问 */
        final List<WorkflowRunNode> nodeRecords = new ArrayList<>();
        /** 保护 finalOutputs 与 nodeRecords 的锁 */
        final Object recordsLock = new Object();

        /** 主线程等待句柄；remainingNodes 归零时 complete */
        final CompletableFuture<Void> done = new CompletableFuture<>();

        DataflowCtx(DagResolver.ResolvedDag dag, WorkflowContext context,
                    String runId, Long runRecordId, SseEmitter emitter) {
            this.dag = dag;
            this.context = context;
            this.runId = runId;
            this.runRecordId = runRecordId;
            this.emitter = emitter;
            this.remainingNodes = new AtomicInteger(dag.getNodeDataMap().size());
        }

        /** 标记一个节点完成（含 skipped）；remainingNodes 归零时唤醒主线程 */
        void markCompleted() {
            if (remainingNodes.decrementAndGet() == 0) {
                done.complete(null);
            }
        }
    }

    /**
     * 按 DAG 数据流驱动执行。
     *
     * @param dag         已解析的 DAG
     * @param context     工作流上下文
     * @param runId       本次运行 ID
     * @param runRecordId 关联的 WorkflowRun 主键
     * @param emitter     SSE 推送器
     * @return 执行产物
     */
    public ExecutionOutcome executeByLevel(DagResolver.ResolvedDag dag, WorkflowContext context,
                                           String runId, Long runRecordId, SseEmitter emitter) {
        DataflowCtx ctx = new DataflowCtx(dag, context, runId, runRecordId, emitter);

        // 注入流式回调：executor 调用 sink.emit(...) 时实时包装为 node_stream 事件推给前端。
        // v1.1：包装一层 clarify_request 拦截器，遇到 Askfor 时同步写入 WorkflowPause 表，
        //      用于前端断线重连后通过 GET /state 恢复 Askfor UI。
        // v2.0：sink 通过 ctx.nodeVisitUuids 查找当前 visitUuid，让 node_stream 事件按 visitUuid 路由
        //      （DAG 模式下同一节点不会并发执行，nodeVisitUuids 中按 nodeId 查找安全）。
        context.setStreamSink(ClarifyHooks.wrap(
                (nodeId, kind, content, toolName) -> {
                    String vUuid = ctx.nodeVisitUuids.get(nodeId);
                    safeSend(emitter, WorkflowRunEvent.nodeStream(runId, nodeId, vUuid, kind, content, toolName));
                },
                runId,
                pauseRepository));

        // === 初始化入度（按 source 去重，与 DagResolver 入度计算规则保持一致） ===
        for (Map.Entry<String, List<DagResolver.EdgeInfo>> e : dag.getIncomingEdges().entrySet()) {
            Set<String> sources = new HashSet<>();
            for (DagResolver.EdgeInfo edge : e.getValue()) {
                sources.add(edge.getSource());
            }
            ctx.pendingInputs.put(e.getKey(), new AtomicInteger(sources.size()));
        }
        for (String nodeId : dag.getNodeDataMap().keySet()) {
            ctx.pendingInputs.computeIfAbsent(nodeId, k -> new AtomicInteger(0));
        }

        // === 所有节点初始都视为活跃；条件分支未选 target 由 tryScheduleSuccessors 标记移除，
        // === SKIPPED 节点的下游由 handleSkippedNode 传播时移除 ===
        for (String nodeId : dag.getNodeDataMap().keySet()) {
            ctx.activeNodes.add(nodeId);
        }
        // 入度为 0 的节点（DAG 起点）提交调度
        for (Map.Entry<String, AtomicInteger> e : ctx.pendingInputs.entrySet()) {
            if (e.getValue().get() == 0) {
                scheduleNode(e.getKey(), ctx);
            }
        }

        // === 主线程等待所有节点完成（含 abort 后的清理） ===
        try {
            ctx.done.get(EXECUTION_TIMEOUT_MINUTES, TimeUnit.MINUTES);
        } catch (TimeoutException e) {
            log.error("[WorkflowEngine] 工作流执行超时（{} 分钟），强制结束: runId={}",
                    EXECUTION_TIMEOUT_MINUTES, runId);
            triggerAbort(ctx, "工作流执行超时");
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("[WorkflowEngine] 工作流执行被中断: runId={}", runId);
        } catch (ExecutionException e) {
            log.error("[WorkflowEngine] 工作流执行异常: runId={}", runId, e);
        }

        String errorMsg = ctx.abortMsgRef.get();
        return new ExecutionOutcome(ctx.finalOutputs, ctx.nodeRecords, errorMsg != null, errorMsg);
    }

    /**
     * 调度单个节点：提交到线程池执行，挂载 whenComplete 回调。
     *
     * <p>并发安全：</p>
     * <ul>
     *   <li>abort 时直接 return，不再提交新任务，未启动节点转交 handleSkippedNode 收尾</li>
     *   <li>重复调度同一节点（理论不应发生）由 inFlightFutures 的 putIfAbsent 检测</li>
     * </ul>
     */
    private void scheduleNode(String nodeId, DataflowCtx ctx) {
        if (ctx.abortFlag.get()) {
            // 已 abort：未启动节点不再调度，直接按 skipped 处理（仅递减 remaining，不写 record/SSE）
            handleSkippedNode(nodeId, ctx);
            return;
        }
        CompletableFuture<NodeExecutionResult> existing = ctx.inFlightFutures.get(nodeId);
        if (existing != null) {
            // 已在飞 / 已处理，避免重复调度
            return;
        }

        // 预生成本次执行 visitUuid，写入 ctx.nodeVisitUuids 供 sink lambda 查找，
        // 并立即推送 nodeRunning（带 visitUuid）让前端先建卡片占位
        String visitUuid = UUID.randomUUID().toString();
        ctx.nodeVisitUuids.put(nodeId, visitUuid);
        safeSend(ctx.emitter, WorkflowRunEvent.nodeRunning(ctx.runId, nodeId, visitUuid));

        CompletableFuture<NodeExecutionResult> future = CompletableFuture.supplyAsync(
                () -> executeOneNode(nodeId, ctx.dag, ctx.context, ctx.runId, ctx.emitter, ctx.activeEdges),
                selectPoolFor(ctx.dag.getNodeTypeMap().get(nodeId)));
        // putIfAbsent 防止并发场景下重复提交（如多前驱同时归零）
        CompletableFuture<NodeExecutionResult> race = ctx.inFlightFutures.putIfAbsent(nodeId, future);
        if (race != null) {
            // 极端竞态：另一线程已先提交，取消当前 future 并复用既有
            future.cancel(false);
            ctx.nodeVisitUuids.remove(nodeId);
            return;
        }
        // whenComplete 中给 result 绑定 visitUuid（无论成功/失败/异常），统一交给 onNodeCompleted
        future.whenComplete((result, ex) -> {
            NodeExecutionResult withUuid = (result != null)
                    ? result.withVisitUuid(visitUuid)
                    : null;
            onNodeCompleted(nodeId, ctx, withUuid, ex);
        });
    }

    /**
     * 节点完成回调：处理结果（成功/失败/异常）、写 records、推 SSE、激活下游。
     *
     * <p>无论 abort 与否，都会执行 markCompleted；abort 时仅清理 inFlightFutures，
     * 不写 records / 不推 SSE / 不激活下游（避免与已发的 workflow_error 乱序）。</p>
     */
    private void onNodeCompleted(String nodeId, DataflowCtx ctx,
                                 NodeExecutionResult result, Throwable ex) {
        ctx.inFlightFutures.remove(nodeId);
        // 取出本节点预生成的 visitUuid（scheduleNode 写入），用于持久化 Visit 与 visitRecorded 推送
        String visitUuid = ctx.nodeVisitUuids.remove(nodeId);
        // 去重：同一节点可能因 cancel + 正常完成进入两次回调
        if (!ctx.completedNodes.add(nodeId)) {
            return;
        }

        // 异常降级为 failed 结果（绑定 visitUuid 保留路由信息）
        if (ex != null) {
            result = NodeExecutionResult.failed(nodeId, "节点并行执行异常: " + ex.getMessage())
                    .withVisitUuid(visitUuid);
        } else if (result == null) {
            result = NodeExecutionResult.failed(nodeId, "节点执行返回 null")
                    .withVisitUuid(visitUuid);
        }

        // abort 后只清理计数，不处理副作用
        if (ctx.abortFlag.get()) {
            ctx.markCompleted();
            return;
        }

        String nodeType = ctx.dag.getNodeTypeMap().get(nodeId);
        JsonNode nodeData = ctx.dag.getNodeDataMap().get(nodeId);
        String label = nodeData.path("label").asText(nodeId);

        // D: envelope 失败自动升级
        // executor 返回 SUCCESS 但 envelope.status=400 时（如 Hermes Agent 把 LLM 429 错误描述当作成功结果返回），
        // 视为业务失败，按 FAILED 处理 failStrategy；原始 failure envelope 已由 executeOneNode 写入 context，下游可通过 envelope.status 感知
        NodeExecutionResult effectiveResult = result;
        if (result.getStatus() == NodeExecutionResult.Status.SUCCESS
                && NodeOutputEnvelope.isEnvelope(result.getOutput())) {
            NodeOutputEnvelope env = NodeOutputEnvelope.fromObject(result.getOutput());
            if (env != null && env.isFailure()) {
                String originalMsg = env.getMessage();
                String errMsg = "节点业务执行失败（envelope.status=400）: "
                        + (originalMsg == null || originalMsg.isEmpty() ? "未提供失败原因" : originalMsg);
                log.warn("[WorkflowEngine] 节点 {} 触发 envelope 失败升级（envelope.status=400）: {}", nodeId, originalMsg);
                effectiveResult = NodeExecutionResult.failed(nodeId, errMsg, result.getLogs())
                        .withVisitUuid(visitUuid);
            }
        }

        // 持锁处理 records / finalOutputs / sortOrder（dataflow 下多 worker 并发进入）
        boolean shouldAbort = false;
        String abortMsg = null;
        WorkflowRunVisit visitSnapshot;
        synchronized (ctx.recordsLock) {
            int sortOrder = ctx.sortOrderSeq.getAndIncrement();
            ctx.nodeRecords.add(buildNodeRecord(ctx.runRecordId, nodeId, nodeType, label,
                    effectiveResult.getStatus().name(), effectiveResult.getOutput(), effectiveResult.getError(),
                    effectiveResult.getSelectedBranch(), effectiveResult.getLogs(),
                    effectiveResult.getContextSnapshot(), sortOrder));
            // 同时构建 Visit 快照（DAG 模式下 iterSeq 永远=1，fromNode/fromVisit=null）
            visitSnapshot = buildDagVisit(ctx.runRecordId, visitUuid, nodeId, nodeType, label,
                    effectiveResult, sortOrder, ctx.context);

            if (effectiveResult.getStatus() == NodeExecutionResult.Status.FAILED) {
                String failStrategy = nodeData.path("failStrategy").asText("abort");
                if ("skip".equals(failStrategy)) {
                    log.info("[WorkflowEngine] 节点 {} 执行失败，策略为跳过，继续工作流", nodeId);
                } else {
                    shouldAbort = true;
                    abortMsg = "节点 " + nodeId + " 执行失败: " + effectiveResult.getError();
                }
            }

            if ("output".equals(nodeType) && effectiveResult.getOutput() != null) {
                // M-4 说明：OutputExecutor 返回的 envelope 可能是 failure（必填字段缺失等），
                // 但在上方 envelope 失败升级逻辑中已被转成 NodeExecutionResult.failed(nodeId, errMsg, logs)，
                // 此时 effectiveResult.getOutput() == null（FAILED 工厂方法不携带 output），
                // 不会进入此分支。本分支仅处理成功 envelope，直接整体作为工作流最终输出。
                ctx.finalOutputs.clear();
                ctx.finalOutputs.putAll(effectiveResult.getOutput());
            }
        }

        // 持久化 Visit（DAG 模式与 AGENTIC 模式统一）
        if (visitRepository != null && visitSnapshot != null) {
            try {
                visitRepository.save(visitSnapshot);
            } catch (Exception e) {
                log.warn("[WorkflowEngine] runId={} Visit 持久化失败：{}", ctx.runId, e.getMessage());
            }
        }

        // 推送 nodeResult（带 visitUuid，前端按 visitUuid 路由） + visitRecorded（提供卡片元数据）
        safeSend(ctx.emitter, WorkflowRunEvent.nodeResult(ctx.runId, effectiveResult));
        if (visitSnapshot != null) {
            safeSend(ctx.emitter, WorkflowRunEvent.visitRecorded(ctx.runId, visitSnapshot));
        }

        // 失败 + abort：跳过下游调度，直接 triggerAbort（未启动节点由 triggerAbort 内 handleSkippedNode 收尾）
        if (shouldAbort) {
            triggerAbort(ctx, abortMsg);
            ctx.markCompleted();
            return;
        }

        // 激活下游（含条件分支入度补偿）
        tryScheduleSuccessors(nodeId, nodeType, effectiveResult, ctx);

        ctx.markCompleted();
    }

    /**
     * 激活后继节点：递减 target 入度，归零则调度。
     *
     * <p>关键规则（与 DagResolver 入度去重保持一致）：</p>
     * <ul>
     *   <li>每个 source 完成时，对其所有 target 各递减 1 次（即使同 source 多条边到同 target）</li>
     *   <li>条件分支节点：未选中分支的 target 仍要递减入度，但 activeEdges 不加</li>
     *   <li>顺序：先 add activeEdges 再 decrement 入度，保证新 worker 启动时能看到活跃边（happens-before）</li>
     * </ul>
     */
    private void tryScheduleSuccessors(String nodeId, String nodeType, NodeExecutionResult result,
                                       DataflowCtx ctx) {
        List<DagResolver.EdgeInfo> outEdges = ctx.dag.getOutgoingEdges().get(nodeId);
        if (outEdges == null || outEdges.isEmpty()) return;

        boolean isConditionBranch = "condition".equals(nodeType) && result.getSelectedBranch() != null;
        // 同一 target 在本节点完成时只递减一次（无论几条边、几个分支）
        Set<String> decreased = new HashSet<>();

        // 条件分支：先把未选 target 从 activeNodes 移除，使其入度归零时走 SKIPPED 路径而非调度
        if (isConditionBranch) {
            for (DagResolver.EdgeInfo edge : outEdges) {
                if (!result.getSelectedBranch().equals(edge.getSourceHandle())) {
                    ctx.activeNodes.remove(edge.getTarget());
                }
            }
        }

        for (DagResolver.EdgeInfo edge : outEdges) {
            boolean edgeActive = !isConditionBranch
                    || result.getSelectedBranch().equals(edge.getSourceHandle());
            if (edgeActive) {
                // 先加活跃边，确保下游 worker 构建工作区时能看到（happens-before）
                ctx.activeEdges.add(WorkflowScopeResolver.activeEdgeKey(
                        edge.getSource(), edge.getSourceHandle(), edge.getTarget()));
            }
            if (decreased.add(edge.getTarget())) {
                AtomicInteger pending = ctx.pendingInputs.get(edge.getTarget());
                if (pending != null && pending.decrementAndGet() == 0) {
                    // 入度归零：决定调度或 SKIPPED
                    if (ctx.activeNodes.contains(edge.getTarget())) {
                        scheduleNode(edge.getTarget(), ctx);
                    } else {
                        handleSkippedNode(edge.getTarget(), ctx);
                    }
                }
            }
        }
    }

    /**
     * 处理 SKIPPED 节点：推送 skipped SSE + 写 record + 继续传播下游入度。
     *
     * <p>触发场景：</p>
     * <ol>
     *   <li>条件分支未选中路径上的节点（前驱完成时发现 target 不在 activeNodes 中）</li>
     *   <li>abort 后未启动的节点（scheduleNode 入口检测 abortFlag 后直接走此分支）</li>
     * </ol>
     */
    private void handleSkippedNode(String nodeId, DataflowCtx ctx) {
        if (!ctx.completedNodes.add(nodeId)) {
            return;  // 已处理（如 abort 时已遍历过）
        }
        // 标记本节点为非活跃，确保其下游入度归零时也走 SKIPPED 而不是误调度
        ctx.activeNodes.remove(nodeId);
        String nodeType = ctx.dag.getNodeTypeMap().get(nodeId);
        JsonNode nodeData = ctx.dag.getNodeDataMap().get(nodeId);
        String label = nodeData.path("label").asText(nodeId);
        // SKIPPED 也分配一个 visitUuid，让前端能给 SKIPPED 节点建卡片
        String visitUuid = UUID.randomUUID().toString();
        NodeExecutionResult skipped = NodeExecutionResult.skipped(nodeId).withVisitUuid(visitUuid);
        // 始终写 record（abort 时也写，让前端能看到所有节点的最终状态）；
        // 但 abort 时不推 SSE（避免与已发的 workflow_error 乱序）
        WorkflowRunVisit visitSnapshot;
        synchronized (ctx.recordsLock) {
            int sortOrder = ctx.sortOrderSeq.getAndIncrement();
            ctx.nodeRecords.add(buildNodeRecord(ctx.runRecordId, nodeId, nodeType, label,
                    "SKIPPED", null, null, null, null, null, sortOrder));
            visitSnapshot = buildDagVisit(ctx.runRecordId, visitUuid, nodeId, nodeType, label,
                    skipped, sortOrder, ctx.context);
        }
        // 持久化 SKIPPED Visit
        if (visitRepository != null && visitSnapshot != null) {
            try {
                visitRepository.save(visitSnapshot);
            } catch (Exception e) {
                log.warn("[WorkflowEngine] runId={} SKIPPED Visit 持久化失败：{}", ctx.runId, e.getMessage());
            }
        }
        if (!ctx.abortFlag.get()) {
            safeSend(ctx.emitter, WorkflowRunEvent.nodeResult(ctx.runId, skipped));
            if (visitSnapshot != null) {
                safeSend(ctx.emitter, WorkflowRunEvent.visitRecorded(ctx.runId, visitSnapshot));
            }
        }

        // 传播下游：本节点 SKIPPED 等价于"前驱已处理"，对下游入度递减；
        // 同时把下游 target 也标记为非活跃（即使其他前驱正常完成，target 也应 SKIPPED）
        List<DagResolver.EdgeInfo> outEdges = ctx.dag.getOutgoingEdges().get(nodeId);
        if (outEdges != null) {
            Set<String> decreased = new HashSet<>();
            for (DagResolver.EdgeInfo edge : outEdges) {
                ctx.activeNodes.remove(edge.getTarget());
                if (decreased.add(edge.getTarget())) {
                    AtomicInteger pending = ctx.pendingInputs.get(edge.getTarget());
                    if (pending != null && pending.decrementAndGet() == 0) {
                        handleSkippedNode(edge.getTarget(), ctx);
                    }
                }
            }
        }
        ctx.markCompleted();
    }

    /**
     * 触发 abort：CAS 去重 + 推送 workflow_error（仅一次）+ 取消在飞 future + 收尾未启动节点。
     */
    private void triggerAbort(DataflowCtx ctx, String msg) {
        if (!ctx.abortFlag.compareAndSet(false, true)) {
            return;  // 已 abort，仅首个失败节点推 workflow_error
        }
        ctx.abortMsgRef.set(msg);
        safeSend(ctx.emitter, WorkflowRunEvent.workflowError(ctx.runId, msg));

        // 取消所有在飞 future（cancel 只能中断响应中断的任务，对已返回的不影响）
        for (Map.Entry<String, CompletableFuture<NodeExecutionResult>> e : ctx.inFlightFutures.entrySet()) {
            e.getValue().cancel(true);
        }

        // 未启动的节点（pendingInputs 未归零、不在 inFlightFutures、不在 completedNodes）按 SKIPPED 收尾
        // 保证 remainingNodes 最终归零，主线程能退出
        for (String nodeId : ctx.pendingInputs.keySet()) {
            if (!ctx.completedNodes.contains(nodeId) && !ctx.inFlightFutures.containsKey(nodeId)) {
                handleSkippedNode(nodeId, ctx);
            }
        }
    }

    /**
     * 执行单个节点（线程池 worker 中调用）。
     * 把节点类型查找、前置条件校验、executor.execute、上下文写入等放在同一个并行任务里。
     * 异常一律封装为 failed 结果返回，不向上抛（避免中断 CompletableFuture 链）。
     */
    private NodeExecutionResult executeOneNode(String nodeId, DagResolver.ResolvedDag dag,
                                               WorkflowContext context, String runId,
                                               SseEmitter emitter, Set<String> activeEdges) {
        String nodeType = dag.getNodeTypeMap().get(nodeId);
        JsonNode nodeData = dag.getNodeDataMap().get(nodeId);

        NodeExecutor executor = executorMap.get(nodeType);
        if (executor == null) {
            String errMsg = "未知节点类型: " + nodeType;
            log.warn("[WorkflowEngine] 节点 {} {}", nodeId, errMsg);
            return NodeExecutionResult.failed(nodeId, errMsg);
        }

        // 注：nodeRunning 已由 scheduleNode 推送（携带预生成的 visitUuid），此处不再重复

        try {
            // 构建节点工作区（基于当前活跃边过滤可达前驱）
            NodeWorkspace workspace = workspaceBuilder.build(nodeId, dag, context, activeEdges);

            // 解析节点输入（显式 mapping + 隐式 fallback）
            NodeInputResolver.ResolvedInputs resolvedInputs = NodeInputResolver.resolveInputs(nodeData, workspace);
            if (resolvedInputs.hasMissing()) {
                String errMsg = "前置条件不满足: 缺少输入 " + String.join(", ", resolvedInputs.getMissing());
                log.warn("[WorkflowEngine] 节点 {} {}", nodeId, errMsg);
                return NodeExecutionResult.failed(nodeId, errMsg);
            }
            if (!resolvedInputs.getWarnings().isEmpty()) {
                resolvedInputs.getWarnings().forEach(w ->
                        log.warn("[WorkflowEngine] 节点 {} 输入解析警告: {}", nodeId, w));
            }
            // 将解析后的输入注入工作区，供执行器通过名称直接读取
            workspace.getVariables().putAll(resolvedInputs.getValues());

            // 显式打印节点输入（仅用户声明的 inputs 字段值，不含整个工作区）
            log.info("[节点输入] {}: {}", nodeTag(nodeId, nodeData), toJsonSafe(resolvedInputs.getValues()));

            // 前置条件校验：文件/目录存在性与类型兼容性
            String precheckError = checkPreconditions(nodeData, workspace, context.getWorkingDir());
            if (precheckError != null) {
                log.warn("[WorkflowEngine] 节点 {} 前置条件不满足: {}", nodeId, precheckError);
                return NodeExecutionResult.failed(nodeId, precheckError);
            }

            NodeExecutionResult result = executor.execute(nodeId, nodeData, context, workspace);
            Map<String, Object> output = result.getOutput();
            if (output != null) {
                context.setNodeOutput(nodeId, output);
            } else if (result.getStatus() == NodeExecutionResult.Status.FAILED) {
                // 失败兜底：executor 返回 failed（无 output）时，构造失败 envelope 写入上下文，
                // 让下游节点能通过 envelope.status 感知失败（保留 failStrategy=abort/skip 行为）
                Map<String, Object> failEnv = NodeOutputEnvelope.failure(result.getError()).toMap();
                context.setNodeOutput(nodeId, failEnv);
            }
            // 显式打印节点输出（仅 envelope.data，即节点自身产出字段，不含整个工作区）
            log.info("[节点输出] {}: {}", nodeTag(nodeId, nodeData), toJsonSafe(unwrapProducedOutput(result)));
            // 节点完成后采集完整输出快照（初始输入 + 各节点 {nodeId}__{varName} 命名空间形式，用于 debug 视图与运行历史回放）
            return result.withContextSnapshot(context.snapshotAllOutputs());
        } catch (Exception e) {
            log.error("[WorkflowEngine] 节点 {} 执行抛出异常: {}", nodeId, e.getMessage(), e);
            return NodeExecutionResult.failed(nodeId, "节点执行异常: " + e.getMessage());
        }
    }

    private WorkflowRunNode buildNodeRecord(Long runId, String nodeId, String nodeType, String label,
                                            String status, Map<String, Object> output, String error,
                                            String selectedBranch, List<ExecutionLog> logs,
                                            Map<String, Object> contextSnapshot, int sortOrder) {
        WorkflowRunNode node = new WorkflowRunNode();
        node.setRunId(runId);
        node.setNodeId(nodeId);
        node.setNodeType(nodeType);
        node.setLabel(label);
        node.setStatus(status);
        node.setSortOrder(sortOrder);
        if (error != null) node.setError(error);
        if (selectedBranch != null) node.setSelectedBranch(selectedBranch);
        try {
            if (output != null && !output.isEmpty()) node.setOutput(MAPPER.writeValueAsString(output));
            if (logs != null && !logs.isEmpty()) node.setLogs(MAPPER.writeValueAsString(logs));
            if (contextSnapshot != null && !contextSnapshot.isEmpty()) {
                node.setContextSnapshot(MAPPER.writeValueAsString(contextSnapshot));
            }
        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
            log.warn("[WorkflowEngine] 序列化节点结果失败: nodeId={}, err={}", nodeId, e.getMessage());
        }
        return node;
    }

    /**
     * 构建 DAG 模式的 Visit 快照。
     * <p>与 AGENTIC 模式的 buildVisit 区别：
     * <ul>
     *   <li>iterSeq 永远 = 1（DAG 模式同节点只执行一次）</li>
     *   <li>fromVisitUuid / fromNodeId = null（DAG 多前驱并行，无法指认单一上游）</li>
     *   <li>decisionJson = null（无 Planner 决策）</li>
     * </ul>
     * </p>
     */
    private WorkflowRunVisit buildDagVisit(Long runRecordId, String visitUuid, String nodeId, String nodeType,
                                            String label, NodeExecutionResult result, int sortOrder,
                                            WorkflowContext context) {
        if (visitUuid == null) {
            // 兜底：极少数场景下（cancel/异常）scheduleNode 未写入 visitUuid，跳过 Visit 持久化
            return null;
        }
        WorkflowRunVisit v = new WorkflowRunVisit();
        v.setRunId(runRecordId);
        v.setVisitUuid(visitUuid);
        v.setNodeId(nodeId);
        v.setNodeType(nodeType);
        v.setNodeLabel(label);
        v.setIterSeq(1);
        v.setFromVisitUuid(null);
        v.setFromNodeId(null);
        try {
            // Visit 工作区 = 当前 context 全局快照（含初始输入 + 各节点命名空间输出）
            if (context != null) {
                v.setVariablesJson(MAPPER.writeValueAsString(context.snapshotAllOutputs()));
            }
        } catch (Exception e) {
            v.setVariablesJson("{}");
        }
        try {
            if (result != null && result.getOutput() != null) {
                v.setOutputJson(MAPPER.writeValueAsString(result.getOutput()));
            }
        } catch (Exception e) {
            v.setOutputJson(null);
        }
        try {
            // logsJson：从 result.getLogs() 序列化，让运行记录页能从 Visit 还原
            // streamingThinking / toolHistory / todos，无需联表 WorkflowRunNode
            if (result != null && result.getLogs() != null && !result.getLogs().isEmpty()) {
                v.setLogsJson(MAPPER.writeValueAsString(result.getLogs()));
            }
        } catch (Exception e) {
            v.setLogsJson(null);
        }
        v.setDecisionJson(null);
        v.setStatus(result != null ? result.getStatus().name() : "FAILED");
        v.setErrorMsg(result != null ? result.getError() : null);
        v.setSortOrder(sortOrder);
        return v;
    }

    /**
     * 前置条件校验：检查节点 data.inputs 中 required=true 的字段是否在工作区中存在
     * - filePath/directoryPath 类型额外校验工作目录中文件/目录是否存在
     * - 类型校验：声明的 type 与工作区实际值的类型必须兼容（参考前端 ioInference.isTypeCompatible）
     * 无 inputs 定义时跳过校验（向后兼容）
     */
    private String checkPreconditions(JsonNode nodeData, NodeWorkspace workspace, Path workingDir) {
        JsonNode inputsNode = nodeData.path("inputs");
        if (!inputsNode.isArray() || inputsNode.isEmpty()) return null;

        List<String> missing = new ArrayList<>();

        for (JsonNode field : inputsNode) {
            boolean required = field.path("required").asBoolean(false);
            if (!required) continue;

            String name = field.path("name").asText("");
            String type = field.path("type").asText("string");
            if (name.isEmpty()) continue;

            Object value = workspace.getVariable(name);
            if (value == null) {
                missing.add(name + " (变量未提供)");
                continue;
            }

            if ("filePath".equals(type) && workingDir != null) {
                if (!java.nio.file.Files.isRegularFile(workingDir.resolve(value.toString()))) {
                    missing.add(name + " (文件不存在: " + value + ")");
                    continue;
                }
            } else if ("directoryPath".equals(type) && workingDir != null) {
                if (!java.nio.file.Files.isDirectory(workingDir.resolve(value.toString()))) {
                    missing.add(name + " (目录不存在: " + value + ")");
                    continue;
                }
            }

            // 类型兼容性校验
            String actualType = NodeTypeUtils.inferTypeName(value);
            if (actualType != null && !NodeTypeUtils.isTypeCompatible(actualType, type)) {
                missing.add(name + " (类型不匹配: 期望 " + type + ", 实际 " + actualType + ")");
            }
        }

        return missing.isEmpty() ? null : "前置条件不满足: " + String.join("; ", missing);
    }

    /**
     * 构造节点标识：nodeName 优先，其次 label，最后 nodeId。
     * 格式：`nodeName[label](nodeId)` / `label(nodeId)` / `nodeId`
     */
    private static String nodeTag(String nodeId, JsonNode nodeData) {
        String nodeName = nodeData.path("nodeName").asText("");
        String label = nodeData.path("label").asText("");
        if (!nodeName.isEmpty() && !label.isEmpty()) {
            return nodeName + "[" + label + "](" + nodeId + ")";
        }
        if (!nodeName.isEmpty()) {
            return nodeName + "(" + nodeId + ")";
        }
        if (!label.isEmpty()) {
            return label + "(" + nodeId + ")";
        }
        return nodeId;
    }

    /**
     * 解包节点产出：成功时返回 envelope.data（即节点自身输出字段），
     * 失败兜底（executor 返回 FAILED 且无 output）时返回 {error: ...} 单字段，
     * 供日志输出使用。不含整个工作区快照。
     */
    private static Map<String, Object> unwrapProducedOutput(NodeExecutionResult result) {
        Map<String, Object> output = result.getOutput();
        if (output != null) {
            NodeOutputEnvelope env = NodeOutputEnvelope.fromObject(output);
            if (env != null) {
                return env.getData() == null ? Map.of() : env.getData();
            }
            return output;
        }
        if (result.getStatus() == NodeExecutionResult.Status.FAILED) {
            return Map.of("error", result.getError() == null ? "" : result.getError());
        }
        return Map.of();
    }

    /**
     * 安全 JSON 序列化：失败时降级为 toString，避免日志记录自身抛异常中断工作流。
     */
    private static String toJsonSafe(Object obj) {
        if (obj == null) return "null";
        try {
            return MAPPER.writeValueAsString(obj);
        } catch (Exception e) {
            return String.valueOf(obj);
        }
    }

    /**
     * 安全发送 SSE 事件
     * <p>加 synchronized：dataflow 下多 worker 并发调用 safeSend
     * （nodeRunning / nodeStream / nodeResult），而 SseEmitter.send 非线程安全。</p>
     */
    private synchronized void safeSend(SseEmitter emitter, WorkflowRunEvent event) {
        // 同时发布到事件总线（外部 API 通过 /stream 订阅消费）
        eventBus.publish(event.getRunId(), event.getType(), event);
        try {
            emitter.send(SseEmitter.event()
                    .name(event.getType())
                    .data(event));
        } catch (IOException e) {
            log.warn("[WorkflowEngine] SSE 推送失败: {}", e.getMessage());
        }
    }

}
