/**
 * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
 * <p>
 * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.gnu.org/licenses/lgpl-3.0.txt
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.easyagents.flow.core.chain.runtime;

import com.easyagents.flow.core.chain.*;
import com.easyagents.flow.core.chain.event.ChainStatusChangeEvent;
import com.easyagents.flow.core.chain.listener.ChainErrorListener;
import com.easyagents.flow.core.chain.listener.ChainEventListener;
import com.easyagents.flow.core.chain.listener.ChainOutputListener;
import com.easyagents.flow.core.chain.listener.NodeErrorListener;
import com.easyagents.flow.core.chain.repository.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;

/**
 * TinyFlow 最新 ChainExecutor
 * <p>
 * 说明:
 * * 负责触发 Chain 执行 / 恢复
 * * 不持有长时间运行的 Chain 实例
 * * 支持 async-only 架构
 */
public class ChainExecutor {

    private static final Logger log = LoggerFactory.getLogger(ChainExecutor.class);
    private static final String NESTED_DEPTH_MEMORY_KEY =
            "__tinyflow.nesting.depth";
    private static final String DEFINITION_CALL_PATH_MEMORY_KEY =
            "__tinyflow.nesting.definitionPath";
    /**
     * 子工作流触发器使用的独立执行通道。
     */
    public static final String CHILD_WORKFLOW_EXECUTION_LANE_PREFIX =
            "child-workflow:";
    private volatile Semaphore rootChildExecutionPermits =
            new Semaphore(32, true);
    private volatile long persistentOutcomePollMillis = 500L;
    private volatile int maxChildExecutionLaneDepth =
            Integer.MAX_VALUE;
    private static final String CHILD_EXECUTION_REFERENCE_KEY =
            "__tinyflow.workflowNode.childExecution";
    /**
     * 进程内定义快照仅作为热点缓存；持久快照负责跨节点和跨进程恢复。
     */
    private static final int MAX_ACTIVE_DEFINITIONS = 1024;
    private final ChainDefinitionRepository definitionRepository;
    private final ChainStateRepository chainStateRepository;
    private final NodeStateRepository nodeStateRepository;
    private final LoopResultRepository loopResultRepository;
    private final ChainDefinitionSnapshotRepository definitionSnapshotRepository;
    private final TriggerScheduler triggerScheduler;
    private final ExecutionBudget executionBudget;
    private final EventManager eventManager = new EventManager();
    /** 等待同步执行结果的任务，按工作流实例 ID 进行常量时间路由。 */
    private final ConcurrentMap<String, CompletableFuture<Map<String, Object>>> pendingExecutions =
            new ConcurrentHashMap<>();
    /**
     * 活跃工作流实例使用的定义快照，避免每个节点触发都重新加载和解析定义。
     */
    private final Map<String, ChainDefinition> activeDefinitions =
            Collections.synchronizedMap(new LinkedHashMap<>(
                    MAX_ACTIVE_DEFINITIONS + 1, 0.75F, true) {
                @Override
                protected boolean removeEldestEntry(
                        Map.Entry<String, ChainDefinition> eldest) {
                    return size() > MAX_ACTIVE_DEFINITIONS;
                }
            });

    public ChainExecutor(ChainDefinitionRepository definitionRepository
            , ChainStateRepository chainStateRepository
            , NodeStateRepository nodeStateRepository
    ) {
        this(definitionRepository,
                chainStateRepository,
                nodeStateRepository,
                new InMemoryLoopResultRepository(),
                new InMemoryChainDefinitionSnapshotRepository(),
                ChainRuntime.triggerScheduler(),
                ExecutionBudget.defaults());
    }


    public ChainExecutor(ChainDefinitionRepository definitionRepository
            , ChainStateRepository chainStateRepository
            , NodeStateRepository nodeStateRepository
            , TriggerScheduler triggerScheduler) {
        this(definitionRepository,
                chainStateRepository,
                nodeStateRepository,
                new InMemoryLoopResultRepository(),
                new InMemoryChainDefinitionSnapshotRepository(),
                triggerScheduler,
                ExecutionBudget.defaults());
    }

    /**
     * 创建使用指定调度器和资源预算的执行器。
     *
     * @param definitionRepository 工作流定义仓储
     * @param chainStateRepository 工作流状态仓储
     * @param nodeStateRepository 节点状态仓储
     * @param triggerScheduler 触发调度器
     * @param executionBudget 平台资源保护预算
     */
    public ChainExecutor(ChainDefinitionRepository definitionRepository
            , ChainStateRepository chainStateRepository
            , NodeStateRepository nodeStateRepository
            , TriggerScheduler triggerScheduler
            , ExecutionBudget executionBudget) {
        this(definitionRepository,
                chainStateRepository,
                nodeStateRepository,
                new InMemoryLoopResultRepository(),
                new InMemoryChainDefinitionSnapshotRepository(),
                triggerScheduler,
                executionBudget);
    }

