feat: 收口工作流对话最终结果展示

- 主时间线只展示顶级工作流最终输出和有限运行进度

- 补齐结构化结果、人工审核交互与实时详情回归

- 优化数组层级和复制操作

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

@@ -10,12 +10,9 @@ import com.easyagents.flow.core.chain.Node;
 import com.easyagents.flow.core.chain.event.ChainStatusChangeEvent;
 import com.easyagents.flow.core.chain.event.EdgeConditionCheckFailedEvent;
 import com.easyagents.flow.core.chain.event.EdgeTriggerEvent;
-import com.easyagents.flow.core.chain.event.LlmStreamEvent;
 import com.easyagents.flow.core.chain.event.NodeEndEvent;
 import com.easyagents.flow.core.chain.event.NodeStartEvent;
 import com.easyagents.flow.core.chain.runtime.ChainExecutor;
-import com.easyagents.flow.core.node.EndNode;
-import com.easyagents.flow.core.node.LlmNode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
@@ -59,7 +56,6 @@ public class WorkflowChatEventStream {
     @PostConstruct
     public void registerListeners() {
         chainExecutor.addEventListener(this::onEvent);
-        chainExecutor.addOutputListener(this::onExplicitOutput);
         chainExecutor.addErrorListener(this::onChainError);
     }
 
@@ -107,10 +103,6 @@ public class WorkflowChatEventStream {
         if (session == null) {
             return;
         }
-        if (event instanceof LlmStreamEvent streamEvent) {
-            session.onLlmDelta(chain, streamEvent);
-            return;
-        }
         if (event instanceof NodeStartEvent nodeStartEvent) {
             session.onNodeStarted(chain, nodeStartEvent);
             return;
@@ -133,24 +125,6 @@ public class WorkflowChatEventStream {
         }
     }
 
-    /**
-     * 转发节点显式发布的每一条输出。
-     *
-     * @param chain 当前工作流
-     * @param node 输出节点
-     * @param outputMessage 输出内容
-     */
-    private void onExplicitOutput(
-            Chain chain,
-            Node node,
-            Object outputMessage
-    ) {
-        StreamSession session = findSession(chain);
-        if (session != null) {
-            session.sendNodeOutput(node, outputMessage);
-        }
-    }
-
     /**
      * 将链级异常发送到客户端。
      *
@@ -229,6 +203,26 @@ public class WorkflowChatEventStream {
         return error.getMessage();
     }
 
+    /**
+     * 去掉顶级工作流结果中的内部状态控制字段。
+     *
+     * @param result 顶级工作流执行结果
+     * @return 可直接交给会话结果区展示的最终输出
+     */
+    static Map<String, Object> visibleFinalOutput(
+            Map<String, Object> result
+    ) {
+        Map<String, Object> visible = new LinkedHashMap<>();
+        if (result != null) {
+            visible.putAll(result);
+        }
+        visible.remove(ChainConsts.CHAIN_STATE_STATUS_KEY);
+        visible.remove(ChainConsts.CHAIN_STATE_MESSAGE_KEY);
+        visible.remove(ChainConsts.NODE_STATE_STATUS_KEY);
+        visible.remove(ChainConsts.SCHEDULE_NEXT_NODE_DISABLED_KEY);
+        return visible;
+    }
+
     /**
      * 单次工作流执行的 SSE 会话。
      */
@@ -237,8 +231,6 @@ public class WorkflowChatEventStream {
         private final SseEmitter emitter;
         private final AtomicLong sequence = new AtomicLong();
         private final AtomicBoolean terminal = new AtomicBoolean(false);
-        private final Map<String, String> activeLlmStreams =
-                new ConcurrentHashMap<>();
         private volatile String executeId;
 
         /**
@@ -259,24 +251,6 @@ public class WorkflowChatEventStream {
             this.executeId = executeId;
         }
 
-        /**
-         * 处理 LLM 文本增量。
-         *
-         * @param chain 当前工作流
-         * @param event LLM 增量事件
-         */
-        private void onLlmDelta(Chain chain, LlmStreamEvent event) {
-            String nodeRunKey = nodeRunKey(chain, event.getNode());
-            activeLlmStreams.put(nodeRunKey, event.getStreamId());
-            String eventType = event.isReasoning()
-                    ? "llm_thinking_delta"
-                    : "llm_delta";
-            send(eventType, nodePayload(event.getNode(), Map.of(
-                    "streamId", event.getStreamId(),
-                    "delta", event.getDelta()
-            )));
-        }
-
         /**
          * 处理节点开始事件。
          *
@@ -295,14 +269,13 @@ public class WorkflowChatEventStream {
         }
 
         /**
-         * 处理节点完成事件并输出所有可见结果。
+         * 处理节点完成事件并更新运行详情。
          *
          * @param chain 当前工作流
          * @param event 节点完成事件
          */
         private void onNodeFinished(Chain chain, NodeEndEvent event) {
             Node node = event.getNode();
-            String streamId = activeLlmStreams.remove(nodeRunKey(chain, node));
             Map<String, Object> data = new LinkedHashMap<>();
             data.put("attemptKey", event.getExecutionAttemptKey());
             data.put("status", event.getStatus() == null
@@ -316,17 +289,7 @@ public class WorkflowChatEventStream {
             if (event.getError() != null) {
                 data.put("error", safeErrorMessage(event.getError()));
             }
-            if (streamId != null) {
-                data.put("streamId", streamId);
-            }
             send("node_finished", nodePayload(node, data));
-
-            if (node instanceof EndNode) {
-                sendNodeOutput(node, visibleEndOutput(event.getResult()));
-            } else if (node instanceof LlmNode && streamId == null) {
-                // 少数模型只在流结束时返回完整文本，仍需向对话区展示结果。
-                sendNodeOutput(node, event.getResult());
-            }
         }
 
         /**
@@ -453,23 +416,17 @@ public class WorkflowChatEventStream {
             Map<String, Object> data = new LinkedHashMap<>();
             data.put("status", status.name());
             data.put("message", chain.getState().getMessage());
+            if (status == ChainStatus.SUCCEEDED) {
+                data.put(
+                        "output",
+                        visibleFinalOutput(chain.getState().getExecuteResult())
+                );
+            }
             send(eventType, data);
             removeSession(this);
             emitter.complete();
         }
 
-        /**
-         * 发送一条用户可见节点输出。
-         *
-         * @param node 输出节点
-         * @param outputMessage 输出内容
-         */
-        private void sendNodeOutput(Node node, Object outputMessage) {
-            send("output", nodePayload(node, Map.of(
-                    "output", outputMessage == null ? Map.of() : outputMessage
-            )));
-        }
-
         /**
          * 发送 SSE 事件。
          *
@@ -515,17 +472,6 @@ public class WorkflowChatEventStream {
             }
         }
 
-        /**
-         * 构建节点本次执行的关联键。
-         *
-         * @param chain 当前工作流
-         * @param node 当前节点
-         * @return 节点运行键
-         */
-        private String nodeRunKey(Chain chain, Node node) {
-            return chain.getStateInstanceId() + ":" + node.getId();
-        }
-
         /**
          * 构建带节点信息的事件数据。
          *
@@ -546,24 +492,5 @@ public class WorkflowChatEventStream {
             return payload;
         }
 
-        /**
-         * 去掉结束节点内部状态控制字段。
-         *
-         * @param result 节点结果
-         * @return 用户可见输出
-         */
-        private Map<String, Object> visibleEndOutput(
-                Map<String, Object> result
-        ) {
-            Map<String, Object> visible = new LinkedHashMap<>();
-            if (result != null) {
-                visible.putAll(result);
-            }
-            visible.remove(ChainConsts.CHAIN_STATE_STATUS_KEY);
-            visible.remove(ChainConsts.CHAIN_STATE_MESSAGE_KEY);
-            visible.remove(ChainConsts.NODE_STATE_STATUS_KEY);
-            visible.remove(ChainConsts.SCHEDULE_NEXT_NODE_DISABLED_KEY);
-            return visible;
-        }
     }
 }

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

@@ -0,0 +1,72 @@
+package tech.easyflow.admin.service.ai;
+
+import com.easyagents.flow.core.chain.ChainConsts;
+import com.easyagents.flow.core.chain.runtime.ChainExecutor;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+/**
+ * {@link WorkflowChatEventStream} 最终输出协议测试。
+ */
+public class WorkflowChatEventStreamTest {
+
+    /**
+     * 验证会话事件流不再订阅节点显式输出。
+     */
+    @Test
+    public void shouldOnlyRegisterLifecycleAndErrorListeners() {
+        ChainExecutor chainExecutor = mock(ChainExecutor.class);
+        WorkflowChatEventStream eventStream =
+                new WorkflowChatEventStream(chainExecutor);
+
+        eventStream.registerListeners();
+
+        verify(chainExecutor).addEventListener(any());
+        verify(chainExecutor).addErrorListener(any());
+        verify(chainExecutor, never()).addOutputListener(any());
+    }
+
+    /**
+     * 验证顶级工作流最终输出会保留业务结构并移除内部控制字段。
+     */
+    @Test
+    public void shouldKeepStructuredFinalOutputAndRemoveInternalFields() {
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("summary", "执行完成");
+        result.put("items", List.of(
+                Map.of("name", "A", "score", 90),
+                Map.of("name", "B", "score", 85)
+        ));
+        result.put(ChainConsts.CHAIN_STATE_STATUS_KEY, "SUCCEEDED");
+        result.put(ChainConsts.CHAIN_STATE_MESSAGE_KEY, "internal");
+        result.put(ChainConsts.NODE_STATE_STATUS_KEY, "SUCCESS");
+        result.put(ChainConsts.SCHEDULE_NEXT_NODE_DISABLED_KEY, true);
+
+        Map<String, Object> visible =
+                WorkflowChatEventStream.visibleFinalOutput(result);
+
+        Assert.assertEquals(visible.get("summary"), "执行完成");
+        Assert.assertEquals(visible.get("items"), result.get("items"));
+        Assert.assertEquals(visible.size(), 2);
+        Assert.assertEquals(result.size(), 6);
+    }
+
+    /**
+     * 验证空执行结果被规范化为空对象。
+     */
+    @Test
+    public void shouldNormalizeMissingFinalOutputToEmptyMap() {
+        Assert.assertTrue(
+                WorkflowChatEventStream.visibleFinalOutput(null).isEmpty()
+        );
+    }
+}