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


import com.easyagents.flow.core.chain.*;
import com.easyagents.flow.core.chain.repository.ChainStateField;
import com.easyagents.flow.core.chain.repository.LoopInputReference;
import com.easyagents.flow.core.chain.repository.NodeStateField;
import com.easyagents.flow.core.chain.runtime.Trigger;
import com.easyagents.flow.core.chain.runtime.TriggerContext;
import com.easyagents.flow.core.chain.runtime.TriggerType;
import com.easyagents.flow.core.chain.runtime.ExecutionBudgetExceededException;
import com.easyagents.flow.core.util.Maps;
import com.easyagents.flow.core.util.StringUtil;

import java.io.Serializable;
import java.lang.reflect.Array;
import java.math.BigDecimal;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 支持纯次数、数组遍历和两者组合限制的显式循环节点。
 */
public class LoopNode extends BaseNode {
    private static final long serialVersionUID = 1L;

    private static final int DIRECT_INDEX_MAX_ITEMS = Math.max(
            1,
            Integer.getInteger(
                    "tinyflow.loop.direct-index.max-items", 64));

    /** 旧版单循环参数。 */
    private Parameter loopVar;
    /** 可选的循环次数参数。 */
    private Parameter loopCount;
    /** 可选的数组输入参数。 */
    private Parameter loopItems;

    /**
     * 获取旧版循环参数。
     *
     * @return 旧版循环参数
     */
    public Parameter getLoopVar() {
        return loopVar;
    }

    /**
     * 设置旧版循环参数。
     *
     * @param loopVar 旧版循环参数
     */
    public void setLoopVar(Parameter loopVar) {
        this.loopVar = loopVar;
    }

    /**
     * 获取显式循环次数参数。
     *
     * @return 循环次数参数
     */
    public Parameter getLoopCount() {
        return loopCount;
    }

    /**
     * 设置显式循环次数参数。
     *
     * @param loopCount 循环次数参数
     */
    public void setLoopCount(Parameter loopCount) {
        this.loopCount = loopCount;
    }

    /**
     * 获取待遍历数组参数。
     *
     * @return 数组参数
     */
    public Parameter getLoopItems() {
        return loopItems;
    }