    /**
     * 创建使用指定调度器、循环结果仓储和资源预算的执行器。
     *
     * @param definitionRepository 工作流定义仓储
     * @param chainStateRepository 工作流状态仓储
     * @param nodeStateRepository 节点状态仓储
     * @param loopResultRepository 循环累计结果仓储
     * @param triggerScheduler 触发调度器
     * @param executionBudget 平台资源保护预算
     */
    public ChainExecutor(ChainDefinitionRepository definitionRepository
            , ChainStateRepository chainStateRepository
            , NodeStateRepository nodeStateRepository
            , LoopResultRepository loopResultRepository
            , TriggerScheduler triggerScheduler
            , ExecutionBudget executionBudget) {
        this(definitionRepository,
                chainStateRepository,
                nodeStateRepository,
                loopResultRepository,
                new InMemoryChainDefinitionSnapshotRepository(),
                triggerScheduler,
                executionBudget);
    }

    /**
     * 创建使用持久定义快照的执行器。
     *
     * @param definitionRepository 工作流定义仓储
     * @param chainStateRepository 工作流状态仓储
     * @param nodeStateRepository 节点状态仓储
     * @param loopResultRepository 循环累计结果仓储
     * @param definitionSnapshotRepository 实例级定义快照仓储
     * @param triggerScheduler 触发调度器
     * @param executionBudget 平台资源保护预算
     */
    public ChainExecutor(ChainDefinitionRepository definitionRepository
            , ChainStateRepository chainStateRepository
            , NodeStateRepository nodeStateRepository
            , LoopResultRepository loopResultRepository
            , ChainDefinitionSnapshotRepository definitionSnapshotRepository
            , TriggerScheduler triggerScheduler
            , ExecutionBudget executionBudget) {
        this.definitionRepository = definitionRepository;
        this.chainStateRepository = chainStateRepository;
        this.nodeStateRepository = nodeStateRepository;
        this.loopResultRepository = loopResultRepository == null
                ? new InMemoryLoopResultRepository()
                : loopResultRepository;
        this.definitionSnapshotRepository = definitionSnapshotRepository == null
                ? new InMemoryChainDefinitionSnapshotRepository()
                : definitionSnapshotRepository;
        this.triggerScheduler = triggerScheduler;
        this.executionBudget = executionBudget == null ? ExecutionBudget.defaults() : executionBudget;
        registerRuntimeCallbacks();
    }


    public Map<String, Object> execute(String definitionId, Map<String, Object> variables) {
        return executeInternal(definitionId, variables, Long.MAX_VALUE, TimeUnit.SECONDS, false);
    }


    public Map<String, Object> execute(String definitionId, Map<String, Object> variables, long timeout, TimeUnit unit) {
        return executeInternal(definitionId, variables, timeout, unit, false);
    }

    /**
     * 同步执行不允许进入人工挂起状态的工作流。
     *
     * <p>该入口适用于 Tool 等无法把工作流恢复协议接回原调用方的同步场景。
     * 工作流一旦进入 {@link ChainStatus#SUSPEND}，实例会被取消并立即返回失败。</p>
     *
     * @param definitionId 工作流定义 ID
     * @param variables 输入变量
     * @return 工作流输出
     * @throws RuntimeException 工作流失败、挂起或执行线程被中断时抛出
     */
    public Map<String, Object> executeWithoutSuspension(
            String definitionId, Map<String, Object> variables) {
        return executeInternal(
                definitionId, variables, Long.MAX_VALUE, TimeUnit.SECONDS, true);
    }

    private Map<String, Object> executeInternal(
            String definitionId,
            Map<String, Object> variables,
            long timeout,
            TimeUnit unit,
            boolean rejectSuspension) {
        Chain chain = createChain(definitionId);
        String stateInstanceId = chain.getStateInstanceId();

        try {
            chain.start(variables);
            Map<String, Object> result = awaitPersistentOutcome(
                    stateInstanceId, timeout, unit, null, rejectSuspension);
            clearDefaultStates(result);
            return result;
        } catch (TimeoutException e) {
            cancel(stateInstanceId, "Execution timed out");
            throw new RuntimeException("Execution timed out", e);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            cancel(stateInstanceId, "Execution interrupted");
            throw new RuntimeException("Execution interrupted", e);
        } catch (Throwable e) {
            throw new RuntimeException("Execution failed", e);
        } finally {
            activeDefinitions.remove(stateInstanceId);
        }
    }

    /**
     * 取消仍在运行的工作流实例。
     *
     * <p>取消状态写入后，未开始的触发器会在执行入口短路；已经完成的外部 I/O 也会在提交
     * 结果前重新检查状态，避免继续推进下游。</p>
     *
     * @param stateInstanceId 工作流实例 ID
     * @param message 取消原因
     * @return 本次是否完成了非终态到取消状态的转换
     */
    public boolean cancel(String stateInstanceId, String message) {
        ChainState state = chainStateRepository.load(stateInstanceId);
        if (state == null) {
            return false;
        }
        ChainDefinition definition = getDefinitionForInstance(state);
        Chain chain = configureChain(definition, stateInstanceId);
        return chain.cancel(message);
    }

    /**
     * 注册工作流调度和同步结果路由回调。
     */
    private void registerRuntimeCallbacks() {
        eventManager.addEventListener(ChainStatusChangeEvent.class, this::completePendingExecution);
        eventManager.addChainErrorListener(this::failPendingExecution);
        triggerScheduler.registerConsumer(this::accept);
        triggerScheduler.registerFailureListener(
                this::failDeadLetteredTrigger);
    }

