diff --git a/agent-flow/src/common/Consts.js b/agent-flow/src/common/Consts.js index d929a08d3a..22369cda70 100644 --- a/agent-flow/src/common/Consts.js +++ b/agent-flow/src/common/Consts.js @@ -10,6 +10,7 @@ export const NODE_STATUS = { RUNNING: 'RUNNING', ERROR: 'ERROR', SUCCESS: 'ARCHIVED', + SKIPPED: 'SKIPPED', DEFAULT: 'DEFAULT', UN_RUNNING: 'UN_RUNNING', TERMINATED: 'TERMINATED', @@ -287,4 +288,4 @@ export const DEFAULT_KNOWLEDGE_RETRIEVAL_NODE_EXTENSIONS = { referenceNode: VIRTUAL_CONTEXT_NODE.id, value: [VIRTUAL_CONTEXT_NODE_VARIABLES.APP_CREATE_BY], }], -}; \ No newline at end of file +}; diff --git a/agent-flow/src/components/flowRunComponent/RunningStatusPanel.jsx b/agent-flow/src/components/flowRunComponent/RunningStatusPanel.jsx index 4efd23b905..45141ba038 100644 --- a/agent-flow/src/components/flowRunComponent/RunningStatusPanel.jsx +++ b/agent-flow/src/components/flowRunComponent/RunningStatusPanel.jsx @@ -143,9 +143,16 @@ SvgComponent.propTypes = { }; const TimeComponent = ({shape, runStatus}) => { + const statusClassMap = { + [NODE_STATUS.SUCCESS]: 'success', + [NODE_STATUS.ERROR]: 'failed', + [NODE_STATUS.RUNNING]: 'running', + [NODE_STATUS.SKIPPED]: 'skipped', + }; + const statusClass = statusClassMap[runStatus]; return (<> -
-
+
+
{shape.cost ? (shape.cost / 1000) + "s" : "0.000s"}
@@ -182,6 +189,17 @@ const getStatus = (nodeStatus) => { return }/>; }, }; + case NODE_STATUS.SKIPPED: + return { + title: t('skipped'), + enableReport: true, + getTime: (shape) => { + return ; + }, + getIcon: () => { + return }/>; + }, + }; case NODE_STATUS.RUNNING: return { title: t('running'), @@ -222,4 +240,4 @@ const getStatus = (nodeStatus) => { } }; -export default RunningStatusPanel; \ No newline at end of file +export default RunningStatusPanel; diff --git a/agent-flow/src/components/flowRunComponent/style.css b/agent-flow/src/components/flowRunComponent/style.css index ffe7b7c01a..06a1ba49c9 100644 --- a/agent-flow/src/components/flowRunComponent/style.css +++ b/agent-flow/src/components/flowRunComponent/style.css @@ -31,6 +31,10 @@ background-color: rgb(198, 57, 57, 0.2); } +.time-text-container-skipped { + background-color: rgba(107, 109, 117, 0.16); +} + .time-text { font-family: SF Pro Display, -apple-system, BlinkMacSystemFont, Segoe Ul, PingFang SC, Hiragino Sans GB, Microsoft YaHei, Helvetica Neue, Helvetica, Arial, sans-serif; font-size: 12px; @@ -52,6 +56,10 @@ color: rgb(198, 57, 57); } +.time-text-skipped { + color: rgb(107, 109, 117); +} + .button-text { color: rgb(77, 77, 77); font-family: SF Pro Display, -apple-system, BlinkMacSystemFont, Segoe Ul, PingFang SC, Hiragino Sans GB, Microsoft YaHei, Helvetica Neue, Helvetica, Arial, sans-serif; @@ -213,4 +221,4 @@ position: absolute; border-radius: 16px; background: rgba(197, 197, 197, 0.13); -} \ No newline at end of file +} diff --git a/agent-flow/src/flow/runners.js b/agent-flow/src/flow/runners.js index 34b8dfc145..eff75a467e 100644 --- a/agent-flow/src/flow/runners.js +++ b/agent-flow/src/flow/runners.js @@ -75,7 +75,7 @@ export const standardRunner = (node) => { return false; } const data = dataList.find(d => d.nodeId === preNode.id); - return data && data.status === NODE_STATUS.SUCCESS; + return data && (data.status === NODE_STATUS.SUCCESS || data.status === NODE_STATUS.SKIPPED); }; return self; @@ -174,4 +174,4 @@ export const inactiveNodeRunner = (node) => { }; return self; -}; \ No newline at end of file +}; diff --git a/agent-flow/src/i18n/en_US.json b/agent-flow/src/i18n/en_US.json index 30b0717997..62d20a917c 100644 --- a/agent-flow/src/i18n/en_US.json +++ b/agent-flow/src/i18n/en_US.json @@ -123,6 +123,7 @@ "running": "Running", "notRunning": "Not running", "runFailed": "Failed", + "skipped": "Skipped", "runResult": "Running Result", "promptName": "Prompt Word", "promptTextarea": "Use {{Variable name}} to add a variable", diff --git a/agent-flow/src/i18n/zh_CN.json b/agent-flow/src/i18n/zh_CN.json index 4e35a09a38..772203ac55 100644 --- a/agent-flow/src/i18n/zh_CN.json +++ b/agent-flow/src/i18n/zh_CN.json @@ -423,6 +423,7 @@ "running": "运行中", "notRunning": "未运行", "runFailed": "运行失败", + "skipped": "已跳过", "runResult": "运行结果", "closeRunResult": "收起运行结果", "invalidCategory": "存在不合法的类目，请先修改", diff --git a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/domain/EndNodeStatus.java b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/domain/EndNodeStatus.java new file mode 100644 index 0000000000..fbd38cbc71 --- /dev/null +++ b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/domain/EndNodeStatus.java @@ -0,0 +1,40 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved. + * This file is a part of the ModelEngine Project. + * Licensed under the MIT License. See License.txt in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package modelengine.fit.jober.aipp.domain; + +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.experimental.SuperBuilder; + +/** + * END节点状态领域对象. + * + * @author 张越 + * @since 2026-04-01 + */ +@EqualsAndHashCode(callSuper = true) +@Data +@SuperBuilder +public class EndNodeStatus extends BaseDomain { + private Long id; + private String traceId; + private String endNodeId; + private String status; + private long startTime; + private long endTime; + private String instanceId; + private String flowDefinitionId; + + /** + * 获取节点执行时间. + * + * @return 执行时间（毫秒）. + */ + public long getExecutionCost() { + return this.endTime - this.startTime; + } +} diff --git a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/fitable/AippFlowEndCallback.java b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/fitable/AippFlowEndCallback.java index 5814971f02..a2ee3a43b4 100644 --- a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/fitable/AippFlowEndCallback.java +++ b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/fitable/AippFlowEndCallback.java @@ -14,23 +14,30 @@ import modelengine.fit.jane.common.entity.OperationContext; import modelengine.fit.jober.aipp.constants.AippConst; import modelengine.fit.jober.aipp.domain.AppBuilderForm; +import modelengine.fit.jober.aipp.domain.AppBuilderFlowGraph; +import modelengine.fit.jober.aipp.domain.EndNodeStatus; import modelengine.fit.jober.aipp.domains.appversion.AppVersion; import modelengine.fit.jober.aipp.domains.appversion.service.AppVersionService; import modelengine.fit.jober.aipp.domains.business.RunContext; import modelengine.fit.jober.aipp.domains.task.AppTask; import modelengine.fit.jober.aipp.domains.task.service.AppTaskService; import modelengine.fit.jober.aipp.domains.taskinstance.AppTaskInstance; +import modelengine.fit.jober.aipp.domains.taskinstance.TaskInstanceUpdateEntity; import modelengine.fit.jober.aipp.domains.taskinstance.service.AppTaskInstanceService; import modelengine.fit.jober.aipp.dto.chat.AppChatRsp; +import modelengine.fit.jober.aipp.enums.NodeType; import modelengine.fit.jober.aipp.entity.AippFlowData; import modelengine.fit.jober.aipp.entity.AippLogData; import modelengine.fit.jober.aipp.enums.AippInstLogType; import modelengine.fit.jober.aipp.enums.MetaInstStatusEnum; import modelengine.fit.jober.aipp.events.InsertConversationEnd; import modelengine.fit.jober.aipp.genericable.AppFlowFinishObserver; +import modelengine.fit.jober.aipp.repository.AppBuilderFlowGraphRepository; +import modelengine.fit.jober.aipp.repository.EndNodeStatusRepository; import modelengine.fit.jober.aipp.service.AippLogService; import modelengine.fit.jober.aipp.service.AppBuilderFormService; import modelengine.fit.jober.aipp.service.AppChatSseService; +import modelengine.fit.jober.aipp.util.ConvertUtils; import modelengine.fit.jober.aipp.util.DataUtils; import modelengine.fit.jober.aipp.util.FormUtils; import modelengine.fit.jober.aipp.util.JsonUtils; @@ -38,6 +45,7 @@ import modelengine.fit.jober.common.exceptions.JobberException; import modelengine.fit.waterflow.domain.enums.FlowTraceStatus; import modelengine.fit.waterflow.spi.FlowCallbackService; +import modelengine.fit.waterflow.spi.lock.DistributedLockProvider; import modelengine.fitframework.annotation.Component; import modelengine.fitframework.annotation.Fit; import modelengine.fitframework.annotation.Fitable; @@ -55,6 +63,9 @@ import modelengine.jade.app.engine.metrics.po.ConversationRecordPo; import modelengine.jade.app.engine.metrics.service.ConversationRecordService; +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; + import java.time.Duration; import java.time.LocalDateTime; import java.time.temporal.ChronoUnit; @@ -62,7 +73,11 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; +import java.util.Set; +import java.util.concurrent.locks.Lock; +import java.util.stream.Collectors; /** * 流程结束回调节点 @@ -79,6 +94,7 @@ public class AippFlowEndCallback implements FlowCallbackService { .put(Constant.DEFAULT, AippInstLogType.MSG) .put(Constant.LLM_OUTPUT, AippInstLogType.META_MSG) .build(); + private static final String TERMINATE_LOCK_PREFIX = "aipp-terminate-lock:"; private final AippLogService aippLogService; private final BrokerClient brokerClient; @@ -91,12 +107,18 @@ public class AippFlowEndCallback implements FlowCallbackService { private final AppTaskInstanceService appTaskInstanceService; private final AppTaskService appTaskService; private final AppVersionService appVersionService; + private final EndNodeStatusRepository endNodeStatusRepository; + private final AppBuilderFlowGraphRepository flowGraphRepository; + private final DistributedLockProvider distributedLockProvider; public AippFlowEndCallback(@Fit AippLogService aippLogService, @Fit BrokerClient brokerClient, @Fit BeanContainer beanContainer, @Fit ConversationRecordService conversationRecordService, @Fit AppBuilderFormService formService, @Fit AppChatSseService appChatSseService, @Fit OutputFormatterChain formatterChain, @Fit AppTaskInstanceService appTaskInstanceService, - @Fit AppTaskService appTaskService, @Fit AppVersionService appVersionService, FitRuntime fitRuntime) { + @Fit AppTaskService appTaskService, @Fit AppVersionService appVersionService, + @Fit EndNodeStatusRepository endNodeStatusRepository, + @Fit AppBuilderFlowGraphRepository flowGraphRepository, + @Fit DistributedLockProvider distributedLockProvider, FitRuntime fitRuntime) { this.formService = formService; this.aippLogService = aippLogService; this.brokerClient = brokerClient; @@ -107,6 +129,9 @@ public AippFlowEndCallback(@Fit AippLogService aippLogService, @Fit BrokerClient this.appTaskInstanceService = appTaskInstanceService; this.appTaskService = appTaskService; this.appVersionService = appVersionService; + this.endNodeStatusRepository = endNodeStatusRepository; + this.flowGraphRepository = flowGraphRepository; + this.distributedLockProvider = distributedLockProvider; this.fitRuntime = fitRuntime; } @@ -125,14 +150,36 @@ public void callback(List> contexts) { .orElseThrow(() -> new JobberException(ErrorCodes.UN_EXCEPTED_ERROR, StringUtils.format("App task[{0}] not found.", versionId))); String aippInstId = ObjectUtils.cast(businessData.get(AippConst.BS_AIPP_INST_ID_KEY)); - this.saveInstance(businessData, versionId, aippInstId, context, appTask); + String parentCallbackId = ObjectUtils.cast(businessData.get(AippConst.PARENT_CALLBACK_ID)); + boolean allowTerminalSignal = StringUtils.isEmpty(parentCallbackId); + String lockKey = TERMINATE_LOCK_PREFIX + aippInstId; + Lock lock = this.distributedLockProvider.get(lockKey); + lock.lock(); + try { + this.insertEndNodeStatus(contexts, aippInstId, appTask.getEntity().getFlowConfigId()); + boolean readyToTerminate = this.shouldSendLastData(contexts, appTask); + this.executeAfterReadyToTerminateCheck(contexts, businessData, versionId, aippInstId, context, appTask, + readyToTerminate, allowTerminalSignal); + } finally { + lock.unlock(); + } + + // 子流程 callback 主流程 + if (StringUtils.isNotEmpty(parentCallbackId)) { + this.brokerClient.getRouter(FlowCallbackService.class, "w8onlgq9xsw13jce4wvbcz3kbmjv3tuw") + .route(new FitableIdFilter(parentCallbackId)) + .format(SerializationFormat.CBOR) + .invoke(contexts); + } + } + + private void executeAfterReadyToTerminateCheck(List> contexts, + Map businessData, String versionId, String aippInstId, + OperationContext context, AppTask appTask, boolean readyToTerminate, boolean allowTerminalSignal) { + this.saveInstance(businessData, versionId, aippInstId, context, appTask, readyToTerminate); String parentInstanceId = ObjectUtils.cast(businessData.get(AippConst.PARENT_INSTANCE_ID)); String appId = ObjectUtils.cast(appTask.getEntity().getAppId()); businessData.put(AippConst.ATTR_APP_ID_KEY, appId); - // 表明流程结果是否需要再经过模型加工，当前场景全为false。 - // 正常情况下应该是在结束节点配上该key并放入businessData中，此处模拟该过程。 - // 如果子流程结束后需要再经过模型加工，子流程结束节点不打印日志；否则子流程结束节点需要打印日志。 - // 如果前一个节点是人工检查节点，并在结束节点reference到了表单，那么这里一定会打印消息。 businessData.put(AippConst.BS_AIPP_OUTPUT_IS_NEEDED_LLM, false); if (businessData.containsKey(AippConst.BS_END_FORM_ID_KEY)) { String endFormId = ObjectUtils.cast(businessData.get(AippConst.BS_END_FORM_ID_KEY)); @@ -148,34 +195,163 @@ public void callback(List> contexts) { returnedLogId = this.saveFormToLog(appId, businessData, endFormId, endFormVersion, formDataMap); } AppChatRsp appChatRsp = AppChatRsp.builder().chatId(chatId).atChatId(atChatId) - .status(FlowTraceStatus.ARCHIVED.name()) + .status(readyToTerminate ? FlowTraceStatus.ARCHIVED.name() : FlowTraceStatus.RUNNING.name()) .answer(Collections.singletonList(AppChatRsp.Answer.builder() .content(formDataMap).type(AippInstLogType.FORM.name()).build())) + .extension(this.buildEndNodeSummary(contexts, appTask)) .instanceId(aippInstId).logId(returnedLogId) .build(); - this.appChatSseService.sendLastData(aippInstId, appChatRsp); + if (readyToTerminate && allowTerminalSignal) { + this.doSendTerminalSignal(aippInstId, appChatRsp); + } else { + this.appChatSseService.send(aippInstId, appChatRsp); + } this.insertConversation(businessData, aippInstId, ObjectUtils.cast(businessData.get("chartsData"))); } else { this.logFinalOutput(businessData, aippInstId); + this.sendTerminalEventIfReady(readyToTerminate && allowTerminalSignal, contexts, appTask, + businessData, context, aippInstId); } + } - // 子流程 callback 主流程 - String parentCallbackId = ObjectUtils.cast(businessData.get(AippConst.PARENT_CALLBACK_ID)); - if (StringUtils.isNotEmpty(parentCallbackId)) { - this.brokerClient.getRouter(FlowCallbackService.class, "w8onlgq9xsw13jce4wvbcz3kbmjv3tuw") - .route(new FitableIdFilter(parentCallbackId)) - .format(SerializationFormat.CBOR) - .invoke(contexts); + private void sendTerminalEventIfReady(boolean readyToTerminate, List> contexts, AppTask appTask, + Map businessData, OperationContext operationContext, String aippInstId) { + if (!readyToTerminate) { + return; } + RunContext runContext = new RunContext(businessData, operationContext); + AppChatRsp terminalRsp = AppChatRsp.builder() + .chatId(runContext.getOriginChatId()) + .atChatId(runContext.getAtChatId()) + .status(FlowTraceStatus.ARCHIVED.name()) + .instanceId(aippInstId) + .answer(Collections.emptyList()) + .extension(this.buildEndNodeSummary(contexts, appTask)) + .build(); + this.doSendTerminalSignal(aippInstId, terminalRsp); + } + + private void sendTerminalSignalWithLock(String aippInstId, AppChatRsp appChatRsp) { + String lockKey = TERMINATE_LOCK_PREFIX + aippInstId; + Lock lock = this.distributedLockProvider.get(lockKey); + if (lock.tryLock()) { + try { + this.appChatSseService.sendLastData(aippInstId, appChatRsp); + } finally { + lock.unlock(); + } + } + // If lock acquisition fails, another thread has already sent the terminal signal, skip. + } + + private void doSendTerminalSignal(String aippInstId, AppChatRsp appChatRsp) { + this.appChatSseService.sendLastData(aippInstId, appChatRsp); + } + + private void insertEndNodeStatus(List> contexts, String aippInstId, String flowConfigId) { + String flowTraceId = DataUtils.getFlowTraceId(contexts); + String currentNodeId = ObjectUtils.cast(contexts.get(0).get(AippConst.BS_NODE_ID_KEY)); + String status = ObjectUtils.cast(contexts.get(0).get("status")); + long currentTime = ConvertUtils.toLong(LocalDateTime.now()); + EndNodeStatus endNodeStatus = EndNodeStatus.builder() + .traceId(flowTraceId) + .endNodeId(currentNodeId) + .status(status) + .startTime(currentTime) + .endTime(currentTime) + .flowDefinitionId(flowConfigId) + .instanceId(aippInstId) + .createAt(LocalDateTime.now()) + .updateAt(LocalDateTime.now()) + .build(); + this.endNodeStatusRepository.insertOne(endNodeStatus); + } + + private boolean shouldSendLastData(List> contexts, AppTask appTask) { + Set configuredEndNodeIds = this.getConfiguredEndNodeIds(appTask); + if (configuredEndNodeIds.isEmpty()) { + log.warn("Cannot resolve configured end nodes. Skip terminal event this round. taskId={0}, flowConfigId={1}", + appTask.getEntity().getTaskId(), appTask.getEntity().getFlowConfigId()); + return false; + } + String flowTraceId = DataUtils.getFlowTraceId(contexts); + Set arrivedEndNodeIds = Optional.ofNullable(this.endNodeStatusRepository.selectByTraceId(flowTraceId)) + .orElse(Collections.emptyList()) + .stream() + .filter(endNodeStatus -> this.isArrivedStatus(endNodeStatus.getStatus())) + .map(EndNodeStatus::getEndNodeId) + .filter(StringUtils::isNotBlank) + .collect(Collectors.toSet()); + String currentNodeId = ObjectUtils.cast(contexts.get(0).get(AippConst.BS_NODE_ID_KEY)); + if (StringUtils.isNotBlank(currentNodeId)) { + arrivedEndNodeIds.add(currentNodeId); + } + return !arrivedEndNodeIds.isEmpty() && arrivedEndNodeIds.containsAll(configuredEndNodeIds); + } + + + private boolean isArrivedStatus(String status) { + return StringUtils.equalsIgnoreCase(status, "ARCHIVED") || StringUtils.equalsIgnoreCase(status, "SKIPPED"); + } + + private Set getConfiguredEndNodeIds(AppTask appTask) { + String flowConfigId = appTask.getEntity().getFlowConfigId(); + if (StringUtils.isBlank(flowConfigId)) { + return Collections.emptySet(); + } + AppBuilderFlowGraph flowGraph = this.flowGraphRepository.selectWithId(flowConfigId); + if (flowGraph == null || StringUtils.isBlank(flowGraph.getAppearance())) { + return Collections.emptySet(); + } + JSONObject appearance = JSONObject.parseObject(flowGraph.getAppearance()); + JSONArray pages = appearance.getJSONArray("pages"); + if (pages == null) { + return Collections.emptySet(); + } + return pages.stream() + .filter(JSONObject.class::isInstance) + .map(JSONObject.class::cast) + .map(page -> page.getJSONArray("shapes")) + .filter(shapes -> shapes != null) + .flatMap(List::stream) + .filter(JSONObject.class::isInstance) + .map(JSONObject.class::cast) + .filter(shape -> StringUtils.equalsIgnoreCase(shape.getString("type"), NodeType.END_NODE.type())) + .map(shape -> shape.getString("id")) + .filter(StringUtils::isNotBlank) + .collect(Collectors.toSet()); + } + + private Map buildEndNodeSummary(List> contexts, AppTask appTask) { + String flowTraceId = DataUtils.getFlowTraceId(contexts); + List> endNodes = Optional.ofNullable(this.endNodeStatusRepository.selectByTraceId(flowTraceId)) + .orElse(Collections.emptyList()) + .stream() + .map(endNodeStatus -> { + Map node = new HashMap<>(); + node.put("nodeId", endNodeStatus.getEndNodeId()); + node.put("status", endNodeStatus.getStatus()); + node.put("startTime", endNodeStatus.getStartTime()); + node.put("endTime", endNodeStatus.getEndTime()); + return node; + }) + .collect(Collectors.toList()); + Map summary = new HashMap<>(); + summary.put("flowTraceId", flowTraceId); + summary.put("configuredEndNodeCount", this.getConfiguredEndNodeIds(appTask).size()); + summary.put("endNodeContexts", endNodes); + return summary; } private void saveInstance(Map businessData, String versionId, String aippInstId, - OperationContext context, AppTask appTask) { - this.appTaskInstanceService.update(AppTaskInstance.asUpdate(versionId, aippInstId) - .setFinishTime(LocalDateTime.now()) - .setStatus(MetaInstStatusEnum.ARCHIVED.name()) - .fetch(businessData, appTask.getEntity().getProperties()) - .build(), context); + OperationContext context, AppTask appTask, boolean readyToTerminate) { + TaskInstanceUpdateEntity updateEntity = AppTaskInstance.asUpdate(versionId, aippInstId) + .fetch(businessData, appTask.getEntity().getProperties()) + .setStatus(readyToTerminate ? MetaInstStatusEnum.ARCHIVED.name() : MetaInstStatusEnum.RUNNING.name()); + if (readyToTerminate) { + updateEntity.setFinishTime(LocalDateTime.now()); + } + this.appTaskInstanceService.update(updateEntity.build(), context); } private String saveFormToLog(String appId, Map businessData, String endFormId, @@ -211,6 +387,10 @@ private void logFinalOutput(Map businessData, String aippInstId) if (!checkEnableLog(businessData)) { logType = AippInstLogType.HIDDEN_MSG; } + // META_MSG 类型已在 llmOutputConsumer 中插入过，避免重复落库 + if (logType == AippInstLogType.META_MSG) { + return; + } this.aippLogService.insertLog(logType.name(), AippLogData.builder().msg(logMsg).build(), businessData); this.beanContainer.all(AppFlowFinishObserver.class) .stream() diff --git a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/fitable/FlowPublishSubscriber.java b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/fitable/FlowPublishSubscriber.java index 2430802395..6ffc2995e6 100644 --- a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/fitable/FlowPublishSubscriber.java +++ b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/fitable/FlowPublishSubscriber.java @@ -57,8 +57,9 @@ public class FlowPublishSubscriber implements FlowPublishService { * @param runtimeInfoService {@link RuntimeInfoService} 对象. */ public FlowPublishSubscriber(AppBuilderRuntimeInfoRepository runtimeInfoRepository, - @Fit ToolExceptionHandle toolExceptionHandle, AppChatSessionService appChatSessionService, - AppChatSseService appChatSSEService, RuntimeInfoService runtimeInfoService) { + @Fit ToolExceptionHandle toolExceptionHandle, + AppChatSessionService appChatSessionService, AppChatSseService appChatSSEService, + RuntimeInfoService runtimeInfoService) { this.runtimeInfoRepository = runtimeInfoRepository; this.toolExceptionHandle = toolExceptionHandle; this.appChatSessionService = appChatSessionService; @@ -88,13 +89,12 @@ public void publishNodeInfo(FlowNodePublishInfo flowNodePublishInfo) { private void stageProcessedHandle(FlowNodePublishInfo flowNodePublishInfo, Map businessData, String aippInstId) { Optional> instanceSession = this.appChatSessionService.getSession(aippInstId); + FlowPublishContext context = flowNodePublishInfo.getFlowContext(); + if (instanceSession.isPresent() && !instanceSession.get().isDebug()) { return; } - FlowPublishContext context = flowNodePublishInfo.getFlowContext(); - String traceId = context.getTraceId(); - String nodeId = flowNodePublishInfo.getNodeId(); - String nodeType = flowNodePublishInfo.getNodeType(); + FlowErrorInfo errorInfo = flowNodePublishInfo.getErrorMsg(); AtomicReference locale = new AtomicReference<>(Locale.getDefault()); instanceSession.ifPresent(session -> locale.set(session.getLocale())); @@ -104,15 +104,15 @@ private void stageProcessedHandle(FlowNodePublishInfo flowNodePublishInfo, Map } private String getStandardValue(Object value) { - notNull(value, "The value cannot be null."); + if (value == null) { + return "null"; + } if (value instanceof String) { return ObjectUtils.cast(value); } @@ -606,4 +608,4 @@ private void sendMsgHandle(String msg, StreamMsgType logType, Map selectByTraceId(String traceId); + + /** + * 根据instanceId查询所有的END节点状态信息. + * + * @param instanceId 实例ID. + * @return END节点状态列表. + */ + List selectByInstanceId(String instanceId); + + /** + * 插入一条数据. + * + * @param endNodeStatusPo END节点状态对象. + */ + void insertOne(EndNodeStatusPo endNodeStatusPo); + + /** + * 获取超期的END节点状态信息唯一标识列表。 + * + * @param expiredDays 表示超期时间的 int。 + * @param limit 表示查询条数的 int。 + * @return END节点状态唯一标识列表. + */ + List getExpiredEndNodeStatuses(int expiredDays, int limit); + + /** + * 根据END节点状态唯一标识列表删除记录。 + * + * @param endNodeStatusIds END节点状态唯一标识列表. + */ + void deleteEndNodeStatuses(List endNodeStatusIds); +} diff --git a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/po/EndNodeStatusPo.java b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/po/EndNodeStatusPo.java new file mode 100644 index 0000000000..ba144d331a --- /dev/null +++ b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/po/EndNodeStatusPo.java @@ -0,0 +1,39 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved. + * This file is a part of the ModelEngine Project. + * Licensed under the MIT License. See License.txt in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package modelengine.fit.jober.aipp.po; + +import lombok.AllArgsConstructor; +import lombok.Builder; +import lombok.Data; +import lombok.NoArgsConstructor; + +import java.time.LocalDateTime; + +/** + * END节点状态记录. + * + * @author 沈维枫 + * @since 2026-04-01 + */ +@Builder +@Data +@NoArgsConstructor +@AllArgsConstructor +public class EndNodeStatusPo { + private Long id; + private String traceId; + private String endNodeId; + private String status; + private long startTime; + private long endTime; + private String instanceId; + private String flowDefinitionId; + + /* ------------ 公共字段 ------------ */ + private LocalDateTime createAt; + private LocalDateTime updateAt; +} diff --git a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/repository/EndNodeStatusRepository.java b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/repository/EndNodeStatusRepository.java new file mode 100644 index 0000000000..3166ad8305 --- /dev/null +++ b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/repository/EndNodeStatusRepository.java @@ -0,0 +1,58 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved. + * This file is a part of the ModelEngine Project. + * Licensed under the MIT License. See License.txt in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package modelengine.fit.jober.aipp.repository; + +import modelengine.fit.jober.aipp.domain.EndNodeStatus; + +import java.util.List; + +/** + * END节点状态相关数据库操作对象. + * + * @author 张越 + * @since 2026-04-01 + */ +public interface EndNodeStatusRepository { + /** + * 根据 traceId 获取END节点状态. + * + * @param traceId 追踪实例运行情况的唯一标识. + * @return END节点状态列表. + */ + List selectByTraceId(String traceId); + + /** + * 根据 instanceId 获取END节点状态. + * + * @param instanceId 实例ID. + * @return END节点状态列表. + */ + List selectByInstanceId(String instanceId); + + /** + * 插入一条数据. + * + * @param endNodeStatus END节点状态. + */ + void insertOne(EndNodeStatus endNodeStatus); + + /** + * 获取END节点状态过期历史记录. + * + * @param expiredDays 表示超期天数的 int. + * @param limit 表示查询条数的 int. + * @return 超期END节点状态id的列表. + */ + List getExpiredEndNodeStatuses(int expiredDays, int limit); + + /** + * 根据END节点状态id列表删除历史记录. + * + * @param endNodeStatusIds 历史记录的id列表. + */ + void deleteEndNodeStatuses(List endNodeStatusIds); +} diff --git a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/repository/impl/EndNodeStatusRepositoryImpl.java b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/repository/impl/EndNodeStatusRepositoryImpl.java new file mode 100644 index 0000000000..7435e4573f --- /dev/null +++ b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/repository/impl/EndNodeStatusRepositoryImpl.java @@ -0,0 +1,68 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved. + * This file is a part of the ModelEngine Project. + * Licensed under the MIT License. See License.txt in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package modelengine.fit.jober.aipp.repository.impl; + +import modelengine.fit.jober.aipp.domain.EndNodeStatus; +import modelengine.fit.jober.aipp.mapper.EndNodeStatusMapper; +import modelengine.fit.jober.aipp.repository.EndNodeStatusRepository; +import modelengine.fit.jober.aipp.serializer.impl.EndNodeStatusSerializer; +import modelengine.fitframework.annotation.Component; +import modelengine.fitframework.util.CollectionUtils; + +import java.util.List; +import java.util.stream.Collectors; + +/** + * EndNodeStatusRepository 实现类. + * + * @author 张越 + * @since 2026-04-01 + */ +@Component +public class EndNodeStatusRepositoryImpl implements EndNodeStatusRepository { + private final EndNodeStatusMapper mapper; + private final EndNodeStatusSerializer serializer; + + public EndNodeStatusRepositoryImpl(EndNodeStatusMapper mapper) { + this.mapper = mapper; + this.serializer = new EndNodeStatusSerializer(); + } + + @Override + public List selectByTraceId(String traceId) { + return this.mapper.selectByTraceId(traceId) + .stream() + .map(this.serializer::deserialize) + .collect(Collectors.toList()); + } + + @Override + public List selectByInstanceId(String instanceId) { + return this.mapper.selectByInstanceId(instanceId) + .stream() + .map(this.serializer::deserialize) + .collect(Collectors.toList()); + } + + @Override + public void insertOne(EndNodeStatus endNodeStatus) { + this.mapper.insertOne(this.serializer.serialize(endNodeStatus)); + } + + @Override + public List getExpiredEndNodeStatuses(int expiredDays, int limit) { + return this.mapper.getExpiredEndNodeStatuses(expiredDays, limit); + } + + @Override + public void deleteEndNodeStatuses(List endNodeStatusIds) { + if (CollectionUtils.isEmpty(endNodeStatusIds)) { + return; + } + this.mapper.deleteEndNodeStatuses(endNodeStatusIds); + } +} diff --git a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/serializer/impl/EndNodeStatusSerializer.java b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/serializer/impl/EndNodeStatusSerializer.java new file mode 100644 index 0000000000..56df00dc90 --- /dev/null +++ b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/serializer/impl/EndNodeStatusSerializer.java @@ -0,0 +1,58 @@ +/*--------------------------------------------------------------------------------------------- + * Copyright (c) 2025 Huawei Technologies Co., Ltd. All rights reserved. + * This file is a part of the ModelEngine Project. + * Licensed under the MIT License. See License.txt in the project root for license information. + *--------------------------------------------------------------------------------------------*/ + +package modelengine.fit.jober.aipp.serializer.impl; + +import modelengine.fit.jober.aipp.domain.EndNodeStatus; +import modelengine.fit.jober.aipp.po.EndNodeStatusPo; +import modelengine.fit.jober.aipp.serializer.BaseSerializer; + +import java.util.Objects; + +/** + * {@link EndNodeStatus} 以及 {@link EndNodeStatusPo} 之间互相转换的序列化器. + * + * @author 张越 + * @since 2026-04-01 + */ +public class EndNodeStatusSerializer implements BaseSerializer { + @Override + public EndNodeStatusPo serialize(EndNodeStatus endNodeStatus) { + if (Objects.isNull(endNodeStatus)) { + return null; + } + return EndNodeStatusPo.builder() + .id(endNodeStatus.getId()) + .traceId(endNodeStatus.getTraceId()) + .endNodeId(endNodeStatus.getEndNodeId()) + .status(endNodeStatus.getStatus()) + .startTime(endNodeStatus.getStartTime()) + .endTime(endNodeStatus.getEndTime()) + .instanceId(endNodeStatus.getInstanceId()) + .flowDefinitionId(endNodeStatus.getFlowDefinitionId()) + .createAt(endNodeStatus.getCreateAt()) + .updateAt(endNodeStatus.getUpdateAt()) + .build(); + } + + @Override + public EndNodeStatus deserialize(EndNodeStatusPo dataObject) { + return Objects.isNull(dataObject) + ? EndNodeStatus.builder().build() + : EndNodeStatus.builder() + .id(dataObject.getId()) + .traceId(dataObject.getTraceId()) + .endNodeId(dataObject.getEndNodeId()) + .status(dataObject.getStatus()) + .startTime(dataObject.getStartTime()) + .endTime(dataObject.getEndTime()) + .instanceId(dataObject.getInstanceId()) + .flowDefinitionId(dataObject.getFlowDefinitionId()) + .createAt(dataObject.getCreateAt()) + .updateAt(dataObject.getUpdateAt()) + .build(); + } +} diff --git a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/service/impl/AippFlowRuntimeInfoServiceImpl.java b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/service/impl/AippFlowRuntimeInfoServiceImpl.java index ad4a71526f..b281187d79 100644 --- a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/service/impl/AippFlowRuntimeInfoServiceImpl.java +++ b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/service/impl/AippFlowRuntimeInfoServiceImpl.java @@ -14,6 +14,7 @@ import modelengine.fit.jober.aipp.domains.task.service.AppTaskService; import modelengine.fit.jober.aipp.domains.taskinstance.AppTaskInstance; import modelengine.fit.jober.aipp.domains.taskinstance.service.AppTaskInstanceService; +import modelengine.fit.jober.aipp.enums.MetaInstStatusEnum; import modelengine.fit.jober.aipp.repository.AppBuilderRuntimeInfoRepository; import modelengine.fit.jober.aipp.service.AippFlowRuntimeInfoService; import modelengine.fit.jober.common.ErrorCodes; @@ -21,14 +22,19 @@ import modelengine.fit.jober.entity.consts.NodeTypes; import modelengine.fit.runtime.entity.NodeInfo; import modelengine.fit.runtime.entity.RuntimeData; -import modelengine.fit.waterflow.domain.enums.FlowNodeStatus; +import modelengine.fit.waterflow.flowsengine.domain.flows.enums.FlowNodeStatus; import modelengine.fitframework.annotation.Component; import modelengine.fitframework.log.Logger; import modelengine.fitframework.util.CollectionUtils; import modelengine.fitframework.util.StringUtils; +import java.util.Arrays; +import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.Optional; +import java.util.Set; import java.util.stream.Collectors; /** @@ -40,6 +46,10 @@ @Component public class AippFlowRuntimeInfoServiceImpl implements AippFlowRuntimeInfoService { private static final Logger LOGGER = Logger.get(AippFlowRuntimeInfoServiceImpl.class); + private static final String NODE_STATUS_RUNNING = "RUNNING"; + private static final Set FINISHED_NODE_STATUS = new HashSet<>( + Arrays.asList(FlowNodeStatus.ARCHIVED.name(), FlowNodeStatus.ERROR.name(), + FlowNodeStatus.TERMINATE.name(),FlowNodeStatus.SKIPPED.name())); private final AppBuilderRuntimeInfoRepository runtimeInfoRepository; private final AppTaskInstanceService appTaskInstanceService; @@ -83,16 +93,19 @@ public Optional getRuntimeData(String aippId, String version, Strin runtimeData.setTraceId(traceId); runtimeData.setAippInstanceId(end.getInstanceId()); runtimeData.setExecuteTime(end.getEndTime() - start.getStartTime()); - runtimeData.setNodeInfos(runtimeInfoList.stream().map(this::toNodeInfo).collect(Collectors.toList())); - runtimeData.setFinished(isFinished(runtimeInfoList)); + runtimeData.setNodeInfos(this.toNodeInfos(runtimeInfoList)); + runtimeData.setFinished(isFinished(instance, runtimeInfoList)); return Optional.of(runtimeData); } - private boolean isFinished(List runtimeInfoList) { - return runtimeInfoList.stream() - .anyMatch(runtimeInfo -> runtimeInfo.getNextPositionId() == null || NodeTypes.END.getType() - .equals(runtimeInfo.getNodeType()) || FlowNodeStatus.ERROR.name() - .equals(runtimeInfo.getStatus())); + private boolean isFinished(AppTaskInstance instance, List runtimeInfoList) { + return instance.getEntity() + .getStatus() + .map(MetaInstStatusEnum::getMetaInstStatus) + .map(status -> status == MetaInstStatusEnum.ARCHIVED || status == MetaInstStatusEnum.ERROR + || status == MetaInstStatusEnum.TERMINATED) + .orElseGet(() -> !runtimeInfoList.isEmpty() && runtimeInfoList.stream() + .allMatch(runtimeInfo -> FINISHED_NODE_STATUS.contains(runtimeInfo.getStatus()))); } private NodeInfo toNodeInfo(AppBuilderRuntimeInfo info) { @@ -107,4 +120,41 @@ private NodeInfo toNodeInfo(AppBuilderRuntimeInfo info) { nodeInfo.setStatus(info.getStatus()); return nodeInfo; } + + private List toNodeInfos(List runtimeInfoList) { + Map> groupedInfos = runtimeInfoList.stream() + .collect(Collectors.groupingBy(AppBuilderRuntimeInfo::getNodeId, LinkedHashMap::new, Collectors.toList())); + return groupedInfos.values().stream().map(this::toAggregatedNodeInfo).collect(Collectors.toList()); + } + + private NodeInfo toAggregatedNodeInfo(List infos) { + AppBuilderRuntimeInfo latestInfo = infos.get(infos.size() - 1); + AppBuilderRuntimeInfo earliestInfo = infos.stream() + .min((left, right) -> Long.compare(left.getStartTime(), right.getStartTime())) + .orElse(latestInfo); + NodeInfo nodeInfo = this.toNodeInfo(latestInfo); + long startTime = earliestInfo.getStartTime(); + long endTime = infos.stream().mapToLong(AppBuilderRuntimeInfo::getEndTime).max().orElse(latestInfo.getEndTime()); + nodeInfo.setStartTime(startTime); + nodeInfo.setRunCost(endTime - startTime); + nodeInfo.setStatus(this.aggregateStatus(infos)); + return nodeInfo; + } + + private String aggregateStatus(List infos) { + List statuses = infos.stream().map(AppBuilderRuntimeInfo::getStatus).collect(Collectors.toList()); + if (statuses.stream().anyMatch(FlowNodeStatus.ERROR.name()::equals)) { + return FlowNodeStatus.ERROR.name(); + } + if (statuses.stream().anyMatch(FlowNodeStatus.TERMINATE.name()::equals)) { + return FlowNodeStatus.TERMINATE.name(); + } + if (statuses.stream().anyMatch(FlowNodeStatus.ARCHIVED.name()::equals)) { + return FlowNodeStatus.ARCHIVED.name(); + } + if (statuses.stream().allMatch(FlowNodeStatus.SKIPPED.name()::equals)) { + return FlowNodeStatus.SKIPPED.name(); + } + return NODE_STATUS_RUNNING; + } } diff --git a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/service/impl/AippLogStreamServiceImpl.java b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/service/impl/AippLogStreamServiceImpl.java index 5c5643bbca..34efb9b2b5 100644 --- a/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/service/impl/AippLogStreamServiceImpl.java +++ b/app-builder/plugins/aipp-plugin/src/main/java/modelengine/fit/jober/aipp/service/impl/AippLogStreamServiceImpl.java @@ -18,9 +18,10 @@ import modelengine.fit.jober.aipp.vo.AippLogVO; import modelengine.fit.jober.common.ErrorCodes; import modelengine.fit.jober.common.exceptions.JobberException; - import modelengine.fit.waterflow.domain.enums.FlowTraceStatus; + import modelengine.fitframework.annotation.Component; +import modelengine.fitframework.log.Logger; import modelengine.fitframework.util.ObjectUtils; import modelengine.fitframework.util.StringUtils; @@ -38,6 +39,8 @@ */ @Component public class AippLogStreamServiceImpl implements AippLogStreamService { + private static final Logger log = Logger.get(AippLogStreamServiceImpl.class); + private static final List OUTPUT_WITH_MSG_WHITE_LIST = Arrays.asList(AippInstLogType.MSG.name(), AippInstLogType.ERROR.name(), AippInstLogType.META_MSG.name(), @@ -56,58 +59,58 @@ public AippLogStreamServiceImpl(AppChatSseService appChatSseService, } @Override - public void send(AippLogVO log) { - if (!log.displayable()) { + public void send(AippLogVO aippLog) { + if (!aippLog.displayable()) { return; } - AppChatRsp appChatRsp = this.buildData(log); + AppChatRsp appChatRsp = this.buildData(aippLog); + if (!appChatRsp.getStatus().equalsIgnoreCase(FlowTraceStatus.RUNNING.name()) && !appChatRsp.getStatus() .equalsIgnoreCase(FlowTraceStatus.READY.name())) { - this.appChatSseService.sendLastData(log.getInstanceId(), appChatRsp); + this.appChatSseService.sendLastData(aippLog.getInstanceId(), appChatRsp); } else { - this.appChatSseService.send(log.getInstanceId(), appChatRsp); + this.appChatSseService.send(aippLog.getInstanceId(), appChatRsp); } } - private AppChatRsp buildData(AippLogVO log) { - String instanceId = log.getInstanceId(); + private AppChatRsp buildData(AippLogVO aippLog) { + String instanceId = aippLog.getInstanceId(); AppTaskInstance instance = this.appTaskInstanceService.getInstanceById(instanceId, null) .orElseThrow(() -> new JobberException(ErrorCodes.UN_EXCEPTED_ERROR, StringUtils.format("App task instance[{0}] not found.", instanceId))); - // 在当前某些情况下，会出现插入log日志，但是不修改instance状态的情况. - // 参考modelengine.fit.jober.aipp.fitable.agent.AippFlowAgent.fetchAgentErrorMsgToMain - String status = log.getLogType().equals(AippInstLogType.ERROR.name()) - ? FlowTraceStatus.ERROR.name() + // 日志流阶段统一透出 RUNNING，避免中间态误触发前端结束动画。 + String status = aippLog.getLogType().equals(AippInstLogType.ERROR.name()) + ? FlowTraceStatus.ERROR.name() : instance.getEntity().getStatus().orElse(null); - AppChatRsp.Answer answer = this.buildAnswer(log); + AppChatRsp.Answer answer = this.buildAnswer(aippLog); Map extensionMap = new HashMap<>(); - extensionMap.put("isEnableLog", log.isEnableLog()); + extensionMap.put("isEnableLog", aippLog.isEnableLog()); return AppChatRsp.builder() - .chatId(log.getChatId()) - .atChatId(log.getAtChatId()) + .chatId(aippLog.getChatId()) + .atChatId(aippLog.getAtChatId()) .status(status) .instanceId(instanceId) .answer(Collections.singletonList(answer)) - .logId(log.getLogId()) + .logId(aippLog.getLogId()) .extension(extensionMap) .build(); } - private AppChatRsp.Answer buildAnswer(AippLogVO log) { + private AppChatRsp.Answer buildAnswer(AippLogVO aippLog) { AppChatRsp.Answer.AnswerBuilder builder = - AppChatRsp.Answer.builder().type(log.getLogType()).msgId(log.getMsgId()); - if (OUTPUT_WITH_MSG_WHITE_LIST.contains(StringUtils.toUpperCase(log.getLogType()))) { - Object msg = JsonUtils.parseObject(log.getLogData()).get("msg"); + AppChatRsp.Answer.builder().type(aippLog.getLogType()).msgId(aippLog.getMsgId()); + if (OUTPUT_WITH_MSG_WHITE_LIST.contains(StringUtils.toUpperCase(aippLog.getLogType()))) { + Object msg = JsonUtils.parseObject(aippLog.getLogData()).get("msg"); if (msg instanceof String) { msg = this.sensitiveFilterTools.filterString(ObjectUtils.cast(msg)); } builder.content(msg); - } else if (JsonUtils.isValidJson(log.getLogData())) { - builder.content(JsonUtils.parseObject(log.getLogData())); + } else if (JsonUtils.isValidJson(aippLog.getLogData())) { + builder.content(JsonUtils.parseObject(aippLog.getLogData())); } else { - builder.content(log.getLogData()); + builder.content(aippLog.getLogData()); } return builder.build(); } diff --git a/app-builder/plugins/aipp-plugin/src/main/resources/mapper/EndNodeStatusMapper.xml b/app-builder/plugins/aipp-plugin/src/main/resources/mapper/EndNodeStatusMapper.xml new file mode 100644 index 0000000000..3c2e0d5326 --- /dev/null +++ b/app-builder/plugins/aipp-plugin/src/main/resources/mapper/EndNodeStatusMapper.xml @@ -0,0 +1,93 @@ + + + + + + + + + + + + + + + + + + id, + trace_id, + end_node_id, + status, + start_time, + end_time, + instance_id, + flow_definition_id, + create_at, + update_at + + + + INSERT INTO app_builder_end_node_status ( + trace_id, + end_node_id, + status, + start_time, + end_time, + instance_id, + flow_definition_id, + create_at, + update_at + ) VALUES ( + #{traceId}, + #{endNodeId}, + #{status}, + #{startTime}, + #{endTime}, + #{instanceId}, + #{flowDefinitionId}, + #{createAt}, + #{updateAt} + ) + ON CONFLICT (trace_id, end_node_id) DO UPDATE SET + status = EXCLUDED.status, + end_time = EXCLUDED.end_time, + update_at = EXCLUDED.update_at + + + + + + + + + + DELETE FROM app_builder_end_node_status + WHERE id IN + + #{item} + + + diff --git a/app-builder/plugins/aipp-plugin/src/main/resources/sql/schema/create_tables/appbuilder_create.sql b/app-builder/plugins/aipp-plugin/src/main/resources/sql/schema/create_tables/appbuilder_create.sql index 6b2aeb9e5b..d65d038ba4 100644 --- a/app-builder/plugins/aipp-plugin/src/main/resources/sql/schema/create_tables/appbuilder_create.sql +++ b/app-builder/plugins/aipp-plugin/src/main/resources/sql/schema/create_tables/appbuilder_create.sql @@ -165,6 +165,44 @@ comment on column app_builder_form_property.data_from is '数据来源'; comment on column app_builder_form_property.in_group is '应用所属的组'; comment on column app_builder_form_property.description is '应用描述'; comment on column app_builder_form_property.default_index is '属性的默认顺序'; +-- End Node Status 表 +-- 用于记录 END 节点的执行状态，解决生产环境下 RuntimeInfo 不记录导致关流判断失效的问题 +-- 创建时间: 2026-04-01 + +CREATE TABLE IF NOT EXISTS app_builder_end_node_status +( + id BIGSERIAL primary key, + trace_id varchar(64) not null, + end_node_id varchar(64) not null, + status varchar(32) not null, + start_time bigint not null, + end_time bigint not null, + instance_id varchar(64) not null, + flow_definition_id varchar(64), + create_at timestamp not null default current_timestamp, + update_at timestamp not null default current_timestamp, + constraint uk_trace_end_node unique (trace_id, end_node_id) +); + +-- 索引 +create index if not exists idx_end_status_trace_id on app_builder_end_node_status(trace_id); +create index if not exists idx_end_status_instance_id on app_builder_end_node_status(instance_id); +create index if not exists idx_end_status_trace_status on app_builder_end_node_status(trace_id, status); + +-- 注释 +comment on table app_builder_end_node_status is 'END节点状态记录表'; +comment on column app_builder_end_node_status.id is '主键'; +comment on column app_builder_end_node_status.trace_id is '流程追踪ID'; +comment on column app_builder_end_node_status.end_node_id is 'END节点ID'; +comment on column app_builder_end_node_status.status is '状态：ARCHIVED或SKIPPED'; +comment on column app_builder_end_node_status.start_time is '开始时间（毫秒时间戳）'; +comment on column app_builder_end_node_status.end_time is '结束时间（毫秒时间戳）'; +comment on column app_builder_end_node_status.instance_id is '实例ID'; +comment on column app_builder_end_node_status.flow_definition_id is '流程定义ID'; +comment on column app_builder_end_node_status.create_at is '创建时间'; +comment on column app_builder_end_node_status.update_at is '更新时间'; + + create table if not exists app_builder_runtime_info ( diff --git a/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/TestUtils.java b/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/TestUtils.java index 527ace7ce0..2cecba4efe 100644 --- a/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/TestUtils.java +++ b/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/TestUtils.java @@ -64,6 +64,8 @@ public static List> buildFlowDataWithExtraConfig(Map infos = new ArrayList<>(); + infos.add(AppBuilderRuntimeInfo.builder() + .startTime(1) + .endTime(10) + .flowDefinitionId("flow1") + .traceId("trace1") + .nodeId("node1") + .instanceId("instance1") + .nodeType(NodeTypes.START.getType()) + .status(FlowNodeStatus.ARCHIVED.name()) + .published(true) + .build()); + infos.add(AppBuilderRuntimeInfo.builder() + .startTime(11) + .endTime(20) + .flowDefinitionId("flow1") + .traceId("trace1") + .nodeId("node2") + .instanceId("instance1") + .nodeType(NodeTypes.STATE.getType()) + .status(FlowNodeStatus.SKIPPED.name()) + .published(true) + .build()); + infos.add(AppBuilderRuntimeInfo.builder() + .startTime(21) + .endTime(30) + .flowDefinitionId("flow1") + .traceId("trace1") + .nodeId("node2") + .instanceId("instance1") + .nodeType(NodeTypes.STATE.getType()) + .status(FlowNodeStatus.ARCHIVED.name()) + .published(true) + .build()); + doReturn(infos).when(this.repository).selectByTraceId(anyString()); + + RuntimeData runtimeData = this.service.getRuntimeData("app1", "", "instance1", new OperationContext()).get(); + + assertEquals(2, runtimeData.getNodeInfos().size()); + assertEquals(FlowNodeStatus.ARCHIVED.name(), runtimeData.getNodeInfos().get(1).getStatus()); + } + + @Test + void shouldAggregateNodeStatusToSkippedWhenAllContextsSkipped() { + doReturn(Optional.of(AppTask.asEntity().setTaskId("version1").setAppId("app1").build())).when( + this.appTaskService).getLatest(anyString(), anyString(), any(OperationContext.class)); + doReturn(Optional.of(AppTaskInstance.asEntity().setFlowTraceId("trace1").build())).when(this.appTaskInstanceService) + .getInstance(anyString(), anyString(), any(OperationContext.class)); + + List infos = new ArrayList<>(); + infos.add(AppBuilderRuntimeInfo.builder() + .startTime(1) + .endTime(10) + .flowDefinitionId("flow1") + .traceId("trace1") + .nodeId("node1") + .instanceId("instance1") + .nodeType(NodeTypes.START.getType()) + .status(FlowNodeStatus.ARCHIVED.name()) + .published(true) + .build()); + infos.add(AppBuilderRuntimeInfo.builder() + .startTime(11) + .endTime(20) + .flowDefinitionId("flow1") + .traceId("trace1") + .nodeId("node2") + .instanceId("instance1") + .nodeType(NodeTypes.STATE.getType()) + .status(FlowNodeStatus.SKIPPED.name()) + .published(true) + .build()); + infos.add(AppBuilderRuntimeInfo.builder() + .startTime(21) + .endTime(30) + .flowDefinitionId("flow1") + .traceId("trace1") + .nodeId("node2") + .instanceId("instance1") + .nodeType(NodeTypes.STATE.getType()) + .status(FlowNodeStatus.SKIPPED.name()) + .published(true) + .build()); + doReturn(infos).when(this.repository).selectByTraceId(anyString()); + + RuntimeData runtimeData = this.service.getRuntimeData("app1", "", "instance1", new OperationContext()).get(); + + assertEquals(2, runtimeData.getNodeInfos().size()); + assertEquals(FlowNodeStatus.SKIPPED.name(), runtimeData.getNodeInfos().get(1).getStatus()); + } + private void mockData() { doReturn(Optional.of(AppTask.asEntity().setTaskId("version1").setAppId("app1").build())).when( this.appTaskService).getLatest(anyString(), anyString(), any(OperationContext.class)); diff --git a/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/fitable/FlowPublishSubscriberTest.java b/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/fitable/FlowPublishSubscriberTest.java index 9310e70e82..33661ea134 100644 --- a/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/fitable/FlowPublishSubscriberTest.java +++ b/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/fitable/FlowPublishSubscriberTest.java @@ -89,8 +89,8 @@ public class FlowPublishSubscriberTest { void setUp() { RuntimeInfoServiceImpl runtimeInfoService = new RuntimeInfoServiceImpl(null, this.appTaskService, this.appTaskInstanceService, this.appVersionService); - this.flowPublishSubscriber = new FlowPublishSubscriber(this.repository, this.toolExceptionHandle, - this.appChatSessionService, null, runtimeInfoService); + this.flowPublishSubscriber = new FlowPublishSubscriber(this.repository, + this.toolExceptionHandle, this.appChatSessionService, null, runtimeInfoService); } /** diff --git a/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/fitable/LlmComponentTest.java b/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/fitable/LlmComponentTest.java index f1b011cd1b..873cd32a50 100644 --- a/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/fitable/LlmComponentTest.java +++ b/app-builder/plugins/aipp-plugin/src/test/java/modelengine/fit/jober/aipp/fitable/LlmComponentTest.java @@ -225,6 +225,8 @@ void shouldOkWhenNoTool() throws InterruptedException { Mockito.verify(this.toolService, times(0)).getTool(any()); } + + @Test void shouldFailedWhenNoTool() throws InterruptedException { // stub @@ -423,4 +425,4 @@ private void prepareModel() { when(this.aippModelCenter.getModelAccessInfo(any(), any(), any())).thenReturn( ModelAccessInfo.builder().tag("tag").build()); } -} \ No newline at end of file +} diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/common/Constant.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/common/Constant.java index 347d1afcab..24dd1291be 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/common/Constant.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/common/Constant.java @@ -164,6 +164,11 @@ public final class Constant { */ public static final String INTERNAL_EXECUTE_INFO_KEY = "executeInfo"; + /** + * 内部跳过标记，用于让未命中的条件分支继续参与后续汇聚。 + */ + public static final String INTERNAL_SKIPPED_SIGNAL_KEY = "skippedSignal"; + /** * 存储系统参数的虚拟节点id */ @@ -231,6 +236,7 @@ public final class Constant { put(FlowNodeStatus.TERMINATE.toString(), CONTEXT_TERMINATE_EXCLUSIVE_STATUS_LIST); put(FlowNodeStatus.ERROR.toString(), CONTEXT_ERROR_EXCLUSIVE_STATUS_LIST); put(FlowNodeStatus.PENDING.toString(), CONTEXT_NONE_TERMINATE_EXCLUSIVE_STATUS_LIST); + put(FlowNodeStatus.SKIPPED.toString(), CONTEXT_NONE_TERMINATE_EXCLUSIVE_STATUS_LIST); put(FlowNodeStatus.READY.toString(), CONTEXT_NONE_TERMINATE_EXCLUSIVE_STATUS_LIST); put(FlowNodeStatus.PROCESSING.toString(), CONTEXT_NONE_TERMINATE_EXCLUSIVE_STATUS_LIST); put(FlowNodeStatus.ARCHIVED.toString(), CONTEXT_ARCHIVED_EXCLUSIVE_STATUS_LIST); diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/FlowRuntimeServiceImpl.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/FlowRuntimeServiceImpl.java index ef0cec0f41..0edc1466be 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/FlowRuntimeServiceImpl.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/FlowRuntimeServiceImpl.java @@ -139,7 +139,8 @@ public class FlowRuntimeServiceImpl implements FlowRuntimeService { private final BrokerClient brokerClient; private final List traceRunningStatus = Arrays.asList(FlowTraceStatus.READY.name(), - FlowNodeStatus.NEW.name(), FlowNodeStatus.PENDING.name(), FlowNodeStatus.RETRYABLE.name()); + FlowNodeStatus.NEW.name(), FlowNodeStatus.PENDING.name(), FlowNodeStatus.SKIPPED.name(), + FlowNodeStatus.RETRYABLE.name()); private final FlowDefinitionQueryService definitionQueryService; @@ -171,7 +172,8 @@ public FlowRuntimeServiceImpl(FlowDefinitionRepo definitionRepo, private static boolean isContextRunning(FlowContextPO flowContextPO) { FlowNodeStatus contextStatus = FlowNodeStatus.valueOf(flowContextPO.getStatus()); return FlowNodeStatus.NEW.equals(contextStatus) || FlowNodeStatus.PENDING.equals(contextStatus) - || FlowNodeStatus.READY.equals(contextStatus) || FlowNodeStatus.RETRYABLE.equals(contextStatus); + || FlowNodeStatus.SKIPPED.equals(contextStatus) || FlowNodeStatus.READY.equals(contextStatus) + || FlowNodeStatus.RETRYABLE.equals(contextStatus); } @Override diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/scheduletasks/RestartContextSchedule.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/scheduletasks/RestartContextSchedule.java index 4ad92dbec9..880299524d 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/scheduletasks/RestartContextSchedule.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/biz/service/scheduletasks/RestartContextSchedule.java @@ -152,9 +152,10 @@ private void restart(From flow, FlowDefinition flowDefinition, List node = flow.findNodeFromFlow(flow, context.getPosition()); node.offer(contexts, (c) -> {}); - } else if (Objects.equals(context.getStatus(), FlowNodeStatus.PENDING) && !flowDefinition.getNodeMap() + } else if ((Objects.equals(context.getStatus(), FlowNodeStatus.PENDING) + || Objects.equals(context.getStatus(), FlowNodeStatus.SKIPPED)) && !flowDefinition.getNodeMap() .containsKey(context.getPosition())) { - // 线上pending状态context + // 线上pending/skipped状态context FlowNode flowNode = flowDefinition.getFromNodeByEvent(context.getPosition()); if (flowNode.belongTo(FlowNodeType.START)) { flow.offer(contexts, (c) -> {}); diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java index 170aa4ebb4..50c7e5a77c 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContext.java @@ -8,12 +8,17 @@ import lombok.Getter; import lombok.Setter; +import lombok.ToString; +import modelengine.fit.waterflow.common.Constant; import modelengine.fit.waterflow.flowsengine.domain.flows.enums.FlowNodeStatus; import modelengine.fit.waterflow.flowsengine.domain.flows.streams.IdGenerator; import java.time.LocalDateTime; +import java.util.HashMap; import java.util.HashSet; import java.util.List; +import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; @@ -22,10 +27,12 @@ * * @since 2023/08/14 */ +@ToString public final class FlowContext extends IdGenerator { /** * 通过from.offer(data)而不是.offer(context)发起的数据会新增一个trace，这个trace会延续到flow end */ + @Getter private final Set traceId; @@ -267,11 +274,17 @@ public List> generate(List data, String position) { /** * fork一个新的context用于一拖多分支，继承当前context的运行元数据，但生成新的contextId。 + * 对于FlowData类型的数据，需要深拷贝以避免多个fork分支共享同一个FlowData引用导致数据相互影响。 * * @return 新的分支context */ public FlowContext fork() { - return this.convertData(this.data); + T dataToFork = this.data; + if (dataToFork instanceof FlowData) { + // FlowData 需要深拷贝，避免多个 fork 分支共享同一个 FlowData 引用导致数据相互影响 + dataToFork = (T) copyFlowData((FlowData) this.data); + } + return this.convertData(dataToFork); } /** @@ -315,6 +328,67 @@ private FlowContext copyContextWithoutID(R data) { context.archivedAt = this.archivedAt; return context; } + /** + * 标记当前context为跳过分支，使其能够穿透普通节点继续传递到后续汇聚点。 + * + * @return 当前context + */ + public FlowContext markSkippedSignal() { + if (this.data instanceof FlowData) { + // Fork 分支间可能共享同一个 FlowData 引用，先复制再打 skipped 标记，避免信号串扰到其他分支。 + FlowData flowData = copyFlowData((FlowData) this.data); + this.data = (T) flowData; + Map passData = Optional.ofNullable(flowData.getPassData()) + .map(HashMap::new) + .orElseGet(HashMap::new); + passData.put(Constant.INTERNAL_SKIPPED_SIGNAL_KEY, true); + flowData.setPassData(passData); + } + return this; + } + + private FlowData copyFlowData(FlowData source) { + return FlowData.builder() + .operator(source.getOperator()) + .startTime(source.getStartTime()) + .businessData(Optional.ofNullable(source.getBusinessData()).map(HashMap::new).orElse(null)) + .contextData(Optional.ofNullable(source.getContextData()).map(HashMap::new).orElse(null)) + .passData(Optional.ofNullable(source.getPassData()).map(HashMap::new).orElse(null)) + .errorMessage(source.getErrorMessage()) + .errorInfo(source.getErrorInfo()) + .build(); + } + + /** + * 当前context是否携带跳过分支信号。 + * + * @return 是否跳过 + */ + public boolean isSkippedSignal() { + if (FlowNodeStatus.SKIPPED.equals(this.status)) { + return true; + } + if (!(this.data instanceof FlowData)) { + return false; + } + FlowData flowData = (FlowData) this.data; + return Boolean.TRUE.equals(Optional.ofNullable(flowData.getPassData()) + .orElseGet(HashMap::new) + .get(Constant.INTERNAL_SKIPPED_SIGNAL_KEY)); + } + + private FlowContext copyContext(R data) { + FlowContext context = new FlowContext<>(this.streamId, this.rootId, data, this.traceId, this.position, + this.parallel, this.parallelMode, LocalDateTime.now()); + context.status = this.status; + context.trans = this.trans; + context.batchId = this.batchId; + context.toBatch = this.toBatch; + context.createAt = this.createAt; + context.updateAt = this.updateAt; + context.archivedAt = this.archivedAt; + return context; + } /** * joined diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowData.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowData.java index cbabd2a0e5..b859ff173a 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowData.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowData.java @@ -10,11 +10,7 @@ import com.alibaba.fastjson.TypeReference; import com.alibaba.fastjson.serializer.SerializerFeature; -import lombok.AllArgsConstructor; -import lombok.Builder; -import lombok.Getter; -import lombok.NoArgsConstructor; -import lombok.Setter; +import lombok.*; import modelengine.fit.waterflow.common.Constant; import modelengine.fit.waterflow.common.utils.ByteArraySerialiseUtilV1; import modelengine.fitframework.log.Logger; @@ -41,6 +37,7 @@ @Builder @NoArgsConstructor @AllArgsConstructor +@ToString public class FlowData { private static final Logger log = Logger.get(FlowData.class); diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextMemoRepo.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextMemoRepo.java index 2f51f593ba..9abcecca50 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextMemoRepo.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextMemoRepo.java @@ -142,7 +142,7 @@ public synchronized List> requestMappingContext(String streamId, List> all = this.contexts.stream() .filter(c -> c.getStreamId().equals(streamId)) .filter(c -> subscriptions.contains(c.getPosition())) - .filter(c -> c.getStatus() == FlowNodeStatus.PENDING) + .filter(c -> c.getStatus() == FlowNodeStatus.PENDING || c.getStatus() == FlowNodeStatus.SKIPPED) .collect(Collectors.toList()); List> filters = filter.process(all); return filters.stream().filter(c -> validator.check(c, filters)).collect(Collectors.toList()); @@ -154,7 +154,7 @@ public synchronized List> requestProducingContext(String streamId List> all = this.contexts.stream() .filter(c -> c.getStreamId().equals(streamId)) .filter(c -> subscriptions.contains(c.getPosition())) - .filter(c -> c.getStatus() == FlowNodeStatus.PENDING) + .filter(c -> c.getStatus() == FlowNodeStatus.PENDING || c.getStatus() == FlowNodeStatus.SKIPPED) .collect(Collectors.toList()); return filter.process(all); } diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextPersistRepo.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextPersistRepo.java index a621adccfd..77d672d9ed 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextPersistRepo.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/repo/flowcontext/FlowContextPersistRepo.java @@ -43,6 +43,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Objects; @@ -300,10 +301,7 @@ public List> getByIds(List ids) { @Override public List> requestMappingContext(String streamId, List subscriptions, Filter filter, Validator validator) { - List traces = this.traceOwnerService.getTraces(); - List pos = contextMapper.findBySubscriptions(streamId, subscriptions, - FlowNodeStatus.PENDING.toString(), traces); - List> all = pos.stream().map(this::serializer).collect(Collectors.toList()); + List> all = getPendingAndSkippedBySubscriptions(streamId, subscriptions, false); List> filters = filter.process(all); return filters.stream().filter(c -> validator.check(c, filters)).collect(Collectors.toList()); } @@ -311,26 +309,34 @@ public List> requestMappingContext(String streamId, List> requestProducingContext(String streamId, List subscriptions, Filter filter) { - List pos; + List> all = getPendingAndSkippedBySubscriptions(streamId, subscriptions, useLimit); + List> result = filter.process(all); + if (result.isEmpty()) { + log.info("[requestProducingContext] Empty contexts. traceIds={}, pos={}, beforeSize={}, afterSize={}.", + StringUtils.join(',', this.traceOwnerService.getTraces()), StringUtils.join(',', subscriptions), + all.size(), result.size()); + } + return result; + } + + private List> getPendingAndSkippedBySubscriptions(String streamId, List subscriptions, + boolean limited) { List traces = this.traceOwnerService.getTraces(); if (traces.isEmpty()) { log.warn("There is no trace owned."); return Collections.emptyList(); } - if (useLimit) { - pos = contextMapper.findSomeBySubscriptions(streamId, subscriptions, FlowNodeStatus.PENDING.toString(), - traces, defaultLimitation); - } else { - pos = contextMapper.findBySubscriptions(streamId, subscriptions, FlowNodeStatus.PENDING.toString(), traces); - } - List> result = - filter.process(pos.stream().map(this::serializer).collect(Collectors.toList())); - if (result.isEmpty()) { - log.info("[requestProducingContext] Empty contexts. traceIds={}, pos={}, beforeSize={}, afterSize={}.", - StringUtils.join(',', traces), StringUtils.join(',', subscriptions), pos.size(), result.size()); + List pending = limited + ? contextMapper.findSomeBySubscriptions(streamId, subscriptions, FlowNodeStatus.PENDING.toString(), + traces, defaultLimitation) + : contextMapper.findBySubscriptions(streamId, subscriptions, FlowNodeStatus.PENDING.toString(), traces); + List skipped = contextMapper.findBySubscriptions(streamId, subscriptions, + FlowNodeStatus.SKIPPED.toString(), traces); + Map contextMap = new LinkedHashMap<>(); + pending.forEach(context -> contextMap.put(context.getContextId(), context)); + skipped.forEach(context -> contextMap.put(context.getContextId(), context)); + return contextMap.values().stream().map(this::serializer).collect(Collectors.toList()); } - return result; - } @Override public List> findByStreamId(String metaId, String version) { diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/FlowConditionNode.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/FlowConditionNode.java index f805feb297..2be17a4554 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/FlowConditionNode.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/FlowConditionNode.java @@ -67,8 +67,9 @@ public class FlowConditionNode extends FlowNode { public FitStream.Processor getProcessor(String streamId, FlowContextRepo repo, FlowContextMessenger messenger, FlowLocks locks) { if (!Optional.ofNullable(this.processor).isPresent()) { + // AIPP 流程启用 condition 分支跳过机制，未匹配分支创建 forked context 并标记为 SKIPPED this.processor = new ConditionsNode<>(streamId, this.metaId, this::conditionalJuster, repo, messenger, - locks, this.type); + locks, this.type, true); this.processor.onError(errorHandler(streamId)); setCallback(this.processor, messenger); setGlobalTrace(this.processor, messenger); diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowGeneralJober.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowGeneralJober.java index 590cb65cbc..ce1dc7ee4f 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowGeneralJober.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowGeneralJober.java @@ -40,6 +40,6 @@ protected List executeJober(List inputs) { } log.info("Remote invoke success,nodeId: {}, fitable id is {}.", this.nodeMetaId, fitableId); } - return convertToFlowData(outputEntities, inputs.get(0)); + return convertToFlowData(outputEntities, inputs); } } diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowHttpJober.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowHttpJober.java index f6c1b18eab..18c9839b17 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowHttpJober.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowHttpJober.java @@ -38,7 +38,7 @@ protected List executeJober(List inputs) { throw new WaterflowException(ex, ErrorCodes.FLOW_HTTP_JOBER_INVOKE_ERROR); } log.info("Remote invoke success,nodeId: {}, fitable id is {}.", this.nodeMetaId, HTTP_JOBER_FITABLE); - return convertToFlowData(outputs, inputs.get(0)); + return convertToFlowData(outputs, inputs); } } diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowJober.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowJober.java index 6040af0e41..e4114ba3ea 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowJober.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowJober.java @@ -43,6 +43,7 @@ import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.LinkedHashSet; import java.util.List; @@ -275,6 +276,31 @@ protected List> getInputs(List inputs) { return contextData; } + /** + * convertToFlowData + * + * @param outputEntities outputEntities + * @param inputs inputs列表 + * @return List + */ + protected List convertToFlowData(List> outputEntities, List inputs) { + List result = new ArrayList<>(); + int inputSize = inputs.size(); + for (int i = 0; i < outputEntities.size(); i++) { + Map output = outputEntities.get(i); + // 按索引顺序从 inputs 中获取对应的 FlowData，如果索引超出则使用第一个 + FlowData input = i < inputSize ? inputs.get(i) : inputs.get(0); + result.add(FlowData.builder() + .operator(input.getOperator()) + .startTime(input.getStartTime()) + .businessData(cast(output.get("businessData"))) + .contextData(new HashMap<>(input.getContextData())) + .passData(cast(output.get("passData"))) + .build()); + } + return result; + } + /** * convertToFlowData * @@ -283,15 +309,7 @@ protected List> getInputs(List inputs) { * @return List */ protected List convertToFlowData(List> outputEntities, FlowData input) { - return outputEntities.stream() - .map(output -> FlowData.builder() - .operator(input.getOperator()) - .startTime(input.getStartTime()) - .businessData(cast(output.get("businessData"))) - .contextData(new HashMap<>(input.getContextData())) - .passData(cast(output.get("passData"))) - .build()) - .collect(Collectors.toList()); + return convertToFlowData(outputEntities, Collections.singletonList(input)); } /** diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowOhScriptJober.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowOhScriptJober.java index e8033cdf6d..122ceaa26d 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowOhScriptJober.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/definitions/nodes/jobers/FlowOhScriptJober.java @@ -40,6 +40,6 @@ protected List executeJober(List inputs) { .format(SerializationFormat.CBOR) .invoke(inputData); log.info("Remote invoke success,nodeId: {}, fitable id is {}.", this.nodeMetaId, OHSCRIPT_FITABLE); - return convertToFlowData(outputEntities, inputs.get(0)); + return convertToFlowData(outputEntities, inputs); } } diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/enums/FlowNodeStatus.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/enums/FlowNodeStatus.java index ab776c1c23..c091bd0f12 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/enums/FlowNodeStatus.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/enums/FlowNodeStatus.java @@ -20,6 +20,7 @@ public enum FlowNodeStatus { NEW, PENDING, + SKIPPED, READY, // 未更新数据库 PROCESSING, // 未更新数据库 ARCHIVED, @@ -27,7 +28,7 @@ public enum FlowNodeStatus { ERROR, RETRYABLE; - private static final Set END_STATUS = new HashSet<>(Arrays.asList(ARCHIVED, ERROR, TERMINATE)); + private static final Set END_STATUS = new HashSet<>(Arrays.asList(ARCHIVED, ERROR, TERMINATE,SKIPPED)); /** * 是否是终态 diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java index f3ff4a07f7..f9ba6f9476 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/To.java @@ -461,10 +461,7 @@ private void process() { } private void handlePreProcessConcurrentConflict() { - List> concurrentConflictContexts = this.preFilter() - .process(repo.getContextsByPosition(this.streamId, - this.froms.stream().map(Identity::getId).collect(Collectors.toList()), - FlowNodeStatus.PENDING.toString())); + List> concurrentConflictContexts = this.preFilter().process(getPendingAndSkippedContexts()); if (CollectionUtils.isEmpty(concurrentConflictContexts) || inParallelMode(concurrentConflictContexts)) { return; } @@ -487,10 +484,7 @@ private List> requestReady() { locks.streamNodeLockKey(this.streamId, this.id, ProcessType.PRE_PROCESS.toString())); lock.lock(); try { - List> contexts = this.preFilter() - .process(repo.getContextsByPosition(this.streamId, - this.froms.stream().map(Identity::getId).collect(Collectors.toList()), - FlowNodeStatus.PENDING.toString())); + List> contexts = this.preFilter().process(getPendingAndSkippedContexts()); contexts = filterTerminate(contexts); if (CollectionUtils.isEmpty(contexts)) { return new ArrayList<>(); @@ -502,6 +496,20 @@ private List> requestReady() { } } + private List> getPendingAndSkippedContexts() { + List fromIds = this.froms.stream().map(Identity::getId).collect(Collectors.toList()); + Map> contextMap = new LinkedHashMap<>(); + for (Object item : repo.getContextsByPosition(this.streamId, fromIds, FlowNodeStatus.PENDING.toString())) { + FlowContext context = ObjectUtils.cast(item); + contextMap.put(context.getId(), context); + } + for (Object item : repo.getContextsByPosition(this.streamId, fromIds, FlowNodeStatus.SKIPPED.toString())) { + FlowContext context = ObjectUtils.cast(item); + contextMap.put(context.getId(), context); + } + return new ArrayList<>(contextMap.values()); + } + @Override public void block(Blocks.Block block) { this.isAuto = false; @@ -699,12 +707,17 @@ public void onProcess(List> pre) { String.join(",", pre.get(0).getTraceId())); return; } + List> executableInputs = filterExecutableInputs(pre); + if (CollectionUtils.isEmpty(executableInputs)) { + this.afterProcess(pre, generateSkippedOutputs(pre)); + return; + } beforeProcess(pre); - if (pre.size() == 1 && pre.get(0).getData() == null) { + if (executableInputs.size() == 1 && executableInputs.get(0).getData() == null) { this.afterProcess(pre, new ArrayList<>()); return; } - List> processInputs = mergeProcessInputs(pre); + List> processInputs = mergeProcessInputs(executableInputs); if (this.isAsyncJob) { beforeAsyncProcess(pre); this.getProcessMode().process(this, processInputs); @@ -758,6 +771,46 @@ private List> mergeProcessInputs(List> pre) { } FlowContext merged = this.merger.merge(pre); return merged != null ? Collections.singletonList(merged) : pre; + + } + + private List> filterExecutableInputs(List> contexts) { + return contexts.stream().filter(context -> !context.isSkippedSignal()).collect(Collectors.toList()); + } + + private List> generateSkippedOutputs(List> pre) { + if (CollectionUtils.isEmpty(pre) || FlowNodeType.END.equals(this.nodeType)) { + return Collections.emptyList(); + } + FlowContext baseContext = pre.get(0); + FlowContext skippedOutput = ObjectUtils.cast(baseContext.generate(ObjectUtils.cast(baseContext.getData()), + this.getId(), LocalDateTime.now()).markSkippedSignal()); + return Collections.singletonList(skippedOutput); + } + + private FlowData mergeFlowData(List> pre) { + FlowData first = ObjectUtils.cast(pre.get(0).getData()); + Map businessData = new HashMap<>(Optional.ofNullable(first.getBusinessData()).orElseGet(HashMap::new)); + Map contextData = new HashMap<>(Optional.ofNullable(first.getContextData()).orElseGet(HashMap::new)); + Map passData = new HashMap<>(Optional.ofNullable(first.getPassData()).orElseGet(HashMap::new)); + + pre.stream().skip(1).map(FlowContext::getData).map(ObjectUtils::cast).forEach(flowData -> { + businessData.putAll(FlowUtil.mergeMaps(businessData, + Optional.ofNullable(flowData.getBusinessData()).orElseGet(HashMap::new))); + contextData.putAll(FlowUtil.mergeMaps(contextData, + Optional.ofNullable(flowData.getContextData()).orElseGet(HashMap::new))); + passData.putAll(FlowUtil.mergeMaps(passData, + Optional.ofNullable(flowData.getPassData()).orElseGet(HashMap::new))); + }); + return FlowData.builder() + .operator(first.getOperator()) + .startTime(first.getStartTime()) + .businessData(businessData) + .contextData(contextData) + .passData(passData) + .errorMessage(first.getErrorMessage()) + .errorInfo(first.getErrorInfo()) + .build(); } private boolean isOwnTrace(List> pre) { @@ -859,6 +912,9 @@ public void afterProcess(List> preContexts, List> this.getRepo().updateProcessStatus(preContexts); LOG.debug("afterProcess after updateProcessStatus"); if (CollectionUtils.isEmpty(after)) { + if (FlowNodeType.END.equals(this.nodeType) && CollectionUtils.isNotEmpty(preContexts)) { + callback(preContexts, ObjectUtils.cast(preContexts)); + } return; } if (FlowNodeType.END.equals(this.nodeType)) { @@ -884,9 +940,17 @@ public void afterProcess(List> preContexts, List> private void callback(List> preContexts, List> after) { LocalDateTime createAt = preContexts.get(0).getCreateAt(); LocalDateTime archivedAt = LocalDateTime.now(); - feedback(after.stream().map(context -> { + List> contextsToFeedback = CollectionUtils.isEmpty(after) + ? preContexts.stream().map(c -> ObjectUtils.>cast(c)).collect(Collectors.toList()) + : after; + feedback(contextsToFeedback.stream().map(context -> { FlowContext callbackContext = context.generate(context.getData(), context.getPosition(), createAt); + if (context.isSkippedSignal()) { + callbackContext.setStatus(FlowNodeStatus.SKIPPED).markSkippedSignal(); + } else { + callbackContext.setStatus(FlowNodeStatus.ARCHIVED); + } callbackContext.setArchivedAt(archivedAt); callbackContext.setNextPositionId(context.getNextPositionId()); return callbackContext; @@ -1241,4 +1305,5 @@ private void handleProcessConcurrentConflict(To to) { /* 多个数据到达后采用的处理方式，ANY表示即到即用，ALL表示所有数据到来才能使用 * */ + } diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/When.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/When.java index 161c5e19d1..d1f5dd8840 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/When.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/When.java @@ -96,11 +96,20 @@ public void cache(List> contexts) { // 将context发送到节点边上，更新为PENDING状态，等待下一个节点处理 // 该过程不产生新的context数据，只更新context的状态 List> converted = contexts.stream() - .map(c -> c.convertData(this.converter.process(c.getData()), c.getId()) - .setPosition(this.getId()) - .setStatus(FlowNodeStatus.PENDING)) + .map(c -> { + FlowNodeStatus edgeStatus = c.isSkippedSignal() ? FlowNodeStatus.SKIPPED : FlowNodeStatus.PENDING; + FlowContext convertedContext = c.convertData(this.converter.process(c.getData()), c.getId()) + .setPosition(this.getId()) + .setStatus(edgeStatus); + if (c.isSkippedSignal()) { + convertedContext.markSkippedSignal(); + } + return convertedContext; + }) .collect(Collectors.toList()); - repo.updateStatus(converted, converted.get(0).getStatus().toString(), converted.get(0).getPosition()); + converted.stream() + .collect(Collectors.groupingBy(FlowContext::getStatus)) + .forEach((status, group) -> repo.updateStatus(group, status.toString(), group.get(0).getPosition())); messenger.send(this.to.isAuto() ? ProcessType.PROCESS : ProcessType.PRE_PROCESS, this.to, converted); } diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/nodes/ConditionsNode.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/nodes/ConditionsNode.java index 63c1976001..754194404a 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/nodes/ConditionsNode.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/domain/flows/streams/nodes/ConditionsNode.java @@ -7,18 +7,25 @@ package modelengine.fit.waterflow.flowsengine.domain.flows.streams.nodes; import modelengine.fit.waterflow.flowsengine.domain.flows.context.FlowContext; +import modelengine.fit.waterflow.flowsengine.domain.flows.context.FlowData; import modelengine.fit.waterflow.flowsengine.domain.flows.context.repo.flowcontext.FlowContextMessenger; import modelengine.fit.waterflow.flowsengine.domain.flows.context.repo.flowcontext.FlowContextRepo; import modelengine.fit.waterflow.flowsengine.domain.flows.context.repo.flowlock.FlowLocks; +import modelengine.fit.waterflow.flowsengine.domain.flows.enums.FlowNodeStatus; import modelengine.fit.waterflow.flowsengine.domain.flows.enums.FlowNodeType; import modelengine.fit.waterflow.flowsengine.domain.flows.streams.FitStream; import modelengine.fit.waterflow.flowsengine.domain.flows.streams.From; import modelengine.fit.waterflow.flowsengine.domain.flows.streams.Processors; import modelengine.fit.waterflow.flowsengine.domain.flows.streams.callbacks.PreSendCallbackInfo; +import modelengine.fitframework.util.CollectionUtils; +import java.util.ArrayList; +import java.util.HashSet; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.concurrent.locks.Lock; import java.util.function.Consumer; import java.util.stream.Collectors; @@ -33,6 +40,12 @@ * @since 2023/08/14 */ public class ConditionsNode extends Node { + /** + * 是否启用 condition 分支跳过机制 + * 启用后，未匹配的分支也会创建 forked context 并标记为 SKIPPED + */ + private boolean enableConditionSkip = false; + /** * 1->1处理节点 * @@ -44,10 +57,26 @@ public class ConditionsNode extends Node { */ public ConditionsNode(String streamId, Processors.Just> processor, FlowContextRepo repo, FlowContextMessenger messenger, FlowLocks locks) { + this(streamId, processor, repo, messenger, locks, false); + } + + /** + * 1->1处理节点 + * + * @param streamId stream流程ID + * @param processor 对应处理器 + * @param repo 上下文持久化repo，默认在内存 + * @param messenger 上下文事件发送器，默认在内存 + * @param locks 流程锁 + * @param enableConditionSkip 是否启用 condition 分支跳过机制 + */ + public ConditionsNode(String streamId, Processors.Just> processor, FlowContextRepo repo, + FlowContextMessenger messenger, FlowLocks locks, boolean enableConditionSkip) { super(streamId, i -> { processor.process(i); return i.getData(); - }, repo, messenger, locks, () -> initFrom(streamId, repo, messenger, locks)); + }, repo, messenger, locks, () -> initFrom(streamId, repo, messenger, locks, enableConditionSkip)); + this.enableConditionSkip = enableConditionSkip; } /** @@ -63,9 +92,41 @@ public ConditionsNode(String streamId, Processors.Just> processor */ public ConditionsNode(String streamId, String nodeId, Processors.Just> processor, FlowContextRepo repo, FlowContextMessenger messenger, FlowLocks locks, FlowNodeType nodeType) { - this(streamId, processor, repo, messenger, locks); + this(streamId, nodeId, processor, repo, messenger, locks, nodeType, false); + } + + /** + * 1->1处理节点 + * + * @param streamId stream流程ID + * @param nodeId stream流程节点ID + * @param processor 对应处理器 + * @param repo 上下文持久化repo，默认在内存 + * @param messenger 上下文事件发送器，默认在内存 + * @param locks 流程锁 + * @param nodeType 节点类型 + * @param enableConditionSkip 是否启用 condition 分支跳过机制 + */ + public ConditionsNode(String streamId, String nodeId, Processors.Just> processor, FlowContextRepo repo, + FlowContextMessenger messenger, FlowLocks locks, FlowNodeType nodeType, boolean enableConditionSkip) { + super(streamId, i -> { + processor.process(i); + return i.getData(); + }, repo, messenger, locks, () -> initFrom(streamId, repo, messenger, locks, enableConditionSkip)); this.id = nodeId; this.nodeType = nodeType; + this.enableConditionSkip = enableConditionSkip; + } + + /** + * 设置是否启用 condition 分支跳过机制 + * + * @param enableConditionSkip 是否启用 + * @return 当前节点实例，支持链式调用 + */ + public ConditionsNode setEnableConditionSkip(boolean enableConditionSkip) { + this.enableConditionSkip = enableConditionSkip; + return this; } /** @@ -77,23 +138,69 @@ public ConditionsNode(String streamId, String nodeId, Processors.Just From initFrom(String streamId, FlowContextRepo repo, FlowContextMessenger messenger, - FlowLocks locks) { + FlowLocks locks, boolean enableConditionSkip) { return new From(streamId, repo, messenger, locks) { @Override public void offer(List> contexts, Consumer> preSendCallback) { + if (CollectionUtils.isEmpty(contexts) || CollectionUtils.isEmpty(this.getSubscriptions())) { + preSendCallback.accept(new PreSendCallbackInfo<>(new LinkedHashMap<>(), contexts)); + return; + } Map, List>> matchedContexts = new LinkedHashMap<>(); - this.getSubscriptions().forEach(w -> { - List> matched = contexts.stream() - .filter(c -> w.getWhether().is(c)) - .peek(c -> c.setNextPositionId(w.getId())) - .collect(Collectors.toList()); - matched.forEach(contexts::remove); - matchedContexts.put(w, matched); + List> forkedContexts = new ArrayList<>(); + Set consumedContextIds = new HashSet<>(); + List> subscriptions = this.getSubscriptions(); + contexts.forEach(context -> { + int matchedIndex = -1; + for (int index = 0; index < subscriptions.size(); index++) { + if (subscriptions.get(index).getWhether().is(context)) { + matchedIndex = index; + break; + } + } + consumedContextIds.add(context.getId()); + for (int index = 0; index < subscriptions.size(); index++) { + FitStream.Subscription subscription = subscriptions.get(index); + boolean useOriginalContext = index == matchedIndex; + boolean isSkippedBranch = index != matchedIndex; + // 根据 enableConditionSkip 标记决定是否创建 skipped context + // enableConditionSkip=true 时，未匹配分支创建 forked context 并标记 skipped + // enableConditionSkip=false 时，非 FlowData 类型的未匹配分支直接忽略 + if (useOriginalContext) { + context.setNextPositionId(subscription.getId()); + matchedContexts.computeIfAbsent(subscription, key -> new ArrayList<>()).add(context); + } else if (isSkippedBranch && enableConditionSkip) { + FlowContext branchContext = context.fork(); + branchContext.setNextPositionId(subscription.getId()); + branchContext.setStatus(FlowNodeStatus.SKIPPED).markSkippedSignal(); + matchedContexts.computeIfAbsent(subscription, key -> new ArrayList<>()).add(branchContext); + forkedContexts.add(branchContext); + } + // enableConditionSkip=false 且未匹配分支：直接忽略，不创建任何 context + } }); - PreSendCallbackInfo callbackInfo = new PreSendCallbackInfo<>(matchedContexts, contexts); + List> unMatchedContexts = contexts.stream() + .filter(context -> !consumedContextIds.contains(context.getId())) + .collect(Collectors.toList()); + if (CollectionUtils.isNotEmpty(forkedContexts)) { + Set traces = forkedContexts.stream() + .flatMap(context -> context.getTraceId().stream()) + .collect(Collectors.toSet()); + Lock lock = this.locks.getDistributedLock(this.locks.streamNodeLockKey(this.getStreamId(), + this.getId(), "ConditionForkContextPool")); + lock.lock(); + try { + this.repo.updateContextPool(forkedContexts, traces); + this.repo.save(forkedContexts); + } finally { + lock.unlock(); + } + } + PreSendCallbackInfo callbackInfo = new PreSendCallbackInfo<>(matchedContexts, unMatchedContexts); preSendCallback.accept(callbackInfo); matchedContexts.forEach(FitStream.Subscription::cache); } diff --git a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/fitable/TraceServiceImpl.java b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/fitable/TraceServiceImpl.java index f142b74fd4..2a0a0202dd 100644 --- a/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/fitable/TraceServiceImpl.java +++ b/app-builder/waterflow/java/waterflow-service/src/main/java/modelengine/fit/waterflow/flowsengine/fitable/TraceServiceImpl.java @@ -92,11 +92,17 @@ public void callback(List> contexts) { contexts.forEach(context -> { String nodeId = getValueOfSpecifyKey(context, Constant.NODE_ID_KEY); FlowNodePublishInfo flowNodePublishInfo = constructFlowNodePublishInfo(context, nodeId, new FlowErrorInfo(), - FlowNodeStatus.ARCHIVED.name()); + resolveCallbackStatus(context)); publishNodeInfo(flowNodePublishInfo); }); } + private static String resolveCallbackStatus(Map context) { + String status = cast(context.get("status")); + return FlowNodeStatus.SKIPPED.name().equals(status) ? FlowNodeStatus.SKIPPED.name() + : FlowNodeStatus.ARCHIVED.name(); + } + @Fitable("modelengine.fit.jober.fitable.FlowInfoException") @Override public void handleException(String nodeId, List> contexts, FlowErrorInfo errorMessage) { diff --git a/app-builder/waterflow/java/waterflow-service/src/test/java/modelengine/fit/waterflow/FlowsDataBaseTest.java b/app-builder/waterflow/java/waterflow-service/src/test/java/modelengine/fit/waterflow/FlowsDataBaseTest.java index e9e2c06d4e..8f9960534f 100644 --- a/app-builder/waterflow/java/waterflow-service/src/test/java/modelengine/fit/waterflow/FlowsDataBaseTest.java +++ b/app-builder/waterflow/java/waterflow-service/src/test/java/modelengine/fit/waterflow/FlowsDataBaseTest.java @@ -442,7 +442,7 @@ protected void assertFlowsExecutorWithConditionNodeFirstBranchTrue(FlowData flow List> contexts, List> allContexts) { assertEquals(1, contexts.size()); assertEquals(ARCHIVED, contexts.get(0).getStatus()); - assertEquals(6, allContexts.size()); + assertEquals(8, allContexts.size()); allContexts.forEach(c -> assertEquals(ARCHIVED, c.getStatus())); Map resultBusinessData = contexts.get(0).getData().getBusinessData(); assertTrue((boolean) getBusinessDataFromChainedKey(resultBusinessData, "cmc.approved")); @@ -461,8 +461,7 @@ protected void assertFlowsExecutorWithConditionNodeFirstFalseBranch(FlowData flo List> contexts, List> allContexts) { assertEquals(1, contexts.size()); assertEquals(ARCHIVED, contexts.get(0).getStatus()); - assertEquals(3, allContexts.size()); - allContexts.forEach(c -> assertEquals(ARCHIVED, c.getStatus())); + assertEquals(6, allContexts.size()); Map resultBusinessData = contexts.get(0).getData().getBusinessData(); assertFalse((boolean) getBusinessDataFromChainedKey(resultBusinessData, "cmc.approved")); assertTrue((boolean) getBusinessDataFromChainedKey(resultBusinessData, "committer.approved")); @@ -480,7 +479,7 @@ protected void assertFlowsExecutorWithConditionNodeSecondFalseBranch(FlowData fl List> contexts, List> allContexts) { assertEquals(1, contexts.size()); assertEquals(ARCHIVED, contexts.get(0).getStatus()); - assertEquals(5, allContexts.size()); + assertEquals(8, allContexts.size()); allContexts.forEach(c -> assertEquals(ARCHIVED, c.getStatus())); Map resultBusinessData = contexts.get(0).getData().getBusinessData(); @@ -651,9 +650,9 @@ protected void assertFlowsExecuteProduceFromMToNWithMinimumSizeOneInSingleThread */ protected void assertFlowsExecuteProduceFromMToNForOfferMultiData(List> contexts, List> allContexts) { - assertEquals(3, contexts.size()); + assertEquals(5, contexts.size()); contexts.forEach(c -> assertEquals(ARCHIVED, c.getStatus())); - assertEquals(14, allContexts.size()); + assertEquals(17, allContexts.size()); allContexts.forEach(c -> assertEquals(ARCHIVED, c.getStatus())); } @@ -680,9 +679,9 @@ protected void assertFlowsExecuteFilterFromMToN(FlowContextRepo repo, */ protected void assertFlowsExecuteProduceFromMToNForOfferOneData(List> contexts, List> allContexts) { - assertEquals(3, contexts.size()); + assertEquals(5, contexts.size()); contexts.forEach(c -> assertEquals(ARCHIVED, c.getStatus())); - assertEquals(10, allContexts.size()); + assertEquals(13, allContexts.size()); allContexts.forEach(c -> assertEquals(ARCHIVED, c.getStatus())); } diff --git a/app-builder/waterflow/java/waterflow-service/src/test/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContextPersistTest.java b/app-builder/waterflow/java/waterflow-service/src/test/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContextPersistTest.java index c869627381..886a5c3608 100644 --- a/app-builder/waterflow/java/waterflow-service/src/test/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContextPersistTest.java +++ b/app-builder/waterflow/java/waterflow-service/src/test/java/modelengine/fit/waterflow/flowsengine/domain/flows/context/FlowContextPersistTest.java @@ -470,7 +470,12 @@ void testFlowsExecutorWithConditionNodeFirstBranchFalse() { List> contexts = FlowsTestUtil.waitSingle( contextSupplier(memRepo, streamId, traceId, flowNode.getMetaId(), FlowNodeStatus.ARCHIVED)); List> all = this.getContextsByTraceWrapper(memRepo, traceId); + for (FlowContext flowDataFlowContext : all) { + System.out.println(flowDataFlowContext.toString()); + } + + System.out.println(all.size()); assertFlowsExecutorWithConditionNodeFirstFalseBranch(flowData, contexts, all); } @@ -526,7 +531,7 @@ void testFlowsExecutorConditionNodeWithError() { contextSupplier(memRepo, streamId, traceId, flowNode.getMetaId(), FlowNodeStatus.ERROR), 1, MAX_WAIT_TIME_MS); List> all = this.getContextsByTraceWrapper(memRepo, traceId); - assertFlowsExecutorStateNodeWithException(flowDefinition, metaId, contexts, all, 3, expectException); + assertFlowsExecutorStateNodeWithException(flowDefinition, metaId, contexts, all, 4, expectException); } } @@ -840,9 +845,8 @@ void testFlowsExecuteProduceFromMToNForOfferOneData() { FlowNode flowNode = flowDefinition.getFlowNode(FlowNodeType.END); List> contexts = FlowsTestUtil.waitSize( - contextSupplier(memRepo, streamId, traceId, flowNode.getMetaId(), FlowNodeStatus.ARCHIVED), 3); + contextSupplier(memRepo, streamId, traceId, flowNode.getMetaId(), FlowNodeStatus.ARCHIVED), 5); List> all = this.getContextsByTraceWrapper(memRepo, traceId); - assertFlowsExecuteProduceFromMToNForOfferOneData(contexts, all); } @@ -878,10 +882,9 @@ void testFlowsExecuteProduceFromMToNForOfferMultiData() { FlowNode flowNode = flowDefinition.getFlowNode(FlowNodeType.END); List> contexts = FlowsTestUtil.waitSize( - contextSupplier(memRepo, streamId, traceId, flowNode.getMetaId(), FlowNodeStatus.ARCHIVED), 3, + contextSupplier(memRepo, streamId, traceId, flowNode.getMetaId(), FlowNodeStatus.ARCHIVED), 5, MAX_WAIT_TIME_MS); List> all = this.getContextsByTraceWrapper(memRepo, traceId); - assertFlowsExecuteProduceFromMToNForOfferMultiData(contexts, all); }