/**
 * 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;

import com.easyagents.flow.core.chain.event.*;
import com.easyagents.flow.core.chain.repository.*;
import com.easyagents.flow.core.chain.runtime.*;
import com.easyagents.flow.core.util.CollectionUtil;
import com.easyagents.flow.core.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;


public class Chain {

    private static final Logger log = LoggerFactory.getLogger(Chain.class);
    private static final ThreadLocal<Chain> EXECUTION_THREAD_LOCAL = new ThreadLocal<>();
    /**
     * 当前节点执行期间的只读状态视图；线程隔离避免并行节点互相覆盖。
     */
    private static final ThreadLocal<NodeExecutionStateView>
            NODE_EXECUTION_STATE_VIEW = new ThreadLocal<>();
    private static final ThreadLocal<Map<String, ChainLock>> HELD_INSTANCE_LOCKS =
            new ThreadLocal<>();
    private static final ThreadLocal<List<Runnable>> DEFERRED_EVENT_ACTIONS = new ThreadLocal<>();


    protected final ChainDefinition definition;
    protected String stateInstanceId;

    //    protected final ChainState state;
    protected ChainStateRepository chainStateRepository;
    protected NodeStateRepository nodeStateRepository;
    protected LoopResultRepository loopResultRepository = new InMemoryLoopResultRepository();
    protected EventManager eventManager;
    protected TriggerScheduler triggerScheduler;
    /**
     * 当前实例锁内已经读取或成功提交的状态快照。
     */
    private ChainState executionStateSnapshot;
    /**
     * 产生当前状态快照的实例锁；锁作用域变化后快照不得继续用于判断或修改。
     */
    private ChainLock executionStateSnapshotLock;
    /**
     * 当前执行实例使用的平台资源保护预算。
     */
    protected ExecutionBudget executionBudget = ExecutionBudget.defaults();
    protected String executionLane;
    protected int nestedDepthBase;

    public static Chain currentChain() {
        return EXECUTION_THREAD_LOCAL.get();
    }

    public Chain(ChainDefinition definition, String stateInstanceId) {
        this.definition = definition;
        this.stateInstanceId = stateInstanceId;
    }

    public void notifyEvent(Event event) {
        deferOrRun(() -> eventManager.notifyEvent(event, this));
    }

    public void setStatusAndNotifyEvent(ChainStatus status) {
        AtomicReference<ChainStatus> before = new AtomicReference<>();
        updateStateSafely(state -> {
            before.set(state.getStatus());
            state.setStatus(status);
            return EnumSet.of(ChainStateField.STATUS);
        });
        notifyEvent(new ChainStatusChangeEvent(this, status, before.get()));
    }

    public void setStatusAndNotifyEvent(String stateInstanceId, ChainStatus status) {
        AtomicReference<ChainStatus> before = new AtomicReference<>();
        updateStateSafely(stateInstanceId, state -> {
            before.set(state.getStatus());
            state.setStatus(status);
            return EnumSet.of(ChainStateField.STATUS);
        });
        notifyEvent(new ChainStatusChangeEvent(this, status, before.get()));
    }

    /**
     * Safely updates the chain state with optimistic locking and retry-on-conflict.
     *
     * @param modifier the modifier that applies changes and declares updated fields
     * @throws ChainUpdateTimeoutException if update cannot succeed within timeout
     */
    public ChainState updateStateSafely(ChainStateModifier modifier) {
        return updateStateSafely(this.stateInstanceId, modifier);
    }


    public ChainState updateStateSafely(String stateInstanceId, ChainStateModifier modifier) {
        return executeWithLock(stateInstanceId, 10, TimeUnit.SECONDS,
                () -> updateStateSafelyLocked(stateInstanceId, modifier));
    }

    /**
     * 在实例锁保护下更新工作流状态。
     *
     * @param stateInstanceId 工作流实例 ID
     * @param modifier 状态修改器
     * @return 更新后的状态
     */
    private ChainState updateStateSafelyLocked(String stateInstanceId, ChainStateModifier modifier) {
        final long timeoutMs = 30_000; // 30 seconds total timeout
        final long maxRetryDelayMs = 100; // Maximum delay between retries

        long startTime = System.currentTimeMillis();
        int attempt = 0;
        ChainState current = null;
        while (System.currentTimeMillis() - startTime < timeoutMs) {
            current = attempt == 0
                    ? reusableExecutionState(stateInstanceId)
                    : null;
            if (current == null) {
                current = chainStateRepository.load(stateInstanceId);
            }
            if (current == null) {
                throw new IllegalStateException("Chain state not found: " + stateInstanceId);
            }

            EnumSet<ChainStateField> updatedFields = modifier.modify(current);
            if (updatedFields == null || updatedFields.isEmpty()) {
                cacheExecutionState(stateInstanceId, current);
                return current; // No actual changes, exit early
            }

            current.setVersion(current.getVersion() + 1);
            updatedFields.add(ChainStateField.VERSION);
            executionBudget.checkHotStateBytes(estimateHotStateBytes(current));
            assertInstanceLockOwned(stateInstanceId);
            if (chainStateRepository.tryUpdate(
                    current,
                    updatedFields,
                    currentLockFencingToken(stateInstanceId),
                    currentFencingClaimId(stateInstanceId),
                    currentClaimGeneration(stateInstanceId))) {
                cacheExecutionState(stateInstanceId, current);
                return current;
            }
            assertTriggerClaimOwned(TriggerContext.getCurrentTrigger());
            if (currentLockFencingToken(stateInstanceId) > 0L) {
                throw new RetryableTriggerException(
                        "Workflow state commit guard rejected: " + stateInstanceId,
                        null);
            }

            // Prepare next retry
            attempt++;
            long nextDelay = calculateNextRetryDelay(attempt, maxRetryDelayMs);
            sleepUninterruptibly(nextDelay);
        }

        // Timeout reached
        assert current != null;
        String msg = String.format(
                "Chain state update timeout after %d ms (instanceId: %s)",
                timeoutMs, current.getInstanceId()
        );
        log.warn(msg);
        throw new ChainUpdateTimeoutException(msg);
    }


    public NodeState updateNodeStateSafely(String nodeId, NodeStateModifier modifier) {
        return this.updateNodeStateSafely(this.stateInstanceId, nodeId, modifier);
    }

    public NodeState updateNodeStateSafely(String stateInstanceId, String nodeId, NodeStateModifier modifier) {
        return executeWithLock(stateInstanceId, 10, TimeUnit.SECONDS,
                () -> updateNodeStateSafelyLocked(stateInstanceId, nodeId, modifier));
    }

    /**
     * 在实例锁保护下更新节点状态。
     *
     * @param stateInstanceId 工作流实例 ID
     * @param nodeId 节点 ID
     * @param modifier 状态修改器
     * @return 更新后的节点状态
     */
    private NodeState updateNodeStateSafelyLocked(String stateInstanceId,
                                                  String nodeId,
                                                  NodeStateModifier modifier) {
        final long timeoutMs = 30_000;
        final long maxRetryDelayMs = 100;
        long startTime = System.currentTimeMillis();
        int attempt = 0;

        while (System.currentTimeMillis() - startTime < timeoutMs) {
            // 1. 仅加载 ChainState 版本，避免为节点状态提交反序列化完整工作流热状态。
            Long chainStateVersion = chainStateRepository.loadVersion(stateInstanceId);
            if (chainStateVersion == null) {
                throw new IllegalStateException("Chain state not found");
            }

            // 2. 加载 NodeState
            NodeState nodeState = nodeStateRepository.load(stateInstanceId, nodeId);
            if (nodeState == null) {
                nodeState = nodeStateRepository.create(
                        stateInstanceId,
                        nodeId,
                        chainStateVersion,
                        currentLockFencingToken(stateInstanceId),
                        currentFencingClaimId(stateInstanceId),
                        currentClaimGeneration(stateInstanceId));
                if (nodeState == null) {
                    throw new IllegalStateException(
                            "Unable to initialize node state: " + stateInstanceId + "/" + nodeId);
                }
            }

            // 3. 应用修改
            EnumSet<NodeStateField> updatedFields = modifier.modify(nodeState);

            if (updatedFields == null || updatedFields.isEmpty()) {
                return nodeState;
            }

            // 4. 尝试更新（传入 chainVersion 保证一致性）
            nodeState.setVersion(nodeState.getVersion() + 1);
            updatedFields.add(NodeStateField.VERSION);
            executionBudget.checkHotStateBytes(estimateValueBytes(
                    Arrays.asList(
                            nodeState.getMemory(),
                            nodeState.getTriggerEdgeIds(),
                            nodeState.getExecuteEdgeIds()),
                    new IdentityHashMap<>(),
                    executionBudget.getMaxHotStateBytes()));
            assertInstanceLockOwned(stateInstanceId);
            if (nodeStateRepository.tryUpdate(
                    nodeState,
                    updatedFields,
                    chainStateVersion,
                    currentLockFencingToken(stateInstanceId),
                    currentFencingClaimId(stateInstanceId),
                    currentClaimGeneration(stateInstanceId))) {
                return nodeState;
            }
            assertTriggerClaimOwned(TriggerContext.getCurrentTrigger());
            if (currentLockFencingToken(stateInstanceId) > 0L) {
                throw new RetryableTriggerException(
                        "Workflow node state commit guard rejected: "
                                + stateInstanceId
                                + "/"
                                + nodeId,
                        null);
            }

            // 5. 退避重试
            attempt++;
            sleepUninterruptibly(calculateNextRetryDelay(attempt, maxRetryDelayMs));
        }

        throw new ChainUpdateTimeoutException("Node state update timeout");
    }


    /**
     * Calculates the next retry delay using exponential backoff with jitter.
     *
     * @param attempt    the current retry attempt (1-based)
     * @param maxDelayMs the maximum delay in milliseconds
     * @return the delay in milliseconds to wait before next retry
     */
    private long calculateNextRetryDelay(int attempt, long maxDelayMs) {
        // Base delay: 10ms * (2^(attempt-1))
        long baseDelay = 10L * (1L << (attempt - 1));

        // Add jitter: ±25% randomness to avoid thundering herd
        double jitterFactor = 0.75 + (Math.random() * 0.5); // [0.75, 1.25)
        long delayWithJitter = (long) (baseDelay * jitterFactor);

        // Clamp between 1ms and maxDelayMs
        return Math.max(1L, Math.min(delayWithJitter, maxDelayMs));
    }

    /**
     * Sleeps for the specified duration, silently ignoring interrupts
     * but preserving the interrupt status.
     *
     * @param millis the length of time to sleep in milliseconds
     */
    private void sleepUninterruptibly(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // Preserve interrupt status
            // Do NOT throw here — we want to continue retrying
        }
    }

    /**
     * 以短路方式估算热状态体积，避免为保护预算额外进行完整序列化。
     *
     * @param state 工作流状态
     * @return 估算字节数
     */
    private long estimateHotStateBytes(ChainState state) {
        long limit = executionBudget.getMaxHotStateBytes();
        if (limit <= 0) {
            return 0L;
        }
        return estimateValueBytes(Arrays.asList(
                state.getMemory(),
                state.getExecuteResult(),
                state.getTriggerEdgeIds(),
                state.getTriggerNodeIds()), new IdentityHashMap<>(), limit);
    }

    /**
     * 递归估算对象图大小，到达预算后立即短路。
     *
     * @param value 当前对象
     * @param visited 已访问对象
     * @param limit 短路阈值
     * @return 估算字节数
     */
    private long estimateValueBytes(Object value,
                                    IdentityHashMap<Object, Boolean> visited,
                                    long limit) {
        if (value == null) {
            return 0L;
        }
        if (value instanceof CharSequence) {
            return (long) ((CharSequence) value).length() * Character.BYTES;
        }
        if (value instanceof byte[]) {
            return ((byte[]) value).length;
        }
        if (value instanceof Number || value instanceof Date) {
            return 16L;
        }
        if (value instanceof Boolean || value instanceof Character) {
            return 2L;
        }
        if (visited.put(value, Boolean.TRUE) != null) {
            return 0L;
        }
        long bytes = 32L;
        if (value instanceof Map) {
            for (Object entryObject : ((Map<?, ?>) value).entrySet()) {
                Map.Entry<?, ?> entry = (Map.Entry<?, ?>) entryObject;
                bytes += estimateValueBytes(entry.getKey(), visited, limit - bytes);
                bytes += estimateValueBytes(entry.getValue(), visited, limit - bytes);
                if (bytes > limit) {
                    return bytes;
                }
            }
        } else if (value instanceof Collection) {
            for (Object item : (Collection<?>) value) {
                bytes += estimateValueBytes(item, visited, limit - bytes);
                if (bytes > limit) {
                    return bytes;
                }
            }
        } else if (value instanceof Iterable) {
            // 任意 Iterable 可能是单次消费流，预算估算不得改变业务输入。
            bytes += 64L;
        } else if (value.getClass().isArray()) {
            int length = java.lang.reflect.Array.getLength(value);
            for (int index = 0; index < length; index++) {
                bytes += estimateValueBytes(java.lang.reflect.Array.get(value, index), visited, limit - bytes);
                if (bytes > limit) {
                    return bytes;
                }
            }
        } else {
            bytes += 64L;
        }
        return bytes;
    }


    public void start(Map<String, Object> variables) {
        Trigger prev = TriggerContext.getCurrentTrigger();
        try {
            // start 可能在 node 里执行一个新的 chain 的情况，
            // 需要清空父级 chain 的 Trigger
            TriggerContext.setCurrentTrigger(null);
            initializeState();
            ensureStarted(variables);
        } finally {
            // 恢复父级 chain 的 Trigger
            TriggerContext.setCurrentTrigger(prev);
        }
    }

    /**
     * 恢复被入口触发器捕获的 READY 启动。
     *
     * <p>入口意图先于 RUNNING 状态保存。若进程在两者之间退出，扫描器重新投递
     * START 触发器时由该方法补齐所有入口意图、恢复初始变量并推进到 RUNNING，
     * 当前触发器随后继续执行，避免被误确认后永久停留在 READY。</p>
     *
     * @param trigger 当前 START 触发器
     * @return 实例已经可以继续执行业务节点时为 {@code true}
     */
    private boolean recoverStart(Trigger trigger) {
        if (trigger == null || trigger.getType() != TriggerType.START) {
            return false;
        }
        return ensureStarted(trigger.getStartVariables());
    }

    /**
     * 幂等持久化入口意图并推进实例启动状态。
     *
     * @param variables 首次启动变量；恢复时可来自首个稳定入口意图
     * @return 实例处于 RUNNING 时为 {@code true}
     */
    private boolean ensureStarted(Map<String, Object> variables) {
        AtomicBoolean active = new AtomicBoolean();
        executeWithLock(stateInstanceId, 10L, TimeUnit.SECONDS, () -> {
            AtomicReference<ChainStatus> before =
                    new AtomicReference<>();
            AtomicBoolean started = new AtomicBoolean();
            ChainState current = chainStateRepository.load(
                    stateInstanceId);
            if (current == null
                    || (current.getStatus() != ChainStatus.READY
                    && current.getStatus() != ChainStatus.RUNNING)) {
                return null;
            }

            List<Node> pendingStartNodes = new ArrayList<>();
            for (Node startNode : definition.getStartNodes()) {
                NodeState nodeState = nodeStateRepository.load(
                        stateInstanceId, startNode.getId());
                if (!isCompletedStartNode(nodeState)) {
                    pendingStartNodes.add(startNode);
                }
            }

            List<String> startTriggerIds = new ArrayList<>();
            if (current.getStatus() == ChainStatus.READY
                    && !pendingStartNodes.isEmpty()) {
                // 首个稳定入口意图携带变量，确保该意图保存后可独立恢复输入。
                startTriggerIds.add(prepareStartNodeLocked(
                        pendingStartNodes.get(0), variables));
            }

            if (current.getStatus() == ChainStatus.READY) {
                updateStateSafely(state -> {
                    before.set(state.getStatus());
                    EnumSet<ChainStateField> fields =
                            EnumSet.noneOf(ChainStateField.class);
                    if (variables != null && !variables.isEmpty()) {
                        state.getMemory().putAll(variables);
                        applyStartParameterAliases(
                                state.getMemory(), variables);
                        fields.add(ChainStateField.MEMORY);
                    }
                    if (StringUtil.noText(
                            state.getChainDefinitionId())) {
                        state.setChainDefinitionId(definition.getId());
                        fields.add(
                                ChainStateField.CHAIN_DEFINITION_ID);
                    }
                    return fields;
                });
            }

            int preparedCount = startTriggerIds.isEmpty() ? 0 : 1;
            for (int index = preparedCount;
                 index < pendingStartNodes.size();
                 index++) {
                startTriggerIds.add(prepareStartNodeLocked(
                        pendingStartNodes.get(index), null));
            }

            if (current.getStatus() == ChainStatus.READY) {
                updateStateSafely(state -> {
                    EnumSet<ChainStateField> fields =
                            EnumSet.of(ChainStateField.STATUS);
                    state.setStatus(ChainStatus.RUNNING);
                    if (state.getStartedAt() <= 0) {
                        state.setStartedAt(
                                System.currentTimeMillis());
                        fields.add(ChainStateField.STARTED_AT);
                    }
                    started.set(true);
                    return fields;
                });
            }
            active.set(true);
            if (started.get()) {
                notifyEvent(new ChainStartEvent(this, variables));
                notifyEvent(new ChainStatusChangeEvent(
                        this, ChainStatus.RUNNING, before.get()));
            }

            // 状态提交后主动领取；崩溃时持久扫描仍会重新投递。
            for (String triggerId : startTriggerIds) {
                getTriggerScheduler().fire(triggerId);
            }
            return null;
        });
        return active.get();
    }

    /**
     * 为开始节点输入参数补齐 `nodeId.paramName` 与 `paramName` 双向别名。
     * 这样既兼容运行表单仅提交裸参数名，也兼容设计器内部统一保存完整引用路径。
     *
     * @param memory    流程内存
     * @param variables 本次注入的变量
     */
    public void applyStartParameterAliases(Map<String, Object> memory, Map<String, Object> variables) {
        if (memory == null || variables == null || variables.isEmpty()) {
            return;
        }
        List<Node> startNodes = definition == null ? Collections.emptyList() : definition.getStartNodes();
        if (startNodes == null || startNodes.isEmpty()) {
            return;
        }
        for (Node startNode : startNodes) {
            if (startNode == null || StringUtil.noText(startNode.getId())) {
                continue;
            }
            List<Parameter> parameters = startNode.getParameters();
            if (parameters == null || parameters.isEmpty()) {
                continue;
            }
            for (Parameter parameter : parameters) {
                if (parameter == null || parameter.getRefType() != RefType.INPUT || StringUtil.noText(parameter.getName())) {
                    continue;
                }
                String parameterName = parameter.getName().trim();
                String scopedName = startNode.getId() + "." + parameterName;
                Object plainValue = variables.get(parameterName);
                Object scopedValue = variables.get(scopedName);

                if (plainValue != null && !memory.containsKey(scopedName)) {
                    memory.put(scopedName, plainValue);
                }
                if (scopedValue != null && !memory.containsKey(parameterName)) {
                    memory.put(parameterName, scopedValue);
                }
            }
        }
    }

    public void executeNode(Node node, Trigger trigger) {
        try {
            EXECUTION_THREAD_LOCAL.set(this);
            assertTriggerClaimOwned(trigger);
            ChainState chainState = loadFreshState();
            bindNodeExecutionState(chainState);

            if (chainState.getStatus() == ChainStatus.READY
                    && trigger.getType() == TriggerType.START) {
                if (!recoverStart(trigger)) {
                    throw new IllegalStateException(
                            "Failed to recover READY workflow start: "
                            + stateInstanceId);
                }
                chainState = loadFreshState();
                bindNodeExecutionState(chainState);
            }

            // 当前处于挂起状态
            if (chainState.getStatus() == ChainStatus.SUSPEND) {
                updateStateSafely(state -> {
                    state.addSuspendNodeId(node.getId());
                    return EnumSet.of(ChainStateField.SUSPEND_NODE_IDS);
                });
                return;
            }
            // 处于非运行状态，比如错误状态
            else if (chainState.getStatus() != ChainStatus.RUNNING) {
                return;
            }

            // 可恢复启动可能重放同一稳定入口触发器；已开始的入口节点直接确认。
            if (trigger.getType() == TriggerType.START) {
                NodeState existing = nodeStateRepository.load(
                        stateInstanceId, node.getId());
                if (isCompletedStartNode(existing)) {
                    return;
                }
            }

            String triggerEdgeId = trigger.getEdgeId();
            if (shouldSkipNode(node, triggerEdgeId)) {
                return;
            }

            Map<String, Object> nodeResult = null;
            Throwable error = null;
            String executionAttemptKey = null;
            try {
                AtomicBoolean nodeStarted = new AtomicBoolean();
                String candidateAttemptKey =
                        currentExecutionAttemptKey(
                                node.getId());
                // 首次创建、状态转换和执行计数在同一实例锁内完成，避免额外读取和首次执行空状态。
                NodeState activeNodeState =
                        updateNodeStateSafely(node.id, s -> {
                            nodeStarted.set(false);
                            EnumSet<NodeStateField> fields = EnumSet.of(
                                    NodeStateField.EXECUTE_COUNT,
                                    NodeStateField.EXECUTE_EDGE_IDS);
                            if (s.getStatus() != NodeStatus.RUNNING) {
                                s.setStatus(NodeStatus.RUNNING);
                                nodeStarted.set(true);
                                fields.add(NodeStateField.STATUS);
                                s.setExecutionAttemptKey(
                                        candidateAttemptKey);
                                fields.add(
                                        NodeStateField
                                                .EXECUTION_ATTEMPT_KEY);
                            }
                            if (node.getJoinMode() == NodeJoinMode.ANY
                                    && node.getCondition() == null) {
                                s.recordTrigger(triggerEdgeId);
                                fields.add(NodeStateField.TRIGGER_COUNT);
                                fields.add(NodeStateField.TRIGGER_EDGE_IDS);
                            }
                            s.recordExecute(triggerEdgeId);
                            return fields;
                        });
                executionAttemptKey =
                        activeNodeState
                                .getExecutionAttemptKey();
                if (nodeStarted.get()) {
                    notifyEvent(new NodeStartEvent(
                            this,
                            node,
                            executionAttemptKey,
                            activeNodeState.getStatus(),
                            getAuditInstanceId()));
                }

                ChainState nodeExecutionState = updateStateSafely(state -> {
                    long childExecutionCount = state.getChildExecutionCount() + 1;
                    executionBudget.checkChildExecutions(childExecutionCount);
                    state.setChildExecutionCount(childExecutionCount);
                    state.addTriggerNodeId(node.id);
                    return EnumSet.of(
                            ChainStateField.TRIGGER_NODE_IDS,
                            ChainStateField.CHILD_EXECUTION_COUNT);
                });
                bindNodeExecutionState(nodeExecutionState);

                executionBudget.checkDuration(chainState.getStartedAt(), System.currentTimeMillis());
                nodeResult = node.execute(this);
                assertTriggerClaimOwned(trigger);
            } catch (TriggerClaimLostException | RetryableTriggerException claimLost) {
                throw claimLost;
            } catch (Throwable throwable) {
                log.error("Node execute error", throwable);
                error = throwable;
            }
            // 结果提交入口会在实例锁内重读状态，统一拦截取消、超时和其他终态。
            handleNodeResult(
                    node,
                    nodeResult,
                    triggerEdgeId,
                    error,
                    executionAttemptKey);
        } finally {
            executionStateSnapshot = null;
            executionStateSnapshotLock = null;
            NODE_EXECUTION_STATE_VIEW.remove();
            EXECUTION_THREAD_LOCAL.remove();
        }
    }

    /**
     * 验证当前触发器仍由本工作线程持有，阻止失去租约的工作线程提交状态。
     *
     * @param trigger 当前触发器
     */
    private void assertTriggerClaimOwned(Trigger trigger) {
        if (triggerScheduler != null) {
            triggerScheduler.assertClaimOwned(trigger);
        }
    }

    public NodeState getNodeState(String nodeId) {
        return getNodeState(this.stateInstanceId, nodeId);
    }

    public NodeState getNodeState(String stateInstanceId, String nodeId) {
        return nodeStateRepository.load(stateInstanceId, nodeId);
    }

    public <T> T executeWithLock(String instanceId, long timeout, TimeUnit unit, Supplier<T> action) {
        Map<String, ChainLock> heldLocks = HELD_INSTANCE_LOCKS.get();
        if (heldLocks != null && heldLocks.containsKey(instanceId)) {
            assertInstanceLockOwned(instanceId);
            return action.get();
        }
        if (heldLocks == null) {
            heldLocks = new HashMap<>();
            HELD_INSTANCE_LOCKS.set(heldLocks);
        }
        Map<String, ChainLock> lockScope = heldLocks;
        try {
            try (ChainLock lock = chainStateRepository.getLock(instanceId, timeout, unit)) {
                if (!lock.isAcquired()) {
                    throw new ChainLockTimeoutException("Failed to acquire lock for instance: " + instanceId);
                }
                lockScope.put(instanceId, lock);
                try {
                    T result = action.get();
                    assertInstanceLockOwned(instanceId);
                    return result;
                } finally {
                    lockScope.remove(instanceId);
                }
            }
        } finally {
            if (lockScope.isEmpty()) {
                HELD_INSTANCE_LOCKS.remove();
            }
        }
    }

    /**
     * 验证当前线程持有的实例锁仍然有效。
     *
     * @param instanceId 工作流实例 ID
     */
    private void assertInstanceLockOwned(String instanceId) {
        ChainLock lock = currentInstanceLock(instanceId);
        if (lock == null || !lock.isValid()) {
            throw new ChainLockTimeoutException(
                    "Workflow instance lock ownership lost: " + instanceId);
        }
    }

    /**
     * 获取当前线程持有的实例锁。
     *
     * @param instanceId 工作流实例 ID
     * @return 当前实例锁；未持有时为 {@code null}
     */
    private ChainLock currentInstanceLock(
            String instanceId) {
        Map<String, ChainLock> heldLocks =
                HELD_INSTANCE_LOCKS.get();
        return heldLocks == null
                ? null
                : heldLocks.get(instanceId);
    }

    /**
     * 显式创建当前工作流实例状态。
     *
     * @return 已存在或新创建的状态
     */
    public ChainState initializeState() {
        ChainState state = chainStateRepository.create(stateInstanceId);
        if (state == null) {
            throw new IllegalStateException("Unable to initialize chain state: " + stateInstanceId);
        }
        if (StringUtil.hasText(state.getChainDefinitionId())
                || definition == null
                || StringUtil.noText(definition.getId())) {
            return state;
        }
        // 定义 ID 必须先于可重放入口触发器持久化，避免恢复时无法定位定义快照。
        return updateStateSafely(current -> {
            if (StringUtil.hasText(current.getChainDefinitionId())) {
                return null;
            }
            current.setChainDefinitionId(definition.getId());
            return EnumSet.of(ChainStateField.CHAIN_DEFINITION_ID);
        });
    }

    private boolean shouldSkipNode(Node node, String edgeId) {
        NodeCondition condition = node.getCondition();
        if (node.getJoinMode() == NodeJoinMode.ANY && condition == null) {
            return false;
        }
        return executeWithLock(stateInstanceId, 10, TimeUnit.SECONDS, () -> {
            NodeState newState = updateNodeStateSafely(node.id, s -> {
                s.recordTrigger(edgeId);
                return EnumSet.of(NodeStateField.TRIGGER_COUNT, NodeStateField.TRIGGER_EDGE_IDS);
            });

            Map<String, Object> prevResult = Collections.emptyMap();
            boolean joinPending = node.getJoinMode() == NodeJoinMode.ALL
                    && !newState.isUpstreamFullyExecuted();
            boolean shouldSkipNode = joinPending
                    || (condition != null
                    && !condition.check(this, newState, prevResult));
            if (shouldSkipNode) {
                updateStateSafely(state -> {
                    return state.addUncheckedNodeId(node.id)
                            ? EnumSet.of(ChainStateField.UNCHECKED_NODE_IDS)
                            : null;
                });
            } else {
                updateStateSafely(state -> {
                    if (state.removeUncheckedNodeId(node.id)) {
                        return EnumSet.of(ChainStateField.UNCHECKED_NODE_IDS);
                    } else {
                        return null;
                    }
                });
            }
            return shouldSkipNode;
        });
    }


    private void handleNodeResult(Node node,
                                  Map<String, Object> prevNodeResult,
                                  String triggerEdgeId,
                                  Throwable error,
                                  String executionAttemptKey) {
        List<Runnable> deferredActions = new ArrayList<>();
        executeWithLock(stateInstanceId, 10, TimeUnit.SECONDS, () -> {
            assertTriggerClaimOwned(TriggerContext.getCurrentTrigger());
            ChainState latestState = loadFreshState();
            cacheExecutionState(
                    stateInstanceId, latestState);
            if (latestState == null) {
                throw new IllegalStateException("Chain state not found: " + stateInstanceId);
            }
            if (latestState.getStatus() != ChainStatus.RUNNING) {
                return null;
            }
            DEFERRED_EVENT_ACTIONS.set(deferredActions);
            try {
                handleNodeResultLocked(
                        node,
                        prevNodeResult,
                        triggerEdgeId,
                        error,
                        executionAttemptKey);
            } finally {
                DEFERRED_EVENT_ACTIONS.remove();
            }
            return null;
        });
        deferredActions.forEach(Runnable::run);
    }

    /**
     * 在实例锁保护下合并节点结果并调度后继节点。
     *
     * @param node 已执行节点
     * @param prevNodeResult 节点输出
     * @param triggerEdgeId 触发边 ID
     * @param error 节点执行异常
     * @param executionAttemptKey 进入节点时捕获的业务尝试键
     */
    private void handleNodeResultLocked(Node node,
                                        Map<String, Object> prevNodeResult,
                                        String triggerEdgeId,
                                        Throwable error,
                                        String executionAttemptKey) {
        ChainStatus finalChainStatus = null;
        NodeStatus finalNodeStatus = null;
        try {
            if (error == null) {
                // 更新 state 数据
                updateStateSafely(state -> {
                    EnumSet<ChainStateField> fields = EnumSet.of(ChainStateField.EXECUTE_RESULT);
                    state.setExecuteResult(prevNodeResult);

                    if (prevNodeResult != null && !prevNodeResult.isEmpty()) {
                        prevNodeResult.forEach((k, v) -> {
                            if (v != null) {
                                state.getMemory().put(node.getId() + "." + k, v);
                            }
                        });
                        fields.add(ChainStateField.MEMORY);
                    }

                    return fields;
                });

                if (node.isRetryEnable() && node.isResetRetryCountAfterNormal()) {
                    updateNodeStateSafely(node.id, state -> {
                        state.setRetryCount(0);
                        return EnumSet.of(NodeStateField.RETRY_COUNT);
                    });
                }

                finalNodeStatus = prevNodeResult == null ? null : (NodeStatus) prevNodeResult.get(ChainConsts.NODE_STATE_STATUS_KEY);

                // 不调度下一个节点，由 node 自行调度，比如 Loop 循环
                Boolean scheduleNextNodeDisabled = prevNodeResult == null ? null : (Boolean) prevNodeResult.get(ChainConsts.SCHEDULE_NEXT_NODE_DISABLED_KEY);
                if (scheduleNextNodeDisabled != null && scheduleNextNodeDisabled) {
                    return;
                }

                // 结束节点
                finalChainStatus = prevNodeResult != null ? (ChainStatus) prevNodeResult.get(ChainConsts.CHAIN_STATE_STATUS_KEY) : null;
                if (finalChainStatus != null && finalChainStatus.isTerminal()) {
                    return;
                }

                // 调度下一个节点
                scheduleNextForNode(node, prevNodeResult, triggerEdgeId);
            } else {
                // 挂起
                if (error instanceof ChainSuspendException) {
                    updateNodeStateSafely(node.id, s -> {
                        s.setStatus(NodeStatus.SUSPEND);
                        return EnumSet.of(NodeStateField.STATUS);
                    });

                    updateStateSafely(s -> {
                        s.addSuspendNodeId(node.getId());
                        s.addSuspendForParameters(((ChainSuspendException) error).getSuspendParameters());
                        return EnumSet.of(ChainStateField.SUSPEND_NODE_IDS, ChainStateField.SUSPEND_FOR_PARAMETERS);
                    });

                    finalNodeStatus = NodeStatus.SUSPEND;
                    finalChainStatus = ChainStatus.SUSPEND;
                }
                // 失败
                else {
                    NodeState newState = updateNodeStateSafely(node.getId(), s -> {
                        s.setStatus(NodeStatus.ERROR);
                        s.setError(new ExceptionSummary(error));
                        return EnumSet.of(NodeStateField.ERROR, NodeStateField.STATUS);
                    });

                    deferOrRun(() -> eventManager.notifyNodeError(error, node, prevNodeResult, this));

                    if (node.isRetryEnable()
                            && node.getMaxRetryCount() > 0
                            && newState.getRetryCount() < node.getMaxRetryCount()) {

                        updateNodeStateSafely(node.getId(), s -> {
                            s.setRetryCount(s.getRetryCount() + 1);
                            return EnumSet.of(NodeStateField.RETRY_COUNT);
                        });

                        scheduleNode(node, triggerEdgeId, TriggerType.RETRY, node.getRetryIntervalMs());
                    } else {
                        finalChainStatus = handleNodeError(node.id, error);
                    }
                }
            }
        } finally {
            // 如果当前的工作流正在执行中，则不发送 NodeEndEvent 事件
            if (finalNodeStatus != NodeStatus.RUNNING) {
                NodeStatus nodeStatus = finalNodeStatus == null ? NodeStatus.SUCCEEDED : finalNodeStatus;
                updateNodeStateSafely(node.id, state -> {
                    state.setStatus(nodeStatus);
                    return EnumSet.of(NodeStateField.STATUS);
                });
                notifyEvent(new NodeEndEvent(
                        this,
                        node,
                        prevNodeResult,
                        error,
                        nodeStatus,
                        executionAttemptKey));
            }

            if (finalChainStatus != null) {
                setStatusAndNotifyEvent(finalChainStatus);

                // chain 执行结束
                if (finalChainStatus.isTerminal()) {
                    notifyEvent(new ChainEndEvent(this));

                    // 执行结束，但是未执行成功，失败和取消等
                    // 更新父级链的状态
                    if (!finalChainStatus.isSuccess()) {
                        ChainState currentState = getState();
                        ChainStatus currentStatus = finalChainStatus;
                        while (currentState != null && StringUtil.hasText(currentState.getParentInstanceId())) {
                            updateStateSafely(currentState.getParentInstanceId(), state -> {
                                state.setStatus(currentStatus);
                                return EnumSet.of(ChainStateField.STATUS);
                            });
                            setStatusAndNotifyEvent(currentState.getParentInstanceId(), currentStatus);
                            currentState = getState(currentState.getParentInstanceId());
                        }
                    }
                }
            }
        }
    }

    /**
     * 为指定节点调度下一次执行
     *
     * @param node      要调度的节点
     * @param result    节点执行结果
     * @param byEdigeId 触发边的ID
     */
    private void scheduleNextForNode(Node node, Map<String, Object> result, String byEdigeId) {
        // 如果节点不支持循环，则直接调度向外的节点
        if (!node.isLoopEnable()) {
            scheduleOutwardNodes(node, result);
            return;
        }

        NodeState nodeState = getNodeState(node.getId());
        int completedLoopCount = Math.addExact(nodeState.getLoopCount(), 1);
        executionBudget.checkIterations(node.getId(), completedLoopCount);

        // 检查循环中断条件，如果满足则调度向外的节点
        NodeCondition breakCondition = node.getLoopBreakCondition();
        boolean shouldBreak = breakCondition != null
                && breakCondition.check(this, nodeState, result);
        if (shouldBreak || completedLoopCount >= node.getMaxLoopCount()) {
            resetNodeLoopCount(node.getId());
            scheduleOutwardNodes(node, result);
            return;
        }

        // 记录已经完成的执行次数；下一次执行的零基索引与该值一致。
        updateNodeStateSafely(node.getId(), s -> {
            s.setLoopCount(completedLoopCount);
            return EnumSet.of(NodeStateField.LOOP_COUNT);
        });

        scheduleNode(node, byEdigeId, TriggerType.LOOP, node.getLoopIntervalMs());
    }

    /**
     * 清理一次节点循环生命周期的计数，保证节点被外层循环再次触发时从零开始。
     *
     * @param nodeId 节点 ID
     */
    private void resetNodeLoopCount(String nodeId) {
        NodeState currentState = getNodeState(nodeId);
        if (currentState.getLoopCount() == 0) {
            return;
        }
        updateNodeStateSafely(nodeId, state -> {
            state.setLoopCount(0);
            return EnumSet.of(NodeStateField.LOOP_COUNT);
        });
    }


    private void scheduleOutwardNodes(Node node, Map<String, Object> result) {
        List<Edge> edges = definition.getOutwardEdge(node.getId());
        if (!CollectionUtil.hasItems(edges)) {
            // 当前节点没有向外的边，则调度父节点（自动回归父节点） 用在 Loop 循环等场景
            if (StringUtil.hasText(node.getParentId())) {
                Node parent = definition.getNodeById(node.getParentId());
                scheduleNode(parent, null, TriggerType.PARENT, 0L);
            }
            return;
        }

        // 检查所有向外的边是不是同一个父节点
        boolean allNotSameParent = false;
        boolean scheduleSuccess = false;
        for (Edge edge : edges) {
            Node nextNode = definition.getNodeById(edge.getTarget());
            if (nextNode == null) {
                throw new ChainException("Invalid edge target: " + edge.getTarget());
            }

            // 如果存在不同父节点的边，则跳过, 比如 Loop 节点可能只有子节点，没有后续的节点
            if (!isSameParent(node, nextNode)) {
                allNotSameParent = true;
                continue;
            }

            allNotSameParent = false;
            EdgeCondition edgeCondition = edge.getCondition();
            if (edgeCondition == null) {
                scheduleNode(nextNode, edge.getId(), TriggerType.NEXT, 0L);
                scheduleSuccess = true;
                continue;
            }

            if (edgeCondition.check(this, edge, result)) {
                updateStateSafely(state -> {
                    if (state.removeUncheckedEdgeId(edge.getId())) {
                        return EnumSet.of(ChainStateField.UNCHECKED_EDGE_IDS);
                    } else {
                        return null;
                    }
                });
                scheduleNode(nextNode, edge.getId(), TriggerType.NEXT, 0L);
                scheduleSuccess = true;
            } else {
                updateStateSafely(state -> {
                    return state.addUncheckedEdgeId(edge.getId())
                            ? EnumSet.of(ChainStateField.UNCHECKED_EDGE_IDS)
                            : null;
                });
                notifyEvent(new EdgeConditionCheckFailedEvent(this, edge, node, result));
            }
        }

        // 如果所有向外的边都不满足条件，则调度父节点（自动回归父节点） 用在 Loop 循环嵌套等场景（Loop 下的第一个节点是 Loop）
        if (allNotSameParent && !scheduleSuccess) {
            if (StringUtil.hasText(node.getParentId())) {
                Node parent = definition.getNodeById(node.getParentId());
                scheduleNode(parent, null, TriggerType.PARENT, 0L);
            }
        }
    }

    /**
     * 判断两个节点是否具有相同的父节点
     *
     * @param node 第一个节点
     * @param next 第二个节点
     * @return 如果两个节点的父节点ID相同则返回true，否则返回false
     */
    private boolean isSameParent(Node node, Node next) {
        // 如果两个节点的父节点ID都为空或空白，则认为是相同父节点
        if (StringUtil.noText(node.getParentId()) && StringUtil.noText(next.getParentId())) {
            return true;
        }

        // 比较两个节点的父节点ID是否相等
        return node.getParentId() != null && node.getParentId().equals(next.getParentId());
    }


    public void scheduleNode(Node node, String edgeId, TriggerType type, long delayMs) {
        scheduleNode(node, edgeId, type, delayMs, null, null);
    }

    /**
     * 调度循环体直属分支并写入稳定的代际游标。
     *
     * @param node 目标节点
     * @param edgeId 触发边 ID
     * @param delayMs 延迟毫秒数
     * @param loopNodeId 循环节点 ID
     * @param cursor 本轮分支游标
     */
    public void scheduleLoopChild(Node node,
                                  String edgeId,
                                  long delayMs,
                                  String loopNodeId,
                                  Trigger.LoopCursor cursor) {
        scheduleNode(node, edgeId, TriggerType.CHILD, delayMs, loopNodeId, cursor);
    }

    /**
     * 创建并持久化节点触发器。
     *
     * @param node 目标节点
     * @param edgeId 触发边 ID
     * @param type 触发类型
     * @param delayMs 延迟毫秒数
     * @param loopNodeId 需要覆盖的循环节点 ID
     * @param cursor 循环代际游标
     */
    private void scheduleNode(Node node,
                              String edgeId,
                              TriggerType type,
                              long delayMs,
                              String loopNodeId,
                              Trigger.LoopCursor cursor) {
        executeWithLock(stateInstanceId, 10L, TimeUnit.SECONDS, () -> {
            scheduleNodeLocked(node, edgeId, type, delayMs, loopNodeId, cursor);
            return null;
        });
    }

    /**
     * 在实例锁保护下创建并持久化节点触发器。
     *
     * @param node 目标节点
     * @param edgeId 触发边 ID
     * @param type 触发类型
     * @param delayMs 延迟毫秒数
     * @param loopNodeId 需要覆盖的循环节点 ID
     * @param cursor 循环代际游标
     */
    private void scheduleNodeLocked(Node node,
                                    String edgeId,
                                    TriggerType type,
                                    long delayMs,
                                    String loopNodeId,
                                    Trigger.LoopCursor cursor) {
        scheduleNodeLocked(
                node, edgeId, type, delayMs, loopNodeId, cursor,
                null, false, true, null);
    }

    /**
     * 在状态切换前持久化可重放的入口触发器意图。
     *
     * @param node 入口节点
     * @param startVariables 首个入口意图携带的启动变量
     * @return 稳定触发器 ID
     */
    private String prepareStartNodeLocked(
            Node node, Map<String, Object> startVariables) {
        String stableId = "start-"
                + UUID.nameUUIDFromBytes(
                (stateInstanceId + '\n' + node.getId())
                        .getBytes(java.nio.charset.StandardCharsets.UTF_8));
        scheduleNodeLocked(
                node, null, TriggerType.START, 1_000L, null, null,
                stableId, true, false, startVariables);
        return stableId;
    }

    /**
     * 判断稳定入口触发器是否已经完成状态提交。
     *
     * <p>RUNNING 不能视为完成：进程可能在执行计数递增后、业务结果提交前崩溃，
     * 此时必须允许同一稳定触发器重放。</p>
     *
     * @param nodeState 入口节点状态
     * @return 已提交终态时为 {@code true}
     */
    private boolean isCompletedStartNode(NodeState nodeState) {
        if (nodeState == null || nodeState.getStatus() == null) {
            return false;
        }
        return nodeState.getStatus() == NodeStatus.SUCCEEDED
                || nodeState.getStatus() == NodeStatus.SUSPEND
                || nodeState.getStatus() == NodeStatus.ERROR
                || nodeState.getStatus() == NodeStatus.FAILED;
    }

    /**
     * 在实例锁保护下创建并持久化节点触发器。
     *
     * @param node 目标节点
     * @param edgeId 触发边 ID
     * @param type 触发类型
     * @param delayMs 延迟毫秒数
     * @param loopNodeId 循环节点 ID
     * @param cursor 循环游标
     * @param stableTriggerId 可选稳定触发器 ID
     * @param onlyIfAbsent 是否仅在仓储中不存在时保存
     * @param requireActive 是否要求实例已处于运行态
     * @param startVariables 首个入口意图携带的启动变量
     */
    private void scheduleNodeLocked(Node node,
                                    String edgeId,
                                    TriggerType type,
                                    long delayMs,
                                    String loopNodeId,
                                    Trigger.LoopCursor cursor,
                                    String stableTriggerId,
                                    boolean onlyIfAbsent,
                                    boolean requireActive,
                                    Map<String, Object> startVariables) {
        if (requireActive && !isExecutionActiveLocked()) {
            return;
        }
        Trigger trigger = new Trigger();
        trigger.setId(stableTriggerId);
        trigger.setStateInstanceId(stateInstanceId);
        trigger.setEdgeId(edgeId);
        trigger.setNodeId(node.getId());
        trigger.setType(type);
        trigger.setExecutionLane(executionLane);
        trigger.setStartVariables(startVariables);
        trigger.setTriggerAt(System.currentTimeMillis() + delayMs);
        Trigger currentTrigger = TriggerContext.getCurrentTrigger();
        String fencingClaimId = currentFencingClaimId();
        long claimGeneration = currentClaimGeneration();
        long lockFencingToken = currentLockFencingToken(stateInstanceId);
        trigger.setRequiredLockFencingToken(lockFencingToken);
        if (claimGeneration > 0L && StringUtil.hasText(fencingClaimId)) {
            trigger.setRequiredFencingClaimId(fencingClaimId);
            trigger.setRequiredFencingToken(claimGeneration);
        }
        if (type == TriggerType.RETRY && currentTrigger != null) {
            String logicalExecutionId = StringUtil.hasText(currentTrigger.getLogicalExecutionId())
                    ? currentTrigger.getLogicalExecutionId()
                    : currentTrigger.getId();
            trigger.setLogicalExecutionId(logicalExecutionId);
        }
        if (currentTrigger != null && !currentTrigger.getLoopCursors().isEmpty()) {
            trigger.setLoopCursors(new LinkedHashMap<>(currentTrigger.getLoopCursors()));
        }
        if (loopNodeId != null && cursor != null) {
            trigger.getLoopCursors().put(loopNodeId, cursor);
        }

        if (edgeId != null) {
            updateStateSafely(state -> {
                state.addTriggerEdgeId(edgeId);
                return EnumSet.of(ChainStateField.TRIGGER_EDGE_IDS);
            });

            notifyEvent(new EdgeTriggerEvent(this, trigger));
        }

        if (onlyIfAbsent) {
            getTriggerScheduler().scheduleIfAbsent(trigger);
        } else {
            getTriggerScheduler().schedule(trigger);
        }
    }

    /**
     * 在已持有实例锁时判断工作流是否仍可推进。
     *
     * <p>优先复用当前触发器内已加载或成功提交的状态快照。实例锁保证此期间其他执行者
     * 无法提交取消、超时或终态，因此无需再次访问状态仓储。</p>
     *
     * @return 当前实例处于运行态时返回 {@code true}
     */
    private boolean isExecutionActiveLocked() {
        assertInstanceLockOwned(stateInstanceId);
        ChainState state = reusableExecutionState(stateInstanceId);
        if (state == null) {
            state = loadFreshState();
            cacheExecutionState(stateInstanceId, state);
        }
        return state != null && state.getStatus() == ChainStatus.RUNNING;
    }

    /**
     * 获取当前触发器认领代际。
     *
     * @param instanceId 工作流实例 ID
     * @return 当前认领代际；非触发器调用为 {@code 0}
     */
    private long currentClaimGeneration(String instanceId) {
        Trigger trigger = TriggerContext.getCurrentTrigger();
        return trigger != null && Objects.equals(instanceId, trigger.getStateInstanceId())
                ? trigger.getFencingToken()
                : 0L;
    }

    /**
     * 获取当前触发器认领代际。
     *
     * @return 当前认领代际；非触发器调用为 {@code 0}
     */
    public long currentClaimGeneration() {
        return currentClaimGeneration(stateInstanceId);
    }

    /**
     * 获取当前线程持有的实例锁 fencing token。
     *
     * @param instanceId 工作流实例 ID
     * @return 当前锁 token；本地锁或未持锁时为 {@code 0}
     */
    private long currentLockFencingToken(String instanceId) {
        Map<String, ChainLock> heldLocks = HELD_INSTANCE_LOCKS.get();
        ChainLock lock = heldLocks == null ? null : heldLocks.get(instanceId);
        return lock == null ? 0L : lock.getFencingToken();
    }

    /**
     * 获取当前线程持有的本实例锁 fencing token。
     *
     * @return 当前锁 token；本地锁或未持锁时为 {@code 0}
     */
    public long currentInstanceLockFencingToken() {
        return currentLockFencingToken(stateInstanceId);
    }

    /**
     * 获取当前触发器认领 ID。
     *
     * @param instanceId 工作流实例 ID
     * @return 当前触发器 ID；非触发器调用为 {@code null}
     */
    private String currentFencingClaimId(String instanceId) {
        Trigger trigger = TriggerContext.getCurrentTrigger();
        return trigger != null && Objects.equals(instanceId, trigger.getStateInstanceId())
                ? trigger.getId()
                : null;
    }

    /**
     * 获取当前触发器认领 ID，供循环结果等派生状态原子提交使用。
     *
     * @return 当前触发器 ID；非触发器调用为 {@code null}
     */
    public String currentFencingClaimId() {
        return currentFencingClaimId(stateInstanceId);
    }

    /**
     * 在实例锁保护下原子追加循环结果。
     *
     * @param resultId 循环结果 ID
     * @param iterationIndex 迭代序号
     * @param outputValues 本轮输出
     */
    public void appendLoopResult(
            String resultId, int iterationIndex, Map<String, Object> outputValues) {
        executeWithLock(stateInstanceId, 10L, TimeUnit.SECONDS, () -> {
            loopResultRepository.append(
                    stateInstanceId,
                    currentLockFencingToken(stateInstanceId),
                    currentFencingClaimId(stateInstanceId),
                    currentClaimGeneration(stateInstanceId),
                    resultId,
                    iterationIndex,
                    outputValues);
            return null;
        });
    }

    /**
     * 使用当前实例锁和触发器认领守卫物化循环输入。
     *
     * @param resultId 循环结果 ID
     * @param items 原始输入
     * @param maxItems 最大元素数
     * @return 已物化元素数量
     */
    public int storeLoopInput(
            String resultId, Iterable<?> items, long maxItems) {
        return loopResultRepository.storeInput(
                stateInstanceId,
                currentLockFencingToken(stateInstanceId),
                currentFencingClaimId(stateInstanceId),
                currentClaimGeneration(stateInstanceId),
                resultId,
                items,
                maxItems);
    }

    /**
     * 使用稳定的触发器认领守卫在实例锁外物化循环输入。
     *
     * <p>实例锁是短临界区资源，其他合法分支提交会推进其 fencing token，
     * 因而锁外长 I/O 仅绑定触发器 claim。发布结果时会重新获取实例锁并校验
     * resultId 与物化代际。</p>
     *
     * @param resultId 循环结果 ID
     * @param items 原始输入
     * @param maxItems 最大元素数
     * @param claimId 当前触发器 ID
     * @param claimGeneration 当前触发器认领代际
     * @return 已物化元素数量
     */
    public int storeLoopInputOutsideLock(
            String resultId,
            Iterable<?> items,
            long maxItems,
            String claimId,
            long claimGeneration) {
        return loopResultRepository.storeInput(
                stateInstanceId,
                0L,
                claimId,
                claimGeneration,
                resultId,
                items,
                maxItems);
    }

    /**
     * 使用稳定触发器认领守卫，在实例锁外接收并分块保存推送式输入。
     *
     * @param resultId 循环结果 ID
     * @param producer 输入生产者
     * @param maxItems 最大元素数
     * @param claimId 当前触发器 ID
     * @param claimGeneration 当前触发器认领代际
     * @return 已物化元素数量
     */
    public int storeProducedLoopInputOutsideLock(
            String resultId,
            LoopResultRepository.InputProducer producer,
            long maxItems,
            String claimId,
            long claimGeneration) {
        return loopResultRepository.storeProducedInput(
                stateInstanceId,
                0L,
                claimId,
                claimGeneration,
                resultId,
                producer,
                maxItems);
    }

    /**
     * 使用当前实例锁和触发器认领守卫清理循环输入。
     *
     * @param resultId 循环结果 ID
     */
    public void removeLoopInput(String resultId) {
        loopResultRepository.removeInput(
                stateInstanceId,
                currentLockFencingToken(stateInstanceId),
                currentFencingClaimId(stateInstanceId),
                currentClaimGeneration(stateInstanceId),
                resultId);
    }

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

    /**
     * 判断当前工作流实例是否仍允许执行和推进。
     *
     * @return 状态存在且为运行中时返回 {@code true}
     */
    private boolean isExecutionActive() {
        // 取消、超时等状态可能由其他线程提交，此处必须绕过执行期快照。
        ChainState state = loadFreshState();
        cacheExecutionState(stateInstanceId, state);
        return state != null && state.getStatus() == ChainStatus.RUNNING;
    }

    /**
     * 从仓储读取最新状态，判断当前实例是否仍允许推进。
     *
     * @return 最新状态为运行中时返回 {@code true}
     */
    public boolean isExecutionActiveNow() {
        return isExecutionActive();
    }

    /**
     * 设置本实例派生触发器使用的执行通道。
     *
     * @param executionLane 执行通道；{@code null} 表示默认通道
     */
    public void setExecutionLane(String executionLane) {
        this.executionLane = executionLane;
    }

    /**
     * 设置由父子工作流调用链贡献的基础嵌套深度。
     *
     * @param nestedDepthBase 非负基础深度
     */
    public void setNestedDepthBase(int nestedDepthBase) {
        this.nestedDepthBase = Math.max(0, nestedDepthBase);
    }

    /**
     * 获取父子工作流调用链贡献的基础嵌套深度。
     *
     * @return 非负基础深度
     */
    public int getNestedDepthBase() {
        return nestedDepthBase;
    }

    /**
     * 生成当前节点副作用操作的稳定幂等键。
     *
     * <p>同一持久化触发器发生租约恢复或重复投递时返回相同键；直接执行单节点且没有触发器
     * 上下文时返回 {@code null}，保持原有每次调用均执行的语义。</p>
     *
     * @param nodeId 当前节点 ID
     * @return 稳定幂等键；无持久化触发器上下文时返回 {@code null}
     */
    public String currentExecutionIdempotencyKey(String nodeId) {
        Trigger trigger = TriggerContext.getCurrentTrigger();
        if (trigger == null || StringUtil.noText(trigger.getId())) {
            return null;
        }
        String logicalExecutionId = StringUtil.hasText(trigger.getLogicalExecutionId())
                ? trigger.getLogicalExecutionId()
                : trigger.getId();
        return stateInstanceId + ":" + nodeId + ":" + logicalExecutionId;
    }

    /**
     * 生成当前节点本次业务尝试的稳定键。
     *
     * <p>基础设施重新投递同一触发器时键保持不变，业务重试生成新触发器时键随之变化，
     * 可用于执行步骤等“每次业务尝试一条”的幂等记录。</p>
     *
     * @param nodeId 当前节点 ID
     * @return 当前业务尝试稳定键；无持久化触发器上下文时返回 {@code null}
     */
    public String currentExecutionAttemptKey(String nodeId) {
        Trigger trigger = TriggerContext.getCurrentTrigger();
        if (trigger == null || StringUtil.noText(trigger.getId())) {
            return null;
        }
        return stateInstanceId + ":" + nodeId + ":" + trigger.getId();
    }

    /**
     * 将当前工作流实例标记为取消。
     *
     * @param message 取消原因
     * @return 本次是否将非终态实例转换为取消状态
     */
    public boolean cancel(String message) {
        AtomicReference<Boolean> changed = new AtomicReference<>(false);
        AtomicReference<ChainStatus> before = new AtomicReference<>();
        updateStateSafely(state -> {
            if (state.getStatus() != null && state.getStatus().isTerminal()) {
                return null;
            }
            before.set(state.getStatus());
            state.setStatus(ChainStatus.CANCELLED);
            state.setMessage(message);
            changed.set(true);
            return EnumSet.of(ChainStateField.STATUS, ChainStateField.MESSAGE);
        });
        if (changed.get()) {
            notifyEvent(new ChainStatusChangeEvent(
                    this, ChainStatus.CANCELLED, before.get()));
            // 取消属于工作流终态，统一发布结束事件供审计、清理等监听器收口。
            notifyEvent(new ChainEndEvent(this));
        }
        return changed.get();
    }

    /**
     * 将不可继续投递的实例幂等收敛为失败终态并发布统一终态事件。
     *
     * <p>该入口用于触发器耗尽或不可恢复错误。状态、错误和消息先在实例锁内提交，
     * 再发布状态、错误和结束事件，使同步等待、审计记录和定义快照清理观察到同一终态。</p>
     *
     * @param cause 最终失败原因
     * @return 本次完成终态转换或实例此前已终止时为 {@code true}
     */
    public boolean failTerminal(Throwable cause) {
        Throwable failure = cause == null
                ? new ChainException("Workflow execution failed")
                : cause;
        AtomicBoolean changed = new AtomicBoolean();
        AtomicReference<ChainStatus> before =
                new AtomicReference<>();
        List<Runnable> deferredActions = new ArrayList<>();
        executeWithLock(stateInstanceId, 10L, TimeUnit.SECONDS, () -> {
            updateStateSafely(state -> {
                if (state.getStatus() != null
                        && state.getStatus().isTerminal()) {
                    return null;
                }
                before.set(state.getStatus());
                state.setStatus(ChainStatus.FAILED);
                state.setError(new ExceptionSummary(failure));
                state.setMessage(failure.getMessage());
                changed.set(true);
                return EnumSet.of(
                        ChainStateField.STATUS,
                        ChainStateField.ERROR,
                        ChainStateField.MESSAGE);
            });
            if (changed.get()) {
                DEFERRED_EVENT_ACTIONS.set(deferredActions);
                try {
                    notifyEvent(new ChainStatusChangeEvent(
                            this, ChainStatus.FAILED, before.get()));
                    deferOrRun(() ->
                            eventManager.notifyChainError(failure, this));
                    notifyEvent(new ChainEndEvent(this));
                } finally {
                    DEFERRED_EVENT_ACTIONS.remove();
                }
            }
            return null;
        });
        deferredActions.forEach(Runnable::run);
        ChainState state = chainStateRepository.load(
                stateInstanceId);
        return changed.get()
                || (state != null
                && state.getStatus() != null
                && state.getStatus().isTerminal());
    }


    private ChainStatus handleNodeError(String nodeId, Throwable throwable) {
        updateNodeStateSafely(nodeId, s -> {
            s.setStatus(NodeStatus.FAILED);
            s.setError(new ExceptionSummary(throwable));
            return EnumSet.of(NodeStateField.ERROR, NodeStateField.STATUS);
        });

        updateStateSafely(state -> {
            state.setError(new ExceptionSummary(throwable));
            return EnumSet.of(ChainStateField.ERROR);
        });

        setStatusAndNotifyEvent(ChainStatus.FAILED);
        deferOrRun(() -> eventManager.notifyChainError(throwable, this));
        return ChainStatus.FAILED;
    }

    /**
     * 在实例锁内仅记录事件动作，离开锁后按原顺序执行监听器 I/O。
     *
     * @param action 事件动作
     */
    private void deferOrRun(Runnable action) {
        List<Runnable> deferredActions = DEFERRED_EVENT_ACTIONS.get();
        if (deferredActions == null) {
            action.run();
        } else {
            deferredActions.add(action);
        }
    }

    public void suspend() {
        setStatusAndNotifyEvent(ChainStatus.SUSPEND);
    }

    public void suspend(Node node) {
        updateStateSafely(state -> {
            state.addSuspendNodeId(node.getId());
            return EnumSet.of(ChainStateField.SUSPEND_NODE_IDS);
        });
        setStatusAndNotifyEvent(ChainStatus.SUSPEND);
    }


    /**
     * 仅在工作流处于暂停状态时恢复执行。
     *
     * <p>状态判断与恢复动作在同一个实例锁内完成，避免并发恢复请求重复注入变量
     * 或把终态实例重新改为运行中。</p>
     *
     * @param variables 恢复时注入的变量
     * @return 本次是否完成了暂停态到运行态的转换
     */
    public boolean resumeIfSuspended(Map<String, Object> variables) {
        return executeWithLock(
                stateInstanceId,
                10L,
                TimeUnit.SECONDS,
                () -> {
                    ChainState current =
                            chainStateRepository.load(stateInstanceId);
                    if (current == null
                            || current.getStatus() != ChainStatus.SUSPEND) {
                        return false;
                    }
                    resumeSuspended(variables);
                    return true;
                });
    }

    /**
     * 恢复暂停中的工作流。
     *
     * @param variables 恢复时注入的变量
     */
    public void resume(Map<String, Object> variables) {
        resumeIfSuspended(variables);
    }

    /**
     * 在调用方持有实例锁且已确认暂停状态后执行恢复动作。
     *
     * @param variables 恢复时注入的变量
     */
    private void resumeSuspended(Map<String, Object> variables) {
        ChainState newState = updateStateSafely(state -> {
            if (variables != null) {
                state.getMemory().putAll(variables);
                return EnumSet.of(ChainStateField.MEMORY);
            } else {
                return null;
            }
        });

        notifyEvent(new ChainResumeEvent(this, variables));
        setStatusAndNotifyEvent(ChainStatus.RUNNING);

        Set<String> suspendNodeIds = newState.getSuspendNodeIds();
        if (suspendNodeIds != null && !suspendNodeIds.isEmpty()) {
            // 移除 suspend 状态，方便二次 suspend 时，不带有旧数据
             updateStateSafely(state -> {
                state.setSuspendNodeIds(null);
                state.setSuspendForParameters(null);
                return EnumSet.of(ChainStateField.SUSPEND_NODE_IDS, ChainStateField.SUSPEND_FOR_PARAMETERS);
            });

            for (String id : suspendNodeIds) {
                Node node = definition.getNodeById(id);
                if (node == null) {
                    throw new ChainException("Node not found: " + id);
                }
                scheduleNode(node, null, TriggerType.RESUME, 0L);
            }
        }
    }

    public void resume() {
        resume(Collections.emptyMap());
    }

    public void output(Node node, Object response) {
        eventManager.notifyOutput(this, node, response);
    }


    public EventManager getEventManager() {
        return eventManager;
    }

    public void setEventManager(EventManager eventManager) {
        this.eventManager = eventManager;
    }

    public ChainStateRepository getChainStateRepository() {
        return chainStateRepository;
    }

    public void setChainStateRepository(ChainStateRepository chainStateRepository) {
        this.chainStateRepository = chainStateRepository;
    }

    public ChainDefinition getDefinition() {
        return definition;
    }

    public TriggerScheduler getTriggerScheduler() {
        if (this.triggerScheduler == null) {
            this.triggerScheduler = ChainRuntime.triggerScheduler();
        }
        return triggerScheduler;
    }

    public void setTriggerScheduler(TriggerScheduler triggerScheduler) {
        this.triggerScheduler = triggerScheduler;
    }

    public NodeStateRepository getNodeStateRepository() {
        return nodeStateRepository;
    }

    public void setNodeStateRepository(NodeStateRepository nodeStateRepository) {
        this.nodeStateRepository = nodeStateRepository;
    }

    /**
     * 获取循环累计结果仓储。
     *
     * @return 循环累计结果仓储
     */
    public LoopResultRepository getLoopResultRepository() {
        return loopResultRepository;
    }

    /**
     * 设置循环累计结果仓储。
     *
     * @param loopResultRepository 循环累计结果仓储；为空时使用进程内实现
     */
    public void setLoopResultRepository(LoopResultRepository loopResultRepository) {
        this.loopResultRepository = loopResultRepository == null
                ? new InMemoryLoopResultRepository()
                : loopResultRepository;
    }

    public String getStateInstanceId() {
        return stateInstanceId;
    }

    /**
     * 获取当前工作流状态。
     *
     * @return 仓储中的最新工作流状态
     */
    public ChainState getState() {
        return loadFreshState();
    }

    /**
     * 获取当前节点执行入口捕获的状态视图。
     *
     * <p>节点业务逻辑和参数解析应使用该方法，保证一次节点执行内引用同一个
     * point-in-time 状态并消除重复仓储 I/O。若当前线程不在节点执行上下文中，
     * 则退化为读取仓储最新状态，不改变外部调用语义。</p>
     *
     * @return 当前节点执行状态视图，或仓储中的最新状态
     */
    public ChainState getExecutionState() {
        NodeExecutionStateView view =
                NODE_EXECUTION_STATE_VIEW.get();
        return view != null && view.chain == this
                ? view.state
                : loadFreshState();
    }

    /**
     * 获取节点审计归属的顶级执行实例 ID。
     *
     * <p>新状态直接读取持久字段。兼容升级前仅保存
     * {@link ChainState#getParentInstanceId()} 的状态时，首次按父链解析并回填，
     * 后续节点无需再次递归读取祖先状态。</p>
     *
     * @return 顶级审计实例 ID
     * @throws IllegalStateException 状态缺失、父链断裂或形成环时抛出
     */
    public String getAuditInstanceId() {
        ChainState current =
                getExecutionState();
        if (current == null) {
            throw new IllegalStateException(
                    "Chain state not found: "
                            + stateInstanceId);
        }
        if (StringUtil.hasText(
                current.getAuditInstanceId())) {
            return current.getAuditInstanceId();
        }

        Set<String> visited =
                new HashSet<>();
        visited.add(current.getInstanceId());
        String resolvedInstanceId =
                current.getInstanceId();
        while (StringUtil.hasText(
                current.getParentInstanceId())) {
            String parentInstanceId =
                    current.getParentInstanceId();
            if (!visited.add(parentInstanceId)) {
                throw new IllegalStateException(
                        "Workflow parent state cycle detected: "
                                + parentInstanceId);
            }
            current = chainStateRepository.load(
                    parentInstanceId);
            if (current == null) {
                throw new IllegalStateException(
                        "Workflow parent state not found: "
                                + parentInstanceId);
            }
            if (StringUtil.hasText(
                    current.getAuditInstanceId())) {
                resolvedInstanceId =
                        current.getAuditInstanceId();
                break;
            }
            resolvedInstanceId =
                    current.getInstanceId();
        }

        String rootInstanceId =
                resolvedInstanceId;
        ChainState updated = updateStateSafely(state -> {
            if (StringUtil.hasText(
                    state.getAuditInstanceId())) {
                return null;
            }
            state.setAuditInstanceId(
                    rootInstanceId);
            return EnumSet.of(
                    ChainStateField.AUDIT_INSTANCE_ID);
        });
        return updated.getAuditInstanceId();
    }

    /**
     * 获取指定实例的最新工作流状态。
     *
     * @param stateInstanceId 工作流实例 ID
     * @return 指定实例的最新状态
     */
    public ChainState getState(String stateInstanceId) {
        return chainStateRepository.load(stateInstanceId);
    }

    /**
     * 从仓储读取当前实例的最新状态。
     *
     * @return 当前实例的最新状态
     */
    private ChainState loadFreshState() {
        return chainStateRepository.load(stateInstanceId);
    }

    /**
     * 在当前触发器执行线程内更新可复用状态快照。
     *
     * @param instanceId 状态实例 ID
     * @param state 最新状态
     */
    private void cacheExecutionState(String instanceId, ChainState state) {
        if (EXECUTION_THREAD_LOCAL.get() == this
                && Objects.equals(stateInstanceId, instanceId)) {
            executionStateSnapshot = state;
            executionStateSnapshotLock =
                    currentInstanceLock(instanceId);
            NodeExecutionStateView view =
                    NODE_EXECUTION_STATE_VIEW.get();
            if (view != null && view.chain == this) {
                view.state = state;
            }
        }
    }

    /**
     * 绑定当前线程单次节点执行所使用的状态视图。
     *
     * @param state 节点入口的状态快照
     */
    private void bindNodeExecutionState(ChainState state) {
        NODE_EXECUTION_STATE_VIEW.set(
                new NodeExecutionStateView(this, state));
    }

    /**
     * 单次节点执行的线程隔离状态视图。
     */
    private static final class NodeExecutionStateView {
        private final Chain chain;
        private ChainState state;

        /**
         * 创建节点执行状态视图。
         *
         * @param chain 所属执行链
         * @param state 节点入口状态
         */
        private NodeExecutionStateView(
                Chain chain, ChainState state) {
            this.chain = chain;
            this.state = state;
        }
    }

    /**
     * 获取当前执行线程可安全尝试提交的状态快照。
     *
     * <p>仅复用由当前同一把实例锁读取或提交的快照。锁外读取和上一临界区快照均返回
     * {@code null}，避免并发取消、超时或变量更新被旧状态遮蔽。</p>
     *
     * @param instanceId 状态实例 ID
     * @return 当前执行快照；不可复用时为 {@code null}
     */
    private ChainState reusableExecutionState(String instanceId) {
        ChainLock currentLock =
                currentInstanceLock(instanceId);
        return EXECUTION_THREAD_LOCAL.get() == this
                && Objects.equals(
                stateInstanceId, instanceId)
                && currentLock != null
                && currentLock.isValid()
                && currentLock
                == executionStateSnapshotLock
                ? executionStateSnapshot
                : null;
    }

    public void setStateInstanceId(String stateInstanceId) {
        this.stateInstanceId = stateInstanceId;
    }

    /**
     * 获取当前实例的执行预算。
     *
     * @return 执行预算
     */
    public ExecutionBudget getExecutionBudget() {
        return executionBudget;
    }

    /**
     * 设置当前实例的执行预算。
     *
     * @param executionBudget 执行预算；为空时恢复宽松缺省值
     */
    public void setExecutionBudget(ExecutionBudget executionBudget) {
        this.executionBudget = executionBudget == null ? ExecutionBudget.defaults() : executionBudget;
    }
}