    /**
     * 配置同步子工作流的容量和持久状态轮询参数。
     *
     * <p>该方法应在执行器对外提供服务前调用。</p>
     *
     * @param rootPermits 根级同步子流程最大并发
     * @param pollMillis 持久终态轮询间隔毫秒数
     * @param laneMaxDepth 已注册独立执行通道覆盖的最大深度
     */
    public void configureChildWorkflowRuntime(
            int rootPermits,
            long pollMillis,
            int laneMaxDepth) {
        if (rootPermits <= 0 || pollMillis <= 0L
                || laneMaxDepth <= 0) {
            throw new IllegalArgumentException(
                    "Child workflow runtime values must be positive");
        }
        this.rootChildExecutionPermits =
                new Semaphore(rootPermits, true);
        this.persistentOutcomePollMillis = pollMillis;
        this.maxChildExecutionLaneDepth = laneMaxDepth;
    }

    /**
     * 将已成功写入死信的触发器对应实例收敛为失败终态。
     *
     * <p>状态更新使用实例锁 fencing token，避免旧节点在锁失效后覆盖新 owner 的
     * 业务终态；终态实例保持原状态。</p>
     *
     * @param trigger 已死信触发器
     * @param failure 最后一次失败
     */
    private boolean failDeadLetteredTrigger(
            Trigger trigger, Throwable failure) {
        if (trigger == null
                || trigger.getStateInstanceId() == null) {
            return true;
        }
        String instanceId = trigger.getStateInstanceId();
        Throwable cause = failure == null
                ? new ChainException(
                "Workflow trigger delivery attempts exhausted: "
                        + trigger.getId())
                : failure;
        try {
            ChainState state =
                    chainStateRepository.load(instanceId);
            if (state == null
                    || (state.getStatus() != null
                    && state.getStatus().isTerminal())) {
                return true;
            }
            ChainDefinition definition =
                    getDefinitionForInstance(state);
            if (definition == null) {
                definition = new ChainDefinition();
                definition.setId(state.getChainDefinitionId());
            }
            Chain chain = configureChain(
                    definition, instanceId);
            return chain.failTerminal(cause);
        } catch (Throwable terminalError) {
            log.error(
                    "Failed to mark dead-lettered workflow terminal, "
                            + "instanceId={}, triggerId={}",
                    instanceId, trigger.getId(), terminalError);
            return false;
        }
    }

    /**
     * 在工作流进入终态时完成对应的同步等待任务。
     *
     * @param event 工作流状态事件
     * @param chain 产生事件的工作流实例
     */
    private void completePendingExecution(Event event, Chain chain) {
        if (!(event instanceof ChainStatusChangeEvent statusChangeEvent)
                || !statusChangeEvent.getStatus().isTerminal()) {
            return;
        }

        String stateInstanceId = chain.getStateInstanceId();
        CompletableFuture<Map<String, Object>> future = pendingExecutions.get(stateInstanceId);
        try {
            if (future != null) {
                ChainState state = chainStateRepository.load(stateInstanceId);
                if (state == null) {
                    throw new ChainException("Chain state not found: " + stateInstanceId);
                }
                @SuppressWarnings("unchecked")
                Map<String, Object> execResult = (Map<String, Object>)
                        loopResultRepository.resolveReferences(state.getExecuteResult());
                future.complete(execResult != null ? execResult : Collections.emptyMap());
            }
        } catch (Exception error) {
            if (future != null) {
                future.completeExceptionally(error);
            }
            log.error(
                    "Failed to complete workflow execution continuation, instanceId={}",
                    stateInstanceId,
                    error);
        } finally {
            activeDefinitions.remove(stateInstanceId);
            definitionSnapshotRepository.remove(stateInstanceId);
        }
    }

    /**
     * 将工作流执行异常传递给对应的同步等待任务。
     *
     * @param error 工作流执行异常
     * @param chain 发生异常的工作流实例
     */
    private void failPendingExecution(Throwable error, Chain chain) {
        String stateInstanceId = chain.getStateInstanceId();
        CompletableFuture<Map<String, Object>> future = pendingExecutions.get(stateInstanceId);
        if (future != null) {
            future.completeExceptionally(error);
        }
        activeDefinitions.remove(stateInstanceId);
        ChainState state = chainStateRepository.load(stateInstanceId);
        if (state != null && state.getStatus().isTerminal()) {
            definitionSnapshotRepository.remove(stateInstanceId);
        }
    }

    /**
     * 清理默认状态
     *
     * @param result 执行结果
     */
    public void clearDefaultStates(Map<String, Object> result) {
        if (result == null || result.isEmpty()) {
            return;
        }
        result.remove(ChainConsts.SCHEDULE_NEXT_NODE_DISABLED_KEY);
        result.remove(ChainConsts.NODE_STATE_STATUS_KEY);
        result.remove(ChainConsts.CHAIN_STATE_STATUS_KEY);
        result.remove(ChainConsts.CHAIN_STATE_MESSAGE_KEY);
    }

    public String executeAsync(String definitionId, Map<String, Object> variables) {
        return executeAsync(definitionId, variables, null);
    }