    /**
     * 设置待遍历数组参数。
     *
     * @param loopItems 数组参数
     */
    public void setLoopItems(Parameter loopItems) {
        this.loopItems = loopItems;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Map<String, Object> execute(Chain chain) {
        MaterializationPlan[] planHolder = new MaterializationPlan[1];
        Map<String, Object> initialResult = chain.executeWithLock(
                chain.getStateInstanceId(),
                10,
                TimeUnit.SECONDS,
                () -> executeLocked(chain, false, planHolder));
        MaterializationPlan plan = planHolder[0];
        if (plan == null) {
            return initialResult;
        }

        int iterableSize = chain.storeLoopInputOutsideLock(
                plan.resultId,
                plan.items,
                materializationLimit(chain),
                plan.claimId,
                plan.claimGeneration);
        return chain.executeWithLock(
                chain.getStateInstanceId(),
                10,
                TimeUnit.SECONDS,
                () -> publishMaterializedInputAndContinue(
                        chain, plan, iterableSize));
    }

    /**
     * 在实例锁内推进一次循环状态机。
     *
     * @param chain 当前工作流
     * @param resumeAfterMaterialization 是否在本次调用中完成了锁外物化
     * @param planHolder 待锁外执行的物化计划容器
     * @return 节点执行结果
     */
    private Map<String, Object> executeLocked(
            Chain chain,
            boolean resumeAfterMaterialization,
            MaterializationPlan[] planHolder) {
        Trigger prevTrigger = TriggerContext.getCurrentTrigger();
        Deque<LoopContext> loopStack = getOrCreateLoopStack(chain);

        LoopContext loopContext;

        // 判断是否是首次进入该 LoopNode（即不是由子节点返回）
        TriggerType triggerType = prevTrigger.getType();
        boolean isFirstEntry = !resumeAfterMaterialization
                && triggerType != TriggerType.PARENT
                && triggerType != TriggerType.SELF;

        if (isFirstEntry) {
            // 首次触发：创建新的 LoopContext 并压入堆栈
            loopContext = new LoopContext();
            loopContext.currentIndex = 0;
            loopContext.resultId = chain.getStateInstanceId() + ":" + UUID.randomUUID();
            // 保存原始触发上下文（用于循环结束后恢复）
            loopStack.offerLast(loopContext);
            int nestedDepth = chain.getNestedDepthBase()
                    + (prevTrigger == null
                    ? 1
                    : prevTrigger.getLoopCursors().size() + 1);
            chain.getExecutionBudget().checkNestedDepth(this.id, nestedDepth);

            chain.updateNodeStateSafely(this.id, state -> {
                state.getMemory().put(buildLoopStackId(), loopStack);
                return EnumSet.of(NodeStateField.MEMORY);
            });

            if (loopStack.size() > 1) {
                // 不执行，等等其他节点唤起
                return Maps.of(ChainConsts.SCHEDULE_NEXT_NODE_DISABLED_KEY, true)
                        .set(ChainConsts.NODE_STATE_STATUS_KEY, NodeStatus.RUNNING);
            }

            // 首次进入循环时，先为循环体暴露空的累计输出，方便子节点在第一轮读取到“空值”而不是缺参异常。
            publishLoopProgress(chain, Collections.emptyMap());
        }
        // 由子节点返回：从堆栈低部获取当前循环上下文
        else {
            if (loopStack.isEmpty()) {
                throw new IllegalStateException("Loop stack is empty when returning from child node.");
            }
            loopContext = loopStack.peekFirst();
        }

        if (loopContext.materializingInput && !resumeAfterMaterialization) {
            String currentClaimId = chain.currentFencingClaimId();
            long currentGeneration = chain.currentClaimGeneration();
            if (Objects.equals(loopContext.materializationClaimId, currentClaimId)
                    && loopContext.materializationClaimGeneration == currentGeneration) {
                return waitingResult();
            }
            // 原 owner 已失去 claim，由新代际使用全新 resultId 接管。
            loopContext.resultId =
                    chain.getStateInstanceId() + ":" + UUID.randomUUID();
            loopContext.materializationClaimId = currentClaimId;
            loopContext.materializationClaimGeneration = currentGeneration;
            persistLoopStack(chain, loopStack);
        }

        migrateLegacyResult(chain, loopContext);
        if (!acceptParentBranch(prevTrigger, loopContext, chain, loopStack)) {
            return Maps.of(ChainConsts.SCHEDULE_NEXT_NODE_DISABLED_KEY, true)
                    .set(ChainConsts.NODE_STATE_STATUS_KEY, NodeStatus.RUNNING);
        }

        int shouldLoopCount;
        boolean storedIterable = false;
        boolean directlyIndexed = false;
        boolean numericLoop = false;
        Object loopValue = null;
        boolean usesExplicitInputs =
                loopCount != null || loopItems != null;
        Integer requestedCount = usesExplicitInputs
                ? resolveRequestedLoopCount(chain, loopContext)
                : null;
        if (loopContext.iterableInputStored) {
            // 已物化输入不再解析原始参数，避免大集合在每轮反序列化和路径解析。
            shouldLoopCount = planIterableIterations(
                    loopContext,
                    loopContext.iterableSize,
                    requestedCount);
            storedIterable = true;
        } else {
            LoopInputReference storedInput =
                    resolveStoredInputReference(chain);
            if (storedInput != null) {
                loopContext.resultId =
                        storedInput.getResultId();
                loopContext.iterableSize =
                        storedInput.getItemCount();
                loopContext.iterableInputStored = true;
                loopContext.inputExternalized = true;
                shouldLoopCount = planIterableIterations(
                        loopContext,
                        storedInput.getItemCount(),
                        requestedCount);
                storedIterable = true;
                persistLoopStack(chain, loopStack);
            } else if (loopItems != null) {
                loopValue = resolveParameterValue(
                        chain, loopItems, "loop items");
                Iterable<?> iterableInput =
                        toIterableInput(loopValue);
                if (iterableInput == null) {
                    throw invalidLoopItemsValue(loopValue);
                }
                int knownSize = knownInputSize(loopValue);
                if (knownSize >= 0) {
                    shouldLoopCount = planIterableIterations(
                            loopContext,
                            knownSize,
                            requestedCount);
                    checkExplicitLoopIterations(
                            chain, shouldLoopCount, true);
                } else {
                    shouldLoopCount = requestedCount == null
                            ? -1
                            : requestedCount;
                }

                if (knownSize >= 0
                        && shouldLoopCount <= DIRECT_INDEX_MAX_ITEMS) {
                    directlyIndexed = true;
                } else {
                    /*
                     * 有次数限制时只物化所需前缀。前缀快照仅属于当前循环，
                     * 不替换上游完整数组，避免影响其他下游节点。
                     */
                    Iterable<?> materializedInput =
                            requestedCount == null
                                    ? iterableInput
                                    : limitIterable(
                                            iterableInput,
                                            requestedCount);
                    boolean externalizeInput =
                            requestedCount == null
                                    || (knownSize >= 0
                                    && requestedCount >= knownSize);
                    loopContext.resultId =
                            chain.getStateInstanceId() + ":" + UUID.randomUUID();
                    loopContext.materializingInput = true;
                    loopContext.materializationClaimId =
                            chain.currentFencingClaimId();
                    loopContext.materializationClaimGeneration =
                            chain.currentClaimGeneration();
                    persistLoopStack(chain, loopStack);
                    if (planHolder == null) {
                        throw new IllegalStateException(
                                "Loop materialization plan holder is unavailable");
                    }
                    planHolder[0] = new MaterializationPlan(
                            loopContext.resultId,
                            materializedInput,
                            loopItems,
                            externalizeInput,
                            loopContext.materializationClaimId,
                            loopContext.materializationClaimGeneration);
                    return waitingResult();
                }
            } else if (loopCount != null) {
                shouldLoopCount = requestedCount;
                numericLoop = true;
                planNumericIterations(loopContext, shouldLoopCount);
            } else if (loopVar != null) {
                loopValue = resolveParameterValue(
                        chain, loopVar, "legacy loop value");
                Iterable<?> iterableInput =
                        toIterableInput(loopValue);
                if (iterableInput != null) {
                    int knownSize = knownInputSize(loopValue);
                    if (knownSize >= 0) {
                        checkExplicitLoopIterations(chain, knownSize, true);
                    }
                    if (knownSize >= 0
                            && knownSize <= DIRECT_INDEX_MAX_ITEMS) {
                        shouldLoopCount = knownSize;
                        directlyIndexed = true;
                    } else {
                        /*
                         * 每次未完成的物化尝试使用新结果 ID。失锁 owner 的旧分块仅会按 TTL
                         * 回收，接管者不会复用或删除其部分数据。
                         */
                        loopContext.resultId =
                                chain.getStateInstanceId() + ":" + UUID.randomUUID();
                        loopContext.materializingInput = true;
                        loopContext.materializationClaimId =
                                chain.currentFencingClaimId();
                        loopContext.materializationClaimGeneration =
                                chain.currentClaimGeneration();
                        persistLoopStack(chain, loopStack);
                        if (planHolder == null) {
                            throw new IllegalStateException(
                                    "Loop materialization plan holder is unavailable");
                        }
                        planHolder[0] = new MaterializationPlan(
                                loopContext.resultId,
                                iterableInput,
                                loopVar,
                                true,
                                loopContext.materializationClaimId,
                                loopContext.materializationClaimGeneration);
                        return waitingResult();
                    }
                } else if (loopValue instanceof Number
                        || loopValue instanceof String) {
                    shouldLoopCount = parseNumericLoopCount(loopValue);
                    numericLoop = true;
                } else {
                    throw invalidLoopValue(loopValue);
                }
            } else {
                throw new IllegalArgumentException(
                        "Loop node requires loop count or loop items");
            }
        }
        checkExplicitLoopIterations(chain, shouldLoopCount, !numericLoop);

        //  不是第一次执行，合并结果到 subResult
        if (loopContext.currentIndex != 0) {
            ChainState subState =
                    chain.getExecutionState();
            Map<String, Object> currentOutputs = collectCurrentOutputValues(subState);
            loopContext.accumulatedBytes += estimateBytes(currentOutputs, new IdentityHashMap<>());
            chain.getExecutionBudget().checkAccumulatedBytes(this.id, loopContext.accumulatedBytes);
            chain.appendLoopResult(
                    loopContext.resultId,
                    loopContext.currentIndex - 1,
                    currentOutputs);
            // 将上一轮最新输出同步到循环节点作用域，供下一轮循环体读取。
            publishLoopProgress(chain, currentOutputs);
        }


        // 执行的次数够了, 恢复父级触发
        if (loopContext.currentIndex >= shouldLoopCount) {
            loopStack.pollFirst();    // 移除最顶部部的 LoopContext
            chain.updateNodeStateSafely(this.id, state -> {
                ConcurrentHashMap<String, Object> memory = state.getMemory();
                memory.put(buildLoopStackId(), loopStack);
                memory.remove(this.id + ".index");
                memory.remove(this.id + ".loopItem");
                return EnumSet.of(NodeStateField.MEMORY);
            });
            if (!loopStack.isEmpty()) {
                chain.scheduleNode(this, null, TriggerType.SELF, 0);
            }
            if (prevTrigger != null) {
                prevTrigger.getLoopCursors().remove(this.id);
            }
            Map<String, Object> completedResult = chain.getLoopResultRepository().references(
                    loopContext.resultId,
                    loopContext.currentIndex,
                    getOutputNames(),
                    getFlattenedOutputNames());
            chain.getLoopResultRepository().releaseActiveCache(
                    loopContext.resultId);
            if (!loopContext.inputExternalized) {
                chain.removeLoopInput(loopContext.resultId);
            }
            return completedResult;
        }

        int loopIndex = loopContext.currentIndex;
        loopContext.currentIndex++;

        persistLoopStack(chain, loopStack);


        if (storedIterable) {
            Object loopItem = chain.getLoopResultRepository().loadInputItem(loopContext.resultId, loopIndex);
            executeLoopChain(chain, loopContext, loopItem);
        } else if (directlyIndexed) {
            executeLoopChain(
                    chain,
                    loopContext,
                    directInputItem(loopValue, loopIndex));
        } else if (numericLoop) {
            executeLoopChain(chain, loopContext, loopIndex);
        } else {
            throw invalidLoopValue(loopValue);
        }

        // 禁用调度下个节点
        return Maps.of(ChainConsts.SCHEDULE_NEXT_NODE_DISABLED_KEY, true)
                .set(ChainConsts.NODE_STATE_STATUS_KEY, NodeStatus.RUNNING);
    }

    /**
     * 在短实例锁内发布锁外物化结果并继续循环状态机。
     *
     * @param chain 当前工作流
     * @param plan 已完成的物化计划
     * @param iterableSize 物化元素数
     * @return 本轮循环执行结果
     */
    private Map<String, Object> publishMaterializedInputAndContinue(
            Chain chain, MaterializationPlan plan, int iterableSize) {
        if (!chain.isExecutionActiveNow()) {
            throw new com.easyagents.flow.core.chain.runtime.TriggerClaimLostException(
                    "loop-materialization-cancelled:" + plan.resultId);
        }
        Deque<LoopContext> loopStack = getOrCreateLoopStack(chain);
        LoopContext context = loopStack.peekFirst();
        if (context == null
                || !context.materializingInput
                || !Objects.equals(context.resultId, plan.resultId)
                || !Objects.equals(
                        context.materializationClaimId, plan.claimId)
                || context.materializationClaimGeneration
                        != plan.claimGeneration) {
            throw new com.easyagents.flow.core.chain.runtime.TriggerClaimLostException(
                    "loop-materialization:" + plan.resultId);
        }
        context.iterableSize = iterableSize;
        context.plannedIterations = iterableSize;
        context.plannedIterationsResolved = true;
        context.iterableInputStored = true;
        context.materializingInput = false;
        context.inputExternalized = plan.externalizeInput
                && externalizeMaterializedInput(
                        chain,
                        plan.sourceParameter,
                        plan.resultId,
                        iterableSize);
        persistLoopStack(chain, loopStack);
        return executeLocked(chain, true, null);
    }

    /**
     * 将热状态中的大型原始输入替换为轻量引用。
     *
     * <p>仅替换参数直接对应的扁平内存键；无法准确定位的嵌套路径保持原值，
     * 以业务兼容性优先。其他节点读取该引用时由仓储透明还原完整列表。</p>
     *
     * @param chain 当前工作流
     * @param resultId 已物化输入 ID
     * @param iterableSize 输入元素数
     * @return 已替换热状态值时为 {@code true}
     */
    private boolean externalizeMaterializedInput(
            Chain chain,
            Parameter sourceParameter,
            String resultId,
            int iterableSize) {
        String ref = sourceParameter == null
                ? null
                : sourceParameter.getRef();
        String name = sourceParameter == null
                ? null
                : sourceParameter.getName();
        AtomicBoolean replaced = new AtomicBoolean();
        chain.updateStateSafely(state -> {
            ConcurrentHashMap<String, Object> memory =
                    state.getMemory();
            String key = StringUtil.hasText(ref)
                    && memory.containsKey(ref)
                    ? ref
                    : (StringUtil.hasText(name)
                    && memory.containsKey(name)
                    ? name
                    : null);
            if (key == null) {
                return null;
            }
            Object current = memory.get(key);
            if (current instanceof LoopInputReference) {
                replaced.set(true);
                return null;
            }
            memory.put(
                    key,
                    new LoopInputReference(
                            resultId, iterableSize));
            replaced.set(true);
            return EnumSet.of(ChainStateField.MEMORY);
        });
        return replaced.get();
    }

    /**
     * 直接识别上游已分页物化的输入引用，避免参数解析边界先还原完整列表。
     *
     * @param chain 当前工作流
     * @return 已物化输入引用；不存在时为 {@code null}
     */
    private LoopInputReference resolveStoredInputReference(
            Chain chain) {
        Parameter sourceParameter =
                loopItems != null ? loopItems : loopVar;
        if (sourceParameter == null) {
            return null;
        }
        Map<String, Object> memory =
                chain.getExecutionState().getMemory();
        String ref = sourceParameter.getRef();
        if (StringUtil.hasText(ref)
                && memory.get(ref)
                instanceof LoopInputReference) {
            return (LoopInputReference) memory.get(ref);
        }
        String name = sourceParameter.getName();
        return StringUtil.hasText(name)
                && memory.get(name)
                instanceof LoopInputReference
                ? (LoopInputReference) memory.get(name)
                : null;
    }

    /**
     * 构造不推进下游的循环等待结果。
     *
     * @return 运行态控制结果
     */
    private Map<String, Object> waitingResult() {
        return Maps.of(ChainConsts.SCHEDULE_NEXT_NODE_DISABLED_KEY, true)
                .set(ChainConsts.NODE_STATE_STATUS_KEY, NodeStatus.RUNNING);
    }

    /**
     * 解析新结构中的循环次数，并在当前循环上下文中固定结果。
     *
     * @param chain 当前工作流
     * @param context 当前循环上下文
     * @return 已配置次数；未配置时返回 {@code null}
     */
    private Integer resolveRequestedLoopCount(
            Chain chain, LoopContext context) {
        if (loopCount == null) {
            return null;
        }
        if (context.requestedCountResolved) {
            return context.requestedCount;
        }
        int count = parseNumericLoopCount(
                resolveParameterValue(
                        chain, loopCount, "loop count"));
        context.requestedCount = count;
        context.requestedCountResolved = true;
        return count;
    }

    /**
     * 解析单个循环参数。
     *
     * @param chain 当前工作流
     * @param parameter 待解析参数
     * @param parameterLabel 错误提示中的参数名称
     * @return 参数运行值
     */
    private Object resolveParameterValue(
            Chain chain,
            Parameter parameter,
            String parameterLabel) {
        if (parameter == null
                || !StringUtil.hasText(parameter.getName())) {
            throw new IllegalArgumentException(
                    parameterLabel + " parameter is missing");
        }
        Map<String, Object> values =
                chain.getExecutionState().resolveParameters(
                        this,
                        Collections.singletonList(parameter));
        return values.get(parameter.getName());
    }

    /**
     * 固定数组循环的实际迭代次数。
     *
     * @param context 当前循环上下文
     * @param iterableSize 数组实际元素数
     * @param requestedCount 可选的最大处理数量
     * @return 本轮实际迭代次数
     */
    private int planIterableIterations(
            LoopContext context,
            int iterableSize,
            Integer requestedCount) {
        if (context.plannedIterationsResolved) {
            return context.plannedIterations;
        }
        int iterations = requestedCount == null
                ? iterableSize
                : Math.min(requestedCount, iterableSize);
        context.plannedIterations = iterations;
        context.plannedIterationsResolved = true;
        return iterations;
    }

    /**
     * 固定纯次数循环的实际迭代次数。
     *
     * @param context 当前循环上下文
     * @param iterations 迭代次数
     */
    private void planNumericIterations(
            LoopContext context, int iterations) {
        if (!context.plannedIterationsResolved) {
            context.plannedIterations = iterations;
            context.plannedIterationsResolved = true;
        }
    }

    /**
     * 将可迭代输入限制为最多读取指定数量元素。
     *
     * @param source 原始可迭代输入
     * @param limit 最大元素数
     * @return 单次消费的有界可迭代输入
     */
    private Iterable<?> limitIterable(
            Iterable<?> source, int limit) {
        return () -> new Iterator<Object>() {
            private final Iterator<?> delegate = source.iterator();
            private int remaining = limit;

            @Override
            public boolean hasNext() {
                return remaining > 0 && delegate.hasNext();
            }

            @Override
            public Object next() {
                if (!hasNext()) {
                    throw new NoSuchElementException();
                }
                remaining--;
                return delegate.next();
            }
        };
    }

    /**
     * 将集合或数组统一转换为单次消费的 Iterable。
     *
     * @param loopValue 循环输入
     * @return 可迭代输入；数值循环返回 {@code null}
     */
    private Iterable<?> toIterableInput(Object loopValue) {
        if (loopValue instanceof Iterable<?>) {
            return (Iterable<?>) loopValue;
        }
        if (loopValue != null && loopValue.getClass().isArray()) {
            int length = Array.getLength(loopValue);
            return () -> new Iterator<Object>() {
                private int index;

                @Override
                public boolean hasNext() {
                    return index < length;
                }

                @Override
                public Object next() {
                    if (!hasNext()) {
                        throw new NoSuchElementException();
                    }
                    return Array.get(loopValue, index++);
                }
            };
        }
        return null;
    }

    /**
     * 获取无需遍历即可确定的输入元素数量。
     *
     * @param loopValue 循环输入
     * @return 已知数量；未知时返回 {@code -1}
     */
    private int knownInputSize(Object loopValue) {
        if (loopValue instanceof Collection<?>) {
            return ((Collection<?>) loopValue).size();
        }
        if (loopValue != null && loopValue.getClass().isArray()) {
            return Array.getLength(loopValue);
        }
        return -1;
    }

    /**
     * 解析数值型循环次数，循环索引从 0 开始并执行到 {@code count - 1}。
     *
     * @param loopValue 数值或数值字符串
     * @return 1～300 的循环总次数
     * @throws IllegalArgumentException 输入不是范围内整数
     */
    private int parseNumericLoopCount(Object loopValue) {
        final int count;
        try {
            count = new BigDecimal(String.valueOf(loopValue).trim())
                    .intValueExact();
        } catch (ArithmeticException | NumberFormatException exception) {
            throw new IllegalArgumentException(
                    "Loop count must be an integer between "
                            + Node.MIN_LOOP_COUNT
                            + " and "
                            + Node.MAX_LOOP_COUNT,
                    exception);
        }
        if (count < Node.MIN_LOOP_COUNT || count > Node.MAX_LOOP_COUNT) {
            throw new IllegalArgumentException(
                    "Loop count must be between "
                            + Node.MIN_LOOP_COUNT
                            + " and "
                            + Node.MAX_LOOP_COUNT
                            + ", but was "
                            + count);
        }
        return count;
    }

    /**
     * 校验显式循环节点的本层迭代次数。
     *
     * @param chain 当前工作流
     * @param iterations 本层计划迭代次数
     * @param allowEmpty 是否允许空集合产生零次迭代
     * @throws IllegalArgumentException 不允许的负数或零次数
     * @throws ExecutionBudgetExceededException 超过 300 次或部署预算
     */
    private void checkExplicitLoopIterations(
            Chain chain, long iterations, boolean allowEmpty) {
        if (iterations < 0L || (!allowEmpty && iterations == 0L)) {
            throw new IllegalArgumentException(
                    "Loop count must be at least " + Node.MIN_LOOP_COUNT);
        }
        if (iterations > Node.MAX_LOOP_COUNT) {
            throw new ExecutionBudgetExceededException(
                    "Loop iteration limit exceeded for node "
                            + this.id
                            + ": "
                            + iterations
                            + " > "
                            + Node.MAX_LOOP_COUNT);
        }
        chain.getExecutionBudget().checkIterations(this.id, iterations);
    }

    /**
     * 计算未知 Iterable 的物化上限，同时服从平台预算和 300 次硬上限。
     *
     * @param chain 当前工作流
     * @return 正数物化上限
     */
    private long materializationLimit(Chain chain) {
        long budgetLimit = chain.getExecutionBudget().getMaxIterations();
        if (budgetLimit <= 0L) {
            return Node.MAX_LOOP_COUNT;
        }
        return Math.min(budgetLimit, Node.MAX_LOOP_COUNT);
    }

    /**
     * 按序号读取小型集合或数组，避免额外 Redis 分块读写。
     *
     * @param loopValue 小型循环输入
     * @param index 元素序号
     * @return 对应元素
     */
    private Object directInputItem(Object loopValue, int index) {
        if (loopValue instanceof List<?>) {
            return ((List<?>) loopValue).get(index);
        }
        if (loopValue != null && loopValue.getClass().isArray()) {
            return Array.get(loopValue, index);
        }
        if (loopValue instanceof Collection<?>) {
            Iterator<?> iterator =
                    ((Collection<?>) loopValue).iterator();
            for (int current = 0; current < index; current++) {
                iterator.next();
            }
            return iterator.next();
        }
        throw invalidLoopValue(loopValue);
    }

    /**
     * 创建循环输入类型错误。
     *
     * @param loopValue 非法输入
     * @return 参数异常
     */
    private IllegalArgumentException invalidLoopValue(Object loopValue) {
        return new IllegalArgumentException(
                "loopValue must be Iterable, array, Number or String, but loopValue is \""
                        + loopValue
                        + "\"");
    }

    /**
     * 创建输入数组类型错误。
     *
     * @param loopValue 非法数组输入
     * @return 参数异常
     */
    private IllegalArgumentException invalidLoopItemsValue(
            Object loopValue) {
        String actualType = loopValue == null
                ? "null"
                : loopValue.getClass().getName();
        return new IllegalArgumentException(
                "Loop items must resolve to Iterable or array, but actual type is "
                        + actualType);
    }


    /**
     * 获取或创建当前节点的 LoopContext 堆栈（每个 LoopNode 实例独立）
     */
    @SuppressWarnings("unchecked")
    private Deque<LoopContext> getOrCreateLoopStack(Chain chain) {
        NodeState nodeState = chain.getNodeState(this.id);
        String key = buildLoopStackId();
        Object stackObj = nodeState.getMemory().get(key);
        Deque<LoopContext> stack;
        if (stackObj instanceof Deque) {
            stack = (Deque<LoopContext>) stackObj;
        } else {
            stack = new ArrayDeque<>();
        }
        return stack;
    }

    private void executeLoopChain(Chain chain, LoopContext loopContext, Object loopItem) {

        chain.updateStateSafely(state -> {
            ConcurrentHashMap<String, Object> memory = state.getMemory();
            memory.put(this.id + ".index", (loopContext.currentIndex - 1));
            memory.put(this.id + ".loopItem", loopItem);
            return EnumSet.of(ChainStateField.MEMORY);
        });


        ChainDefinition definition = chain.getDefinition();
        List<ChainDefinition.LoopChildDispatch> childDispatches =
                definition.getLoopChildDispatches(this.id);
        if (childDispatches.isEmpty()) {
            throw new IllegalStateException("Loop node has no executable child branch: " + this.id);
        }
        loopContext.expectedReturnCount = childDispatches.size();
        loopContext.getCompletedBranchIds().clear();
        for (ChainDefinition.LoopChildDispatch dispatch :
                childDispatches) {
            Trigger.LoopCursor cursor = new Trigger.LoopCursor(
                    loopContext.resultId,
                    loopContext.currentIndex - 1,
                    dispatch.getBranchId());
            chain.scheduleLoopChild(
                    dispatch.getNode(),
                    dispatch.getEdgeId(),
                    0,
                    this.id,
                    cursor);
        }
    }

    /**
     * 校验父分支回调的代际和分支屏障。
     *
     * @param trigger 当前触发器
     * @param context 当前循环上下文
     * @param chain 当前工作流
     * @param loopStack 循环上下文栈
     * @return 当前回调是否为本轮最后一个有效分支
     */
    private boolean acceptParentBranch(Trigger trigger,
                                       LoopContext context,
                                       Chain chain,
                                       Deque<LoopContext> loopStack) {
        if (trigger == null || trigger.getType() != TriggerType.PARENT) {
            return true;
        }
        Trigger.LoopCursor cursor = trigger.getLoopCursors().get(this.id);
        if (cursor == null) {
            return true;
        }
        if (!Objects.equals(context.resultId, cursor.getResultId())) {
            return false;
        }
        int expectedIndex = context.currentIndex - 1;
        if (cursor.getIterationIndex() < expectedIndex) {
            return false;
        }
        if (cursor.getIterationIndex() > expectedIndex) {
            throw new IllegalStateException("Future loop callback index: " + cursor.getIterationIndex());
        }
        if (!context.getCompletedBranchIds().add(cursor.getBranchId())) {
            return false;
        }
        persistLoopStack(chain, loopStack);
        if (context.getCompletedBranchIds().size() < Math.max(1, context.expectedReturnCount)) {
            return false;
        }
        context.getCompletedBranchIds().clear();
        return true;
    }

    /**
     * 将旧版本内嵌累计结果惰性迁移到分块仓储。
     *
     * @param chain 当前工作流
     * @param context 循环上下文
     */
    @SuppressWarnings("unchecked")
    private void migrateLegacyResult(Chain chain, LoopContext context) {
        if (context.resultId != null) {
            return;
        }
        context.resultId = chain.getStateInstanceId() + ":" + UUID.randomUUID();
        if (context.subResult != null && !context.subResult.isEmpty()) {
            int migratedCount = 0;
            for (Object value : context.subResult.values()) {
                if (value instanceof List) {
                    migratedCount = Math.max(migratedCount, ((List<?>) value).size());
                }
            }
            for (int index = 0; index < migratedCount; index++) {
                Map<String, Object> outputs = new LinkedHashMap<>();
                for (Map.Entry<String, Object> entry : context.subResult.entrySet()) {
                    if (entry.getValue() instanceof List
                            && index < ((List<?>) entry.getValue()).size()) {
                        outputs.put(entry.getKey(), ((List<Object>) entry.getValue()).get(index));
                    }
                }
                chain.appendLoopResult(
                        context.resultId,
                        index,
                        outputs);
            }
        }
        context.subResult = null;
    }

    /**
     * 保存紧凑循环上下文。
     *
     * @param chain 当前工作流
     * @param loopStack 循环上下文栈
     */
    private void persistLoopStack(Chain chain, Deque<LoopContext> loopStack) {
        chain.updateNodeStateSafely(this.id, state -> {
            state.getMemory().put(buildLoopStackId(), loopStack);
            return EnumSet.of(NodeStateField.MEMORY);
        });
    }


    /**
     * 把子流程执行的结果填充到主流程的输出参数中
     *
     * @param toResult 主流程的输出参数
     * @param subState 子流程的
     */
    private Map<String, Object> collectCurrentOutputValues(ChainState subState) {
        Map<String, Object> currentOutputs = new LinkedHashMap<>();
        List<Parameter> outputDefs = getOutputDefs();
        if (outputDefs != null) {
            for (Parameter outputDef : outputDefs) {
                Object value = null;

                //引用
                if (outputDef.getRefType() == RefType.REF) {
                    value = subState.resolveValue(outputDef.getRef());
                }
                //固定值
                else if (outputDef.getRefType() == RefType.FIXED) {
                    value = outputDef.getValue();
                }

                currentOutputs.put(outputDef.getName(), value);
            }
        }
        return currentOutputs;
    }


    /**
     * 获取循环输出名称并保持定义顺序。
     *
     * @return 输出名称列表
     */
    private List<String> getOutputNames() {
        List<String> outputNames = new ArrayList<>();
        List<Parameter> outputDefs = getOutputDefs();
        if (outputDefs != null) {
            for (Parameter outputDef : outputDefs) {
                outputNames.add(outputDef.getName());
            }
        }
        return outputNames;
    }

    /**
     * 获取启用一层扁平聚合的输出名称。
     *
     * @return 保持定义顺序的输出名称集合
     */
    private Set<String> getFlattenedOutputNames() {
        Set<String> outputNames = new LinkedHashSet<>();
        List<Parameter> outputDefs = getOutputDefs();
        if (outputDefs != null) {
            for (Parameter outputDef : outputDefs) {
                if (outputDef.isFlattenAggregation()) {
                    outputNames.add(outputDef.getName());
                }
            }
        }
        return outputNames;
    }

    /**
     * 估算本轮输出占用字节数，用于宽松的累计结果失控保护。
     *
     * @param value   待估算值
     * @param visited 已访问对象集合，防止循环引用
     * @return 估算字节数
     */
    private long estimateBytes(Object value, IdentityHashMap<Object, Boolean> visited) {
        if (value == null) {
            return 0L;
        }
        if (value instanceof String) {
            return (long) ((String) 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 = 0L;
        if (value instanceof Map) {
            for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet()) {
                bytes += estimateBytes(entry.getKey(), visited);
                bytes += estimateBytes(entry.getValue(), visited);
            }
        } else if (value instanceof Collection) {
            for (Object item : (Collection<?>) value) {
                bytes += estimateBytes(item, visited);
            }
        } else if (value instanceof Iterable) {
            bytes = 64L;
        } else if (value.getClass().isArray()) {
            int length = Array.getLength(value);
            for (int index = 0; index < length; index++) {
                bytes += estimateBytes(Array.get(value, index), visited);
            }
        } else {
            bytes = 64L;
        }
        return bytes;
    }


    /**
     * 将循环当前累计值同步到父循环节点作用域。
     * <p>
     * 循环体内读取 `loopNodeId.outputName` 时，拿到的是上一轮该输出的最新值；
     * 循环结束后，handleNodeResult 会再把最终聚合结果（列表）覆盖回同名作用域。
     */
    private void publishLoopProgress(Chain chain, Map<String, Object> currentOutputs) {
        List<Parameter> outputDefs = getOutputDefs();
        if (outputDefs == null || outputDefs.isEmpty()) {
            return;
        }

        chain.updateStateSafely(state -> {
            ConcurrentHashMap<String, Object> memory = state.getMemory();
            for (Parameter outputDef : outputDefs) {
                String key = this.id + "." + outputDef.getName();
                Object value = currentOutputs.get(outputDef.getName());
                memory.put(key, value == null ? "" : value);
            }
            return EnumSet.of(ChainStateField.MEMORY);
        });
    }


    private String buildLoopStackId() {
        return this.getId() + "__loop__context";
    }


    public static class LoopContext implements Serializable {
        private static final long serialVersionUID = 5258356831772776243L;

        int currentIndex;
        String resultId;
        Map<String, Object> subResult;
        boolean iterableInputStored;
        boolean inputExternalized;
        boolean materializingInput;
        String materializationClaimId;
        long materializationClaimGeneration;
        int iterableSize;
        long accumulatedBytes;
        /** 是否已固定显式次数输入。 */
        boolean requestedCountResolved;
        /** 当前循环固定后的次数输入。 */
        int requestedCount;
        /** 是否已固定实际迭代次数。 */
        boolean plannedIterationsResolved;
        /** 当前循环固定后的实际迭代次数。 */
        int plannedIterations;
        int expectedReturnCount = 1;
        Set<String> completedBranchIds = new LinkedHashSet<>();

        public int getCurrentIndex() {
            return currentIndex;
        }

        public void setCurrentIndex(int currentIndex) {
            this.currentIndex = currentIndex;
        }

        public String getResultId() {
            return resultId;
        }

        public void setResultId(String resultId) {
            this.resultId = resultId;
        }

        public Map<String, Object> getSubResult() {
            return subResult;
        }

        public void setSubResult(Map<String, Object> subResult) {
            this.subResult = subResult;
        }

        public long getAccumulatedBytes() {
            return accumulatedBytes;
        }

        public void setAccumulatedBytes(long accumulatedBytes) {
            this.accumulatedBytes = accumulatedBytes;
        }

        public boolean isIterableInputStored() {
            return iterableInputStored;
        }

        public void setIterableInputStored(boolean iterableInputStored) {
            this.iterableInputStored = iterableInputStored;
        }

        /**
         * @return 原始输入是否已从热状态替换为轻量引用
         */
        public boolean isInputExternalized() {
            return inputExternalized;
        }

        /**
         * 设置原始输入外置状态。
         *
         * @param inputExternalized 是否已替换为轻量引用
         */
        public void setInputExternalized(
                boolean inputExternalized) {
            this.inputExternalized = inputExternalized;
        }

        /**
         * @return 是否正在锁外物化输入
         */
        public boolean isMaterializingInput() {
            return materializingInput;
        }

        /**
         * 设置锁外物化状态。
         *
         * @param materializingInput 是否正在物化
         */
        public void setMaterializingInput(boolean materializingInput) {
            this.materializingInput = materializingInput;
        }

        public int getIterableSize() {
            return iterableSize;
        }

        public void setIterableSize(int iterableSize) {
            this.iterableSize = iterableSize;
        }

        /**
         * 获取当前循环固定后的实际迭代次数。
         *
         * @return 实际迭代次数；尚未规划时返回 {@code null}
         */
        public Integer getPlannedIterations() {
            return plannedIterationsResolved
                    ? plannedIterations
                    : null;
        }

        public int getExpectedReturnCount() {
            return expectedReturnCount;
        }

        public void setExpectedReturnCount(int expectedReturnCount) {
            this.expectedReturnCount = expectedReturnCount;
        }

        public Set<String> getCompletedBranchIds() {
            if (completedBranchIds == null) {
                completedBranchIds = new LinkedHashSet<>();
            }
            return completedBranchIds;
        }

        public void setCompletedBranchIds(Set<String> completedBranchIds) {
            this.completedBranchIds = completedBranchIds;
        }

    }

    /**
     * 一次锁外循环输入物化所需的不可变守卫快照。
     */
    private static final class MaterializationPlan {

        private final String resultId;
        private final Iterable<?> items;
        private final Parameter sourceParameter;
        private final boolean externalizeInput;
        private final String claimId;
        private final long claimGeneration;

        /**
         * 创建物化计划。
         *
         * @param resultId 唯一结果 ID
         * @param items 输入元素
         * @param sourceParameter 输入来源参数
         * @param externalizeInput 是否允许替换上游完整输入
         * @param claimId 触发器 ID
         * @param claimGeneration 触发器代际
         */
        private MaterializationPlan(
                String resultId,
                Iterable<?> items,
                Parameter sourceParameter,
                boolean externalizeInput,
                String claimId,
                long claimGeneration) {
            this.resultId = resultId;
            this.items = items;
            this.sourceParameter = sourceParameter;
            this.externalizeInput = externalizeInput;
            this.claimId = claimId;
            this.claimGeneration = claimGeneration;
        }
    }
}
