From c7c301d3b95bf8f0e22cf6b6d8c00ffa86267dd7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Tue, 8 Sep 2026 10:49:02 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E7=BB=9F=E4=B8=80=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E4=B8=89=E4=B8=AA=E5=87=BA=E5=8F=A3=E7=9A=84=E9=94=99?= =?UTF-8?q?=E8=AF=AF=E5=8F=8D=E9=A6=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 关联 EASY-2，补齐安全错误、节点展示与执行标识 - 保留历史可读摘要并验证接口、SSE 与界面兼容 --- .../controller/ai/WorkflowChatController.java | 1 + .../controller/ai/WorkflowController.java | 2 + .../service/ai/WorkflowChatEventStream.java | 80 ++++++--- .../service/ai/WorkflowPublicChatService.java | 33 +--- .../ai/WorkflowChatEventStreamTest.java | 53 +++++- .../ai/WorkflowPublicChatServiceTest.java | 2 + .../controller/PublicWorkflowController.java | 2 + .../dto/PublicWorkflowNodeStatus.java | 7 +- .../dto/PublicWorkflowStatusError.java | 9 + .../PublicWorkflowStatusSanitizer.java | 147 ++++------------ .../PublicWorkflowStatusSanitizerTest.java | 48 ++++- .../ai/easyagentsflow/entity/ChainInfo.java | 4 + .../ai/easyagentsflow/entity/NodeInfo.java | 4 + .../entity/WorkflowExecutionError.java | 32 ++++ .../listener/ChainEventListenerForSave.java | 10 +- .../service/TinyFlowService.java | 53 +----- .../service/WorkflowExecutionErrorMapper.java | 77 ++++++++ .../WorkflowApiUploadLifecycleService.java | 2 +- .../service/TinyFlowServiceTest.java | 52 +++++- .../WorkflowFailureDiagnosticsTest.java | 165 ++++++++++++++++++ easyflow-ui-admin/app/src/api/request.ts | 10 +- .../app/src/api/sseRequestLifecycle.test.ts | 22 ++- .../app/src/api/sseRequestLifecycle.ts | 21 +++ .../src/views/ai/workflow/WorkflowList.vue | 2 +- .../ai/workflow/components/ExecResult.vue | 3 +- .../ai/workflow/components/SingleRun.vue | 57 ++++-- .../workflow/components/WorkflowChatPage.vue | 115 ++++++++---- .../components/WorkflowErrorDetail.vue | 44 +++++ .../ai/workflow/components/WorkflowSteps.vue | 48 +++-- .../components/__tests__/ExecResult.test.ts | 45 ++--- .../components/__tests__/SingleRun.test.ts | 78 +++++++++ .../__tests__/WorkflowSteps.test.ts | 75 ++++++++ .../workflowExecutionDetails.test.ts | 106 +++++++++++ .../components/workflowExecutionDetails.ts | 94 +++++++--- .../components/workflowExecutionError.ts | 26 +++ 35 files changed, 1171 insertions(+), 358 deletions(-) create mode 100644 easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/WorkflowExecutionError.java create mode 100644 easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowExecutionErrorMapper.java create mode 100644 easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowFailureDiagnosticsTest.java create mode 100644 easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowErrorDetail.vue create mode 100644 easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/SingleRun.test.ts create mode 100644 easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionError.ts diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowChatController.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowChatController.java index 36450a4f..4cab2975 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowChatController.java +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowChatController.java @@ -224,6 +224,7 @@ public class WorkflowChatController { Map detail = new LinkedHashMap<>(); detail.put("record", recordView); detail.put("steps", stepViews); + detail.put("runtime", eventStream.runtimeView(executeId)); return Result.ok(detail); } diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowController.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowController.java index c5b3d916..336a4b29 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowController.java +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowController.java @@ -1,5 +1,6 @@ package tech.easyflow.admin.controller.ai; +import tech.easyflow.ai.easyagentsflow.service.WorkflowExecutionErrorMapper; import cn.dev33.satoken.annotation.SaCheckPermission; import cn.dev33.satoken.stp.StpUtil; import cn.hutool.core.io.IoUtil; @@ -261,6 +262,7 @@ public class WorkflowController extends BaseCurdController res = chainExecutor.executeNode(workflowId.toString(), nodeId, variables); return Result.ok(res); } diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowChatEventStream.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowChatEventStream.java index cc997bc8..cbfebe27 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowChatEventStream.java +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowChatEventStream.java @@ -4,9 +4,14 @@ import com.alibaba.fastjson.JSON; import com.easyagents.flow.core.chain.Chain; import com.easyagents.flow.core.chain.ChainConsts; import com.easyagents.flow.core.chain.ChainStatus; +import com.easyagents.flow.core.chain.ChainState; import com.easyagents.flow.core.chain.Edge; import com.easyagents.flow.core.chain.Event; import com.easyagents.flow.core.chain.Node; +import com.easyagents.flow.core.chain.NodeStatus; +import com.easyagents.flow.core.chain.ExceptionSummary; +import tech.easyflow.ai.easyagentsflow.entity.WorkflowExecutionError; +import tech.easyflow.ai.easyagentsflow.service.WorkflowExecutionErrorMapper; import com.easyagents.flow.core.chain.event.ChainStatusChangeEvent; import com.easyagents.flow.core.chain.event.EdgeConditionCheckFailedEvent; import com.easyagents.flow.core.chain.event.EdgeTriggerEvent; @@ -39,6 +44,41 @@ import java.util.concurrent.atomic.AtomicLong; @Service public class WorkflowChatEventStream { + public Map runtimeView(String executeId) { + try { + ChainState state = chainExecutor.getChainStateRepository() + .load(executeId); + if (state == null || state.getStatus() == null) { + return Map.of(); + } + Map view = new LinkedHashMap<>(); + view.put("status", state.getStatus().name()); + view.put("statusValue", state.getStatus().getValue()); + WorkflowExecutionError error = WorkflowExecutionErrorMapper.chain(state.getError(), state.getStatus()); + view.put("error", error); + view.put("message", state.getStatus() == ChainStatus.SUSPEND ? state.getMessage() + : WorkflowExecutionErrorMapper.summary(error)); + if (state.getStatus() == ChainStatus.SUSPEND) { + view.put("parameters", state.getSuspendForParameters()); + } + if (state.getStatus() == ChainStatus.SUCCEEDED) { + view.put( + "output", + WorkflowChatEventStream.visibleFinalOutput( + state.getExecuteResult()) + ); + } + return view; + } catch (RuntimeException error) { + log.warn( + "failed to load public workflow runtime state, executeId={}", + executeId, + error + ); + return Map.of(); + } + } + private static final Logger log = LoggerFactory.getLogger(WorkflowChatEventStream.class); private static final long SSE_TIMEOUT_MILLIS = 30L * 60L * 1000L; @@ -211,9 +251,9 @@ public class WorkflowChatEventStream { StreamSession session = findSession(chain); if (session != null && Objects.equals(chain.getStateInstanceId(), session.executeId)) { - session.send("execution_error", Map.of( - "message", safeErrorMessage(error) - )); + WorkflowExecutionError detail = WorkflowExecutionErrorMapper.map( + chain.getState().getError(), true, null, null, false); + session.send("execution_error", Map.of("message", detail.getMessage(), "error", detail)); } } @@ -269,20 +309,6 @@ public class WorkflowChatEventStream { } } - /** - * 读取适合返回给用户的异常信息。 - * - * @param error 异常 - * @return 非空异常信息 - */ - private String safeErrorMessage(Throwable error) { - if (error == null || error.getMessage() == null - || error.getMessage().isBlank()) { - return "工作流执行失败"; - } - return error.getMessage(); - } - /** * 去掉顶级工作流结果中的内部状态控制字段。 * @@ -431,8 +457,13 @@ public class WorkflowChatEventStream { data.put("output", event.getResult() == null ? Map.of() : event.getResult()); - if (event.getError() != null) { - data.put("error", safeErrorMessage(event.getError())); + if (event.getErrorSummary() != null) { + WorkflowExecutionError detail = WorkflowExecutionErrorMapper.node(event.getErrorSummary(), + event.getStatus() == null ? NodeStatus.FAILED : event.getStatus(), node.getId(), node.getName()); + if (detail != null) { + data.put("error", detail.getMessage()); + data.put("errorDetail", detail); + } } send("node_finished", nodePayload(node, data)); } @@ -561,6 +592,11 @@ public class WorkflowChatEventStream { Map data = new LinkedHashMap<>(); data.put("status", status.name()); data.put("message", chain.getState().getMessage()); + WorkflowExecutionError detail = WorkflowExecutionErrorMapper.chain(chain.getState().getError(), status); + if (detail != null) { + data.put("error", detail); + data.put("message", WorkflowExecutionErrorMapper.summary(detail)); + } if (status == ChainStatus.SUCCEEDED) { data.put( "output", @@ -614,8 +650,12 @@ public class WorkflowChatEventStream { */ private void fail(Throwable error) { if (terminal.compareAndSet(false, true)) { + WorkflowExecutionError detail = WorkflowExecutionErrorMapper.map( + error == null ? null : new ExceptionSummary(error), true, null, null, false); send("execution_failed", Map.of( - "message", safeErrorMessage(error) + "status", ChainStatus.FAILED.name(), + "message", detail.getMessage(), + "error", detail )); removeSession(this); if (connected.compareAndSet(true, false)) { diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatService.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatService.java index 7d3fc1cc..673dfac5 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatService.java +++ b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatService.java @@ -161,7 +161,7 @@ public class WorkflowPublicChatService { .eq(WorkflowExecStep::getRecordId, record.getId()) .orderBy(WorkflowExecStep::getStartTime, true) )); - return buildExecutionDetail(record, steps, runtimeView(executeId)); + return buildExecutionDetail(record, steps, eventStream.runtimeView(executeId)); } /** @@ -292,35 +292,4 @@ public class WorkflowPublicChatService { /** * 构建刷新恢复所需的最小 Runtime 视图。 */ - private Map runtimeView(String executeId) { - try { - ChainState state = chainExecutor.getChainStateRepository() - .load(executeId); - if (state == null || state.getStatus() == null) { - return Map.of(); - } - Map view = new LinkedHashMap<>(); - view.put("status", state.getStatus().name()); - view.put("statusValue", state.getStatus().getValue()); - view.put("message", state.getMessage()); - if (state.getStatus() == ChainStatus.SUSPEND) { - view.put("parameters", state.getSuspendForParameters()); - } - if (state.getStatus() == ChainStatus.SUCCEEDED) { - view.put( - "output", - WorkflowChatEventStream.visibleFinalOutput( - state.getExecuteResult()) - ); - } - return view; - } catch (RuntimeException error) { - log.warn( - "failed to load public workflow runtime state, executeId={}", - executeId, - error - ); - return Map.of(); - } - } } diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowChatEventStreamTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowChatEventStreamTest.java index 7c3afc98..3dd7dd8b 100644 --- a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowChatEventStreamTest.java +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowChatEventStreamTest.java @@ -1,6 +1,13 @@ package tech.easyflow.admin.service.ai; -import com.easyagents.flow.core.chain.ChainConsts; +import com.easyagents.flow.core.chain.*; +import com.easyagents.flow.core.chain.repository.*; +import com.easyagents.flow.core.chain.runtime.InMemoryTriggerStore; +import com.easyagents.flow.core.chain.runtime.TriggerScheduler; +import com.easyagents.flow.core.node.StartNode; +import com.alibaba.fastjson.JSON; +import com.alibaba.fastjson.JSONObject; +import java.util.concurrent.*; import com.easyagents.flow.core.chain.runtime.ChainExecutor; import org.testng.Assert; import org.testng.annotations.Test; @@ -138,6 +145,50 @@ public class WorkflowChatEventStreamTest { Assert.assertEquals(cleanupCount.get(), 1); } + @Test + public void terminalMustCarryReasonAfterFailedNode() throws Exception { + ChainDefinition definition = new ChainDefinition(); definition.setId("sse-test"); + StartNode start = new StartNode(); start.setId("start"); definition.addNode(start); + Node failed = new Node() { + @Override public Map execute(Chain chain) { + throw new WorkflowExecutionException(WorkflowErrorReason.MODEL_UNAVAILABLE, "PRIVATE_PROVIDER_BODY"); + } + }; + failed.setId("llm"); failed.setName("模型分析"); definition.addNode(failed); + Edge edge = new Edge(); edge.setId("edge"); edge.setSource("start"); edge.setTarget("llm"); definition.addEdge(edge); + TriggerScheduler scheduler = new TriggerScheduler(new InMemoryTriggerStore(), Executors.newSingleThreadScheduledExecutor(), Executors.newFixedThreadPool(2), 1000); + ChainExecutor executor = new ChainExecutor(id -> definition, new InMemoryChainStateRepository(), new InMemoryNodeStateRepository(), scheduler); + List events = new CopyOnWriteArrayList<>(); + CountDownLatch complete = new CountDownLatch(1); + SseEmitter emitter = new SseEmitter() { + @Override public void send(SseEventBuilder event) { + event.build().forEach(data -> { + String value = String.valueOf(data.getData()); + if (value.startsWith("{")) events.add(JSON.parseObject(value)); + }); + } + @Override public void complete() { complete.countDown(); } + }; + WorkflowChatEventStream stream = new WorkflowChatEventStream(executor) { + @Override SseEmitter createEmitter() { return emitter; } + }; + try { + stream.registerListeners(); stream.start("sse-test", Map.of()); + Assert.assertTrue(complete.await(5, TimeUnit.SECONDS)); + List terminals = events.stream().filter(e -> "execution_failed".equals(e.getString("type"))).toList(); + Assert.assertEquals(terminals.size(), 1); + JSONObject detail = terminals.get(0).getJSONObject("data").getJSONObject("error"); + Assert.assertEquals(detail.getString("reasonCode"), "MODEL_UNAVAILABLE"); + Assert.assertEquals(detail.getString("nodeId"), "llm"); + JSONObject ended = events.stream().filter(e -> "node_finished".equals(e.getString("type")) && "llm".equals(e.getJSONObject("data").getString("nodeId"))).findFirst().orElseThrow(); + Assert.assertEquals(ended.getJSONObject("data").getString("status"), "FAILED"); + Assert.assertTrue(events.indexOf(ended) < events.indexOf(terminals.get(0))); + Assert.assertTrue(ended.getJSONObject("data").get("error") instanceof String); + Assert.assertEquals(ended.getJSONObject("data").getJSONObject("errorDetail").getString("reasonCode"), "MODEL_UNAVAILABLE"); + Assert.assertFalse(JSON.toJSONString(events).contains("PRIVATE_PROVIDER_BODY")); + } finally { stream.shutdown(); scheduler.shutdown(); } + } + private static final class CapturingSseEmitter extends SseEmitter { private Runnable completion; diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatServiceTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatServiceTest.java index 7f8a1ca2..e7ab6cb3 100644 --- a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatServiceTest.java +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatServiceTest.java @@ -123,6 +123,8 @@ public class WorkflowPublicChatServiceTest { when(fixture.chainExecutor.getChainStateRepository()) .thenReturn(repository); when(repository.load("execution-1")).thenReturn(state); + Map runtimeSnapshot = new WorkflowChatEventStream(fixture.chainExecutor).runtimeView("execution-1"); + when(fixture.eventStream.runtimeView("execution-1")).thenReturn(runtimeSnapshot); Map detail = fixture.service.detail( "share-key", visitorId(), "execution-1"); diff --git a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/controller/PublicWorkflowController.java b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/controller/PublicWorkflowController.java index 14f383d4..c011393a 100644 --- a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/controller/PublicWorkflowController.java +++ b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/controller/PublicWorkflowController.java @@ -1,5 +1,6 @@ package tech.easyflow.publicapi.controller; +import tech.easyflow.ai.easyagentsflow.service.WorkflowExecutionErrorMapper; import cn.dev33.satoken.annotation.SaCheckPermission; import cn.dev33.satoken.stp.StpUtil; import com.easyagents.flow.core.chain.runtime.ChainExecutor; @@ -125,6 +126,7 @@ public class PublicWorkflowController { if (StpUtil.isLogin()) { variables.put(Constants.LOGIN_USER_KEY, SaTokenUtil.getLoginAccount()); } + WorkflowExecutionErrorMapper.installRequestProfile(); Map res = chainExecutor.executeNode(workflowId.toString(), nodeId, variables); return Result.ok(res); } diff --git a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/dto/PublicWorkflowNodeStatus.java b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/dto/PublicWorkflowNodeStatus.java index 0916379a..6b0f9aa9 100644 --- a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/dto/PublicWorkflowNodeStatus.java +++ b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/dto/PublicWorkflowNodeStatus.java @@ -22,5 +22,10 @@ public record PublicWorkflowNodeStatus( PublicWorkflowExecutionStatus status, String message, Map result, - List suspendForParameters) implements Serializable { + List suspendForParameters, + PublicWorkflowStatusError error) implements Serializable { + public PublicWorkflowNodeStatus(String nodeId, String nodeName, PublicWorkflowExecutionStatus status, + String message, Map result, List suspendForParameters) { + this(nodeId, nodeName, status, message, result, suspendForParameters, null); + } } diff --git a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/dto/PublicWorkflowStatusError.java b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/dto/PublicWorkflowStatusError.java index c5fa9977..317dfa16 100644 --- a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/dto/PublicWorkflowStatusError.java +++ b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/dto/PublicWorkflowStatusError.java @@ -10,6 +10,7 @@ public class PublicWorkflowStatusError implements Serializable { private static final long serialVersionUID = 1L; private final String code; + private final String reasonCode; private final String message; private final String nodeId; private final String nodeName; @@ -30,7 +31,13 @@ public class PublicWorkflowStatusError implements Serializable { String nodeId, String nodeName, boolean retryable) { + this(code, null, message, nodeId, nodeName, retryable); + } + + public PublicWorkflowStatusError(String code, String reasonCode, String message, + String nodeId, String nodeName, boolean retryable) { this.code = code; + this.reasonCode = reasonCode; this.message = message; this.nodeId = nodeId; this.nodeName = nodeName; @@ -46,6 +53,8 @@ public class PublicWorkflowStatusError implements Serializable { return code; } + public String getReasonCode() { return reasonCode; } + /** * 获取安全消息。 * diff --git a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/service/PublicWorkflowStatusSanitizer.java b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/service/PublicWorkflowStatusSanitizer.java index d601981b..4f7601b2 100644 --- a/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/service/PublicWorkflowStatusSanitizer.java +++ b/easyflow-api/easyflow-api-public/src/main/java/tech/easyflow/publicapi/service/PublicWorkflowStatusSanitizer.java @@ -1,9 +1,10 @@ package tech.easyflow.publicapi.service; import org.springframework.stereotype.Service; -import org.springframework.util.StringUtils; import tech.easyflow.ai.easyagentsflow.entity.ChainInfo; import tech.easyflow.ai.easyagentsflow.entity.NodeInfo; +import tech.easyflow.ai.easyagentsflow.entity.WorkflowExecutionError; +import tech.easyflow.ai.easyagentsflow.service.WorkflowExecutionErrorMapper; import tech.easyflow.publicapi.dto.PublicWorkflowChainStatus; import tech.easyflow.publicapi.dto.PublicWorkflowExecutionStatus; import tech.easyflow.publicapi.dto.PublicWorkflowNodeStatus; @@ -12,130 +13,44 @@ import tech.easyflow.publicapi.dto.PublicWorkflowStatusError; import java.util.LinkedHashMap; import java.util.Map; -/** - * 将内部工作流执行错误转换为 Public API 安全状态。 - */ +/** 复用公共错误规范，兼容没有结构化错误的旧状态。 */ @Service public class PublicWorkflowStatusSanitizer { - - private static final String CHAIN_FAILED_MESSAGE = - "工作流执行失败，请检查输入或稍后重试"; - private static final String NODE_FAILED_MESSAGE = - "节点执行失败，请检查输入或稍后重试"; - - /** - * 复制执行状态并移除异常类名、底层地址和内部错误详情。 - * - * @param source 内部执行状态 - * @return 可公开状态 - */ public PublicWorkflowChainStatus sanitize(ChainInfo source) { - if (source == null) { - throw new IllegalArgumentException( - "source must not be null"); - } - PublicWorkflowExecutionStatus chainStatus = - PublicWorkflowExecutionStatus.fromChainStatus( - source.getStatus()); - - Map safeNodes = - new LinkedHashMap<>(); + if (source == null) throw new IllegalArgumentException("source must not be null"); + PublicWorkflowExecutionStatus status = PublicWorkflowExecutionStatus.fromChainStatus(source.getStatus()); + Map nodes = new LinkedHashMap<>(); PublicWorkflowStatusError firstNodeError = null; if (source.getNodes() != null) { - for (Map.Entry entry - : source.getNodes().entrySet()) { - PublicWorkflowNodeStatus safeNode = copyNode( - entry.getValue()); - safeNodes.put(entry.getKey(), safeNode); - if (firstNodeError == null - && StringUtils.hasText(safeNode.message())) { - firstNodeError = new PublicWorkflowStatusError( - "NODE_EXECUTION_FAILED", - safeNode.message(), - safeNode.nodeId(), - safeNode.nodeName(), - isRetryable(safeNode.status())); - } + for (var entry : source.getNodes().entrySet()) { + NodeInfo node = entry.getValue(); + if (node == null) continue; + PublicWorkflowExecutionStatus nodeStatus = PublicWorkflowExecutionStatus.fromNodeStatus(node.getStatus()); + PublicWorkflowStatusError error = copyError(node.getError(), false, nodeStatus, node.getNodeId(), node.getNodeName(), status.isTerminal()); + nodes.put(entry.getKey(), new PublicWorkflowNodeStatus(node.getNodeId(), node.getNodeName(), nodeStatus, + error == null ? null : error.getMessage(), node.getResult(), node.getSuspendForParameters(), error)); + if (firstNodeError == null && error != null) firstNodeError = error; } } - - String message = null; - PublicWorkflowStatusError error = null; - if (StringUtils.hasText(source.getMessage())) { - message = chainMessage(chainStatus); - error = new PublicWorkflowStatusError( - "WORKFLOW_EXECUTION_FAILED", - message, - firstNodeError == null - ? null - : firstNodeError.getNodeId(), - firstNodeError == null - ? null - : firstNodeError.getNodeName(), - isRetryable(chainStatus)); - } else if (firstNodeError != null) { - error = firstNodeError; - } - return new PublicWorkflowChainStatus( - source.getExecuteId(), - chainStatus, - chainStatus.isTerminal(), - message, - source.getResult(), - safeNodes, - error); + PublicWorkflowStatusError error = copyError(source.getError(), true, status, + firstNodeError == null ? null : firstNodeError.getNodeId(), + firstNodeError == null ? null : firstNodeError.getNodeName(), status.isTerminal()); + // 暂态节点错误仍可查询；成功、取消和挂起不携带旧错误。 + if (error == null && status == PublicWorkflowExecutionStatus.RUNNING) error = firstNodeError; + return new PublicWorkflowChainStatus(source.getExecuteId(), status, status.isTerminal(), + error == null ? null : error.getMessage(), source.getResult(), nodes, error); } - /** - * 复制并脱敏单个节点状态。 - * - * @param source 内部节点状态 - * @return 安全节点状态 - */ - private PublicWorkflowNodeStatus copyNode(NodeInfo source) { - if (source == null) { - return new PublicWorkflowNodeStatus( - null, - null, - PublicWorkflowExecutionStatus.UNKNOWN, - null, - null, - null); + private PublicWorkflowStatusError copyError(WorkflowExecutionError source, boolean workflow, + PublicWorkflowExecutionStatus status, String nodeId, String nodeName, boolean executionTerminal) { + if (status != PublicWorkflowExecutionStatus.FAILED && status != PublicWorkflowExecutionStatus.ERROR) return null; + if (source != null) { + nodeId = source.getNodeId(); + nodeName = source.getNodeName(); } - return new PublicWorkflowNodeStatus( - source.getNodeId(), - source.getNodeName(), - PublicWorkflowExecutionStatus.fromNodeStatus( - source.getStatus()), - StringUtils.hasText(source.getMessage()) - ? NODE_FAILED_MESSAGE - : null, - source.getResult(), - source.getSuspendForParameters()); - } - - /** - * 根据工作流状态生成安全消息。 - * - * @param status 可读状态 - * @return 安全消息 - */ - private String chainMessage( - PublicWorkflowExecutionStatus status) { - if (status == PublicWorkflowExecutionStatus.CANCELLED) { - return "工作流执行已取消"; - } - return CHAIN_FAILED_MESSAGE; - } - - /** - * 判断执行状态是否仍可能由运行时继续处理。 - * - * @param status 可读状态 - * @return 是否可重试 - */ - private boolean isRetryable( - PublicWorkflowExecutionStatus status) { - return status == PublicWorkflowExecutionStatus.ERROR; + WorkflowExecutionError safe = WorkflowExecutionErrorMapper.fromReason(source == null ? null : source.getReasonCode(), workflow, nodeId, nodeName, + status == PublicWorkflowExecutionStatus.ERROR && !executionTerminal); + return new PublicWorkflowStatusError(safe.getCode(), safe.getReasonCode(), safe.getMessage(), + safe.getNodeId(), safe.getNodeName(), safe.isRetryable()); } } diff --git a/easyflow-api/easyflow-api-public/src/test/java/tech/easyflow/publicapi/service/PublicWorkflowStatusSanitizerTest.java b/easyflow-api/easyflow-api-public/src/test/java/tech/easyflow/publicapi/service/PublicWorkflowStatusSanitizerTest.java index 6640bda6..e68d0e71 100644 --- a/easyflow-api/easyflow-api-public/src/test/java/tech/easyflow/publicapi/service/PublicWorkflowStatusSanitizerTest.java +++ b/easyflow-api/easyflow-api-public/src/test/java/tech/easyflow/publicapi/service/PublicWorkflowStatusSanitizerTest.java @@ -1,6 +1,8 @@ package tech.easyflow.publicapi.service; import com.easyagents.flow.core.chain.ChainStatus; +import com.easyagents.flow.core.chain.WorkflowErrorReason; +import tech.easyflow.ai.easyagentsflow.entity.WorkflowExecutionError; import com.easyagents.flow.core.chain.NodeStatus; import org.junit.Assert; import org.junit.Test; @@ -39,11 +41,11 @@ public class PublicWorkflowStatusSanitizerTest { PublicWorkflowChainStatus result = sanitizer.sanitize(source); Assert.assertEquals( - "工作流执行失败，请检查输入或稍后重试", + WorkflowErrorReason.NODE_EXECUTION_FAILED.getDefaultMessage(), result.message()); Assert.assertFalse(result.message().contains("minio")); Assert.assertEquals( - "节点执行失败，请检查输入或稍后重试", + WorkflowErrorReason.NODE_EXECUTION_FAILED.getDefaultMessage(), result.nodes().get("node-1").message()); Assert.assertEquals("node-1", result.error().getNodeId()); Assert.assertFalse(result.error().isRetryable()); @@ -101,4 +103,46 @@ public class PublicWorkflowStatusSanitizerTest { PublicWorkflowExecutionStatus.RUNNING, result.nodes().get("node-1").status()); } + @Test + public void shouldReturnStructuredReasonWithoutRequestedNodes() { + for (WorkflowErrorReason reason : WorkflowErrorReason.values()) { + ChainInfo source = new ChainInfo(); + source.setStatus(ChainStatus.FAILED.getValue()); + source.setError(new WorkflowExecutionError("WORKFLOW_EXECUTION_FAILED", reason.getCode(), + "raw provider body must not escape", "llm", "分析", false)); + var result = sanitizer.sanitize(source); + Assert.assertEquals(reason.getCode(), result.error().getReasonCode()); + Assert.assertEquals(reason.getDefaultMessage(), result.message()); + Assert.assertEquals("llm", result.error().getNodeId()); + Assert.assertTrue(result.nodes().isEmpty()); + } + } + + @Test + public void successfulRetryMustNotExposeStaleErrors() { + ChainInfo source = new ChainInfo(); + source.setStatus(ChainStatus.SUCCEEDED.getValue()); + source.setMessage("stale error"); + NodeInfo node = new NodeInfo(); + node.setNodeId("llm"); node.setStatus(NodeStatus.SUCCEEDED.getValue()); node.setMessage("old failure"); + source.setNodes(Map.of("llm", node)); + Assert.assertNull(sanitizer.sanitize(source).error()); + Assert.assertNull(sanitizer.sanitize(source).nodes().get("llm").message()); + } + @Test + public void terminalWorkflowMustNotAdvertiseNodeRetry() { + for (ChainStatus status : new ChainStatus[]{ChainStatus.RUNNING, ChainStatus.FAILED, ChainStatus.CANCELLED}) { + ChainInfo source = new ChainInfo(); + source.setStatus(status.getValue()); + NodeInfo node = new NodeInfo(); + node.setNodeId("llm"); + node.setStatus(NodeStatus.ERROR.getValue()); + node.setError(new WorkflowExecutionError("NODE_EXECUTION_FAILED", "MODEL_TIMEOUT", + "raw error", "llm", "模型分析", true)); + source.setNodes(Map.of("llm", node)); + var result = sanitizer.sanitize(source); + Assert.assertEquals(status == ChainStatus.RUNNING, result.nodes().get("llm").error().isRetryable()); + } + } + } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/ChainInfo.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/ChainInfo.java index 6a5cb2cf..b6579ded 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/ChainInfo.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/ChainInfo.java @@ -23,6 +23,10 @@ public class ChainInfo implements Serializable { * 消息，错误时显示 */ private String message; + private WorkflowExecutionError error; + + public WorkflowExecutionError getError() { return error; } + public void setError(WorkflowExecutionError error) { this.error = error; } /** * 执行结果 */ diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/NodeInfo.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/NodeInfo.java index 5a3b4a8e..56fd2fed 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/NodeInfo.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/NodeInfo.java @@ -28,6 +28,10 @@ public class NodeInfo implements Serializable { * 消息，错误时显示 */ private String message; + private WorkflowExecutionError error; + + public WorkflowExecutionError getError() { return error; } + public void setError(WorkflowExecutionError error) { this.error = error; } /** * 执行结果 */ diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/WorkflowExecutionError.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/WorkflowExecutionError.java new file mode 100644 index 00000000..4d305f7b --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/entity/WorkflowExecutionError.java @@ -0,0 +1,32 @@ +package tech.easyflow.ai.easyagentsflow.entity; + +import java.io.Serializable; + +/** 三个运行出口共用的安全错误信息。 */ +public class WorkflowExecutionError implements Serializable { + private static final long serialVersionUID = 1L; + + private final String code; + private final String reasonCode; + private final String message; + private final String nodeId; + private final String nodeName; + private final boolean retryable; + + public WorkflowExecutionError(String code, String reasonCode, String message, + String nodeId, String nodeName, boolean retryable) { + this.code = code; + this.reasonCode = reasonCode; + this.message = message; + this.nodeId = nodeId; + this.nodeName = nodeName; + this.retryable = retryable; + } + + public String getCode() { return code; } + public String getReasonCode() { return reasonCode; } + public String getMessage() { return message; } + public String getNodeId() { return nodeId; } + public String getNodeName() { return nodeName; } + public boolean isRetryable() { return retryable; } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/listener/ChainEventListenerForSave.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/listener/ChainEventListenerForSave.java index 21b59ea9..8a4c9140 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/listener/ChainEventListenerForSave.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/listener/ChainEventListenerForSave.java @@ -10,6 +10,7 @@ import org.slf4j.LoggerFactory; import org.springframework.dao.DuplicateKeyException; import org.springframework.stereotype.Component; import tech.easyflow.ai.easyagentsflow.event.WorkflowExecutionAuditEvent; +import tech.easyflow.ai.easyagentsflow.service.WorkflowExecutionErrorMapper; import tech.easyflow.ai.easyagentsflow.event.WorkflowExecutionAuditProducer; import tech.easyflow.ai.easyagentsflow.repository.FrozenWorkflowDefinitionRegistry; import tech.easyflow.ai.easyagentsflow.support.PublishedWorkflowDefinitionIds; @@ -111,7 +112,7 @@ public class ChainEventListenerForSave implements ChainEventListener { state.getExecuteResult())); ExceptionSummary error = state.getError(); if (error != null) { - record.setErrorInfo(error.getRootCauseClass() + " --> " + error.getRootCauseMessage()); + record.setErrorInfo(WorkflowExecutionErrorMapper.summary(WorkflowExecutionErrorMapper.chain(error, state.getStatus()))); } sendAuditEvent( WorkflowExecutionAuditEvent.Type.CHAIN_ENDED, @@ -209,14 +210,13 @@ public class ChainEventListenerForSave implements ChainEventListener { step.setEndTime(new Date()); step.setStatus(nodeStatus.getValue()); ExceptionSummary error = - event.getError() == null + event.getErrorSummary() == null ? (legacyNodeState == null ? null : legacyNodeState.getError()) - : new ExceptionSummary( - event.getError()); + : event.getErrorSummary(); if (error != null) { - step.setErrorInfo(error.getRootCauseClass() + " --> " + error.getRootCauseMessage()); + step.setErrorInfo(WorkflowExecutionErrorMapper.summary(WorkflowExecutionErrorMapper.node(error, nodeStatus, node.getId(), node.getName()))); } sendAuditEvent( WorkflowExecutionAuditEvent.Type.NODE_ENDED, diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/TinyFlowService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/TinyFlowService.java index 54128ed1..c44ea8fc 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/TinyFlowService.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/TinyFlowService.java @@ -1,6 +1,5 @@ package tech.easyflow.ai.easyagentsflow.service; -import com.easyagents.document.core.exception.DocumentParseException; import com.easyagents.flow.core.chain.ChainState; import com.easyagents.flow.core.chain.ExceptionSummary; import com.easyagents.flow.core.chain.NodeState; @@ -8,11 +7,9 @@ import com.easyagents.flow.core.chain.NodeStatus; import com.easyagents.flow.core.chain.repository.ChainStateRepository; import com.easyagents.flow.core.chain.repository.NodeStateRepository; import com.easyagents.flow.core.chain.runtime.ChainExecutor; -import com.easyagents.flow.core.code.impl.JavascriptExecutionException; import org.springframework.stereotype.Component; import tech.easyflow.ai.easyagentsflow.entity.ChainInfo; import tech.easyflow.ai.easyagentsflow.entity.NodeInfo; -import tech.easyflow.common.util.StringUtil; import tech.easyflow.common.web.exceptions.BusinessException; import javax.annotation.Resource; @@ -59,10 +56,8 @@ public class TinyFlowService { ? Map.of() : resolvedNodeNames; for (NodeInfo node : nodes) { - if (node != null - && StringUtil.noText(node.getNodeName())) { - node.setNodeName(nodeNames.get(node.getNodeId())); - } + if (node == null) continue; + node.setNodeName(nodeNames.get(node.getNodeId())); processNodeState(executeId, node, chainState, nodeStateRepository); res.getNodes().put(node.getNodeId(), node); } @@ -100,9 +95,8 @@ public class TinyFlowService { res.setExecuteId(executeId); res.setStatus(chainState.getStatus().getValue()); ExceptionSummary chainError = chainState.getError(); - if (chainError != null) { - res.setMessage(formatError(chainError)); - } + res.setError(WorkflowExecutionErrorMapper.chain(chainError, chainState.getStatus())); + res.setMessage(WorkflowExecutionErrorMapper.summary(res.getError())); Map executeResult = chainState.getExecuteResult(); if (executeResult != null && !executeResult.isEmpty()) { @SuppressWarnings("unchecked") @@ -127,12 +121,9 @@ public class TinyFlowService { ? NodeStatus.READY.getValue() : nodeState.getStatus().getValue()); - if (nodeState != null) { - ExceptionSummary error = nodeState.getError(); - if (error != null) { - node.setMessage(formatError(error)); - } - } + node.setError(nodeState == null ? null : WorkflowExecutionErrorMapper.node( + nodeState.getError(), nodeState.getStatus(), nodeId, node.getNodeName(), chainState.getStatus().isTerminal())); + node.setMessage(WorkflowExecutionErrorMapper.summary(node.getError())); Map nodeExecuteResult = chainState.getNodeExecuteResult(nodeId); if (nodeExecuteResult != null && !nodeExecuteResult.isEmpty()) { @@ -151,34 +142,4 @@ public class TinyFlowService { } } - /** - * 将执行异常转换为试运行界面可读的错误信息。 - * - * @param error 持久化的异常摘要 - * @return 可展示的错误信息 - */ - private String formatError(ExceptionSummary error) { - if (JavascriptExecutionException.class.getName() - .equals(error.getExceptionClass()) - && StringUtil.hasText(error.getMessage())) { - return error.getMessage(); - } - String rootClass = StringUtil.hasText(error.getRootCauseClass()) - ? error.getRootCauseClass() - : error.getExceptionClass(); - String rootMessage = StringUtil.hasText(error.getRootCauseMessage()) - ? error.getRootCauseMessage() - : error.getMessage(); - if (DocumentParseException.class.getName().equals(rootClass) - && StringUtil.hasText(rootMessage)) { - return rootMessage; - } - if (StringUtil.noText(rootClass)) { - return rootMessage; - } - if (StringUtil.noText(rootMessage)) { - return rootClass; - } - return rootClass + " --> " + rootMessage; - } } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowExecutionErrorMapper.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowExecutionErrorMapper.java new file mode 100644 index 00000000..a90d7c8c --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowExecutionErrorMapper.java @@ -0,0 +1,77 @@ +package tech.easyflow.ai.easyagentsflow.service; + +import com.easyagents.flow.core.chain.ChainStatus; +import com.easyagents.flow.core.chain.ExceptionSummary; +import com.easyagents.flow.core.chain.NodeStatus; +import com.easyagents.flow.core.chain.WorkflowErrorReason; +import com.easyagents.flow.core.chain.WorkflowExecutionException; +import org.springframework.web.context.request.RequestContextHolder; +import org.springframework.web.context.request.ServletRequestAttributes; +import tech.easyflow.ai.easyagentsflow.entity.WorkflowExecutionError; +import tech.easyflow.common.web.error.RequestErrorProfile; +import tech.easyflow.common.web.error.WebErrorMapping; + +import java.util.Map; +import java.util.LinkedHashMap; + +/** 只通过稳定原因码生成外部信息，原始 cause 与响应正文不进入展示或审计摘要。 */ +public final class WorkflowExecutionErrorMapper { + private WorkflowExecutionErrorMapper() { } + + public static WorkflowExecutionError chain(ExceptionSummary error, ChainStatus status) { + if (status != ChainStatus.FAILED && status != ChainStatus.ERROR) return null; + return map(error, true, null, null, status == ChainStatus.ERROR); + } + + public static WorkflowExecutionError node(ExceptionSummary error, NodeStatus status, String nodeId, String nodeName) { + return node(error, status, nodeId, nodeName, false); + } + + public static WorkflowExecutionError node(ExceptionSummary error, NodeStatus status, String nodeId, String nodeName, + boolean executionTerminal) { + if (status != NodeStatus.FAILED && status != NodeStatus.ERROR) return null; + return map(error, false, nodeId, nodeName, status == NodeStatus.ERROR && !executionTerminal); + } + + public static WorkflowExecutionError map(ExceptionSummary error, boolean workflow, + String nodeId, String nodeName, boolean retryable) { + if (error != null && error.getNodeId() != null) { + nodeId = error.getNodeId(); + nodeName = error.getNodeName() == null ? nodeName : error.getNodeName(); + } + return fromReason(error == null ? null : error.getErrorCode(), workflow, nodeId, nodeName, retryable); + } + + public static WorkflowExecutionError fromReason(String reasonCode, boolean workflow, + String nodeId, String nodeName, boolean retryable) { + WorkflowErrorReason reason = WorkflowErrorReason.fromCode(reasonCode); + if (reason == null) { + reason = nodeId == null ? WorkflowErrorReason.WORKFLOW_INTERNAL_ERROR : WorkflowErrorReason.NODE_EXECUTION_FAILED; + } + return new WorkflowExecutionError(workflow ? "WORKFLOW_EXECUTION_FAILED" : "NODE_EXECUTION_FAILED", + reason.getCode(), reason.getDefaultMessage(), nodeId, nodeName, retryable); + } + + public static String summary(WorkflowExecutionError error) { + if (error == null) return null; + String name = error.getNodeName(); + return name == null || name.isBlank() ? error.getMessage() : "「" + name + "」：" + error.getMessage(); + } + + /** 单节点同步执行沿用全局 HTTP 错误处理，并保留原请求的其他错误契约。 */ + public static void installRequestProfile() { + if (!(RequestContextHolder.getRequestAttributes() instanceof ServletRequestAttributes attributes)) return; + var request = attributes.getRequest(); + Object previous = request.getAttribute(RequestErrorProfile.ATTRIBUTE_NAME); + request.setAttribute(RequestErrorProfile.ATTRIBUTE_NAME, (RequestErrorProfile) (req, exception) -> { + if (exception instanceof WorkflowExecutionException failure) { + WorkflowExecutionError error = map(new ExceptionSummary(failure), false, null, null, false); + Map data = new LinkedHashMap<>(); + data.put("error", error); + if (failure.getChainId() != null) data.put("executeId", failure.getChainId()); + return new WebErrorMapping(500, 500, error.getMessage(), data); + } + return previous instanceof RequestErrorProfile profile ? profile.map(req, exception) : null; + }); + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/upload/WorkflowApiUploadLifecycleService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/upload/WorkflowApiUploadLifecycleService.java index cb8bd3f2..46208d5c 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/upload/WorkflowApiUploadLifecycleService.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/upload/WorkflowApiUploadLifecycleService.java @@ -485,7 +485,7 @@ public class WorkflowApiUploadLifecycleService { return new BusinessException( 500, 50001, - "文件存储处理失败，请联系管理员并提供 requestId", + "文件存储处理失败", error); } diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/TinyFlowServiceTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/TinyFlowServiceTest.java index dd79ac45..d314dddf 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/TinyFlowServiceTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/TinyFlowServiceTest.java @@ -2,6 +2,8 @@ package tech.easyflow.ai.easyagentsflow.service; import com.easyagents.document.core.exception.DocumentParseException; import com.easyagents.flow.core.chain.ChainState; +import com.easyagents.flow.core.chain.WorkflowErrorReason; +import com.easyagents.flow.core.chain.WorkflowExecutionException; import com.easyagents.flow.core.chain.ChainStatus; import com.easyagents.flow.core.chain.ExceptionSummary; import com.easyagents.flow.core.chain.NodeState; @@ -205,7 +207,7 @@ public class TinyFlowServiceTest { * @throws Exception 测试依赖注入失败时抛出 */ @Test - public void shouldExposeJavascriptExecutionMessage() + public void shouldHideRawJavascriptExceptionDetails() throws Exception { ChainExecutor chainExecutor = mock(ChainExecutor.class); ChainStateRepository chainStateRepository = @@ -237,9 +239,9 @@ public class TinyFlowServiceTest { ChainInfo result = service.getChainStatus( EXECUTE_ID, List.of(node(NodeStatus.READY))); - Assert.assertEquals(message, result.getMessage()); + Assert.assertEquals(WorkflowErrorReason.WORKFLOW_INTERNAL_ERROR.getDefaultMessage(), result.getMessage()); Assert.assertEquals( - message, + WorkflowErrorReason.WORKFLOW_INTERNAL_ERROR.getDefaultMessage(), result.getNodes().get(NODE_ID).getMessage()); } @@ -249,7 +251,7 @@ public class TinyFlowServiceTest { * @throws Exception 测试依赖注入失败时抛出 */ @Test - public void shouldExposeDocumentParseMessageWithoutExceptionClass() + public void shouldHideUnclassifiedDocumentCauseDetails() throws Exception { ChainExecutor chainExecutor = mock(ChainExecutor.class); ChainStateRepository chainStateRepository = @@ -274,7 +276,47 @@ public class TinyFlowServiceTest { ChainInfo result = service.getChainStatus(EXECUTE_ID, null); - Assert.assertEquals(message, result.getMessage()); + Assert.assertEquals(WorkflowErrorReason.WORKFLOW_INTERNAL_ERROR.getDefaultMessage(), result.getMessage()); + } + + @Test + public void shouldKeepFailureNodeWithEmptyNodeSelection() throws Exception { + ChainExecutor executor = mock(ChainExecutor.class); + ChainStateRepository states = mock(ChainStateRepository.class); + when(executor.getChainStateRepository()).thenReturn(states); + ChainState state = new ChainState(); + state.setStatus(ChainStatus.FAILED); + state.setError(new ExceptionSummary(new WorkflowExecutionException(WorkflowErrorReason.MODEL_RATE_LIMITED, + "raw credentials"), EXECUTE_ID, NODE_ID, "模型分析")); + when(states.load(EXECUTE_ID)).thenReturn(state); + ChainInfo result = service(executor).getChainStatus(EXECUTE_ID, List.of()); + Assert.assertTrue(result.getNodes().isEmpty()); + Assert.assertEquals(NODE_ID, result.getError().getNodeId()); + Assert.assertEquals("MODEL_RATE_LIMITED", result.getError().getReasonCode()); + Assert.assertFalse(result.getMessage().contains("credentials")); + } + + + @Test + public void terminalWorkflowMustStopAdvertisingPendingRetry() throws Exception { + ChainExecutor executor = mock(ChainExecutor.class); + ChainStateRepository states = mock(ChainStateRepository.class); + NodeStateRepository nodes = mock(NodeStateRepository.class); + when(executor.getChainStateRepository()).thenReturn(states); + when(executor.getNodeStateRepository()).thenReturn(nodes); + NodeState failedAttempt = new NodeState(); + failedAttempt.setStatus(NodeStatus.ERROR); + failedAttempt.setError(new ExceptionSummary(new WorkflowExecutionException( + WorkflowErrorReason.MODEL_TIMEOUT, "private cause"), EXECUTE_ID, NODE_ID, "模型分析")); + when(nodes.load(EXECUTE_ID, NODE_ID)).thenReturn(failedAttempt); + TinyFlowService service = service(executor); + for (ChainStatus status : new ChainStatus[]{ChainStatus.RUNNING, ChainStatus.FAILED, ChainStatus.CANCELLED}) { + ChainState state = new ChainState(); + state.setStatus(status); + when(states.load(EXECUTE_ID)).thenReturn(state); + ChainInfo result = service.getChainStatus(EXECUTE_ID, List.of(node(NodeStatus.READY))); + Assert.assertEquals(status == ChainStatus.RUNNING, result.getNodes().get(NODE_ID).getError().isRetryable()); + } } /** diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowFailureDiagnosticsTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowFailureDiagnosticsTest.java new file mode 100644 index 00000000..97f1e7c0 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowFailureDiagnosticsTest.java @@ -0,0 +1,165 @@ +package tech.easyflow.ai.easyagentsflow.service; + +import ch.qos.logback.classic.Logger; +import ch.qos.logback.classic.spi.ILoggingEvent; +import ch.qos.logback.core.read.ListAppender; +import com.easyagents.flow.core.chain.*; +import com.easyagents.flow.core.chain.event.ChainEndEvent; +import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository; +import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository; +import com.easyagents.flow.core.chain.runtime.*; +import com.easyagents.flow.core.node.StartNode; +import jakarta.servlet.http.HttpServletRequest; +import org.junit.Assert; +import org.junit.Test; +import org.slf4j.LoggerFactory; +import org.springframework.web.context.request.RequestContextHolder; +import org.springframework.web.context.request.ServletRequestAttributes; +import tech.easyflow.ai.easyagentsflow.entity.WorkflowExecutionError; +import tech.easyflow.common.web.error.RequestErrorProfile; +import tech.easyflow.common.web.error.WebErrorMapping; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.*; + +public class WorkflowFailureDiagnosticsTest { + @Test + public void singleRunResponseMustMatchLoggedExecutionId() { + Logger logger = (Logger) LoggerFactory.getLogger(ChainExecutor.class); + ListAppender logs = new ListAppender<>(); + logs.start(); + logger.addAppender(logs); + HttpServletRequest request = request(); + RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request)); + try (Fixture fixture = new Fixture()) { + WorkflowExecutionErrorMapper.installRequestProfile(); + RequestErrorProfile profile = (RequestErrorProfile) request.getAttribute(RequestErrorProfile.ATTRIBUTE_NAME); + String previousId = null; + for (int run = 0; run < 2; run++) { + WorkflowExecutionException failure = Assert.assertThrows(WorkflowExecutionException.class, + () -> fixture.executor.executeNode("diagnostics", "worker", Map.of("privateInput", "do-not-log"))); + WebErrorMapping response = profile.map(request, failure); + Map data = (Map) response.data(); + String executeId = (String) data.get("executeId"); + Assert.assertNotNull(executeId); + Assert.assertNotEquals(previousId, executeId); + Assert.assertEquals(failure.getChainId(), executeId); + Assert.assertEquals(500, response.httpStatus()); + WorkflowExecutionError error = (WorkflowExecutionError) data.get("error"); + Assert.assertEquals("MODEL_TIMEOUT", error.getReasonCode()); + Assert.assertEquals("worker", error.getNodeId()); + ILoggingEvent event = logs.list.get(run); + assertLogContext(event, executeId); + Assert.assertTrue(event.getFormattedMessage().contains("attemptKey=" + executeId + ":worker:single")); + previousId = executeId; + } + } finally { + RequestContextHolder.resetRequestAttributes(); + logger.detachAppender(logs); + logs.stop(); + } + } + + @Test + public void optionalExecutionIdMustPreserveExistingRequestMapping() { + HttpServletRequest request = request(); + WebErrorMapping fallback = new WebErrorMapping(400, 400, "原请求错误", null); + request.setAttribute(RequestErrorProfile.ATTRIBUTE_NAME, (RequestErrorProfile) (req, failure) -> fallback); + RequestContextHolder.setRequestAttributes(new ServletRequestAttributes(request)); + try { + WorkflowExecutionErrorMapper.installRequestProfile(); + RequestErrorProfile profile = (RequestErrorProfile) request.getAttribute(RequestErrorProfile.ATTRIBUTE_NAME); + Assert.assertSame(fallback, profile.map(request, new IllegalArgumentException())); + WebErrorMapping response = profile.map(request, + new WorkflowExecutionException(WorkflowErrorReason.INPUT_INVALID, "internal")); + Map data = (Map) response.data(); + Assert.assertFalse(data.containsKey("executeId")); + Assert.assertEquals("INPUT_INVALID", ((WorkflowExecutionError) data.get("error")).getReasonCode()); + } finally { + RequestContextHolder.resetRequestAttributes(); + } + } + + @Test + public void concurrentExecutionsAndRetriesMustHaveDistinctLogContexts() throws Exception { + Logger logger = (Logger) LoggerFactory.getLogger(Chain.class); + ListAppender logs = new ListAppender<>(); + logs.start(); + logger.addAppender(logs); + try (Fixture fixture = new Fixture()) { + CountDownLatch ended = new CountDownLatch(2); + fixture.executor.addEventListener((event, chain) -> { + if (event instanceof ChainEndEvent) ended.countDown(); + }); + String first = fixture.executor.executeAsync("diagnostics", Map.of()); + String second = fixture.executor.executeAsync("diagnostics", Map.of()); + Assert.assertTrue(ended.await(5, TimeUnit.SECONDS)); + for (String executeId : List.of(first, second)) { + List attempts = logs.list.stream() + .filter(event -> event.getFormattedMessage().contains("executeId=" + executeId + ",")) + .toList(); + Assert.assertEquals(2, attempts.size()); + attempts.forEach(event -> assertLogContext(event, executeId)); + Assert.assertNotEquals(attempts.get(0).getArgumentArray()[4], attempts.get(1).getArgumentArray()[4]); + Assert.assertNotNull(attempts.get(0).getArgumentArray()[4]); + } + } finally { + logger.detachAppender(logs); + logs.stop(); + } + } + + private HttpServletRequest request() { + HttpServletRequest request = mock(HttpServletRequest.class); + Map attributes = new HashMap<>(); + when(request.getAttribute(anyString())).thenAnswer(call -> attributes.get(call.getArgument(0))); + doAnswer(call -> { + attributes.put(call.getArgument(0), call.getArgument(1)); + return null; + }).when(request).setAttribute(anyString(), any()); + return request; + } + + private void assertLogContext(ILoggingEvent event, String executeId) { + String message = event.getFormattedMessage(); + Assert.assertTrue(message.contains("executeId=" + executeId + ",")); + Assert.assertTrue(message.contains("chainInstanceId=" + executeId + ",")); + Assert.assertTrue(message.contains("nodeId=worker, nodeName=模型分析,")); + Assert.assertFalse(message.contains("do-not-log")); + Assert.assertNotNull(event.getThrowableProxy()); + } + + private static class Fixture implements AutoCloseable { + final TriggerScheduler scheduler = new TriggerScheduler(new InMemoryTriggerStore(), + Executors.newSingleThreadScheduledExecutor(), Executors.newFixedThreadPool(3), 1000); + final ChainExecutor executor; + + Fixture() { + ChainDefinition definition = new ChainDefinition(); + definition.setId("diagnostics"); + StartNode start = new StartNode(); start.setId("start"); definition.addNode(start); + Node worker = new Node() { + public Map execute(Chain chain) { + throw new WorkflowExecutionException(WorkflowErrorReason.MODEL_TIMEOUT, "synthetic cause"); + } + }; + worker.setId("worker"); worker.setName("模型分析"); + worker.setRetryEnable(true); worker.setMaxRetryCount(1); worker.setRetryIntervalMs(5); + definition.addNode(worker); + Edge edge = new Edge(); edge.setId("start-worker"); edge.setSource("start"); edge.setTarget("worker"); + definition.addEdge(edge); + executor = new ChainExecutor(id -> definition, new InMemoryChainStateRepository(), + new InMemoryNodeStateRepository(), scheduler); + } + + public void close() { scheduler.shutdown(); } + } +} diff --git a/easyflow-ui-admin/app/src/api/request.ts b/easyflow-ui-admin/app/src/api/request.ts index 0da7eabc..0e6081b3 100644 --- a/easyflow-ui-admin/app/src/api/request.ts +++ b/easyflow-ui-admin/app/src/api/request.ts @@ -25,7 +25,11 @@ import { } from '#/utils/workflow-share-context'; import { refreshTokenApi } from './core'; -import { isInactiveSseRequest } from './sseRequestLifecycle'; +import { + isInactiveSseRequest, + readSseRequestError, + SseRequestError, +} from './sseRequestLifecycle'; const { apiURL } = useAppConfig(import.meta.env, import.meta.env.PROD); const ERROR_MESSAGE_DEDUP_WINDOW = 800; @@ -235,7 +239,7 @@ export class SseClient { }); if (!res.ok) { - const error = new Error(`HTTP ${res.status}: ${res.statusText}`); + const error = await readSseRequestError(res); options?.onError?.(error); return; } @@ -258,7 +262,7 @@ export class SseClient { } } showErrorOnce(errorMessage); - options?.onError?.(new Error(errorMessage)); + options?.onError?.(new SseRequestError(res.status, errorMessage)); return; } diff --git a/easyflow-ui-admin/app/src/api/sseRequestLifecycle.test.ts b/easyflow-ui-admin/app/src/api/sseRequestLifecycle.test.ts index 3d7ec743..aaa76208 100644 --- a/easyflow-ui-admin/app/src/api/sseRequestLifecycle.test.ts +++ b/easyflow-ui-admin/app/src/api/sseRequestLifecycle.test.ts @@ -1,6 +1,10 @@ import { describe, expect, it } from 'vitest'; -import { isInactiveSseRequest } from './sseRequestLifecycle'; +import { + isInactiveSseRequest, + readSseRequestError, + SseRequestError, +} from './sseRequestLifecycle'; describe('sseRequestLifecycle', () => { it('treats an explicit abort as an inactive request', () => { @@ -17,3 +21,19 @@ describe('sseRequestLifecycle', () => { expect(isInactiveSseRequest(controller.signal, 1, 1)).toBe(false); }); }); + +it('keeps HTTP rejection distinguishable from transport interruption', async () => { + for (const status of [400, 403, 500]) { + const error = await readSseRequestError( + new Response(JSON.stringify({ message: '运行请求被拒绝' }), { status }), + ); + expect(error).toBeInstanceOf(SseRequestError); + expect(error.status).toBe(status); + expect(error.message).toBe('运行请求被拒绝'); + } + const error = await readSseRequestError( + new Response('PRIVATE_GATEWAY_BODY', { status: 502 }), + ); + expect(error.message).toContain('502'); + expect(error.message).not.toContain('PRIVATE_GATEWAY_BODY'); +}); diff --git a/easyflow-ui-admin/app/src/api/sseRequestLifecycle.ts b/easyflow-ui-admin/app/src/api/sseRequestLifecycle.ts index 90a48667..a128f0d1 100644 --- a/easyflow-ui-admin/app/src/api/sseRequestLifecycle.ts +++ b/easyflow-ui-admin/app/src/api/sseRequestLifecycle.ts @@ -8,3 +8,24 @@ export function isInactiveSseRequest( ) { return signal.aborted || currentRequestId !== requestId; } +/** 服务端明确拒绝请求，与已经建立的事件流断线区分。 */ +export class SseRequestError extends Error { + constructor( + public readonly status: number, + message: string, + ) { + super(message); + this.name = 'SseRequestError'; + } +} + +export async function readSseRequestError(response: Response) { + let message = `请求失败（HTTP ${response.status}）`; + try { + const body = await response.json(); + if (typeof body?.message === 'string') message = body.message; + } catch { + // 网关可能返回 HTML；只展示状态，不把原始正文当作错误文案。 + } + return new SseRequestError(response.status, message); +} diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowList.vue b/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowList.vue index 1191ed72..2acd1c49 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowList.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowList.vue @@ -987,7 +987,7 @@ const apiDocMarkdown = computed(() => { lines.push(`| 413 | 41301 | 文件、文件数量或请求总量超限 |`); lines.push(`| 415 | 41501 | 顶层 Content-Type 缺失或不支持 |`); lines.push(`| 415 | 41502 | metadata Part 未使用 application/json |`); - lines.push(`| 500 | 50001 | 服务端内部错误；携带 requestId 联系管理员 |`); + lines.push(`| 500 | 50001 | 服务端内部错误 |`); lines.push(`| 503 | 50301 | 文件存储暂时不可用，可稍后重试 |`); lines.push(``); lines.push( diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/ExecResult.vue b/easyflow-ui-admin/app/src/views/ai/workflow/components/ExecResult.vue index 02a54e4a..597d7705 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/ExecResult.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/ExecResult.vue @@ -52,8 +52,7 @@ watch( success.value = true; } if (newVal.status === 21) { - ElMessage.error($t('message.fail')); - result.value = newVal.message; + result.value = newVal.result || ''; success.value = false; } }, diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/SingleRun.vue b/easyflow-ui-admin/app/src/views/ai/workflow/components/SingleRun.vue index 18611fc6..48303108 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/SingleRun.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/SingleRun.vue @@ -1,6 +1,8 @@ + + diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowSteps.vue b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowSteps.vue index 8230c07e..cf3913bf 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowSteps.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowSteps.vue @@ -9,7 +9,6 @@ import { VideoPause, } from '@element-plus/icons-vue'; import { - ElAlert, ElButton, ElCollapse, ElCollapseItem, @@ -20,6 +19,8 @@ import { import ShowJson from '#/components/json/ShowJson.vue'; import WorkflowFormItem from '#/views/ai/workflow/components/WorkflowFormItem.vue'; +import WorkflowErrorDetail from './WorkflowErrorDetail.vue'; + export interface WorkflowStepsProps { workflowId: any; nodeJson: any; @@ -42,7 +43,7 @@ const confirmBtnLoading = ref(false); const chainErrMsg = ref(''); function shouldAutoExpandStatus(status: unknown) { - return [1, 5, 20, 21].includes(Number(status)); + return [1, 5, 10, 20, 21].includes(Number(status)); } function handleManualExpansionChange() { @@ -75,6 +76,7 @@ function hasNodeStateChanged(previous: any, current: any) { if (hasNodePayloadChanged(previous?.result, current?.result)) { return true; } + if (hasNodePayloadChanged(previous?.error, current?.error)) return true; return hasNodePayloadChanged( previous?.suspendForParameters, current?.suspendForParameters, @@ -93,9 +95,11 @@ watch( confirmBtnLoading.value = false; } let autoExpandNodeId: string | undefined; - const failedNodeId = Object.keys(currentNodes).find( - (nodeId) => Number(currentNodes[nodeId]?.status) === 21, - ); + const failedNodeId = + newVal.error?.nodeId || + Object.keys(currentNodes).find( + (nodeId) => Number(currentNodes[nodeId]?.status) === 21, + ); for (const nodeId in currentNodes) { const previousNodeState = nodeStatusMap.value[nodeId]; const currentNodeState = currentNodes[nodeId]; @@ -162,6 +166,18 @@ const displayNodes = computed(() => { ...nodeStatusMap.value[node.key], })); }); +const showChainError = computed(() => { + const failedNodeId = props.pollingData?.error?.nodeId; + return ( + chainErrMsg.value && + (!failedNodeId || + !displayNodes.value.some( + (node) => + (node.key === failedNodeId || node.error?.nodeId === failedNodeId) && + (node.error || node.message), + )) + ); +}); // 动态设置 Ref 的辅助函数 const setFormRef = (el: any, key: string) => { if (el) { @@ -213,13 +229,11 @@ function handleConfirm(node: any) { @@ -288,7 +305,12 @@ function handleConfirm(node: any) {
- + +
diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/ExecResult.test.ts b/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/ExecResult.test.ts index 1dd4fad2..d309ac78 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/ExecResult.test.ts +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/ExecResult.test.ts @@ -1,50 +1,33 @@ import { mount } from '@vue/test-utils'; -import { describe, expect, it, vi } from 'vitest'; +import { describe, expect, it } from 'vitest'; import ExecResult from '../ExecResult.vue'; -vi.mock('#/locales', () => ({ - $t: (key: string) => key, -})); - describe('execResult', () => { - it('结束节点执行失败时展示工作流错误信息', async () => { + it('失败时保留部分输出，结果区不重复显示节点错误', async () => { const wrapper = mount(ExecResult, { - props: { - initSignal: false, - nodeJson: [ - { - original: { - data: { - outputDefs: [], - }, - type: 'endNode', - }, - }, - ], - pollingData: undefined, - workflowId: 'workflow-1', - }, + props: { workflowId: 'test', nodeJson: [] }, global: { stubs: { - ShowJson: { - props: ['value'], - template: '
{{ value }}
', - }, + ShowJson: { props: ['value'], template: '
{{ value }}
' }, + ElEmpty: true, }, }, }); - + await wrapper.setProps({ + pollingData: { status: 21, message: '模型不存在' }, + }); + expect(wrapper.text()).not.toContain('模型不存在'); await wrapper.setProps({ pollingData: { - message: 'JavaScript 执行失败（第 3 行，第 5 列）：boom', status: 21, + message: '模型不存在', + result: { output: '部分输出' }, }, }); - - expect(wrapper.get('[data-test="show-json"]').text()).toContain( - 'JavaScript 执行失败（第 3 行，第 5 列）：boom', - ); + expect(wrapper.text()).toContain('部分输出'); + expect(wrapper.text()).not.toContain('模型不存在'); + wrapper.unmount(); }); }); diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/SingleRun.test.ts b/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/SingleRun.test.ts new file mode 100644 index 00000000..faed6b0a --- /dev/null +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/SingleRun.test.ts @@ -0,0 +1,78 @@ +import { flushPromises, mount } from '@vue/test-utils'; + +import { describe, expect, it, vi } from 'vitest'; + +import SingleRun from '../SingleRun.vue'; + +const { post, copy } = vi.hoisted(() => ({ post: vi.fn(), copy: vi.fn() })); +vi.mock('#/api/request', () => ({ api: { post } })); +vi.mock('#/utils/clipboard-feedback', () => ({ copyTextWithFeedback: copy })); + +describe('singleRun', () => { + it('copies the current execution id and clears it before a new run', async () => { + const detail = { + code: 'NODE_EXECUTION_FAILED', + reasonCode: 'MODEL_NOT_FOUND', + message: '模型不存在', + nodeId: 'llm', + nodeName: '模型分析', + retryable: false, + }; + const rejectWith = (executeId?: string) => ({ + errorCode: 500, + message: detail.message, + data: { error: detail, executeId }, + }); + post.mockRejectedValueOnce(rejectWith('execution-first')); + const wrapper = mount(SingleRun, { + props: { + workflowId: 'test', + node: { id: 'llm', type: 'llmNode', data: { userPrompt: 'test' } }, + }, + global: { stubs: { ShowJson: true, WorkflowFormItem: true } }, + }); + const findButton = (copyButton: boolean) => { + const button = wrapper + .findAll('button') + .find((item) => item.text().includes('复制排查信息') === copyButton); + if (!button) throw new Error('Expected button was not rendered'); + return button; + }; + const run = () => findButton(false); + const copyButton = () => findButton(true); + const copiedContext = () => { + const call = copy.mock.lastCall; + if (!call) + throw new Error('Expected diagnostic information to be copied'); + return JSON.parse(call[0]); + }; + await run().trigger('click'); + await flushPromises(); + await copyButton().trigger('click'); + expect(copiedContext()).toMatchObject({ + executeId: 'execution-first', + reasonCode: 'MODEL_NOT_FOUND', + }); + + let rejectNext!: (error: unknown) => void; + post.mockReturnValueOnce( + new Promise((_resolve, reject) => { + rejectNext = reject; + }), + ); + await run().trigger('click'); + await flushPromises(); + expect(wrapper.text()).not.toContain('复制排查信息'); + rejectNext(rejectWith('execution-second')); + await flushPromises(); + await copyButton().trigger('click'); + expect(copiedContext().executeId).toBe('execution-second'); + + post.mockRejectedValueOnce(rejectWith()); + await run().trigger('click'); + await flushPromises(); + await copyButton().trigger('click'); + expect(copiedContext()).not.toHaveProperty('executeId'); + wrapper.unmount(); + }); +}); diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/WorkflowSteps.test.ts b/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/WorkflowSteps.test.ts index d3470b89..d7998b47 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/WorkflowSteps.test.ts +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/__tests__/WorkflowSteps.test.ts @@ -323,4 +323,79 @@ describe('workflowSteps', () => { expect(wrapper.findAll('workflow-form-item-stub')).toHaveLength(1); expect(wrapper.findAll('button.el-button')).toHaveLength(1); }); + it('错误仅在失败节点内展示一次，同时保留部分输出和自动展开', async () => { + const wrapper = mountWorkflowSteps(); + const error = { + code: 'WORKFLOW_EXECUTION_FAILED', + reasonCode: 'MODEL_TIMEOUT', + message: '模型响应超时', + nodeId: 'node-b', + nodeName: '节点 B', + retryable: false, + }; + await wrapper.setProps({ + pollingData: { + executeId: 'run-1', + status: 21, + message: error.message, + error, + nodes: { + 'node-b': { + status: 21, + error, + message: error.message, + result: { partial: '保留输出' }, + }, + }, + }, + }); + expect(wrapper.text()).toContain('MODEL_TIMEOUT'); + expect(wrapper.text()).toContain('模型响应超时'); + expect(wrapper.findComponent({ name: 'ShowJson' }).exists()).toBe(true); + expect(wrapper.text()).toContain('复制排查信息'); + expect(wrapper.text().match(/MODEL_TIMEOUT/g)).toHaveLength(1); + expect(wrapper.text()).not.toContain('定位失败节点'); + expect(getCollapseItems(wrapper)[1]?.classes()).toContain('is-active'); + wrapper.unmount(); + }); + it('没有可展示的失败节点时仍保留工作流错误', async () => { + const wrapper = mountWorkflowSteps(); + await wrapper.setProps({ + pollingData: { + status: 21, + message: '工作流内部执行异常', + error: { + code: 'WORKFLOW_EXECUTION_FAILED', + reasonCode: 'WORKFLOW_INTERNAL_ERROR', + message: '工作流内部执行异常', + retryable: false, + }, + nodes: {}, + }, + }); + expect(wrapper.text()).toContain('WORKFLOW_INTERNAL_ERROR'); + wrapper.unmount(); + }); + it('工作流终止后不把遗留错误尝试展示为等待重试', async () => { + const wrapper = mountWorkflowSteps(); + await wrapper.setProps({ + pollingData: { + status: 21, + nodes: { + 'node-b': { + status: 10, + error: { + code: 'NODE_EXECUTION_FAILED', + reasonCode: 'MODEL_TIMEOUT', + message: '模型响应超时', + retryable: false, + }, + }, + }, + }, + }); + expect(wrapper.text()).toContain('已停止'); + expect(wrapper.text()).not.toContain('等待重试'); + wrapper.unmount(); + }); }); diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionDetails.test.ts b/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionDetails.test.ts index b5ada4d0..88adfe54 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionDetails.test.ts +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionDetails.test.ts @@ -8,6 +8,38 @@ import { } from './workflowExecutionDetails'; describe('workflowExecutionDetails', () => { + it('restores runtime error detail only on the final failed attempt', () => { + const error = { + code: 'WORKFLOW_EXECUTION_FAILED', + reasonCode: 'MODEL_NOT_FOUND', + message: '模型不存在', + nodeId: 'llm', + nodeName: '模型分析', + retryable: false, + }; + const steps = hydrateWorkflowExecutionSteps( + [ + { nodeId: 'llm', attemptKey: 'old', status: 10, errorInfo: '早先超时' }, + { + nodeId: 'llm', + attemptKey: 'final', + status: 21, + errorInfo: '模型不存在', + }, + ], + error, + ); + expect(steps[0]?.errorDetail).toBeUndefined(); + expect(steps[1]?.errorDetail).toEqual(error); + expect( + hydrateWorkflowExecutionSteps([{ nodeId: 'llm', status: 21 }], error)[0], + ).toMatchObject({ error: error.message, errorDetail: error }); + expect( + hydrateWorkflowExecutionSteps([ + { nodeId: 'llm', status: 21, errorInfo: '模型不存在' }, + ])[0]?.error, + ).toBe('模型不存在'); + }); it('keeps loop attempts separate and completes each output', () => { const first = reduceWorkflowExecutionSteps([], { data: { @@ -132,4 +164,78 @@ describe('workflowExecutionDetails', () => { expect(formatExecutionValue(value, true)).toContain('最终答案'); expect(formatExecutionValue(value, true)).not.toContain('内部推理'); }); + it('keeps retrying distinct and clears current error on success', () => { + const retry = reduceWorkflowExecutionSteps([], { + data: { + nodeId: 'llm', + status: 'ERROR', + error: '限流', + errorDetail: { retryable: true }, + }, + eventId: 'retry', + type: 'node_finished', + }); + expect(retry[0]?.status).toBe('retrying'); + const success = reduceWorkflowExecutionSteps(retry, { + data: { nodeId: 'llm', status: 'SUCCEEDED', output: { text: 'ok' } }, + eventId: 'success', + type: 'node_finished', + }); + expect(success[0]).toMatchObject({ + status: 'completed', + output: { text: 'ok' }, + }); + expect(success[0]?.error).toBeUndefined(); + expect(success[0]?.errorDetail).toBeUndefined(); + }); + + it('does not blame parallel siblings for the first failure', () => { + let steps: ReturnType = []; + for (const nodeId of ['a', 'b']) { + steps = reduceWorkflowExecutionSteps(steps, { + data: { nodeId }, + eventId: nodeId, + type: 'node_started', + }); + } + const finalized = finalizeWorkflowExecutionSteps(steps, 'failed', 200, { + code: 'WORKFLOW_EXECUTION_FAILED', + reasonCode: 'MODEL_TIMEOUT', + message: '超时', + nodeId: 'a', + retryable: false, + }); + expect(finalized[0]?.status).toBe('failed'); + expect(finalized[1]?.status).toBe('cancelled'); + }); + it('preserves a failed attempt when a new retry succeeds', () => { + const retry = reduceWorkflowExecutionSteps([], { + data: { + nodeId: 'llm', + attemptKey: 'llm:1', + status: 'ERROR', + error: '限流', + errorDetail: { retryable: true }, + }, + eventId: '1', + type: 'node_finished', + }); + const started = reduceWorkflowExecutionSteps(retry, { + data: { nodeId: 'llm', attemptKey: 'llm:2' }, + eventId: '2', + type: 'node_started', + }); + const success = reduceWorkflowExecutionSteps(started, { + data: { nodeId: 'llm', attemptKey: 'llm:2', status: 'SUCCEEDED' }, + eventId: '3', + type: 'node_finished', + }); + const completed = finalizeWorkflowExecutionSteps(success, 'completed', 200); + expect(completed[0]).toMatchObject({ + status: 'failed', + errorDetail: { retryable: false }, + }); + expect(completed[1]?.status).toBe('completed'); + expect(completed[1]?.error).toBeUndefined(); + }); }); diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionDetails.ts b/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionDetails.ts index f342269d..28bbf4c0 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionDetails.ts +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionDetails.ts @@ -1,3 +1,5 @@ +import type { WorkflowExecutionError } from './workflowExecutionError'; + import { sanitizeWorkflowOutputForDisplay } from './workflowFinalOutput'; export interface WorkflowExecutionTrace { @@ -10,6 +12,7 @@ export type WorkflowExecutionStepStatus = | 'cancelled' | 'completed' | 'failed' + | 'retrying' | 'running' | 'waiting'; @@ -18,6 +21,7 @@ export interface WorkflowExecutionStepView { duration?: number; endTime?: number; error?: string; + errorDetail?: WorkflowExecutionError; hasInput: boolean; hasOutput: boolean; input?: unknown; @@ -58,6 +62,17 @@ export function reduceWorkflowExecutionSteps( const stepIndex = findStepIndex(current, attemptKey, nodeId); if (event.type === 'node_started') { + current = current.map((step) => + step.nodeId === nodeId && step.status === 'retrying' + ? { + ...step, + status: 'failed', + errorDetail: step.errorDetail + ? { ...step.errorDetail, retryable: false } + : undefined, + } + : step, + ); const startTime = numberValue(data.startedAt) ?? now; const nextStep: WorkflowExecutionStepView = { attemptKey: attemptKey || undefined, @@ -83,6 +98,8 @@ export function reduceWorkflowExecutionSteps( next[stepIndex] = { ...existingStep, ...nextStep, + error: undefined, + errorDetail: undefined, traces: existingStep.traces, }; return next; @@ -120,6 +137,7 @@ export function reduceWorkflowExecutionSteps( startTime === undefined ? undefined : Math.max(0, endTime - startTime), endTime, error: textValue(data.error) || undefined, + errorDetail: data.errorDetail, hasOutput: hasOwn(data, 'output'), output: data.output, status: resolveLiveStatus(data.status, data.error), @@ -139,30 +157,43 @@ export function reduceWorkflowExecutionSteps( */ export function hydrateWorkflowExecutionSteps( steps: unknown, + error?: WorkflowExecutionError, ): WorkflowExecutionStepView[] { if (!Array.isArray(steps)) { return []; } - return steps.map((step: Record, index) => ({ - attemptKey: textValue(step.attemptKey) || undefined, - duration: numberValue(step.execTime), - endTime: timeValue(step.endTime), - error: textValue(step.errorInfo) || undefined, - hasInput: step.input !== undefined && step.input !== null, - hasOutput: step.output !== undefined && step.output !== null, - input: parseWorkflowExecutionValue(step.input), - key: - textValue(step.attemptKey) || - textValue(step.id) || - `${textValue(step.nodeId) || 'node'}:${index}`, - nodeId: textValue(step.nodeId), - nodeName: - textValue(step.nodeName) || textValue(step.nodeId) || '工作流节点', - output: parseWorkflowExecutionValue(step.output), - startTime: timeValue(step.startTime), - status: resolvePersistedStatus(step.status), - traces: [], - })); + const result: WorkflowExecutionStepView[] = steps.map( + (step: Record, index) => ({ + attemptKey: textValue(step.attemptKey) || undefined, + duration: numberValue(step.execTime), + endTime: timeValue(step.endTime), + error: textValue(step.errorInfo) || undefined, + hasInput: step.input !== undefined && step.input !== null, + hasOutput: step.output !== undefined && step.output !== null, + input: parseWorkflowExecutionValue(step.input), + key: + textValue(step.attemptKey) || + textValue(step.id) || + `${textValue(step.nodeId) || 'node'}:${index}`, + nodeId: textValue(step.nodeId), + nodeName: + textValue(step.nodeName) || textValue(step.nodeId) || '工作流节点', + output: parseWorkflowExecutionValue(step.output), + startTime: timeValue(step.startTime), + status: resolvePersistedStatus(step.status), + traces: [], + }), + ); + if (error?.nodeId) { + const failedStep = [...result] + .reverse() + .find((step) => step.nodeId === error.nodeId && step.status === 'failed'); + if (failedStep) { + failedStep.errorDetail = error; + failedStep.error ||= error.message; + } + } + return result; } /** @@ -172,13 +203,22 @@ export function finalizeWorkflowExecutionSteps( steps: WorkflowExecutionStepView[], status: 'cancelled' | 'completed' | 'failed', now = Date.now(), + error?: WorkflowExecutionError, ): WorkflowExecutionStepView[] { let changed = false; const next = steps.map((step) => { - if (step.status !== 'running' && step.status !== 'waiting') { + if (!['retrying', 'running', 'waiting'].includes(step.status)) { return step; } changed = true; + let finalStatus: WorkflowExecutionStepStatus = status; + if (step.status === 'retrying') finalStatus = 'failed'; + else if (status === 'failed' && error?.nodeId !== step.nodeId) + finalStatus = 'cancelled'; + let detail = step.errorDetail + ? { ...step.errorDetail, retryable: false } + : undefined; + if (error?.nodeId === step.nodeId) detail = error; return { ...step, duration: @@ -186,7 +226,9 @@ export function finalizeWorkflowExecutionSteps( ? step.duration : Math.max(0, now - step.startTime), endTime: now, - status, + status: finalStatus, + error: error?.nodeId === step.nodeId ? error.message : step.error, + errorDetail: detail, }; }); return changed ? next : steps; @@ -272,7 +314,9 @@ function resolveLiveStatus( error: unknown, ): WorkflowExecutionStepStatus { const normalized = textValue(status).toUpperCase(); - if (error || normalized === 'ERROR' || normalized === 'FAILED') { + if (normalized === 'ERROR') return 'retrying'; + if (normalized === 'SUCCEEDED') return 'completed'; + if (error || normalized === 'FAILED') { return 'failed'; } if (normalized === 'SUSPEND') { @@ -289,7 +333,9 @@ function resolvePersistedStatus(status: unknown): WorkflowExecutionStepStatus { case '5': { return 'waiting'; } - case '10': + case '10': { + return 'failed'; + } case '21': { return 'failed'; } diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionError.ts b/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionError.ts new file mode 100644 index 00000000..173882e3 --- /dev/null +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/workflowExecutionError.ts @@ -0,0 +1,26 @@ +export interface WorkflowExecutionError { + code: string; + message: string; + nodeId?: string; + nodeName?: string; + reasonCode: string; + retryable: boolean; +} + +/** 只复制排查所需的公开字段，避免带出输入、输出及原始异常。 */ +export function formatWorkflowErrorContext( + error: WorkflowExecutionError, + executeId?: string, +) { + return JSON.stringify( + { + executeId, + nodeId: error.nodeId, + nodeName: error.nodeName, + reasonCode: error.reasonCode, + message: error.message, + }, + null, + 2, + ); +}