    /**
     * 异步启动工作流，并在首个节点开始前暴露执行实例 ID。
     *
     * <p>回调用于提前注册流式事件接收器，保证高速工作流不会在调用方拿到执行 ID
     * 之前丢失开始事件或首批输出。</p>
     *
     * @param definitionId 工作流定义 ID
     * @param variables 工作流输入变量
     * @param beforeStart 启动前回调；可为 {@code null}
     * @return 执行实例 ID
     */
    public String executeAsync(
            String definitionId,
            Map<String, Object> variables,
            Consumer<String> beforeStart) {
        Chain chain = createChain(definitionId);
        try {
            if (beforeStart != null) {
                beforeStart.accept(chain.getStateInstanceId());
            }
            chain.start(variables);
            return chain.getStateInstanceId();
        } catch (RuntimeException | Error error) {
            activeDefinitions.remove(chain.getStateInstanceId());
            definitionSnapshotRepository.remove(chain.getStateInstanceId());
            throw error;
        }
    }

    /**
     * 在独立触发执行通道中同步执行子工作流。
     *
     * <p>调用者继续获得与历史实现一致的同步结果；子流程自身的节点触发器在独立
     * worker lane 执行，因此父节点等待不会占满子流程所需的普通节点工作线程。
     * 根级子流程并发受宽松许可保护，防止异常调用一次创建过多等待线程。</p>
     *
     * @param definitionId 子流程定义 ID
     * @param variables 子流程输入
     * @param parentChain 父流程
     * @param parentNodeId 父工作流节点 ID
     * @return 子流程输出
     */
    public Map<String, Object> executeChild(
            String definitionId,
            Map<String, Object> variables,
            Chain parentChain,
            String parentNodeId) {
        Objects.requireNonNull(parentChain, "parentChain required");
        if (parentNodeId == null || parentNodeId.isBlank()) {
            throw new IllegalArgumentException("parentNodeId required");
        }
        ChainState parentState = chainStateRepository.load(
                parentChain.getStateInstanceId());
        int parentDepth = readNestedDepth(parentState);
        int childDepth = parentDepth + 1;
        executionBudget.checkNestedDepth(parentNodeId, childDepth);
        if (childDepth > maxChildExecutionLaneDepth) {
            throw new ExecutionBudgetExceededException(
                    "Child workflow depth "
                            + childDepth
                            + " exceeds registered execution lanes "
                            + maxChildExecutionLaneDepth);
        }
        List<String> callPath = readDefinitionCallPath(
                parentState, parentChain.getDefinition().getId());
        String canonicalChildId = canonicalDefinitionId(definitionId);
        if (callPath.contains(canonicalChildId)) {
            throw new ExecutionBudgetExceededException(
                    "Recursive workflow call detected at "
                            + parentNodeId
                            + ": "
                            + String.join(" -> ", callPath)
                            + " -> "
                            + canonicalChildId);
        }
        List<String> childCallPath = new ArrayList<>(callPath);
        childCallPath.add(canonicalChildId);
        Trigger currentTrigger = TriggerContext.getCurrentTrigger();
        String invocationId = currentTrigger == null
                ? "direct:" + UUID.randomUUID()
                : (currentTrigger.getLogicalExecutionId() == null
                || currentTrigger.getLogicalExecutionId().isBlank()
                ? currentTrigger.getId()
                : currentTrigger.getLogicalExecutionId());
        boolean rootPermit = false;
        try {
            if (parentDepth == 0) {
                if (!rootChildExecutionPermits.tryAcquire(
                        1L, TimeUnit.SECONDS)) {
                    throw new RetryableTriggerException(
                            "子工作流并发繁忙，请稍后重试", null);
                }
                rootPermit = true;
            }
            ChildExecutionReference reference =
                    prepareChildExecution(
                            definitionId,
                            parentChain,
                            parentNodeId,
                            invocationId,
                            childDepth,
                            childCallPath);
            startChildIfReady(reference, variables);
            Map<String, Object> result = awaitPersistentOutcome(
                    reference.childInstanceId(),
                    Long.MAX_VALUE,
                    TimeUnit.SECONDS,
                    parentChain);
            markChildExecutionCompleted(
                    parentChain, parentNodeId, reference);
            clearDefaultStates(result);
            return result;
        } catch (InterruptedException error) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(
                    "Child workflow execution interrupted", error);
        } catch (TimeoutException impossible) {
            throw new IllegalStateException(
                    "Unexpected child workflow timeout", impossible);
        } finally {
            if (rootPermit) {
                rootChildExecutionPermits.release();
            }
        }
    }

    /**
     * 在父实例短锁内复用或创建持久子流程关联。
     *
     * @param definitionId 子流程定义 ID
     * @param parentChain 父流程
     * @param parentNodeId 父节点 ID
     * @param invocationId 父节点逻辑执行 ID
     * @param childDepth 子流程综合深度
     * @param childCallPath 子流程调用链
     * @return 持久子流程关联
     */
    private ChildExecutionReference prepareChildExecution(
            String definitionId,
            Chain parentChain,
            String parentNodeId,
            String invocationId,
            int childDepth,
            List<String> childCallPath) {
        return parentChain.executeWithLock(
                parentChain.getStateInstanceId(),
                10L,
                TimeUnit.SECONDS,
                () -> {
                    NodeState nodeState =
                            parentChain.updateNodeStateSafely(
                                    parentNodeId, state -> null);
                    Object existingValue = nodeState.getMemory().get(
                            CHILD_EXECUTION_REFERENCE_KEY);
                    if (existingValue instanceof ChildExecutionReference) {
                        ChildExecutionReference existing =
                                (ChildExecutionReference) existingValue;
                        if (Objects.equals(
                                existing.invocationId(), invocationId)) {
                            return existing;
                        }
                        ChainState existingChild = chainStateRepository.load(
                                existing.childInstanceId());
                        if (existingChild != null
                                && (existingChild.getStatus() == null
                                || !existingChild.getStatus().isTerminal())) {
                            throw new RetryableTriggerException(
                                    "前一次子工作流仍在执行", null);
                        }
                    }
                    Chain child = createChain(definitionId);
                    child.updateStateSafely(state -> {
                        state.getMemory().put(
                                NESTED_DEPTH_MEMORY_KEY, childDepth);
                        state.getMemory().put(
                                DEFINITION_CALL_PATH_MEMORY_KEY,
                                new ArrayList<>(childCallPath));
                        return EnumSet.of(ChainStateField.MEMORY);
                    });
                    ChildExecutionReference created =
                            new ChildExecutionReference(
                                    invocationId,
                                    child.getStateInstanceId(),
                                    childDepth,
                                    false);
                    parentChain.updateNodeStateSafely(
                            parentNodeId,
                            state -> {
                                state.getMemory().put(
                                        CHILD_EXECUTION_REFERENCE_KEY,
                                        created);
                                return EnumSet.of(NodeStateField.MEMORY);
                            });
                    return created;
                });
    }

    /**
     * 在子实例仍为 READY 时幂等启动。
     *
     * @param reference 子流程关联
     * @param variables 子流程输入
     */
    private void startChildIfReady(
            ChildExecutionReference reference,
            Map<String, Object> variables) {
        ChainState childState = chainStateRepository.load(
                reference.childInstanceId());
        if (childState == null) {
            throw new ChainException(
                    "Child chain state not found: "
                            + reference.childInstanceId());
        }
        ChainDefinition definition =
                getDefinitionForInstance(childState);
        if (definition == null) {
            throw new ChainException(
                    "Child chain definition not found: "
                            + reference.childInstanceId());
        }
        Chain child = configureChain(
                definition, reference.childInstanceId());
        child.setExecutionLane(childExecutionLane(reference.depth()));
        child.setNestedDepthBase(reference.depth());
        child.executeWithLock(
                reference.childInstanceId(),
                10L,
                TimeUnit.SECONDS,
                () -> {
                    ChainState latest = chainStateRepository.load(
                            reference.childInstanceId());
                    if (latest != null
                            && (latest.getStatus() == ChainStatus.READY
                            || latest.getStatus()
                            == ChainStatus.RUNNING)) {
                        child.start(variables);
                    }
                    return null;
                });
    }

    /**
     * 标记当前父节点关联已观察到子流程终态。
     *
     * @param parentChain 父流程
     * @param parentNodeId 父节点 ID
     * @param reference 子流程关联
     */
    private void markChildExecutionCompleted(
            Chain parentChain,
            String parentNodeId,
            ChildExecutionReference reference) {
        parentChain.updateNodeStateSafely(parentNodeId, state -> {
            Object current = state.getMemory().get(
                    CHILD_EXECUTION_REFERENCE_KEY);
            if (!(current instanceof ChildExecutionReference)
                    || !Objects.equals(
                    ((ChildExecutionReference) current).invocationId(),
                    reference.invocationId())) {
                return null;
            }
            state.getMemory().put(
                    CHILD_EXECUTION_REFERENCE_KEY,
                    new ChildExecutionReference(
                            reference.invocationId(),
                            reference.childInstanceId(),
                            reference.depth(),
                            true));
            return EnumSet.of(NodeStateField.MEMORY);
        });
    }

    /**
     * 按综合嵌套深度生成独立子流程通道。
     *
     * @param depth 综合嵌套深度
     * @return 通道名
     */
    public static String childExecutionLane(int depth) {
        return CHILD_WORKFLOW_EXECUTION_LANE_PREFIX
                + Math.max(1, depth);
    }

    /**
     * 读取实例持久化的子工作流嵌套深度。
     *
     * @param state 工作流状态
     * @return 非负嵌套深度
     */
    private int readNestedDepth(ChainState state) {
        if (state == null) {
            return 0;
        }
        Object value = state.getMemory().get(NESTED_DEPTH_MEMORY_KEY);
        return value instanceof Number
                ? Math.max(0, ((Number) value).intValue())
                : 0;
    }

    /**
     * 读取并规范化实例的工作流定义调用链。
     *
     * @param state 工作流状态
     * @param currentDefinitionId 当前定义 ID
     * @return 从根定义到当前定义的调用链
     */
    private List<String> readDefinitionCallPath(
            ChainState state, String currentDefinitionId) {
        List<String> path = new ArrayList<>();
        Object value = state == null
                ? null
                : state.getMemory().get(DEFINITION_CALL_PATH_MEMORY_KEY);
        if (value instanceof Collection<?>) {
            for (Object item : (Collection<?>) value) {
                if (item != null) {
                    path.add(canonicalDefinitionId(String.valueOf(item)));
                }
            }
        }
        if (path.isEmpty() && currentDefinitionId != null) {
            path.add(canonicalDefinitionId(currentDefinitionId));
        }
        return path;
    }

    /**
     * 将发布态和草稿态的同一工作流规范化为统一调用身份。
     *
     * @param definitionId 定义 ID
     * @return 规范化定义 ID
     */
    private String canonicalDefinitionId(String definitionId) {
        if (definitionId == null) {
            return "";
        }
        String normalized = definitionId.trim();
        return normalized.startsWith("published:")
                ? normalized.substring("published:".length())
                : normalized;
    }

    /**
     * 轮询持久状态等待工作流终态，允许任意集群实例执行实际触发器。
     *
     * @param stateInstanceId 工作流实例 ID
     * @param timeout 超时数值
     * @param unit 超时单位
     * @return 已解析业务结果
     * @throws InterruptedException 等待线程被中断
     * @throws TimeoutException 超时
     */
    @SuppressWarnings("unchecked")
    private Map<String, Object> awaitPersistentOutcome(
            String stateInstanceId,
            long timeout,
            TimeUnit unit,
            Chain parentChain)
            throws InterruptedException, TimeoutException {
        return awaitPersistentOutcome(
                stateInstanceId, timeout, unit, parentChain, false);
    }

    private Map<String, Object> awaitPersistentOutcome(
            String stateInstanceId,
            long timeout,
            TimeUnit unit,
            Chain parentChain,
            boolean rejectSuspension)
            throws InterruptedException, TimeoutException {
        Objects.requireNonNull(unit, "time unit required");
        long timeoutNanos = timeout == Long.MAX_VALUE
                ? Long.MAX_VALUE
                : Math.max(0L, unit.toNanos(timeout));
        long startedAt = System.nanoTime();
        while (true) {
            if (parentChain != null) {
                ChainState parentState = chainStateRepository.load(
                        parentChain.getStateInstanceId());
                if (parentState == null
                        || (parentState.getStatus() != null
                        && parentState.getStatus().isTerminal())) {
                    cancel(
                            stateInstanceId,
                            "Parent workflow is no longer running");
                    throw new ChainException(
                            "Parent workflow ended while child was running");
                }
                Trigger owner = TriggerContext.getCurrentTrigger();
                if (owner != null) {
                    // 认领丢失只终止旧 owner 的等待；durable child 留给新 owner 复用。
                    triggerScheduler.assertClaimOwned(owner);
                }
            }
            ChainState state = chainStateRepository.load(stateInstanceId);
            if (state == null) {
                throw new ChainException(
                        "Chain state not found: " + stateInstanceId);
            }
            ChainStatus status = state.getStatus();
            if (rejectSuspension && status == ChainStatus.SUSPEND) {
                cancel(stateInstanceId, "Suspended workflow is not supported by this caller");
                throw new ChainException(
                        "Workflow suspended and requires external input: "
                                + stateInstanceId);
            }
            if (status != null && status.isTerminal()) {
                if (!status.isSuccess()) {
                    ExceptionSummary error = state.getError();
                    throw new ChainException(
                            error == null
                                    ? "Workflow ended with status " + status
                                    : error.getMessage());
                }
                Map<String, Object> result =
                        (Map<String, Object>)
                                loopResultRepository.resolveReferences(
                                        state.getExecuteResult());
                return result == null
                        ? Collections.emptyMap()
                        : result;
            }
            if (timeoutNanos != Long.MAX_VALUE
                    && System.nanoTime() - startedAt >= timeoutNanos) {
                throw new TimeoutException(
                        "Workflow execution timed out: "
                                + stateInstanceId);
            }
            Thread.sleep(persistentOutcomePollMillis);
        }
    }

    /**
     * 父工作流节点与子实例之间的持久关联。
     *
     * @param invocationId 父节点逻辑执行 ID
     * @param childInstanceId 子流程实例 ID
     * @param depth 子流程综合嵌套深度
     * @param completed 是否已观察到终态
     */
    private record ChildExecutionReference(
            String invocationId,
            String childInstanceId,
            int depth,
            boolean completed) implements Serializable {

        private static final long serialVersionUID = 1L;
    }


    /**
     * 执行指定节点的业务逻辑
     *
     * @param definitionId 流程定义ID，用于标识哪个流程定义
     * @param nodeId       节点ID，用于标识要执行的具体节点
     * @param variables    执行上下文变量集合，包含节点执行所需的参数和数据
     * @return 执行结果映射表，包含节点执行后的输出数据
     */
    public Map<String, Object> executeNode(String definitionId, String nodeId, Map<String, Object> variables) {
        ChainDefinition chainDefinitionById = definitionRepository.getChainDefinitionById(definitionId);
        Node node = chainDefinitionById.getNodeById(nodeId);
        Chain temp = createChain(chainDefinitionById);
        try {
            temp.initializeState();
            if (variables != null && !variables.isEmpty()) {
                temp.updateStateSafely(s -> {
                    s.getMemory().putAll(variables);
                    temp.applyStartParameterAliases(s.getMemory(), variables);
                    return EnumSet.of(ChainStateField.MEMORY);
                });
            }
            return node.execute(temp);
        } finally {
            activeDefinitions.remove(temp.getStateInstanceId());
            definitionSnapshotRepository.remove(temp.getStateInstanceId());
        }
    }


    /**
     * 获取指定节点的参数列表
     *
     * @param definitionId 链定义ID，用于定位具体的链定义
     * @param nodeId       节点ID，用于在链定义中定位具体节点
     * @return 返回指定节点的参数列表
     */
    public List<Parameter> getNodeParameters(String definitionId, String nodeId) {
        ChainDefinition chainDefinitionById = definitionRepository.getChainDefinitionById(definitionId);
        Node node = chainDefinitionById.getNodeById(nodeId);
        return node.getParameters();
    }


    public void resumeAsync(String stateInstanceId) {
        this.resumeAsync(stateInstanceId, Collections.emptyMap());
    }


    public void resumeAsync(String stateInstanceId, Map<String, Object> variables) {
        ChainState state = chainStateRepository.load(stateInstanceId);
        if (state == null) {
            return;
        }

        ChainDefinition definition = getDefinitionForInstance(state);
        if (definition == null) {
            return;
        }

        Chain chain = configureChain(definition, state.getInstanceId());
        chain.resume(variables);
    }

    /**
     * 仅在工作流实例处于暂停状态时恢复执行。
     *
     * <p>状态判断和恢复由 {@link Chain} 在同一个实例锁内完成，可安全处理并发恢复请求。</p>
     *
     * @param stateInstanceId 工作流实例 ID
     * @param variables 恢复时注入的变量
     * @return 本次是否完成了暂停态到运行态的转换
     */
    public boolean resumeAsyncIfSuspended(
            String stateInstanceId,
            Map<String, Object> variables) {
        ChainState state = chainStateRepository.load(stateInstanceId);
        if (state == null) {
            return false;
        }
        ChainDefinition definition = getDefinitionForInstance(state);
        if (definition == null) {
            return false;
        }
        Chain chain = configureChain(
                definition,
                state.getInstanceId(),
                state);
        return chain.resumeIfSuspended(variables);
    }

    /**
     * 获取工作流实例启动时定义快照中的节点名称。
     *
     * @param stateInstanceId 工作流实例 ID
     * @return 按定义顺序排列的节点 ID 与名称；实例或定义不存在时返回空映射
     */
    public Map<String, String> getInstanceNodeNames(
            String stateInstanceId) {
        ChainState state = chainStateRepository.load(stateInstanceId);
        if (state == null) {
            return Collections.emptyMap();
        }
        return getInstanceNodeNames(state);
    }

    /**
     * 使用调用方已经加载的状态获取实例定义快照中的节点名称。
     *
     * @param state 已加载的工作流状态
     * @return 按定义顺序排列的节点 ID 与名称；状态或定义不存在时返回空映射
     */
    public Map<String, String> getInstanceNodeNames(
            ChainState state) {
        if (state == null) {
            return Collections.emptyMap();
        }
        ChainDefinition definition = getDefinitionForInstance(state);
        if (definition == null
                || definition.getNodes() == null
                || definition.getNodes().isEmpty()) {
            return Collections.emptyMap();
        }
        Map<String, String> nodeNames = new LinkedHashMap<>();
        for (Node node : definition.getNodes()) {
            if (node == null
                    || node.getId() == null
                    || node.getId().isBlank()) {
                continue;
            }
            String nodeName = node.getName();
            nodeNames.put(
                    node.getId(),
                    nodeName == null || nodeName.isBlank()
                            ? node.getId()
                            : nodeName);
        }
        return Collections.unmodifiableMap(nodeNames);
    }


    private Chain createChain(String definitionId) {
        ChainDefinition definition = definitionRepository.getChainDefinitionById(definitionId);
        if (definition == null) {
            throw new RuntimeException("Chain definition not found");
        }
        return createChain(definition);
    }

    /**
     * 使用已加载的定义创建工作流实例，避免同一次调用重复读取定义。
     *
     * @param definition 已加载的工作流定义
     * @return 已完成运行时依赖配置的工作流实例
     */
    private Chain createChain(ChainDefinition definition) {
        String stateInstanceId = UUID.randomUUID().toString();
        activeDefinitions.put(stateInstanceId, definition);
        try {
            definitionSnapshotRepository.save(stateInstanceId, definition);
            Chain chain = configureChain(definition, stateInstanceId);
            chain.initializeState();
            return chain;
        } catch (RuntimeException | Error error) {
            activeDefinitions.remove(stateInstanceId);
            try {
                definitionSnapshotRepository.remove(stateInstanceId);
            } catch (RuntimeException cleanupError) {
                error.addSuppressed(cleanupError);
            }
            throw error;
        }
    }

    /**
     * 为工作流实例配置共享运行时依赖。
     *
     * @param definition      工作流定义
     * @param stateInstanceId 状态实例 ID
     * @return 配置完成的工作流实例
     */
    private Chain configureChain(ChainDefinition definition, String stateInstanceId) {
        return configureChain(definition, stateInstanceId, null);
    }

    /**
     * 为工作流实例配置共享运行时依赖，并复用调用方已经加载的状态。
     *
     * @param definition 工作流定义
     * @param stateInstanceId 状态实例 ID
     * @param persistedState 已加载状态；为 {@code null} 时按需读取
     * @return 配置完成的工作流实例
     */
    private Chain configureChain(
            ChainDefinition definition,
            String stateInstanceId,
            ChainState persistedState) {
        Chain chain = new Chain(definition, stateInstanceId);
        chain.setTriggerScheduler(triggerScheduler);
        chain.setChainStateRepository(chainStateRepository);
        chain.setNodeStateRepository(nodeStateRepository);
        chain.setLoopResultRepository(loopResultRepository);
        chain.setEventManager(eventManager);
        chain.setExecutionBudget(executionBudget);
        ChainState state = persistedState == null
                ? chainStateRepository.load(stateInstanceId)
                : persistedState;
        int nestedDepth = readNestedDepth(state);
        chain.setNestedDepthBase(nestedDepth);
        if (nestedDepth > 0) {
            chain.setExecutionLane(childExecutionLane(nestedDepth));
        }
        return chain;
    }

    /**
     * 获取工作流实例启动时使用的定义快照。
     *
     * @param state 工作流状态
     * @return 活跃定义快照；当前实例首次由本节点接管时从仓储加载
     */
    private ChainDefinition getDefinitionForInstance(ChainState state) {
        String stateInstanceId = state.getInstanceId();
        ChainDefinition definition = activeDefinitions.get(stateInstanceId);
        if (definition != null) {
            return definition;
        }
        ChainDefinition loaded = definitionSnapshotRepository.load(stateInstanceId);
        String definitionId = state.getChainDefinitionId();
        if (loaded == null
                && definitionId != null
                && !definitionId.isBlank()) {
            // 兼容升级前已经启动、尚未持久化定义快照的实例。
            loaded = definitionRepository.getChainDefinitionById(definitionId);
        }
        if (loaded == null) {
            return null;
        }
        synchronized (activeDefinitions) {
            ChainDefinition existing = activeDefinitions.get(stateInstanceId);
            if (existing != null) {
                return existing;
            }
            activeDefinitions.put(stateInstanceId, loaded);
            return loaded;
        }
    }


    private void accept(Trigger trigger, ExecutorService worker) {
        ChainState state = chainStateRepository.load(trigger.getStateInstanceId());
        if (state == null) {
            // 状态已过期或被清理时，该触发器已经失去业务目标，直接确认避免无限热重放。
            return;
        }
        if (state.getStatus() != null && state.getStatus().isTerminal()) {
            // 终态不可再次执行；直接确认迟到或重复触发器，避免重新加载已清理的定义快照。
            return;
        }


        ChainDefinition definition = getDefinitionForInstance(state);
        if (definition == null) {
            throw new NonRetryableTriggerException(
                    "Chain definition not found: " + state.getChainDefinitionId());
        }

        Chain chain = configureChain(
                definition, trigger.getStateInstanceId(), state);

        String nodeId = trigger.getNodeId();
        if (nodeId == null) {
            throw new NonRetryableTriggerException("Node ID not found in trigger.");
        }

        Node node = definition.getNodeById(nodeId);
        if (node == null) {
            throw new NonRetryableTriggerException(
                    "Node not found in definition(id: " + definition.getId() + ")");
        }

        chain.executeNode(node, trigger);
    }


    public synchronized void addEventListener(Class<? extends Event> eventClass, ChainEventListener listener) {
        eventManager.addEventListener(eventClass, listener);
    }

    public synchronized void addEventListener(ChainEventListener listener) {
        eventManager.addEventListener(listener);
    }

    public synchronized void removeEventListener(ChainEventListener listener) {
        eventManager.removeEventListener(listener);
    }

    public synchronized void removeEventListener(Class<? extends Event> eventClass, ChainEventListener listener) {
        eventManager.removeEventListener(eventClass, listener);
    }

    public synchronized void addErrorListener(ChainErrorListener listener) {
        eventManager.addChainErrorListener(listener);
    }

    public synchronized void removeErrorListener(ChainErrorListener listener) {
        eventManager.removeChainErrorListener(listener);
    }

    public synchronized void addNodeErrorListener(NodeErrorListener listener) {
        eventManager.addNodeErrorListener(listener);
    }

    public synchronized void removeNodeErrorListener(NodeErrorListener listener) {
        eventManager.removeNodeErrorListener(listener);
    }

    public void addOutputListener(ChainOutputListener outputListener) {
        eventManager.addOutputListener(outputListener);
    }

    public ChainDefinitionRepository getDefinitionRepository() {
        return definitionRepository;
    }

    public ChainStateRepository getChainStateRepository() {
        return chainStateRepository;
    }

    public NodeStateRepository getNodeStateRepository() {
        return nodeStateRepository;
    }

    public TriggerScheduler getTriggerScheduler() {
        return triggerScheduler;
    }

    /**
     * 在查询/API 边界透明还原循环结果引用。
     *
     * @param value 可能包含引用的值
     * @return 业务可见值
     */
    public Object resolveResultReferences(Object value) {
        return loopResultRepository.resolveReferences(value);
    }

    public EventManager getEventManager() {
        return eventManager;
    }
}
