feat(XL13): 归档工作流对话运行界面

- 接入发布快照优先与未发布草稿受控运行

- 支持文本和思考流式输出、循环多输出及实时运行详情

- 完成聊天分享、图片输入、中止与清空重来

- 补充后端与前端定向回归测试

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

@@ -0,0 +1,353 @@
+package tech.easyflow.admin.controller.ai;
+
+import com.easyagents.flow.core.chain.ChainStatus;
+import com.easyagents.flow.core.chain.runtime.ChainExecutor;
+import com.mybatisflex.core.query.QueryWrapper;
+import jakarta.servlet.http.HttpServletRequest;
+import org.springframework.http.MediaType;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+import tech.easyflow.admin.service.ai.WorkflowChatEventStream;
+import tech.easyflow.ai.easyagentsflow.entity.WorkflowCheckStage;
+import tech.easyflow.ai.easyagentsflow.service.WorkflowCheckService;
+import tech.easyflow.ai.easyagentsflow.service.WorkflowRunningParameterResolver;
+import tech.easyflow.ai.easyagentsflow.support.PublishedWorkflowDefinitionIds;
+import tech.easyflow.ai.entity.Workflow;
+import tech.easyflow.ai.entity.WorkflowExecResult;
+import tech.easyflow.ai.entity.WorkflowExecStep;
+import tech.easyflow.ai.enums.PublishStatus;
+import tech.easyflow.ai.service.WorkflowExecResultService;
+import tech.easyflow.ai.service.WorkflowExecStepService;
+import tech.easyflow.ai.service.WorkflowService;
+import tech.easyflow.ai.service.WorkflowShareService;
+import tech.easyflow.ai.share.WorkflowSharePolicy;
+import tech.easyflow.ai.utils.WorkFlowUtil;
+import tech.easyflow.common.constant.Constants;
+import tech.easyflow.common.domain.Result;
+import tech.easyflow.common.entity.LoginAccount;
+import tech.easyflow.common.satoken.util.SaTokenUtil;
+import tech.easyflow.common.web.exceptions.BusinessException;
+import tech.easyflow.common.web.jsonbody.JsonBody;
+import tech.easyflow.system.enums.CategoryResourceType;
+import tech.easyflow.system.enums.ResourceAction;
+import tech.easyflow.system.service.ResourceAccessService;
+
+import javax.annotation.Resource;
+import java.math.BigInteger;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 工作流管理端与分享端的对话运行接口。
+ */
+@RestController
+@RequestMapping("/api/v1/workflowChat")
+public class WorkflowChatController {
+
+    @Resource
+    private WorkflowService workflowService;
+    @Resource
+    private WorkflowShareService workflowShareService;
+    @Resource
+    private WorkflowCheckService workflowCheckService;
+    @Resource
+    private WorkflowRunningParameterResolver parameterResolver;
+    @Resource
+    private ResourceAccessService resourceAccessService;
+    @Resource
+    private WorkflowChatEventStream eventStream;
+    @Resource
+    private ChainExecutor chainExecutor;
+    @Resource
+    private WorkflowExecResultService execResultService;
+    @Resource
+    private WorkflowExecStepService execStepService;
+
+    /**
+     * 获取工作流的对话运行描述和输入表单。
+     *
+     * @param workflowId 工作流 ID
+     * @param request HTTP 请求
+     * @return 对话运行描述
+     */
+    @GetMapping("/descriptor")
+    public Result<Map<String, Object>> descriptor(
+            BigInteger workflowId,
+            HttpServletRequest request
+    ) {
+        Workflow workflow = loadRunnableWorkflow(workflowId, request);
+        workflowCheckService.checkOrThrow(
+                workflow.getContent(),
+                WorkflowCheckStage.PRE_EXECUTE,
+                workflow.getId()
+        );
+        Map<String, Object> descriptor =
+                parameterResolver.buildRunningParametersView(workflow);
+        if (descriptor == null) {
+            throw new BusinessException("工作流输入配置无法解析");
+        }
+        descriptor.put("workflowId", workflow.getId());
+        descriptor.put("publishStatus", workflow.getPublishStatus());
+        descriptor.put("shareable", isStrictlyPublished(workflow));
+        return Result.ok(descriptor);
+    }
+
+    /**
+     * 启动工作流，并流式返回全部可见输出。
+     *
+     * @param workflowId 工作流 ID
+     * @param variables 工作流运行变量
+     * @param request HTTP 请求
+     * @return 工作流 SSE 事件流
+     */
+    @PostMapping(
+            value = "/run",
+            produces = MediaType.TEXT_EVENT_STREAM_VALUE
+    )
+    public SseEmitter run(
+            @JsonBody(value = "workflowId", required = true)
+            BigInteger workflowId,
+            @JsonBody("variables") Map<String, Object> variables,
+            HttpServletRequest request
+    ) {
+        Workflow workflow = loadRunnableWorkflow(workflowId, request);
+        workflowCheckService.checkOrThrow(
+                workflow.getContent(),
+                WorkflowCheckStage.PRE_EXECUTE,
+                workflow.getId()
+        );
+        Map<String, Object> normalizedVariables =
+                parameterResolver.normalizeRuntimeVariables(
+                        workflow.getContent(),
+                        variables
+                );
+        LoginAccount account = SaTokenUtil.getLoginAccount();
+        normalizedVariables.put(Constants.LOGIN_USER_KEY, account);
+        normalizedVariables.put(
+                WorkFlowUtil.CREATED_KEY_MEMORY_KEY,
+                hasChatShareKey(request)
+                        ? WorkFlowUtil.WORKFLOW_CHAT_SHARE
+                        : WorkFlowUtil.WORKFLOW_CHAT
+        );
+        return eventStream.start(
+                isStrictlyPublished(workflow)
+                        ? PublishedWorkflowDefinitionIds.published(workflowId.toString())
+                        : workflowId.toString(),
+                normalizedVariables
+        );
+    }
+
+    /**
+     * 取消当前用户发起的工作流执行。
+     *
+     * @param executeId 执行实例 ID
+     * @return 是否完成取消状态转换
+     */
+    @PostMapping("/cancel")
+    public Result<Boolean> cancel(
+            @JsonBody(value = "executeId", required = true)
+            String executeId
+    ) {
+        assertExecutionOwnership(executeId);
+        return Result.ok(chainExecutor.cancel(executeId, "用户已中止运行"));
+    }
+
+    /**
+     * 恢复当前用户发起并等待确认的工作流执行。
+     *
+     * @param executeId 执行实例 ID
+     * @param confirmParams 确认参数
+     * @return 空结果
+     */
+    @PostMapping("/resume")
+    public Result<Void> resume(
+            @JsonBody(value = "executeId", required = true)
+            String executeId,
+            @JsonBody("confirmParams")
+            Map<String, Object> confirmParams
+    ) {
+        WorkflowExecResult record = assertExecutionOwnership(executeId);
+        if (record.getStatus() != null
+                && (record.getStatus() == ChainStatus.SUCCEEDED.getValue()
+                || record.getStatus() == ChainStatus.FAILED.getValue()
+                || record.getStatus() == ChainStatus.CANCELLED.getValue())) {
+            throw new BusinessException("当前工作流执行已结束");
+        }
+        chainExecutor.resumeAsync(
+                executeId,
+                confirmParams == null
+                        ? new LinkedHashMap<>()
+                        : new LinkedHashMap<>(confirmParams)
+        );
+        return Result.ok();
+    }
+
+    /**
+     * 获取当前用户工作流执行的运行详情。
+     *
+     * @param executeId 执行实例 ID
+     * @return 执行记录和有序节点步骤
+     */
+    @GetMapping("/execution")
+    public Result<Map<String, Object>> detail(String executeId) {
+        WorkflowExecResult record = assertExecutionOwnership(executeId);
+        List<WorkflowExecStep> steps = execStepService.list(
+                QueryWrapper.create()
+                        .eq(WorkflowExecStep::getRecordId, record.getId())
+                        .orderBy(WorkflowExecStep::getStartTime, true)
+        );
+        List<Map<String, Object>> stepViews = new ArrayList<>(steps.size());
+        for (WorkflowExecStep step : steps) {
+            Map<String, Object> view = new LinkedHashMap<>();
+            view.put("id", step.getId());
+            view.put("attemptKey", step.getExecKey());
+            view.put("nodeId", step.getNodeId());
+            view.put("nodeName", step.getNodeName());
+            view.put("input", step.getInput());
+            view.put("output", step.getOutput());
+            view.put("status", step.getStatus());
+            view.put("errorInfo", step.getErrorInfo());
+            view.put("startTime", step.getStartTime());
+            view.put("endTime", step.getEndTime());
+            view.put("execTime", step.getExecTime());
+            stepViews.add(view);
+        }
+
+        Map<String, Object> recordView = new LinkedHashMap<>();
+        recordView.put("executeId", record.getExecKey());
+        recordView.put("workflowId", record.getWorkflowId());
+        recordView.put("title", record.getTitle());
+        recordView.put("status", record.getStatus());
+        recordView.put("input", record.getInput());
+        recordView.put("output", record.getOutput());
+        recordView.put("errorInfo", record.getErrorInfo());
+        recordView.put("startTime", record.getStartTime());
+        recordView.put("endTime", record.getEndTime());
+        recordView.put("execTime", record.getExecTime());
+
+        Map<String, Object> detail = new LinkedHashMap<>();
+        detail.put("record", recordView);
+        detail.put("steps", stepViews);
+        return Result.ok(detail);
+    }
+
+    /**
+     * 加载可运行工作流，并校验直接访问或对话分享权限。
+     * 管理端直接运行时，已发布工作流使用发布快照，未发布工作流使用当前内容；
+     * 分享运行始终要求严格发布快照。
+     *
+     * @param workflowId 工作流 ID
+     * @param request HTTP 请求
+     * @return 可运行工作流视图
+     */
+    private Workflow loadRunnableWorkflow(
+            BigInteger workflowId,
+            HttpServletRequest request
+    ) {
+        if (workflowId == null) {
+            throw new BusinessException("工作流ID不能为空");
+        }
+        LoginAccount account = SaTokenUtil.getLoginAccount();
+        Workflow current = workflowService.getById(workflowId);
+        if (current == null) {
+            throw new BusinessException("工作流不存在");
+        }
+        boolean sharedRequest = hasChatShareKey(request);
+        if (sharedRequest) {
+            workflowShareService.assertChatShareAccess(
+                    request.getHeader(
+                            WorkflowSharePolicy.CHAT_SHARE_KEY_HEADER
+                    ),
+                    workflowId,
+                    account.getTenantId()
+            );
+        } else {
+            resourceAccessService.assertAccess(
+                    CategoryResourceType.WORKFLOW,
+                    current,
+                    ResourceAction.USE,
+                    "无权限运行工作流"
+            );
+        }
+
+        Workflow published = workflowService.getPublishedById(workflowId);
+        if (isStrictlyPublished(published)) {
+            return published;
+        }
+        if (sharedRequest) {
+            throw new BusinessException(
+                    409,
+                    409,
+                    "工作流尚未发布或已下线"
+            );
+        }
+        return current;
+    }
+
+    /**
+     * 校验执行记录属于当前用户发起的工作流对话。
+     *
+     * @param executeId 执行实例 ID
+     * @return 执行记录
+     */
+    private WorkflowExecResult assertExecutionOwnership(String executeId) {
+        if (executeId == null || executeId.isBlank()) {
+            throw new BusinessException("执行ID不能为空");
+        }
+        WorkflowExecResult record = execResultService.getByExecKey(executeId);
+        if (record == null) {
+            throw new BusinessException("工作流执行记录不存在，请稍后重试");
+        }
+        LoginAccount account = SaTokenUtil.getLoginAccount();
+        boolean chatSource = WorkFlowUtil.WORKFLOW_CHAT.equals(
+                record.getCreatedKey()
+        ) || WorkFlowUtil.WORKFLOW_CHAT_SHARE.equals(record.getCreatedKey());
+        if (!chatSource
+                || account.getId() == null
+                || !account.getId().toString().equals(
+                record.getCreatedBy()
+        )) {
+            throw new BusinessException(
+                    403,
+                    403,
+                    "无权限访问当前工作流执行记录"
+            );
+        }
+        return record;
+    }
+
+    /**
+     * 判断请求是否携带对话分享密钥。
+     *
+     * @param request HTTP 请求
+     * @return 携带非空对话分享密钥时返回 {@code true}
+     */
+    private boolean hasChatShareKey(HttpServletRequest request) {
+        String shareKey = request == null
+                ? null
+                : request.getHeader(
+                WorkflowSharePolicy.CHAT_SHARE_KEY_HEADER
+        );
+        return shareKey != null && !shareKey.isBlank();
+    }
+
+    /**
+     * 判断工作流是否可按严格发布快照运行。
+     *
+     * @param workflow 工作流
+     * @return 已发布且存在快照时返回 {@code true}
+     */
+    private boolean isStrictlyPublished(Workflow workflow) {
+        return workflow != null
+                && PublishStatus.PUBLISHED.getCode().equals(
+                workflow.getPublishStatus()
+        )
+                && workflow.getPublishedSnapshotJson() != null
+                && !workflow.getPublishedSnapshotJson().isEmpty();
+    }
+}

easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowShareController.java

@@ -30,7 +30,7 @@ import java.net.URISyntaxException;
 import java.util.Map;
 
 /**
- * 工作流协作分享管理接口。
+ * 工作流分享管理接口。
  */
 @RestController
 @RequestMapping("/api/v1/workflowShare")
@@ -49,7 +49,7 @@ public class WorkflowShareController {
     private KnowledgeShareAuditService knowledgeShareAuditService;
 
     /**
-     * 创建或刷新工作流协作分享链接。
+     * 创建或刷新已发布工作流的对话分享链接。
      *
      * @param request HTTP 请求
      * @param workflowId 工作流 ID
@@ -72,7 +72,7 @@ public class WorkflowShareController {
                 "无权限分享工作流"
         );
         LoginAccount loginAccount = SaTokenUtil.getLoginAccount();
-        WorkflowShareCreateResult result = workflowShareService.createUrlShare(
+        WorkflowShareCreateResult result = workflowShareService.createChatShare(
                 workflowId,
                 loginAccount.getTenantId(),
                 loginAccount.getDeptId(),
@@ -81,8 +81,8 @@ public class WorkflowShareController {
         );
         knowledgeShareAuditService.log(
                 loginAccount.getId(),
-                "创建工作流协作分享",
-                "WORKFLOW_SHARE_CREATE",
+                "创建工作流对话分享",
+                "WORKFLOW_CHAT_SHARE_CREATE",
                 request.getRequestURI(),
                 Map.of("workflowId", workflowId, "shareId", result.getId())
         );
@@ -90,7 +90,7 @@ public class WorkflowShareController {
     }
 
     /**
-     * 解析当前 URL 分享指向的工作流。
+     * 解析当前对话分享指向的工作流。
      *
      * @param request HTTP 请求
      * @return 工作流标识
@@ -98,8 +98,8 @@ public class WorkflowShareController {
     @GetMapping("/resolve")
     public Result<Map<String, BigInteger>> resolveUrlShare(HttpServletRequest request) {
         LoginAccount loginAccount = SaTokenUtil.getLoginAccount();
-        WorkflowShare share = workflowShareService.resolveUrlShare(
-                request.getHeader(WorkflowSharePolicy.SHARE_KEY_HEADER),
+        WorkflowShare share = workflowShareService.resolveChatShare(
+                request.getHeader(WorkflowSharePolicy.CHAT_SHARE_KEY_HEADER),
                 loginAccount.getTenantId()
         );
         return Result.ok(Map.of("workflowId", share.getWorkflowId()));

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

@@ -0,0 +1,569 @@
+package tech.easyflow.admin.service.ai;
+
+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.Edge;
+import com.easyagents.flow.core.chain.Event;
+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;
+import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
+
+import javax.annotation.PostConstruct;
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * 将工作流运行事件转换为对话页可消费的 SSE 事件流。
+ */
+@Service
+public class WorkflowChatEventStream {
+
+    private static final Logger log =
+            LoggerFactory.getLogger(WorkflowChatEventStream.class);
+    private static final long SSE_TIMEOUT_MILLIS = 30L * 60L * 1000L;
+
+    private final ChainExecutor chainExecutor;
+    private final Map<String, StreamSession> sessions =
+            new ConcurrentHashMap<>();
+
+    /**
+     * 创建工作流对话事件流服务。
+     *
+     * @param chainExecutor 工作流执行器
+     */
+    public WorkflowChatEventStream(ChainExecutor chainExecutor) {
+        this.chainExecutor = chainExecutor;
+    }
+
+    /**
+     * 注册工作流全局事件监听器。
+     */
+    @PostConstruct
+    public void registerListeners() {
+        chainExecutor.addEventListener(this::onEvent);
+        chainExecutor.addOutputListener(this::onExplicitOutput);
+        chainExecutor.addErrorListener(this::onChainError);
+    }
+
+    /**
+     * 启动工作流并返回其 SSE 连接。
+     *
+     * @param definitionId 工作流定义 ID
+     * @param variables 运行变量
+     * @return SSE 连接
+     */
+    public SseEmitter start(String definitionId, Map<String, Object> variables) {
+        SseEmitter emitter = new SseEmitter(SSE_TIMEOUT_MILLIS);
+        StreamSession session = new StreamSession(emitter);
+        emitter.onTimeout(() -> disconnect(session, "运行连接超时"));
+        emitter.onError(error -> disconnect(session, "运行连接已断开"));
+        emitter.onCompletion(() -> removeSession(session));
+
+        try {
+            chainExecutor.executeAsync(
+                    definitionId,
+                    variables,
+                    executeId -> {
+                        session.attach(executeId);
+                        sessions.put(executeId, session);
+                        session.send("execution_started", Map.of(
+                                "executeId", executeId
+                        ));
+                    }
+            );
+        } catch (RuntimeException | Error error) {
+            session.fail(error);
+            throw error;
+        }
+        return emitter;
+    }
+
+    /**
+     * 将工作流事件转发到对应执行流。
+     *
+     * @param event 工作流事件
+     * @param chain 当前工作流
+     */
+    private void onEvent(Event event, Chain chain) {
+        StreamSession session = findSession(chain);
+        if (session == null) {
+            return;
+        }
+        if (event instanceof LlmStreamEvent streamEvent) {
+            session.onLlmDelta(chain, streamEvent);
+            return;
+        }
+        if (event instanceof NodeStartEvent nodeStartEvent) {
+            session.onNodeStarted(chain, nodeStartEvent);
+            return;
+        }
+        if (event instanceof NodeEndEvent nodeEndEvent) {
+            session.onNodeFinished(chain, nodeEndEvent);
+            return;
+        }
+        if (event instanceof EdgeTriggerEvent edgeTriggerEvent) {
+            session.onEdgeTriggered(chain, edgeTriggerEvent);
+            return;
+        }
+        if (event instanceof EdgeConditionCheckFailedEvent failedEvent) {
+            session.onEdgeConditionFailed(chain, failedEvent);
+            return;
+        }
+        if (event instanceof ChainStatusChangeEvent statusEvent
+                && Objects.equals(chain.getStateInstanceId(), session.executeId)) {
+            session.onStatusChanged(chain, statusEvent.getStatus());
+        }
+    }
+
+    /**
+     * 转发节点显式发布的每一条输出。
+     *
+     * @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);
+        }
+    }
+
+    /**
+     * 将链级异常发送到客户端。
+     *
+     * @param error 链级异常
+     * @param chain 当前工作流
+     */
+    private void onChainError(Throwable error, Chain chain) {
+        StreamSession session = findSession(chain);
+        if (session != null
+                && Objects.equals(chain.getStateInstanceId(), session.executeId)) {
+            session.send("execution_error", Map.of(
+                    "message", safeErrorMessage(error)
+            ));
+        }
+    }
+
+    /**
+     * 查找顶级执行对应的事件流会话。
+     *
+     * @param chain 当前工作流
+     * @return 流会话；不存在时为 {@code null}
+     */
+    private StreamSession findSession(Chain chain) {
+        if (chain == null) {
+            return null;
+        }
+        String auditInstanceId = chain.getAuditInstanceId();
+        if (auditInstanceId != null && !auditInstanceId.isBlank()) {
+            StreamSession session = sessions.get(auditInstanceId);
+            if (session != null) {
+                return session;
+            }
+        }
+        return sessions.get(chain.getStateInstanceId());
+    }
+
+    /**
+     * 处理 SSE 连接异常，并取消尚未结束的工作流。
+     *
+     * @param session 流会话
+     * @param message 取消原因
+     */
+    private void disconnect(StreamSession session, String message) {
+        if (session == null || session.terminal.get()) {
+            return;
+        }
+        String executeId = session.executeId;
+        removeSession(session);
+        if (executeId != null) {
+            chainExecutor.cancel(executeId, message);
+        }
+    }
+
+    /**
+     * 移除流会话。
+     *
+     * @param session 流会话
+     */
+    private void removeSession(StreamSession session) {
+        if (session != null && session.executeId != null) {
+            sessions.remove(session.executeId, session);
+        }
+    }
+
+    /**
+     * 读取适合返回给用户的异常信息。
+     *
+     * @param error 异常
+     * @return 非空异常信息
+     */
+    private String safeErrorMessage(Throwable error) {
+        if (error == null || error.getMessage() == null
+                || error.getMessage().isBlank()) {
+            return "工作流执行失败";
+        }
+        return error.getMessage();
+    }
+
+    /**
+     * 单次工作流执行的 SSE 会话。
+     */
+    private final class StreamSession {
+
+        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;
+
+        /**
+         * 创建流会话。
+         *
+         * @param emitter SSE 发送器
+         */
+        private StreamSession(SseEmitter emitter) {
+            this.emitter = emitter;
+        }
+
+        /**
+         * 绑定执行实例。
+         *
+         * @param executeId 执行实例 ID
+         */
+        private void attach(String executeId) {
+            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()
+            )));
+        }
+
+        /**
+         * 处理节点开始事件。
+         *
+         * @param chain 当前工作流
+         * @param event 节点开始事件
+         */
+        private void onNodeStarted(Chain chain, NodeStartEvent event) {
+            Node node = event.getNode();
+            Map<String, Object> data = new LinkedHashMap<>();
+            data.put("attemptKey", event.getExecutionAttemptKey());
+            data.put("nodeClass", node.getClass().getSimpleName());
+            data.put("chainInstanceId", chain.getStateInstanceId());
+            data.put("startedAt", System.currentTimeMillis());
+            data.put("input", resolveNodeInput(chain, node));
+            send("node_started", nodePayload(node, data));
+        }
+
+        /**
+         * 处理节点完成事件并输出所有可见结果。
+         *
+         * @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
+                    ? null
+                    : event.getStatus().name());
+            data.put("chainInstanceId", chain.getStateInstanceId());
+            data.put("finishedAt", System.currentTimeMillis());
+            data.put("output", event.getResult() == null
+                    ? Map.of()
+                    : event.getResult());
+            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());
+            }
+        }
+
+        /**
+         * 记录命中条件的流转分支。
+         *
+         * @param chain 当前工作流
+         * @param event 边触发事件
+         */
+        private void onEdgeTriggered(Chain chain, EdgeTriggerEvent event) {
+            if (event.getTrigger() == null) {
+                return;
+            }
+            Edge edge = chain.getDefinition().getEdgeById(
+                    event.getTrigger().getEdgeId());
+            if (edge == null || edge.getCondition() == null) {
+                return;
+            }
+            Node sourceNode = chain.getDefinition().getNodeById(
+                    edge.getSource());
+            sendEdgeTrace(chain, sourceNode, edge, "matched");
+        }
+
+        /**
+         * 记录未命中条件的流转分支。
+         *
+         * @param chain 当前工作流
+         * @param event 条件未命中事件
+         */
+        private void onEdgeConditionFailed(
+                Chain chain,
+                EdgeConditionCheckFailedEvent event
+        ) {
+            sendEdgeTrace(chain, event.getNode(), event.getEdge(), "skipped");
+        }
+
+        /**
+         * 发送节点的条件判断轨迹。
+         *
+         * @param chain 当前工作流
+         * @param sourceNode 条件来源节点
+         * @param edge 被判断的边
+         * @param outcome 判断结果
+         */
+        private void sendEdgeTrace(
+                Chain chain,
+                Node sourceNode,
+                Edge edge,
+                String outcome
+        ) {
+            if (sourceNode == null || edge == null) {
+                return;
+            }
+            Node targetNode = chain.getDefinition().getNodeById(
+                    edge.getTarget());
+            Map<String, Object> data = new LinkedHashMap<>();
+            data.put(
+                    "attemptKey",
+                    chain.currentExecutionAttemptKey(sourceNode.getId()));
+            data.put("kind", "condition");
+            data.put("outcome", outcome);
+            data.put("edgeId", edge.getId());
+            data.put("targetNodeId", edge.getTarget());
+            data.put(
+                    "targetNodeName",
+                    targetNode == null ? edge.getTarget() : targetNode.getName());
+            send("node_trace", nodePayload(sourceNode, data));
+        }
+
+        /**
+         * 解析节点本次执行实际使用的输入。
+         *
+         * @param chain 当前工作流
+         * @param node 当前节点
+         * @return 可序列化的节点输入
+         */
+        private Map<String, Object> resolveNodeInput(Chain chain, Node node) {
+            try {
+                return chain.getExecutionState()
+                        .resolveParametersPreservingReferences(node);
+            } catch (RuntimeException error) {
+                log.warn(
+                        "Failed to resolve workflow node input, "
+                                + "executeId={}, nodeId={}",
+                        executeId,
+                        node == null ? null : node.getId(),
+                        error
+                );
+                return Map.of();
+            }
+        }
+
+        /**
+         * 处理工作流状态变化。
+         *
+         * @param chain 当前工作流
+         * @param status 新状态
+         */
+        private void onStatusChanged(Chain chain, ChainStatus status) {
+            if (status == null) {
+                return;
+            }
+            if (status == ChainStatus.SUSPEND) {
+                Map<String, Object> data = new LinkedHashMap<>();
+                data.put("message", chain.getState().getMessage());
+                data.put(
+                        "parameters",
+                        chain.getState().getSuspendForParameters()
+                );
+                send("execution_waiting", data);
+                return;
+            }
+            if (!status.isTerminal()) {
+                send("execution_status", Map.of("status", status.name()));
+                return;
+            }
+            if (!terminal.compareAndSet(false, true)) {
+                return;
+            }
+            String eventType = switch (status) {
+                case SUCCEEDED -> "execution_finished";
+                case CANCELLED -> "execution_cancelled";
+                default -> "execution_failed";
+            };
+            Map<String, Object> data = new LinkedHashMap<>();
+            data.put("status", status.name());
+            data.put("message", chain.getState().getMessage());
+            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 事件。
+         *
+         * @param type 事件类型
+         * @param data 事件数据
+         */
+        private void send(String type, Map<String, ?> data) {
+            long nextSequence = sequence.incrementAndGet();
+            Map<String, Object> payload = new LinkedHashMap<>();
+            payload.put("eventId", executeId + ":" + nextSequence);
+            payload.put("sequence", nextSequence);
+            payload.put("executeId", executeId);
+            payload.put("type", type);
+            payload.put("data", data);
+            try {
+                emitter.send(SseEmitter.event()
+                        .id(String.valueOf(nextSequence))
+                        .name("workflow")
+                        // 显式发送 JSON 文本，避免全局 CBOR 转换器将 SSE data 编码为二进制。
+                        .data(JSON.toJSONString(payload)));
+            } catch (IOException | IllegalStateException error) {
+                log.debug(
+                        "workflow chat stream disconnected, executeId={}",
+                        executeId,
+                        error
+                );
+                disconnect(this, "运行连接已断开");
+            }
+        }
+
+        /**
+         * 在启动失败时关闭 SSE 会话。
+         *
+         * @param error 启动异常
+         */
+        private void fail(Throwable error) {
+            if (terminal.compareAndSet(false, true)) {
+                send("execution_failed", Map.of(
+                        "message", safeErrorMessage(error)
+                ));
+                removeSession(this);
+                emitter.completeWithError(error);
+            }
+        }
+
+        /**
+         * 构建节点本次执行的关联键。
+         *
+         * @param chain 当前工作流
+         * @param node 当前节点
+         * @return 节点运行键
+         */
+        private String nodeRunKey(Chain chain, Node node) {
+            return chain.getStateInstanceId() + ":" + node.getId();
+        }
+
+        /**
+         * 构建带节点信息的事件数据。
+         *
+         * @param node 当前节点
+         * @param values 业务数据
+         * @return 事件数据
+         */
+        private Map<String, Object> nodePayload(
+                Node node,
+                Map<String, ?> values
+        ) {
+            Map<String, Object> payload = new LinkedHashMap<>();
+            payload.put("nodeId", node == null ? null : node.getId());
+            payload.put("nodeName", node == null ? null : node.getName());
+            if (values != null) {
+                payload.putAll(values);
+            }
+            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;
+        }
+    }
+}