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 new file mode 100644 index 00000000..e2091282 --- /dev/null +++ b/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> descriptor( + BigInteger workflowId, + HttpServletRequest request + ) { + Workflow workflow = loadRunnableWorkflow(workflowId, request); + workflowCheckService.checkOrThrow( + workflow.getContent(), + WorkflowCheckStage.PRE_EXECUTE, + workflow.getId() + ); + Map 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 variables, + HttpServletRequest request + ) { + Workflow workflow = loadRunnableWorkflow(workflowId, request); + workflowCheckService.checkOrThrow( + workflow.getContent(), + WorkflowCheckStage.PRE_EXECUTE, + workflow.getId() + ); + Map 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 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 resume( + @JsonBody(value = "executeId", required = true) + String executeId, + @JsonBody("confirmParams") + Map 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> detail(String executeId) { + WorkflowExecResult record = assertExecutionOwnership(executeId); + List steps = execStepService.list( + QueryWrapper.create() + .eq(WorkflowExecStep::getRecordId, record.getId()) + .orderBy(WorkflowExecStep::getStartTime, true) + ); + List> stepViews = new ArrayList<>(steps.size()); + for (WorkflowExecStep step : steps) { + Map 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 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 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(); + } +} diff --git a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowShareController.java b/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowShareController.java index 3c3ebb39..205ff6ac 100644 --- a/easyflow-api/easyflow-api-admin/src/main/java/tech/easyflow/admin/controller/ai/WorkflowShareController.java +++ b/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> 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())); 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 new file mode 100644 index 00000000..e5ba2740 --- /dev/null +++ b/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 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 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 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 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 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 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 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 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 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 data) { + long nextSequence = sequence.incrementAndGet(); + Map 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 nodePayload( + Node node, + Map values + ) { + Map 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 visibleEndOutput( + Map result + ) { + Map 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; + } + } +} diff --git a/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/WorkflowChatControllerTest.java b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/WorkflowChatControllerTest.java new file mode 100644 index 00000000..bc0e62b8 --- /dev/null +++ b/easyflow-api/easyflow-api-admin/src/test/java/tech/easyflow/admin/controller/ai/WorkflowChatControllerTest.java @@ -0,0 +1,326 @@ +package tech.easyflow.admin.controller.ai; + +import com.easyagents.flow.core.chain.runtime.ChainExecutor; +import jakarta.servlet.http.HttpServletRequest; +import org.mockito.MockedStatic; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import org.testng.Assert; +import org.testng.annotations.Test; +import tech.easyflow.admin.service.ai.WorkflowChatEventStream; +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.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.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.system.service.ResourceAccessService; + +import java.lang.reflect.Field; +import java.math.BigInteger; +import java.util.LinkedHashMap; +import java.util.Locale; +import java.util.Map; + +import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * {@link WorkflowChatController} 运行来源与分享边界测试。 + */ +public class WorkflowChatControllerTest { + + /** + * 验证登录用户可以进入并运行未发布工作流，同时描述信息禁止分享。 + */ + @Test + public void shouldRunDraftForAuthenticatedRequestWithoutSharing() { + ControllerFixture fixture = fixture(workflow(PublishStatus.DRAFT, "draft-content", false)); + when(fixture.parameterResolver.buildRunningParametersView(fixture.current)) + .thenReturn(new LinkedHashMap<>()); + when(fixture.parameterResolver.normalizeRuntimeVariables( + eq("draft-content"), + anyMap() + )).thenReturn(new LinkedHashMap<>()); + when(fixture.eventStream.start(eq("1"), anyMap())).thenReturn(new SseEmitter()); + + try (MockedStatic login = login(fixture.account)) { + Result> descriptor = fixture.controller.descriptor( + BigInteger.ONE, + request(Map.of()) + ); + fixture.controller.run(BigInteger.ONE, Map.of(), request(Map.of())); + + Assert.assertEquals(descriptor.getData().get("shareable"), false); + Assert.assertEquals( + descriptor.getData().get("publishStatus"), + PublishStatus.DRAFT.getCode() + ); + } + + verify(fixture.parameterResolver).buildRunningParametersView(fixture.current); + verify(fixture.eventStream).start(eq("1"), anyMap()); + } + + /** + * 验证已发布工作流的管理端运行继续读取发布快照。 + */ + @Test + public void shouldRunPublishedSnapshotForAuthenticatedRequest() { + Workflow current = workflow(PublishStatus.PUBLISHED, "draft-content", true); + Workflow published = workflow(PublishStatus.PUBLISHED, "published-content", true); + ControllerFixture fixture = fixture(current, published); + when(fixture.parameterResolver.buildRunningParametersView(published)) + .thenReturn(new LinkedHashMap<>()); + when(fixture.parameterResolver.normalizeRuntimeVariables( + eq("published-content"), + anyMap() + )).thenReturn(new LinkedHashMap<>()); + when(fixture.eventStream.start( + eq(PublishedWorkflowDefinitionIds.published("1")), + anyMap() + )).thenReturn(new SseEmitter()); + + try (MockedStatic login = login(fixture.account)) { + Result> descriptor = fixture.controller.descriptor( + BigInteger.ONE, + request(Map.of()) + ); + fixture.controller.run(BigInteger.ONE, Map.of(), request(Map.of())); + + Assert.assertEquals(descriptor.getData().get("shareable"), true); + } + + verify(fixture.parameterResolver).buildRunningParametersView(published); + verify(fixture.eventStream).start( + eq(PublishedWorkflowDefinitionIds.published("1")), + anyMap() + ); + } + + /** + * 验证分享访问仍拒绝未发布工作流。 + */ + @Test + public void shouldRejectDraftWorkflowFromShareRequest() { + ControllerFixture fixture = fixture(workflow(PublishStatus.DRAFT, "draft-content", false)); + HttpServletRequest request = request(Map.of( + WorkflowSharePolicy.CHAT_SHARE_KEY_HEADER.toLowerCase(Locale.ROOT), + "share-key" + )); + + try (MockedStatic login = login(fixture.account)) { + Assert.expectThrows( + BusinessException.class, + () -> fixture.controller.descriptor(BigInteger.ONE, request) + ); + } + + verify(fixture.workflowShareService).assertChatShareAccess( + "share-key", + BigInteger.ONE, + BigInteger.ONE + ); + verify(fixture.resourceAccessService, never()).assertAccess( + org.mockito.ArgumentMatchers.any(), + org.mockito.ArgumentMatchers.any(), + org.mockito.ArgumentMatchers.any(), + org.mockito.ArgumentMatchers.anyString() + ); + } + + /** + * 创建使用同一当前视图和发布视图的测试夹具。 + * + * @param current 当前工作流 + * @return 控制器测试夹具 + */ + private ControllerFixture fixture(Workflow current) { + return fixture(current, current); + } + + /** + * 创建控制器测试夹具。 + * + * @param current 当前工作流 + * @param published 发布工作流视图 + * @return 控制器测试夹具 + */ + private ControllerFixture fixture(Workflow current, Workflow published) { + WorkflowService workflowService = mock(WorkflowService.class); + WorkflowShareService workflowShareService = mock(WorkflowShareService.class); + WorkflowCheckService workflowCheckService = mock(WorkflowCheckService.class); + WorkflowRunningParameterResolver parameterResolver = + mock(WorkflowRunningParameterResolver.class); + ResourceAccessService resourceAccessService = mock(ResourceAccessService.class); + WorkflowChatEventStream eventStream = mock(WorkflowChatEventStream.class); + WorkflowChatController controller = new WorkflowChatController(); + setField(controller, "workflowService", workflowService); + setField(controller, "workflowShareService", workflowShareService); + setField(controller, "workflowCheckService", workflowCheckService); + setField(controller, "parameterResolver", parameterResolver); + setField(controller, "resourceAccessService", resourceAccessService); + setField(controller, "eventStream", eventStream); + setField(controller, "chainExecutor", mock(ChainExecutor.class)); + setField(controller, "execResultService", mock(WorkflowExecResultService.class)); + setField(controller, "execStepService", mock(WorkflowExecStepService.class)); + when(workflowService.getById(BigInteger.ONE)).thenReturn(current); + when(workflowService.getPublishedById(BigInteger.ONE)).thenReturn(published); + + LoginAccount account = new LoginAccount(); + account.setId(BigInteger.ONE); + account.setTenantId(BigInteger.ONE); + return new ControllerFixture( + controller, + current, + workflowService, + workflowShareService, + parameterResolver, + resourceAccessService, + eventStream, + account + ); + } + + /** + * 创建工作流测试视图。 + * + * @param publishStatus 发布状态 + * @param content 工作流内容 + * @param withSnapshot 是否包含发布快照 + * @return 工作流测试视图 + */ + private Workflow workflow( + PublishStatus publishStatus, + String content, + boolean withSnapshot + ) { + Workflow workflow = new Workflow(); + workflow.setId(BigInteger.ONE); + workflow.setContent(content); + workflow.setPublishStatus(publishStatus.getCode()); + if (withSnapshot) { + workflow.setPublishedSnapshotJson(Map.of("content", content)); + } + return workflow; + } + + /** + * 创建登录账号静态模拟。 + * + * @param account 登录账号 + * @return 静态模拟句柄 + */ + private MockedStatic login(LoginAccount account) { + MockedStatic login = mockStatic(SaTokenUtil.class); + login.when(SaTokenUtil::getLoginAccount).thenReturn(account); + return login; + } + + /** + * 创建仅提供请求头能力的轻量 Servlet 请求代理。 + * + * @param headers 小写请求头映射 + * @return HTTP 请求代理 + */ + private HttpServletRequest request(Map headers) { + return (HttpServletRequest) java.lang.reflect.Proxy.newProxyInstance( + getClass().getClassLoader(), + new Class[]{HttpServletRequest.class}, + (proxy, method, args) -> { + if ("getHeader".equals(method.getName())) { + String name = String.valueOf(args[0]).toLowerCase(Locale.ROOT); + return headers.get(name); + } + return defaultValue(method.getReturnType()); + } + ); + } + + /** + * 通过反射设置控制器依赖。 + * + * @param target 目标对象 + * @param fieldName 字段名 + * @param value 字段值 + */ + private void setField(Object target, String fieldName, Object value) { + try { + Field field = target.getClass().getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + } catch (ReflectiveOperationException exception) { + throw new IllegalStateException("设置测试字段失败: " + fieldName, exception); + } + } + + /** + * 返回代理方法所需的基础类型默认值。 + * + * @param returnType 返回类型 + * @return 默认值 + */ + private Object defaultValue(Class returnType) { + if (!returnType.isPrimitive()) { + return null; + } + if (boolean.class == returnType) { + return false; + } + if (char.class == returnType) { + return '\0'; + } + if (byte.class == returnType) { + return (byte) 0; + } + if (short.class == returnType) { + return (short) 0; + } + if (int.class == returnType) { + return 0; + } + if (long.class == returnType) { + return 0L; + } + if (float.class == returnType) { + return 0F; + } + return 0D; + } + + /** + * 控制器及其测试依赖夹具。 + * + * @param controller 控制器 + * @param current 当前工作流 + * @param workflowService 工作流服务 + * @param workflowShareService 工作流分享服务 + * @param parameterResolver 参数解析器 + * @param resourceAccessService 资源权限服务 + * @param eventStream 事件流服务 + * @param account 登录账号 + */ + private record ControllerFixture( + WorkflowChatController controller, + Workflow current, + WorkflowService workflowService, + WorkflowShareService workflowShareService, + WorkflowRunningParameterResolver parameterResolver, + ResourceAccessService resourceAccessService, + WorkflowChatEventStream eventStream, + LoginAccount account + ) { + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/llm/LlmProviderImpl.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/llm/LlmProviderImpl.java index 57e5e203..f6e3b1e4 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/llm/LlmProviderImpl.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/llm/LlmProviderImpl.java @@ -18,7 +18,15 @@ public class LlmProviderImpl implements LlmProvider { private static final Logger log = LoggerFactory.getLogger(LlmProviderImpl.class); @Resource private ModelService modelService; + @Resource + private WorkflowImageSourceResolver workflowImageSourceResolver; + /** + * 根据模型标识创建工作流聊天模型适配器。 + * + * @param modelId 模型标识 + * @return 工作流 LLM；模型不存在时返回 {@code null} + */ @Override public Llm getChatModel(Object modelId) { Model model = modelService.getModelInstance(new BigInteger(modelId.toString())); @@ -28,6 +36,7 @@ public class LlmProviderImpl implements LlmProvider { } EasyAgentsLlm llm = new EasyAgentsLlm(); llm.setChatModel(model.toChatModel()); + llm.setImageInputResolver(workflowImageSourceResolver); return llm; } } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/llm/WorkflowImageSourceResolver.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/llm/WorkflowImageSourceResolver.java new file mode 100644 index 00000000..5312c7db --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/llm/WorkflowImageSourceResolver.java @@ -0,0 +1,584 @@ +package tech.easyflow.ai.easyagentsflow.llm; + +import com.easyagents.core.util.ImageUtil; +import com.easyagents.flow.support.provider.ImageInputResolver; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; +import tech.easyflow.common.filestorage.FileStorageService; +import tech.easyflow.common.web.exceptions.BusinessException; + +import javax.imageio.ImageIO; +import javax.imageio.ImageReader; +import javax.imageio.stream.ImageInputStream; +import java.awt.image.BufferedImage; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.InputStream; +import java.net.Inet4Address; +import java.net.Inet6Address; +import java.net.InetAddress; +import java.net.URI; +import java.net.URISyntaxException; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import java.util.Base64; +import java.util.Iterator; +import java.util.Locale; +import java.util.Map; + +/** + * 在模型调用前读取、校验并转换工作流图片输入。 + */ +@Component +public class WorkflowImageSourceResolver implements ImageInputResolver { + + static final long MAX_IMAGE_BYTES = 10L * 1024 * 1024; + static final long MAX_IMAGE_PIXELS = 40_000_000L; + private static final int MAX_REDIRECTS = 3; + private static final Duration CONNECT_TIMEOUT = Duration.ofSeconds(8); + private static final Duration REQUEST_TIMEOUT = Duration.ofSeconds(15); + private static final String DATA_URI_PREFIX = "data:image/"; + private static final String BASE64_MARKER = ";base64,"; + + private final FileStorageService fileStorageService; + private final HttpClient httpClient; + + /** + * 创建工作流图片解析器。 + * + * @param fileStorageService 文件存储服务 + */ + @Autowired + public WorkflowImageSourceResolver( + @Qualifier("default") FileStorageService fileStorageService) { + this(fileStorageService, HttpClient.newBuilder() + .connectTimeout(CONNECT_TIMEOUT) + .followRedirects(HttpClient.Redirect.NEVER) + .version(HttpClient.Version.HTTP_1_1) + .build()); + } + + /** + * 创建使用指定 HTTP 客户端的图片解析器，供隔离测试使用。 + * + * @param fileStorageService 文件存储服务 + * @param httpClient HTTP 客户端 + */ + WorkflowImageSourceResolver(FileStorageService fileStorageService, HttpClient httpClient) { + this.fileStorageService = fileStorageService; + this.httpClient = httpClient; + } + + /** + * 解析图片输入并返回带 MIME 的 Data URI。 + * + * @param imageInput 图片 URL、文件或结构化图片描述 + * @return 可供模型消费的 Data URI + * @throws BusinessException 图片不可读取、格式不支持或超过限制时抛出 + */ + @Override + public String resolve(Object imageInput) { + if (imageInput instanceof File file) { + return process(readLocalFile(file)); + } + if (imageInput instanceof String value) { + return resolveString(value); + } + if (imageInput instanceof Map imageMap) { + return resolveMap(imageMap); + } + throw new BusinessException("图片输入格式不受支持"); + } + + /** + * 解析字符串形式的旧版图片输入。 + * + * @param value 图片 URL 或 Data URI + * @return 规范化 Data URI + */ + private String resolveString(String value) { + String normalized = trimToNull(value); + if (!StringUtils.hasText(normalized)) { + throw new BusinessException("图片输入不能为空"); + } + if (normalized.startsWith(DATA_URI_PREFIX)) { + return process(decodeDataUri(normalized)); + } + return process(download(normalized)); + } + + /** + * 解析结构化图片描述。 + * + * @param imageMap 图片描述 + * @return 规范化 Data URI + */ + private String resolveMap(Map imageMap) { + String sourceType = trimObjectToNull(imageMap.get("sourceType")); + String filePath = trimObjectToNull(imageMap.get("filePath")); + if (!StringUtils.hasText(sourceType)) { + sourceType = StringUtils.hasText(filePath) ? "upload" : "url"; + } + if ("url".equals(sourceType)) { + return process(download(trimObjectToNull(imageMap.get("url")))); + } + if (!"upload".equals(sourceType) && !"resource".equals(sourceType)) { + throw new BusinessException("图片 sourceType 不受支持"); + } + if (!StringUtils.hasText(filePath)) { + throw new BusinessException("图片缺少 filePath"); + } + try (InputStream input = fileStorageService.readStream(filePath)) { + return process(readBounded(input)); + } catch (BusinessException exception) { + throw exception; + } catch (Exception exception) { + throw new BusinessException(400, 1, "图片读取失败", exception); + } + } + + /** + * 安全下载外部图片，并在每次重定向后重新校验目标地址。 + * + * @param value 外部图片 URL + * @return 图片字节 + */ + private byte[] download(String value) { + URI current = parseRemoteUri(value); + for (int redirectCount = 0; redirectCount <= MAX_REDIRECTS; redirectCount++) { + validateRemoteUri(current); + HttpRequest request = HttpRequest.newBuilder(current) + .timeout(REQUEST_TIMEOUT) + .header("Accept", "image/png,image/jpeg,image/webp,image/gif,image/bmp") + .GET() + .build(); + try { + HttpResponse response = + httpClient.send(request, HttpResponse.BodyHandlers.ofInputStream()); + int status = response.statusCode(); + if (status >= 300 && status < 400) { + closeQuietly(response.body()); + if (redirectCount == MAX_REDIRECTS) { + throw new BusinessException("图片 URL 重定向次数超过限制"); + } + String location = response.headers().firstValue("location") + .orElseThrow(() -> new BusinessException("图片 URL 重定向缺少目标地址")); + current = current.resolve(location); + continue; + } + if (status < 200 || status >= 300) { + closeQuietly(response.body()); + throw new BusinessException("图片 URL 请求失败，HTTP 状态码: " + status); + } + try (InputStream input = response.body()) { + return readBounded(input); + } + } catch (BusinessException exception) { + throw exception; + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new BusinessException(400, 1, "图片 URL 请求被中断", exception); + } catch (Exception exception) { + throw new BusinessException(400, 1, "图片 URL 请求失败", exception); + } + } + throw new BusinessException("图片 URL 请求失败"); + } + + /** + * 校验远程图片 URI，阻止访问本机、内网和云元数据地址。 + * + * @param uri 待访问 URI + * @throws BusinessException URI 不安全时抛出 + */ + void validateRemoteUri(URI uri) { + if (uri == null + || (!"http".equalsIgnoreCase(uri.getScheme()) + && !"https".equalsIgnoreCase(uri.getScheme()))) { + throw new BusinessException("图片 URL 仅支持 HTTP/HTTPS"); + } + if (uri.getUserInfo() != null) { + throw new BusinessException("图片 URL 不能包含用户信息"); + } + String host = trimToNull(uri.getHost()); + if (!StringUtils.hasText(host)) { + throw new BusinessException("图片 URL 缺少有效主机"); + } + String normalizedHost = host.toLowerCase(Locale.ROOT); + if ("localhost".equals(normalizedHost) + || normalizedHost.endsWith(".localhost") + || "metadata.google.internal".equals(normalizedHost)) { + throw new BusinessException("图片 URL 不能访问本机或云元数据地址"); + } + try { + InetAddress[] addresses = InetAddress.getAllByName(host); + if (addresses.length == 0) { + throw new BusinessException("图片 URL 主机无法解析"); + } + for (InetAddress address : addresses) { + if (!isPublicAddress(address)) { + throw new BusinessException("图片 URL 不能访问内网或保留地址"); + } + } + } catch (BusinessException exception) { + throw exception; + } catch (Exception exception) { + throw new BusinessException(400, 1, "图片 URL 主机解析失败", exception); + } + } + + /** + * 判断解析后的地址是否为允许访问的公网地址。 + * + * @param address IP 地址 + * @return 是否为公网地址 + */ + private boolean isPublicAddress(InetAddress address) { + if (address.isAnyLocalAddress() + || address.isLoopbackAddress() + || address.isLinkLocalAddress() + || address.isSiteLocalAddress() + || address.isMulticastAddress()) { + return false; + } + byte[] bytes = address.getAddress(); + if (address instanceof Inet4Address && bytes.length == 4) { + int first = bytes[0] & 0xff; + int second = bytes[1] & 0xff; + int third = bytes[2] & 0xff; + int fourth = bytes[3] & 0xff; + return first != 0 + && first != 10 + && first != 127 + && !(first == 168 && second == 63 && third == 129 && fourth == 16) + && !(first == 169 && second == 254) + && !(first == 100 && second >= 64 && second <= 127) + && !(first == 172 && second >= 16 && second <= 31) + && !(first == 192 && (second == 0 || second == 168)) + && !(first == 198 && (second == 18 || second == 19)) + && first < 224; + } + if (address instanceof Inet6Address && bytes.length == 16) { + int first = bytes[0] & 0xff; + return (first & 0xfe) != 0xfc; + } + return false; + } + + /** + * 校验图片格式、尺寸并规范化不兼容格式。 + * + * @param source 原始图片字节 + * @return 带 MIME 的 Data URI + */ + private String process(byte[] source) { + if (source.length == 0) { + throw new BusinessException("图片内容为空"); + } + ImageFormat format = detectFormat(source); + try { + Dimensions dimensions = format == ImageFormat.WEBP + ? webpDimensions(source) + : imageIoDimensions(source); + validateDimensions(dimensions); + + byte[] normalized = source; + String mimeType = format.mimeType; + if (format == ImageFormat.GIF || format == ImageFormat.BMP) { + BufferedImage image = ImageIO.read(new ByteArrayInputStream(source)); + if (image == null) { + throw new BusinessException("图片内容无法解析"); + } + ByteArrayOutputStream output = new ByteArrayOutputStream(); + if (!ImageIO.write(image, "png", output)) { + throw new BusinessException("图片格式转换失败"); + } + normalized = output.toByteArray(); + mimeType = "image/png"; + } + if (normalized.length > MAX_IMAGE_BYTES) { + throw new BusinessException("处理后的图片不能超过 10 MiB"); + } + return ImageUtil.imageBytesToDataUri(normalized, mimeType); + } catch (BusinessException exception) { + throw exception; + } catch (Exception exception) { + throw new BusinessException(400, 1, "图片处理失败", exception); + } + } + + /** + * 根据文件签名识别真实图片格式。 + * + * @param bytes 图片字节 + * @return 图片格式 + */ + private ImageFormat detectFormat(byte[] bytes) { + if (bytes.length >= 8 + && bytes[0] == (byte) 0x89 && bytes[1] == 0x50 + && bytes[2] == 0x4e && bytes[3] == 0x47) { + return ImageFormat.PNG; + } + if (bytes.length >= 3 + && bytes[0] == (byte) 0xff && bytes[1] == (byte) 0xd8 + && bytes[2] == (byte) 0xff) { + return ImageFormat.JPEG; + } + if (bytes.length >= 6) { + String header = ascii(bytes, 0, 6); + if ("GIF87a".equals(header) || "GIF89a".equals(header)) { + return ImageFormat.GIF; + } + } + if (bytes.length >= 2 && bytes[0] == 'B' && bytes[1] == 'M') { + return ImageFormat.BMP; + } + if (bytes.length >= 12 + && "RIFF".equals(ascii(bytes, 0, 4)) + && "WEBP".equals(ascii(bytes, 8, 4))) { + return ImageFormat.WEBP; + } + throw new BusinessException("仅支持 PNG、JPG、JPEG、WebP、GIF、BMP 图片"); + } + + /** + * 使用 ImageIO 读取图片尺寸。 + * + * @param bytes 图片字节 + * @return 图片尺寸 + * @throws IOException 无法读取图片时抛出 + */ + private Dimensions imageIoDimensions(byte[] bytes) throws IOException { + try (ImageInputStream input = + ImageIO.createImageInputStream(new ByteArrayInputStream(bytes))) { + Iterator readers = ImageIO.getImageReaders(input); + if (!readers.hasNext()) { + throw new BusinessException("图片内容无法解析"); + } + ImageReader reader = readers.next(); + try { + reader.setInput(input, true, true); + return new Dimensions(reader.getWidth(0), reader.getHeight(0)); + } finally { + reader.dispose(); + } + } + } + + /** + * 读取 WebP 图片尺寸。 + * + * @param bytes 图片字节 + * @return 图片尺寸 + */ + private Dimensions webpDimensions(byte[] bytes) { + int offset = 12; + while (offset + 8 <= bytes.length) { + String chunk = ascii(bytes, offset, 4); + long size = Integer.toUnsignedLong(littleEndianInt(bytes, offset + 4)); + int data = offset + 8; + if (size > bytes.length - data) { + break; + } + int chunkSize = (int) size; + if ("VP8X".equals(chunk) && chunkSize >= 10) { + return new Dimensions( + 1 + littleEndian24(bytes, data + 4), + 1 + littleEndian24(bytes, data + 7)); + } + if ("VP8 ".equals(chunk) && chunkSize >= 10 + && bytes[data + 3] == (byte) 0x9d + && bytes[data + 4] == 0x01 && bytes[data + 5] == 0x2a) { + return new Dimensions( + littleEndian16(bytes, data + 6) & 0x3fff, + littleEndian16(bytes, data + 8) & 0x3fff); + } + if ("VP8L".equals(chunk) && chunkSize >= 5 && bytes[data] == 0x2f) { + int b1 = unsigned(bytes[data + 1]); + int b2 = unsigned(bytes[data + 2]); + int b3 = unsigned(bytes[data + 3]); + int b4 = unsigned(bytes[data + 4]); + return new Dimensions( + 1 + ((b1 | b2 << 8) & 0x3fff), + 1 + (((b2 >> 6) | b3 << 2 | b4 << 10) & 0x3fff)); + } + long nextOffset = (long) data + chunkSize + (chunkSize & 1); + if (nextOffset > Integer.MAX_VALUE) { + break; + } + offset = (int) nextOffset; + } + throw new BusinessException("WebP 图片内容无法解析"); + } + + /** + * 校验图片像素数量。 + * + * @param dimensions 图片尺寸 + */ + private void validateDimensions(Dimensions dimensions) { + if (dimensions.width <= 0 || dimensions.height <= 0) { + throw new BusinessException("图片尺寸无效"); + } + long pixels = (long) dimensions.width * dimensions.height; + if (pixels > MAX_IMAGE_PIXELS) { + throw new BusinessException("图片不能超过 4000 万像素"); + } + } + + /** + * 读取文件并应用大小上限。 + * + * @param file 本地文件 + * @return 文件字节 + */ + private byte[] readLocalFile(File file) { + if (file == null || !file.isFile()) { + throw new BusinessException("图片文件不存在"); + } + try (InputStream input = java.nio.file.Files.newInputStream(file.toPath())) { + return readBounded(input); + } catch (BusinessException exception) { + throw exception; + } catch (Exception exception) { + throw new BusinessException(400, 1, "图片文件读取失败", exception); + } + } + + /** + * 流式读取图片，并拒绝超过 10 MiB 的内容。 + * + * @param input 图片输入流 + * @return 图片字节 + * @throws IOException 读取失败时抛出 + */ + private byte[] readBounded(InputStream input) throws IOException { + if (input == null) { + throw new BusinessException("图片读取失败"); + } + ByteArrayOutputStream output = new ByteArrayOutputStream(); + byte[] buffer = new byte[8192]; + long total = 0L; + int read; + while ((read = input.read(buffer)) != -1) { + total += read; + if (total > MAX_IMAGE_BYTES) { + throw new BusinessException("单张图片不能超过 10 MiB"); + } + output.write(buffer, 0, read); + } + return output.toByteArray(); + } + + /** + * 解码并校验图片 Data URI。 + * + * @param dataUri 图片 Data URI + * @return 图片字节 + */ + private byte[] decodeDataUri(String dataUri) { + int markerIndex = dataUri.indexOf(BASE64_MARKER); + if (markerIndex <= DATA_URI_PREFIX.length()) { + throw new BusinessException("图片 Data URI 格式不正确"); + } + String encoded = dataUri.substring(markerIndex + BASE64_MARKER.length()); + long maxEncodedLength = (MAX_IMAGE_BYTES + 2L) / 3L * 4L; + if (encoded.length() > maxEncodedLength + 2L) { + throw new BusinessException("单张图片不能超过 10 MiB"); + } + try { + byte[] bytes = Base64.getDecoder().decode(encoded); + if (bytes.length > MAX_IMAGE_BYTES) { + throw new BusinessException("单张图片不能超过 10 MiB"); + } + return bytes; + } catch (IllegalArgumentException exception) { + throw new BusinessException(400, 1, "图片 Data URI 编码无效", exception); + } + } + + /** + * 解析远程 URI。 + * + * @param value 原始 URL + * @return URI + */ + private URI parseRemoteUri(String value) { + if (!StringUtils.hasText(value)) { + throw new BusinessException("图片 URL 不能为空"); + } + try { + return new URI(value.trim()); + } catch (URISyntaxException exception) { + throw new BusinessException(400, 1, "图片 URL 格式不正确", exception); + } + } + + private String trimObjectToNull(Object value) { + return trimToNull(value == null ? null : String.valueOf(value)); + } + + private String trimToNull(String value) { + return StringUtils.hasText(value) ? value.trim() : null; + } + + private String ascii(byte[] bytes, int offset, int length) { + return new String(bytes, offset, length, StandardCharsets.US_ASCII); + } + + private int littleEndian16(byte[] bytes, int offset) { + return unsigned(bytes[offset]) | unsigned(bytes[offset + 1]) << 8; + } + + private int littleEndian24(byte[] bytes, int offset) { + return unsigned(bytes[offset]) + | unsigned(bytes[offset + 1]) << 8 + | unsigned(bytes[offset + 2]) << 16; + } + + private int littleEndianInt(byte[] bytes, int offset) { + return unsigned(bytes[offset]) + | unsigned(bytes[offset + 1]) << 8 + | unsigned(bytes[offset + 2]) << 16 + | unsigned(bytes[offset + 3]) << 24; + } + + private int unsigned(byte value) { + return value & 0xff; + } + + private void closeQuietly(InputStream input) { + if (input == null) { + return; + } + try { + input.close(); + } catch (IOException ignored) { + // 响应已失败，关闭异常不覆盖原始业务错误。 + } + } + + private record Dimensions(int width, int height) { + } + + private enum ImageFormat { + PNG("image/png"), + JPEG("image/jpeg"), + WEBP("image/webp"), + GIF("image/gif"), + BMP("image/bmp"); + + private final String mimeType; + + ImageFormat(String mimeType) { + this.mimeType = mimeType; + } + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowRunningParameterResolver.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowRunningParameterResolver.java index 7e580edb..acfe9ba5 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowRunningParameterResolver.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/easyagentsflow/service/WorkflowRunningParameterResolver.java @@ -17,6 +17,7 @@ import java.util.Collection; import java.util.LinkedHashMap; import java.util.LinkedHashSet; import java.util.List; +import java.util.Locale; import java.util.Map; import java.util.Set; @@ -37,6 +38,7 @@ public class WorkflowRunningParameterResolver { private static final int FILE_MAX_COUNT = 10; private static final long FILE_MAX_SINGLE_SIZE = 20L * 1024 * 1024; private static final long FILE_MAX_TOTAL_SIZE = 50L * 1024 * 1024; + private static final long IMAGE_MAX_SINGLE_SIZE = 10L * 1024 * 1024; @Resource private ChainParser chainParser; @@ -103,14 +105,15 @@ public class WorkflowRunningParameterResolver { return normalized; } for (Parameter parameter : startParameters) { - if (!isFileParameter(parameter)) { - continue; - } String name = trimToNull(parameter.getName()); if (!StringUtils.hasText(name) || !normalized.containsKey(name)) { continue; } - normalized.put(name, normalizeFileVariableValue(normalized.get(name), name)); + if (isFileParameter(parameter)) { + normalized.put(name, normalizeFileVariableValue(normalized.get(name), name)); + } else if (isImageParameter(parameter)) { + normalized.put(name, normalizeImageVariableValue(normalized.get(name), name)); + } } return normalized; } @@ -146,13 +149,21 @@ public class WorkflowRunningParameterResolver { List> schema = new ArrayList<>(); Set seenKeys = new LinkedHashSet<>(); boolean hasExplicitSchema = rawSchema != null; + Map parameterByName = new LinkedHashMap<>(); + for (Parameter parameter : parameters) { + String parameterName = trimToNull(parameter == null ? null : parameter.getName()); + if (StringUtils.hasText(parameterName)) { + parameterByName.put(parameterName, parameter); + } + } boolean hasSystemParameter = parameters.stream().anyMatch(parameter -> SYSTEM_START_PARAM_NAME.equals(trimToNull(parameter == null ? null : parameter.getName())) ); if (rawSchema != null && !rawSchema.isEmpty()) { for (int i = 0; i < rawSchema.size(); i++) { JSONObject field = rawSchema.getJSONObject(i); - Map normalized = normalizeStartFormField(field, null); + String fieldKey = trimToNull(field == null ? null : field.getString("key")); + Map normalized = normalizeStartFormField(field, parameterByName.get(fieldKey)); if (normalized == null) { continue; } @@ -202,6 +213,10 @@ public class WorkflowRunningParameterResolver { boolean systemReserved = SYSTEM_START_PARAM_NAME.equals(key) || (field != null && Boolean.TRUE.equals(field.getBoolean("systemReserved"))); String type = resolveStartFormFieldType(field == null ? null : field.getString("type"), parameter, systemReserved); + String contentType = resolveStartFormContentType(field, parameter, type, systemReserved); + if ("file".equals(contentType)) { + type = "file"; + } List options = resolveFieldOptions(field, parameter, type); Map normalized = new LinkedHashMap<>(); @@ -212,6 +227,7 @@ public class WorkflowRunningParameterResolver { SYSTEM_START_PARAM_NAME.equals(key) ? "用户问题" : key )); normalized.put("type", type); + normalized.put("contentType", contentType); normalized.put("required", systemReserved || (field != null && Boolean.TRUE.equals(field.getBoolean("required"))) || (parameter != null && parameter.isRequired())); normalized.put("placeholder", trimToDefault( @@ -230,6 +246,40 @@ public class WorkflowRunningParameterResolver { return normalized; } + /** + * 解析开始表单字段的数据内容类型，并兼容旧版仅通过字段类型表达文件输入的配置。 + * + * @param field 字段 Schema + * @param parameter 旧版参数定义 + * @param fieldType 表单字段类型 + * @param systemReserved 是否系统入口字段 + * @return 归一化后的数据内容类型 + */ + private String resolveStartFormContentType(JSONObject field, + Parameter parameter, + String fieldType, + boolean systemReserved) { + if (systemReserved) { + return "text"; + } + if ("file".equals(fieldType)) { + return "file"; + } + String requested = trimToNull(field == null ? null : field.getString("contentType")); + String parameterContentType = parameter == null ? null : trimToNull(parameter.getContentType()); + if ("image".equals(parameterContentType) + && (!StringUtils.hasText(requested) || "text".equals(requested))) { + return "image"; + } + if (!StringUtils.hasText(requested)) { + requested = parameterContentType; + } + return switch (requested == null ? "" : requested) { + case "image", "video", "audio", "file", "other" -> requested; + default -> "text"; + }; + } + private Object resolveDefaultValue(JSONObject field, Parameter parameter, String type) { Object rawDefaultValue = field == null ? null : field.get("defaultValue"); if (rawDefaultValue != null) { @@ -329,6 +379,117 @@ public class WorkflowRunningParameterResolver { || "file".equalsIgnoreCase(trimToNull(String.valueOf(parameter.getDataType()))); } + /** + * 判断参数是否为图片输入参数。 + * + * @param parameter 参数定义 + * @return 是否图片参数 + */ + private boolean isImageParameter(Parameter parameter) { + return parameter != null && "image".equals(trimToNull(parameter.getContentType())); + } + + /** + * 将图片运行值归一化为单图描述对象。 + * + * @param value 原始图片值 + * @param parameterName 参数名 + * @return 归一化后的图片描述；空值返回 {@code null} + */ + private Object normalizeImageVariableValue(Object value, String parameterName) { + if (value == null + || (value instanceof String stringValue && !StringUtils.hasText(stringValue))) { + return null; + } + if (value instanceof Collection) { + throw new BusinessException("图片参数 " + parameterName + " 仅支持单张图片"); + } + if (value instanceof String stringValue) { + String normalized = stringValue.trim(); + if (isHttpUrl(normalized)) { + Map image = new LinkedHashMap<>(); + image.put("sourceType", "url"); + image.put("url", normalized); + return image; + } + throw new BusinessException("图片参数 " + parameterName + " 仅支持 HTTP/HTTPS 图片 URL"); + } + if (!(value instanceof Map imageMap)) { + throw new BusinessException("图片参数 " + parameterName + " 的输入格式不正确"); + } + + String sourceType = trimObjectToNull(imageMap.get("sourceType")); + String filePath = trimObjectToNull(imageMap.get("filePath")); + String url = trimObjectToNull(imageMap.get("url")); + // 兼容旧版没有 sourceType 的文件对象和 URL 对象。 + if (!StringUtils.hasText(sourceType)) { + sourceType = StringUtils.hasText(filePath) ? "upload" : "url"; + } + if ("url".equals(sourceType)) { + if (!isHttpUrl(url)) { + throw new BusinessException("图片参数 " + parameterName + " 缺少有效的 HTTP/HTTPS URL"); + } + Map normalized = new LinkedHashMap<>(); + normalized.put("sourceType", "url"); + normalized.put("url", url); + return normalized; + } + if (!"upload".equals(sourceType) && !"resource".equals(sourceType)) { + throw new BusinessException("图片参数 " + parameterName + " 的 sourceType 不受支持"); + } + + String fileName = trimObjectToNull(imageMap.get("fileName")); + if (!StringUtils.hasText(fileName)) { + throw new BusinessException("图片参数 " + parameterName + " 缺少 fileName"); + } + if (!StringUtils.hasText(filePath)) { + throw new BusinessException("图片参数 " + parameterName + " 缺少 filePath"); + } + Long size = parseLong(imageMap.get("size")); + if (size != null && size > IMAGE_MAX_SINGLE_SIZE) { + throw new BusinessException("图片参数 " + parameterName + " 中图片不能超过 10 MiB"); + } + + Map normalized = new LinkedHashMap<>(); + normalized.put("sourceType", sourceType); + normalized.put("fileName", fileName); + normalized.put("filePath", filePath); + copyOptionalImageField(imageMap, normalized, "contentType"); + if (size != null) { + normalized.put("size", size); + } + copyOptionalImageField(imageMap, normalized, "url"); + return normalized; + } + + /** + * 复制图片描述中的可选非空字段。 + * + * @param source 原始图片描述 + * @param target 归一化图片描述 + * @param key 字段名 + */ + private void copyOptionalImageField(Map source, Map target, String key) { + String value = trimObjectToNull(source.get(key)); + if (StringUtils.hasText(value)) { + target.put(key, value); + } + } + + /** + * 判断字符串是否为 HTTP 或 HTTPS URL。 + * + * @param value 待判断值 + * @return 是否为受支持 URL + */ + private boolean isHttpUrl(String value) { + if (!StringUtils.hasText(value)) { + return false; + } + String lowerValue = value.toLowerCase(Locale.ROOT); + return lowerValue.startsWith("http://") || lowerValue.startsWith("https://"); + } + /** * 将单文件或多文件运行值归一化为文件对象数组。 * diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/WorkflowShare.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/WorkflowShare.java index 05860c21..f604af9b 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/WorkflowShare.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/entity/WorkflowShare.java @@ -23,6 +23,9 @@ public class WorkflowShare implements Serializable { @Column(comment = "工作流ID") private BigInteger workflowId; + @Column(comment = "分享用途") + private String sharePurpose; + @Column(comment = "分享密钥哈希") private String shareKeyHash; @@ -86,6 +89,24 @@ public class WorkflowShare implements Serializable { this.workflowId = workflowId; } + /** + * 获取分享用途。 + * + * @return 分享用途 + */ + public String getSharePurpose() { + return sharePurpose; + } + + /** + * 设置分享用途。 + * + * @param sharePurpose 分享用途 + */ + public void setSharePurpose(String sharePurpose) { + this.sharePurpose = sharePurpose; + } + /** * 获取分享密钥哈希。 * diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/enums/WorkflowSharePurpose.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/enums/WorkflowSharePurpose.java new file mode 100644 index 00000000..90115535 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/enums/WorkflowSharePurpose.java @@ -0,0 +1,17 @@ +package tech.easyflow.ai.enums; + +/** + * 工作流分享用途。 + */ +public enum WorkflowSharePurpose { + + /** + * 历史协作编辑分享。 + */ + COLLABORATION, + + /** + * 已发布工作流对话运行分享。 + */ + CHAT +} diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/WorkflowShareService.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/WorkflowShareService.java index ccef161f..aa9f484c 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/WorkflowShareService.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/WorkflowShareService.java @@ -29,6 +29,24 @@ public interface WorkflowShareService extends IService { String baseUrl ); + /** + * 创建或刷新工作流的唯一对话分享链接。 + * + * @param workflowId 工作流 ID + * @param tenantId 租户 ID + * @param deptId 部门 ID + * @param operatorId 操作人账号 ID + * @param baseUrl 工作流对话分享基础 URL + * @return 创建结果 + */ + WorkflowShareCreateResult createChatShare( + BigInteger workflowId, + BigInteger tenantId, + BigInteger deptId, + BigInteger operatorId, + String baseUrl + ); + /** * 校验分享密钥是否可访问指定工作流。 * @@ -51,4 +69,27 @@ public interface WorkflowShareService extends IService { * @return 有效分享记录 */ WorkflowShare resolveUrlShare(String shareKey, BigInteger tenantId); + + /** + * 校验对话分享密钥是否可访问指定工作流。 + * + * @param shareKey 原始分享密钥 + * @param workflowId 工作流 ID + * @param tenantId 当前登录租户 ID + * @return 有效对话分享记录 + */ + WorkflowShare assertChatShareAccess( + String shareKey, + BigInteger workflowId, + BigInteger tenantId + ); + + /** + * 校验对话分享密钥并解析目标工作流。 + * + * @param shareKey 原始分享密钥 + * @param tenantId 当前登录租户 ID + * @return 有效对话分享记录 + */ + WorkflowShare resolveChatShare(String shareKey, BigInteger tenantId); } diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/WorkflowShareServiceImpl.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/WorkflowShareServiceImpl.java index 04ace823..3f185fc3 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/WorkflowShareServiceImpl.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/service/impl/WorkflowShareServiceImpl.java @@ -8,6 +8,8 @@ import org.springframework.transaction.support.TransactionTemplate; import tech.easyflow.ai.entity.Workflow; import tech.easyflow.ai.entity.WorkflowShare; import tech.easyflow.ai.enums.KnowledgeShareStatus; +import tech.easyflow.ai.enums.PublishStatus; +import tech.easyflow.ai.enums.WorkflowSharePurpose; import tech.easyflow.ai.mapper.WorkflowShareMapper; import tech.easyflow.ai.service.WorkflowService; import tech.easyflow.ai.service.WorkflowShareService; @@ -53,9 +55,63 @@ public class WorkflowShareServiceImpl extends ServiceImpl { @@ -66,7 +122,9 @@ public class WorkflowShareServiceImpl extends ServiceImpl activeShares = list(QueryWrapper.create() .eq(WorkflowShare::getWorkflowId, workflowId) + .eq(WorkflowShare::getSharePurpose, purpose.name()) .eq(WorkflowShare::getStatus, KnowledgeShareStatus.ENABLED.name())); for (WorkflowShare activeShare : activeShares) { WorkflowShare update = new WorkflowShare(); @@ -205,4 +327,17 @@ public class WorkflowShareServiceImpl extends ServiceImpl ALLOWED_REQUESTS = Set.of( permissionKey("GET", "/api/v1/workflow/detail", ResourceAction.READ), permissionKey("GET", "/api/v1/workflow/getRunningParameters", ResourceAction.READ), @@ -66,6 +72,19 @@ public final class WorkflowSharePolicy { return new Date(createdAt.getTime() + DEFAULT_EXPIRE_DURATION.toMillis()); } + /** + * 计算对话分享默认过期时间。 + * + * @param createdAt 创建时间 + * @return 创建后 7 天的时间 + */ + public static Date defaultChatExpiresAt(Date createdAt) { + if (createdAt == null) { + throw new IllegalArgumentException("创建时间不能为空"); + } + return new Date(createdAt.getTime() + DEFAULT_CHAT_EXPIRE_DURATION.toMillis()); + } + /** * 判断 HTTP 请求是否位于工作流协作授权白名单。 * diff --git a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/utils/WorkFlowUtil.java b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/utils/WorkFlowUtil.java index e7b5b396..47944d45 100644 --- a/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/utils/WorkFlowUtil.java +++ b/easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/utils/WorkFlowUtil.java @@ -13,6 +13,10 @@ public class WorkFlowUtil { public final static String USER_KEY = "user"; public final static String API_KEY = "API_KEY"; + /** 管理端工作流对话执行来源。 */ + public final static String WORKFLOW_CHAT = "WORKFLOW_CHAT"; + /** 工作流对话分享执行来源。 */ + public final static String WORKFLOW_CHAT_SHARE = "WORKFLOW_CHAT_SHARE"; public final static String WORKFLOW_KEY = "workflow"; public final static String CREATED_KEY_MEMORY_KEY = "workflowCreatedKey"; diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/llm/WorkflowImageSourceResolverTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/llm/WorkflowImageSourceResolverTest.java new file mode 100644 index 00000000..620e3755 --- /dev/null +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/llm/WorkflowImageSourceResolverTest.java @@ -0,0 +1,156 @@ +package tech.easyflow.ai.easyagentsflow.llm; + +import org.junit.Assert; +import org.junit.Test; +import org.springframework.web.multipart.MultipartFile; +import tech.easyflow.common.filestorage.FileStorageService; +import tech.easyflow.common.web.exceptions.BusinessException; + +import javax.imageio.ImageIO; +import java.awt.image.BufferedImage; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.InputStream; +import java.net.URI; +import java.net.http.HttpClient; +import java.util.Map; + +/** + * 工作流图片源解析器测试。 + */ +public class WorkflowImageSourceResolverTest { + + /** + * 验证存储中的 PNG 图片转换为完整 Data URI。 + * + * @throws Exception 图片构造失败 + */ + @Test + public void shouldResolveStoredPngToDataUri() throws Exception { + byte[] png = imageBytes("png"); + WorkflowImageSourceResolver resolver = resolver(Map.of("/images/a.png", png)); + + String dataUri = resolver.resolve(Map.of( + "sourceType", "upload", + "fileName", "a.png", + "filePath", "/images/a.png")); + + Assert.assertTrue(dataUri.startsWith("data:image/png;base64,")); + Assert.assertArrayEquals( + png, + java.util.Base64.getDecoder().decode(dataUri.substring(dataUri.indexOf(',') + 1))); + } + + /** + * 验证 BMP 图片会在模型调用前规范化为 PNG。 + * + * @throws Exception 图片构造失败 + */ + @Test + public void shouldNormalizeBmpToPng() throws Exception { + WorkflowImageSourceResolver resolver = + resolver(Map.of("/images/a.bmp", imageBytes("bmp"))); + + String dataUri = resolver.resolve(Map.of( + "sourceType", "resource", + "fileName", "a.bmp", + "filePath", "/images/a.bmp")); + + Assert.assertTrue(dataUri.startsWith("data:image/png;base64,")); + } + + /** + * 验证旧版 Data URI 会经过真实图片校验后继续使用。 + * + * @throws Exception 图片构造失败 + */ + @Test + public void shouldValidateLegacyDataUri() throws Exception { + byte[] png = imageBytes("png"); + String input = "data:image/png;base64," + + java.util.Base64.getEncoder().encodeToString(png); + + String dataUri = resolver(Map.of()).resolve(input); + + Assert.assertTrue(dataUri.startsWith("data:image/png;base64,")); + } + + /** + * 验证本机、内网和云元数据地址会被拦截。 + */ + @Test + public void shouldRejectUnsafeRemoteAddresses() { + WorkflowImageSourceResolver resolver = resolver(Map.of()); + + assertUnsafe(resolver, "http://127.0.0.1/image.png"); + assertUnsafe(resolver, "http://192.168.1.2/image.png"); + assertUnsafe(resolver, "http://168.63.129.16/metadata/instance"); + assertUnsafe(resolver, "http://169.254.169.254/latest/meta-data"); + assertUnsafe(resolver, "http://metadata.google.internal/image.png"); + } + + private static void assertUnsafe(WorkflowImageSourceResolver resolver, String value) { + try { + resolver.validateRemoteUri(URI.create(value)); + Assert.fail("expected BusinessException for " + value); + } catch (BusinessException expected) { + Assert.assertTrue(expected.getMessage().contains("不能访问")); + } + } + + private static WorkflowImageSourceResolver resolver(Map files) { + return new WorkflowImageSourceResolver( + new InMemoryStorage(files), + HttpClient.newHttpClient()); + } + + private static byte[] imageBytes(String format) throws Exception { + BufferedImage image = new BufferedImage(2, 2, BufferedImage.TYPE_INT_RGB); + ByteArrayOutputStream output = new ByteArrayOutputStream(); + Assert.assertTrue(ImageIO.write(image, format, output)); + return output.toByteArray(); + } + + /** + * 测试用内存文件存储。 + */ + private static final class InMemoryStorage implements FileStorageService { + + private final Map files; + + private InMemoryStorage(Map files) { + this.files = files; + } + + @Override + public String save(MultipartFile file) { + throw new UnsupportedOperationException(); + } + + @Override + public void delete(String path) { + throw new UnsupportedOperationException(); + } + + @Override + public String save(File file, String prePath) { + throw new UnsupportedOperationException(); + } + + @Override + public InputStream readStream(String path) { + byte[] bytes = files.get(path); + if (bytes == null) { + throw new IllegalArgumentException("missing file: " + path); + } + return new ByteArrayInputStream(bytes); + } + + @Override + public long getFileSize(String path) { + byte[] bytes = files.get(path); + return bytes == null ? 0 : bytes.length; + } + } +} diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowRunningParameterResolverTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowRunningParameterResolverTest.java index 77ca2826..febd7435 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowRunningParameterResolverTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/easyagentsflow/service/WorkflowRunningParameterResolverTest.java @@ -43,16 +43,34 @@ public class WorkflowRunningParameterResolverTest { fileField.put("key", "attachments"); fileField.put("label", "附件"); fileField.put("type", "file"); + fileField.put("contentType", "file"); fileField.put("required", false); schema.add(fileField); + JSONObject imageField = new JSONObject(); + imageField.put("key", "preview_image"); + imageField.put("label", "预览图"); + imageField.put("type", "text"); + imageField.put("contentType", "text"); + imageField.put("required", false); + schema.add(imageField); + JSONObject meta = new JSONObject(); meta.put("title", "问答入口"); meta.put("description", "请先填写信息"); meta.put("submitText", "立即开始"); startData.put("startFormMeta", meta); startData.put("startFormSchema", schema); - startData.put("parameters", startParameters()); + JSONArray parameters = startParameters(); + JSONObject imageParameter = new JSONObject(); + imageParameter.put("name", "preview_image"); + imageParameter.put("dataType", "Object"); + imageParameter.put("refType", "input"); + imageParameter.put("contentType", "image"); + imageParameter.put("formType", "input"); + imageParameter.put("formLabel", "预览图"); + parameters.add(imageParameter); + startData.put("parameters", parameters); Workflow workflow = workflow( workflowJson( @@ -68,11 +86,16 @@ public class WorkflowRunningParameterResolverTest { Assert.assertNotNull(result); Assert.assertEquals("问答入口", ((Map) result.get("startFormMeta")).get("title")); List> fields = (List>) result.get("startFormSchema"); - Assert.assertEquals(2, fields.size()); + Assert.assertEquals(3, fields.size()); Assert.assertEquals("user_input", fields.get(0).get("key")); Assert.assertEquals("text", fields.get(0).get("type")); + Assert.assertEquals("text", fields.get(0).get("contentType")); Assert.assertEquals("attachments", fields.get(1).get("key")); Assert.assertEquals("file", fields.get(1).get("type")); + Assert.assertEquals("file", fields.get(1).get("contentType")); + Assert.assertEquals("preview_image", fields.get(2).get("key")); + Assert.assertEquals("text", fields.get(2).get("type")); + Assert.assertEquals("image", fields.get(2).get("contentType")); } /** @@ -213,6 +236,77 @@ public class WorkflowRunningParameterResolverTest { } } + /** + * 旧版图片 URL 应归一化为 URL 图片描述。 + * + * @throws Exception 反射注入失败 + */ + @Test + public void testNormalizeRuntimeVariablesShouldNormalizeLegacyImageUrl() throws Exception { + WorkflowRunningParameterResolver resolver = newResolver(); + Map variables = new LinkedHashMap<>(); + variables.put("image_input", "https://example.com/image.png"); + + Map normalized = resolver.normalizeRuntimeVariables( + workflowContentWithImageStartParameter(), + variables + ); + + Assert.assertTrue(normalized.get("image_input") instanceof Map); + Map image = (Map) normalized.get("image_input"); + Assert.assertEquals("url", image.get("sourceType")); + Assert.assertEquals("https://example.com/image.png", image.get("url")); + } + + /** + * 运行入口不应接收 Data URI，避免 Base64 写入工作流状态和审计参数。 + * + * @throws Exception 反射注入失败 + */ + @Test + public void testNormalizeRuntimeVariablesShouldRejectImageDataUri() throws Exception { + WorkflowRunningParameterResolver resolver = newResolver(); + Map variables = new LinkedHashMap<>(); + variables.put("image_input", "data:image/png;base64,AQID"); + + try { + resolver.normalizeRuntimeVariables(workflowContentWithImageStartParameter(), variables); + Assert.fail("expected BusinessException"); + } catch (BusinessException exception) { + Assert.assertEquals( + "图片参数 image_input 仅支持 HTTP/HTTPS 图片 URL", + exception.getMessage()); + } + } + + /** + * 图片参数应允许 10 MiB 边界并拒绝更大的声明值。 + * + * @throws Exception 反射注入失败 + */ + @Test + public void testNormalizeRuntimeVariablesShouldEnforceImageLimit() throws Exception { + WorkflowRunningParameterResolver resolver = newResolver(); + Map variables = new LinkedHashMap<>(); + variables.put("image_input", imageValue(10L * 1024L * 1024L)); + + Map normalized = resolver.normalizeRuntimeVariables( + workflowContentWithImageStartParameter(), + variables + ); + Assert.assertEquals("upload", ((Map) normalized.get("image_input")).get("sourceType")); + + variables.put("image_input", imageValue(10L * 1024L * 1024L + 1L)); + try { + resolver.normalizeRuntimeVariables(workflowContentWithImageStartParameter(), variables); + Assert.fail("expected BusinessException"); + } catch (BusinessException exception) { + Assert.assertEquals( + "图片参数 image_input 中图片不能超过 10 MiB", + exception.getMessage()); + } + } + private static WorkflowRunningParameterResolver newResolver() throws Exception { WorkflowRunningParameterResolver resolver = new WorkflowRunningParameterResolver(); ChainParser parser = ChainParser.builder() @@ -247,6 +341,26 @@ public class WorkflowRunningParameterResolverTest { ); } + private static String workflowContentWithImageStartParameter() { + JSONObject startData = data("开始"); + JSONArray parameters = new JSONArray(); + JSONObject imageField = new JSONObject(); + imageField.put("name", "image_input"); + imageField.put("dataType", "Object"); + imageField.put("refType", "input"); + imageField.put("contentType", "image"); + imageField.put("formType", "input"); + parameters.add(imageField); + startData.put("parameters", parameters); + return workflowJson( + array( + node("s1", "startNode", null, startData), + node("e1", "endNode", null, data("结束")) + ), + array(edge("e1", "s1", "e1")) + ); + } + private static JSONArray startParameters() { JSONArray parameters = new JSONArray(); @@ -298,6 +412,16 @@ public class WorkflowRunningParameterResolverTest { return value; } + private static Map imageValue(long size) { + Map value = new LinkedHashMap<>(); + value.put("sourceType", "upload"); + value.put("fileName", "image.png"); + value.put("filePath", "/files/image.png"); + value.put("size", size); + value.put("contentType", "image/png"); + return value; + } + private static void setField(Object target, String fieldName, Object value) throws Exception { Field field = WorkflowRunningParameterResolver.class.getDeclaredField(fieldName); field.setAccessible(true); diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowShareMigrationContractTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowShareMigrationContractTest.java index 0207865b..60c0b145 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowShareMigrationContractTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowShareMigrationContractTest.java @@ -20,7 +20,9 @@ public class WorkflowShareMigrationContractTest { */ @Test public void migrationShouldCreateWorkflowShareContracts() throws Exception { - String sql = migrationSql(); + String sql = migrationSql( + "V34__mysql_workflow_share_and_approval_reason.sql" + ); assertTrue(sql.contains("ADD COLUMN `application_reason` VARCHAR(500)")); assertTrue(sql.contains("ADD COLUMN `revision` INT NOT NULL DEFAULT 0")); @@ -34,22 +36,41 @@ public class WorkflowShareMigrationContractTest { } /** - * 读取工作区中的 V34 MySQL 迁移。 + * 验证 V39 为协作分享和对话分享建立用途隔离。 * + * @throws Exception 迁移文件不可读时抛出 + */ + @Test + public void migrationShouldSeparateChatSharePurpose() throws Exception { + String sql = migrationSql("V39__mysql_workflow_chat_share.sql"); + + assertTrue(sql.contains( + "ADD COLUMN `share_purpose` VARCHAR(32) NOT NULL DEFAULT 'COLLABORATION'" + )); + assertTrue(sql.contains("`idx_workflow_share_purpose_status`")); + assertTrue(sql.contains( + "(`workflow_id`, `share_purpose`, `status`)" + )); + } + + /** + * 读取工作区中的指定 MySQL 迁移。 + * + * @param fileName 迁移文件名 * @return 迁移 SQL * @throws Exception 迁移文件不存在或不可读时抛出 */ - private String migrationSql() throws Exception { + private String migrationSql(String fileName) throws Exception { Path root = Path.of(System.getProperty("maven.multiModuleProjectDirectory", Path.of(System.getProperty("user.dir")).toAbsolutePath().toString())); while (root != null) { Path migration = root.resolve("easyflow-starter/easyflow-starter-all/src/main/resources/" - + "db/migration/mysql/V34__mysql_workflow_share_and_approval_reason.sql"); + + "db/migration/mysql/" + fileName); if (Files.isRegularFile(migration)) { return Files.readString(migration, StandardCharsets.UTF_8); } root = root.getParent(); } - throw new IllegalStateException("找不到 V34 工作流分享与审批说明迁移"); + throw new IllegalStateException("找不到工作流分享迁移: " + fileName); } } diff --git a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowSharePolicyTest.java b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowSharePolicyTest.java index 2e111a88..a26388b9 100644 --- a/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowSharePolicyTest.java +++ b/easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/share/WorkflowSharePolicyTest.java @@ -36,6 +36,21 @@ public class WorkflowSharePolicyTest { Assert.assertEquals(30 * 60 * 1_000L, expiresAt.getTime() - createdAt.getTime()); } + /** + * 验证对话分享默认在创建七天后过期。 + */ + @Test + public void shouldExpireChatShareSevenDaysAfterCreation() { + Date createdAt = new Date(1_000L); + + Date expiresAt = WorkflowSharePolicy.defaultChatExpiresAt(createdAt); + + Assert.assertEquals( + 7L * 24L * 60L * 60L * 1_000L, + expiresAt.getTime() - createdAt.getTime() + ); + } + /** * 验证分享授权仅覆盖编辑、运行和发布所需接口。 */ diff --git a/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V39__mysql_workflow_chat_share.sql b/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V39__mysql_workflow_chat_share.sql new file mode 100644 index 00000000..0442835d --- /dev/null +++ b/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V39__mysql_workflow_chat_share.sql @@ -0,0 +1,10 @@ +SET NAMES utf8mb4; + +ALTER TABLE `tb_workflow_share` + ADD COLUMN `share_purpose` VARCHAR(32) NOT NULL DEFAULT 'COLLABORATION' + COMMENT '分享用途：COLLABORATION/CHAT' AFTER `workflow_id`; + +DROP INDEX `idx_workflow_share_status` ON `tb_workflow_share`; + +CREATE INDEX `idx_workflow_share_purpose_status` + ON `tb_workflow_share` (`workflow_id`, `share_purpose`, `status`); diff --git a/easyflow-ui-admin/app/src/utils/workflow-share-context.ts b/easyflow-ui-admin/app/src/utils/workflow-share-context.ts index 29e649e3..0803e3b8 100644 --- a/easyflow-ui-admin/app/src/utils/workflow-share-context.ts +++ b/easyflow-ui-admin/app/src/utils/workflow-share-context.ts @@ -1,9 +1,9 @@ import { readScopedRouteQueryParam } from './share-route-context'; /** - * 工作流协作分享请求头。 + * 工作流对话分享请求头。 */ -export const WORKFLOW_SHARE_HEADER = 'X-Workflow-Share-Key'; +export const WORKFLOW_SHARE_HEADER = 'X-Workflow-Chat-Share-Key'; interface WorkflowShareResolutionOptions { currentWorkflowId?: null | T; @@ -18,19 +18,14 @@ interface WorkflowShareHeaderOptions { requestUrl?: string; } -const WORKFLOW_SHARE_ROUTES = ['/share/workflow', '/ai/workflow/design']; +const WORKFLOW_SHARE_ROUTES = ['/share/workflow']; const WORKFLOW_SHARE_REQUESTS = [ - ['GET', '/api/v1/workflow/detail'], - ['GET', '/api/v1/workflow/getRunningParameters'], - ['GET', '/api/v1/workflow/publishApprovalRequirement'], + ['GET', '/api/v1/workflowChat/descriptor'], + ['GET', '/api/v1/workflowChat/execution'], ['GET', '/api/v1/workflowShare/resolve'], - ['POST', '/api/v1/workflow/check'], - ['POST', '/api/v1/workflow/getChainStatus'], - ['POST', '/api/v1/workflow/resume'], - ['POST', '/api/v1/workflow/runAsync'], - ['POST', '/api/v1/workflow/singleRun'], - ['POST', '/api/v1/workflow/submitPublishApproval'], - ['POST', '/api/v1/workflow/update'], + ['POST', '/api/v1/workflowChat/cancel'], + ['POST', '/api/v1/workflowChat/resume'], + ['POST', '/api/v1/workflowChat/run'], ] as const; /** diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/RunPage.vue b/easyflow-ui-admin/app/src/views/ai/workflow/RunPage.vue index badf4b07..7e741b00 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/RunPage.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/RunPage.vue @@ -1,134 +1,7 @@ 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 0fe4fbaf..27e29749 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowList.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowList.vue @@ -202,7 +202,9 @@ const actions: ActionButton[] = [ text: $t('button.share'), permission: '/api/v1/workflow/save', placement: 'menu', - disabled: (row: any) => sharingWorkflowId.value === row.id, + disabled: (row: any) => + row.publishStatus !== 'PUBLISHED' || + sharingWorkflowId.value === row.id, loading: (row: any) => sharingWorkflowId.value === row.id, onClick: (row: any) => { shareWorkflow(row); diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowShareView.vue b/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowShareView.vue index eea536ee..0fa52bc5 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowShareView.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/WorkflowShareView.vue @@ -1,7 +1,7 @@ diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowChatPage.vue b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowChatPage.vue new file mode 100644 index 00000000..e4024a7a --- /dev/null +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowChatPage.vue @@ -0,0 +1,1711 @@ + + + + + diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowForm.vue b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowForm.vue index 1445f83a..bee8db98 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowForm.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowForm.vue @@ -10,6 +10,7 @@ import { api } from '#/api/request'; import { $t } from '#/locales'; import WorkflowFormItem from './WorkflowFormItem.vue'; +import { resolveWorkflowFormParameters } from './workflowFormParameters'; export type WorkflowFormProps = { onAsyncExecute?: (values: any) => void; @@ -49,30 +50,9 @@ const startFormMeta = computed(() => { submitText: String(meta.submitText || '').trim() || $t('button.run'), }; }); -const parameters = computed(() => { - const schema = Array.isArray(props.workflowParams?.startFormSchema) - ? props.workflowParams.startFormSchema - : []; - if (schema.length === 0) { - return props.workflowParams.parameters || []; - } - return schema.map((field: any) => { - const type = String(field.type || '').trim() || 'text'; - return { - name: field.key, - formLabel: field.label || field.key, - formDescription: field.description || '', - formPlaceholder: field.placeholder || '', - required: Boolean(field.required), - defaultValue: field.defaultValue, - enums: Array.isArray(field.options) ? field.options : [], - contentType: type === 'file' ? 'file' : 'text', - formType: type === 'text' ? 'input' : type === 'file' ? 'input' : type, - dataType: - type === 'checkbox' ? 'Array' : type === 'file' ? 'File' : 'String', - }; - }); -}); +const parameters = computed(() => + resolveWorkflowFormParameters(props.workflowParams), +); watch( parameters, (items) => { diff --git a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowFormItem.vue b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowFormItem.vue index 3fe067cb..a1754d1c 100644 --- a/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowFormItem.vue +++ b/easyflow-ui-admin/app/src/views/ai/workflow/components/WorkflowFormItem.vue @@ -11,6 +11,9 @@ import { import { $t } from '#/locales'; import ChooseResource from '#/views/ai/resource/ChooseResource.vue'; import WorkflowFileInput from '#/views/ai/workflow/components/WorkflowFileInput.vue'; +import WorkflowImageInput from '#/views/ai/workflow/components/WorkflowImageInput.vue'; + +import { hasWorkflowImageValue } from './workflowImageValue'; const props = defineProps({ parameters: { @@ -37,7 +40,7 @@ function getContentType(item: any) { return 'text'; } function isResource(contentType: any) { - return ['audio', 'image', 'video'].includes(contentType); + return ['audio', 'video'].includes(contentType); } function isFileContentType(contentType: any) { return contentType === 'file'; @@ -61,6 +64,14 @@ function buildRules(item: any) { { required: true, validator: (_rule: any, value: any, callback: any) => { + if (getContentType(item) === 'image') { + callback( + hasWorkflowImageValue(value) + ? undefined + : new Error($t('message.required')), + ); + return; + } if (Array.isArray(value)) { callback(value.length > 0 ? undefined : new Error($t('message.required'))); return; @@ -144,6 +155,12 @@ function choose(data: any, propName: string) { @update:model-value="(val) => updateParam(item.name, val)" /> +