EasyFlow-Tech-Dev/EasyFlow
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 统一工作流三个出口的错误反馈

Browse Source 
- 关联 EASY-2，补齐安全错误、节点展示与执行标识

- 保留历史可读摘要并验证接口、SSE 与界面兼容
This commit is contained in:
陈子默
2026-09-08 10:48:36 +08:00
parent e938ecda80
commit dd14af2733
35 changed files with 1171 additions and 358 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowChatController.java
View File

@@ -224,6 +224,7 @@ public class WorkflowChatController {
Map<String, Object> detail = new LinkedHashMap<>();
detail.put("record", recordView);
detail.put("steps", stepViews);
detail.put("runtime", eventStream.runtimeView(executeId));
return Result.ok(detail);
}

2
easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowController.java
View File

@@ -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<WorkflowService, Work
if (StpUtil.isLogin()) {
variables.put(Constants.LOGIN_USER_KEY, SaTokenUtil.getLoginAccount());
}
WorkflowExecutionErrorMapper.installRequestProfile();
Map<String, Object> res = chainExecutor.executeNode(workflowId.toString(), nodeId, variables);
return Result.ok(res);
}

80
easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowChatEventStream.java
View File

@@ -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<String, Object> runtimeView(String executeId) {
try {
ChainState state = chainExecutor.getChainStateRepository()
.load(executeId);
if (state == null || state.getStatus() == null) {
return Map.of();
}
Map<String, Object> 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<String, Object> 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)) {

33
easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/service/ai/WorkflowPublicChatService.java
View File

@@ -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<String, Object> runtimeView(String executeId) {
try {
ChainState state = chainExecutor.getChainStateRepository()
.load(executeId);
if (state == null || state.getStatus() == null) {
return Map.of();
}
Map<String, Object> 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();
}
}
}

53
easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowChatEventStreamTest.java
View File

@@ -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<String, Object> 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<JSONObject> 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<JSONObject> 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;

2
easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/service/ai/WorkflowPublicChatServiceTest.java
View File

@@ -123,6 +123,8 @@ public class WorkflowPublicChatServiceTest {
when(fixture.chainExecutor.getChainStateRepository())
.thenReturn(repository);
when(repository.load("execution-1")).thenReturn(state);
Map<String, Object> runtimeSnapshot = new WorkflowChatEventStream(fixture.chainExecutor).runtimeView("execution-1");
when(fixture.eventStream.runtimeView("execution-1")).thenReturn(runtimeSnapshot);
Map<String, Object> detail = fixture.service.detail(
"share-key", visitorId(), "execution-1");