/*
 * Copyright 2025 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.alibaba.cloud.ai.studio.admin.builder.controller;

import com.alibaba.cloud.ai.studio.runtime.domain.workflow.CommonParam;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.Edge;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.InvokeSourceEnum;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.Node;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.NodeResult;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.NodeStatusEnum;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.NodeTypeEnum;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.ParamSourceEnum;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.ApiTaskMsg;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.ApiTaskRunRequest;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.InitRequest;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.ProcessGetRequest;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.ProcessGetResponse;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.TaskPartGraphRequest;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.TaskPartGraphResponse;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.TaskResumeRequest;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.TaskResumeResponse;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.TaskRunParam;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.TaskRunRequest;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.TaskRunResponse;
import com.alibaba.cloud.ai.studio.runtime.domain.workflow.debug.TaskStopRequest;
import com.alibaba.cloud.ai.studio.runtime.exception.BizException;
import com.alibaba.cloud.ai.studio.runtime.enums.ErrorCode;
import com.alibaba.cloud.ai.studio.runtime.domain.RequestContext;
import com.alibaba.cloud.ai.studio.runtime.domain.Result;
import com.alibaba.cloud.ai.studio.runtime.domain.app.ApplicationVersion;
import com.alibaba.cloud.ai.studio.runtime.utils.JsonUtils;
import com.alibaba.cloud.ai.studio.core.base.service.AppService;
import com.alibaba.cloud.ai.studio.core.context.RequestContextHolder;
import com.alibaba.cloud.ai.studio.core.base.manager.RedisManager;
import com.alibaba.cloud.ai.studio.core.workflow.WorkflowConfig;
import com.alibaba.cloud.ai.studio.core.workflow.WorkflowContext;
import com.alibaba.cloud.ai.studio.core.utils.common.BeanCopierUtils;
import com.alibaba.cloud.ai.studio.core.utils.common.IdGenerator;
import com.alibaba.cloud.ai.studio.core.utils.LogUtils;
import com.alibaba.cloud.ai.studio.core.utils.concurrent.ThreadPoolUtils;
import com.alibaba.cloud.ai.studio.core.utils.common.VariableUtils;
import com.alibaba.cloud.ai.studio.core.workflow.WorkflowInnerService;
import com.alibaba.cloud.ai.studio.core.workflow.processor.impl.IteratorExecuteProcessor;
import com.alibaba.cloud.ai.studio.core.workflow.runtime.WorkflowExecuteManager;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import static com.alibaba.cloud.ai.studio.core.base.constants.CacheConstants.WORKFLOW_TASK_CONTEXT_PREFIX;
import static com.alibaba.cloud.ai.studio.core.workflow.constants.WorkflowConstants.SYS_QUERY_KEY;
import static com.alibaba.cloud.ai.studio.core.utils.LogUtils.FAIL;

/**
 * REST controller for workflow operations in the SAA Studio platform. This controller
 * provides comprehensive workflow management capabilities including: 1. Workflow Debug
 * Operations: - Execute workflow tasks in debug mode - Monitor task execution progress -
 * Initialize debug parameters - Resume paused tasks - Execute partial workflow graphs 2.
 * Workflow Execution Operations: - Stream workflow execution events using SSE - Execute
 * workflow tasks with real-time monitoring
 *
 * The controller supports both synchronous and asynchronous (streaming) execution modes,
 * providing flexibility for different use cases. It integrates with Redis for context
 * management and supports workspace-based isolation.
 *
 * @since 1.0.0.3
 */
@Slf4j
@RestController
@Tag(name = "workflow")
@RequestMapping("/console/v1/apps")
public class WorkflowController {

	private final RedisManager redisManager;

	private final AppService appService;

	private final WorkflowExecuteManager workflowExecuteManager;

	private final WorkflowInnerService workflowInnerService;

	public WorkflowController(RedisManager redisManager, AppService appService,
			WorkflowExecuteManager workflowExecuteManager, WorkflowInnerService workflowInnerService) {
		this.redisManager = redisManager;
		this.appService = appService;
		this.workflowExecuteManager = workflowExecuteManager;
		this.workflowInnerService = workflowInnerService;
	}

	/**
	 * Executes a workflow task in debug mode. This endpoint allows running workflow tasks
	 * with debug capabilities, enabling step-by-step execution and detailed monitoring.
	 * @param request Task execution request containing: - appId: Application identifier -
	 * version: Application version (defaults to "latest") - inputs: Input parameters for
	 * the workflow - conversationId: Session identifier for tracking
	 * @return TaskRunResponse containing: - taskId: Unique identifier for the executed
	 * task - conversationId: Session identifier - requestId: Request tracking identifier
	 */
	@PostMapping(value = { "/workflow/debug/run-task" })
	public Result<TaskRunResponse> runDebugTask(@RequestBody TaskRunRequest request) {
		if (request == null || StringUtils.isBlank(request.getAppId())) {
			LogUtils.monitor("WorkflowService", "runTask", System.currentTimeMillis(), FAIL, request,
					"request or appId is null");
			throw new BizException(ErrorCode.MISSING_PARAMS.toError("request or appId is null"));
		}
		String version = request.getVersion();
		if (version == null) {
			version = "latest";
		}
		ApplicationVersion appVersion = appService.getAppVersion(request.getAppId(), version);
		WorkflowContext workflowContext = new WorkflowContext();
		workflowContext.setInvokeSource(InvokeSourceEnum.console.getCode());
		TaskRunResponse response = workflowExecuteManager.runTask(appVersion, request.getInputs(),
				request.getConversationId(), workflowContext);
		return Result.success(response);
	}

	/**
	 * Retrieves the current execution status of a workflow task. This endpoint provides
	 * detailed information about the task's progress, including node execution status and
	 * results.
	 * @param request Process status request containing: - taskId: Task identifier to
	 * query
	 * @return ProcessGetResponse containing: - taskStatus: Current status of the task -
	 * nodeResults: List of node execution results - taskResults: Overall task execution
	 * results - errorInfo: Error information if task failed
	 */
	@PostMapping(value = { "/workflow/debug/get-task-process" })
	public Result<ProcessGetResponse> getDebugProcess(@RequestBody ProcessGetRequest request) {
		long start = System.currentTimeMillis();

		RequestContext context = RequestContextHolder.getRequestContext();
		context.setStartTime(System.currentTimeMillis());

		try {
			WorkflowContext wfContext = redisManager
				.get(WORKFLOW_TASK_CONTEXT_PREFIX + context.getWorkspaceId() + "_" + request.getTaskId());
			if (wfContext == null) {
				log.info(
						"log used for query debug task result, taskId:{}, RequestId:{}, costTime:{}, getProcessStatus:{}",
						request.getTaskId(), context.getRequestId(), System.currentTimeMillis() - start, "error");
				return Result.error(ErrorCode.WORKFLOW_NODE_DEBUG_FAIL);
			}
			ProcessGetResponse response = constructProcessGetResponse(request.getTaskId(), wfContext);
			List<NodeResult> filterNodeResult = response.getNodeResults()
				.stream()
				.filter(nodeResult -> nodeResult != null
						&& !nodeResult.getNodeStatus().equals(NodeStatusEnum.SKIP.getCode()))
				.collect(Collectors.toList());
			response.setNodeResults(filterNodeResult);
			return Result.success(response);
		}
		catch (Exception e) {
			log.error("WorkflowController getProcess error requestId:{}, costTime:{}", context.getRequestId(),
					System.currentTimeMillis() - start, e);
		}
		return Result.error(ErrorCode.WORKFLOW_DEBUG_GET_PROCESS_FAIL);
	}

	/**
	 * Initializes workflow debug parameters. This endpoint prepares the necessary
	 * parameters for workflow debugging, including system and user-defined parameters.
	 * @param request Initialization request containing: - appId: Application identifier -
	 * version: Application version (defaults to "latest")
	 * @return List of TaskRunParam containing: - key: Parameter key - type: Parameter
	 * type - source: Parameter source (system/user) - required: Whether parameter is
	 * required
	 */
	@PostMapping(value = { "/workflow/debug/init" })
	public Result<List<TaskRunParam>> debugInit(@RequestBody InitRequest request) {
		long start = System.currentTimeMillis();

		RequestContext context = RequestContextHolder.getRequestContext();
		context.setStartTime(System.currentTimeMillis());
		List<TaskRunParam> initTaskRunParams = Lists.newArrayList();
		try {
			String version = request.getVersion();
			if (version == null) {
				version = "latest";
			}
			// Get application details
			ApplicationVersion appVersion = appService.getAppVersion(request.getAppId(), version);
			WorkflowConfig config = JsonUtils.fromJson(appVersion.getConfig(), WorkflowConfig.class);
			List<Node> nodes = config.getNodes();
			if (CollectionUtils.isNotEmpty(nodes)) {
				Optional<Node> startOptional = nodes.stream()
					.filter(node -> node.getType().equals(NodeTypeEnum.START.getCode()))
					.findAny();
				if (startOptional.isPresent()) {
					Node startNode = startOptional.get();
					List<Node.OutputParam> outputParams = startNode.getConfig().getOutputParams();
					if (CollectionUtils.isEmpty(outputParams)) {
						initTaskRunParams = Lists.newArrayList();
					}
					else {
						initTaskRunParams.addAll(outputParams.stream().map(outputParam -> {
							TaskRunParam taskRunParam = BeanCopierUtils.copy(outputParam, TaskRunParam.class);
							taskRunParam.setSource(ParamSourceEnum.user.name());
							taskRunParam.setRequired(false);
							return taskRunParam;
						}).collect(Collectors.toList()));
					}
					// Add system parameters
					TaskRunParam queryParam = new TaskRunParam();
					queryParam.setKey(SYS_QUERY_KEY);
					queryParam.setDesc("User query");
					queryParam.setType("String");
					queryParam.setRequired(false);
					queryParam.setSource(ParamSourceEnum.sys.name());
					initTaskRunParams.add(queryParam);
				}
			}
			return Result.success(initTaskRunParams);
		}
		catch (Exception e) {
			log.error("WorkflowController debugInit error requestId:{}, costTime:{}", context.getRequestId(),
					System.currentTimeMillis() - start, e);
		}
		return Result.error(ErrorCode.WORKFLOW_DEBUG_INIT_FAIL);
	}

	/**
	 * Constructs process status response from workflow context.
	 * @param taskId Task ID
	 * @param context Workflow context
	 * @return Process status response
	 */
	private ProcessGetResponse constructProcessGetResponse(String taskId, WorkflowContext context) {
		ProcessGetResponse response = new ProcessGetResponse();
		// Construct input parameters
		response.setTaskStatus(context.getTaskStatus());
		response.setTaskId(taskId);
		response.setRequestId(context.getRequestId());
		response.setConversationId(context.getConversationId());
		ConcurrentHashMap<String, NodeResult> nodeResultMap = context.getNodeResultMap();
		List<NodeResult> list = Lists.newArrayList();
		CopyOnWriteArrayList<String> executeOrderList = context.getExecuteOrderList();
		List<ProcessGetResponse.ProcessOutput> processOutputList = Lists.newArrayList();
		Set<String> nodeIdSet = Sets.newHashSet();
		executeOrderList.stream().forEach(nodeId -> {
			if (!nodeIdSet.contains(nodeId)) {
				NodeResult currentNodeResult = nodeResultMap.get(nodeId);
				if (currentNodeResult == null) {
					return;
				}
				list.add(currentNodeResult);
				if ((NodeTypeEnum.OUTPUT.getCode().equals(currentNodeResult.getNodeType())
						|| NodeTypeEnum.END.getCode().equals(currentNodeResult.getNodeType()))
						&& (NodeStatusEnum.EXECUTING.getCode().equals(currentNodeResult.getNodeStatus())
								|| NodeStatusEnum.SUCCESS.getCode().equals(currentNodeResult.getNodeStatus()))) {
					// For output nodes, directly concatenate the content
					ProcessGetResponse.ProcessOutput processOutput = new ProcessGetResponse.ProcessOutput();
					processOutput.setNodeId(currentNodeResult.getNodeId());
					processOutput.setNodeName(currentNodeResult.getNodeName());
					processOutput.setNodeType(currentNodeResult.getNodeType());
					processOutput.setNodeStatus(currentNodeResult.getNodeStatus());
					processOutput.setNodeContent(currentNodeResult.getOutput());
					processOutputList.add(processOutput);
				}
				else if (NodeTypeEnum.INPUT.getCode().equals(currentNodeResult.getNodeType())) {
					if (StringUtils.isNotBlank(currentNodeResult.getParentNodeId())) {
						int index = currentNodeResult.getIndex();
						for (int i = 0; i < index; i++) {
							ProcessGetResponse.ProcessOutput processOutput = fetchProcessOutput(currentNodeResult);
							processOutput.setNodeStatus(NodeStatusEnum.SUCCESS.getCode());
							processOutput.setIndex(i);
							processOutputList.add(processOutput);
						}
						ProcessGetResponse.ProcessOutput processOutput = fetchProcessOutput(currentNodeResult);
						processOutput.setIndex(index);
						processOutput.setNodeStatus(currentNodeResult.getNodeStatus());
						processOutputList.add(processOutput);
					}
					else {
						ProcessGetResponse.ProcessOutput processOutput = fetchProcessOutput(currentNodeResult);
						processOutputList.add(processOutput);
					}
				}
			}
			nodeIdSet.add(nodeId);
		});
		response.setNodeResults(list);
		// Add task results
		response.setTaskResults(processOutputList);
		if (NodeStatusEnum.SUCCESS.getCode().equals(context.getTaskStatus())) {
			response.setTaskExecTime((System.currentTimeMillis() - context.getStartTime()) + "ms");
		}
		if (NodeStatusEnum.FAIL.getCode().equals(context.getTaskStatus())) {
			response.setTaskExecTime((System.currentTimeMillis() - context.getStartTime()) + "ms");
			response.setErrorCode(context.getErrorCode());
			response.setErrorInfo(context.getErrorInfo());
		}
		return response;
	}

	/**
	 * Fetches process output from node result.
	 * @param currentNodeResult Current node execution result
	 * @return Process output details
	 */
	private ProcessGetResponse.ProcessOutput fetchProcessOutput(NodeResult currentNodeResult) {
		ProcessGetResponse.ProcessOutput processOutput = new ProcessGetResponse.ProcessOutput();
		processOutput.setNodeId(currentNodeResult.getNodeId());
		processOutput.setNodeName(currentNodeResult.getNodeName());
		processOutput.setNodeType(currentNodeResult.getNodeType());
		processOutput.setNodeStatus(currentNodeResult.getNodeStatus());
		processOutput.setNodeContent(Lists.newArrayList());
		processOutput.setParentNodeId(currentNodeResult.getParentNodeId());
		if (currentNodeResult.getInput() != null) {
			List<Node.OutputParam> outputParams = JsonUtils.fromJsonToList(currentNodeResult.getInput(),
					Node.OutputParam.class);
			if (currentNodeResult.getOutput() != null) {
				Map<Object, Object> outputMap = JsonUtils.fromJsonToMap(currentNodeResult.getOutput());
				outputParams.stream().forEach(outputParam -> {
					outputParam.setValue(outputMap.get(outputParam.getKey()));
				});
			}
			processOutput.setNodeContent(outputParams);
		}
		return processOutput;
	}

	/**
	 * Resumes a paused workflow task. This endpoint allows continuing execution of a
	 * previously paused workflow task, with the ability to provide new input parameters.
	 * @param request Task resume request containing: - taskId: Task identifier to resume
	 * - resumeNodeId: Node identifier to resume from - resumeParentId: Parent node
	 * identifier (for nested nodes) - inputParams: New input parameters for the resumed
	 * execution
	 * @return TaskResumeResponse containing: - taskId: Task identifier - requestId:
	 * Request tracking identifier
	 */
	@PostMapping(value = { "/workflow/debug/resume-task" })
	public Result<TaskResumeResponse> resumeDebugTask(@RequestBody TaskResumeRequest request) {
		long start = System.currentTimeMillis();
		RequestContext context = RequestContextHolder.getRequestContext();
		try {
			if (request == null || request.getTaskId() == null || request.getResumeNodeId() == null) {
				throw new BizException(ErrorCode.MISSING_PARAMS.toError("request or appId is null"));
			}
			WorkflowContext wfContext = redisManager
				.get(WORKFLOW_TASK_CONTEXT_PREFIX + context.getWorkspaceId() + "_" + request.getTaskId());
			if (wfContext == null) {
				return Result.error(IdGenerator.uuid(), ErrorCode.WORKFLOW_CONFIG_INVALID.toError("taskId not exists"));
			}
			Node inputNode = null;
			// 组装输入节点内容
			if (StringUtils.isBlank(request.getResumeParentId())) {
				Optional<Node> inputNodeOptional = wfContext.getWorkflowConfig()
					.getNodes()
					.stream()
					.filter(node -> node.getId().equals(request.getResumeNodeId()))
					.findFirst();
				if (inputNodeOptional.isPresent()) {
					inputNode = inputNodeOptional.get();
				}
			}
			else {
				Optional<Node> inputNodeOptional = wfContext.getWorkflowConfig()
					.getNodes()
					.stream()
					.filter(node -> node.getId().equals(request.getResumeParentId()))
					.findFirst();
				if (inputNodeOptional.isPresent()) {
					Node parentNode = inputNodeOptional.get();
					IteratorExecuteProcessor.NodeParam nodeParam = JsonUtils
						.fromMap(parentNode.getConfig().getNodeParam(), IteratorExecuteProcessor.NodeParam.class);
					Optional<Node> subInputNodeOptional = nodeParam.getBlock()
						.getNodes()
						.stream()
						.filter(node -> node.getId().equals(request.getResumeNodeId()))
						.findFirst();
					if (subInputNodeOptional.isPresent()) {
						inputNode = subInputNodeOptional.get();
					}
				}

			}

			if (inputNode == null) {
				return Result.error(IdGenerator.uuid(),
						ErrorCode.WORKFLOW_CONFIG_INVALID.toError("resume_node_id not exists"));
			}
			NodeResult nodeResultFront = wfContext.getNodeResultMap().get(request.getResumeNodeId());
			if (nodeResultFront == null || nodeResultFront.getNodeStatus().equals(NodeStatusEnum.PAUSE.getCode())) {
				NodeResult nodeResult = new NodeResult();
				nodeResult.setNodeId(inputNode.getId());
				nodeResult.setNodeName(inputNode.getName());
				nodeResult.setNodeType(inputNode.getType());
				nodeResult.setUsages(null);
				if (nodeResultFront != null) {
					nodeResult.setBatches(nodeResultFront.getBatches());
				}
				List<Node.OutputParam> outputParams = inputNode.getConfig().getOutputParams();
				nodeResult.setInput(JsonUtils.toJson(outputParams));
				if (CollectionUtils.isNotEmpty(outputParams)) {
					List<CommonParam> inputParams = request.getInputParams();
					Map<String, Object> inputMap = Maps.newHashMap();
					if (CollectionUtils.isNotEmpty(inputParams)) {
						inputParams.stream().forEach(param -> inputMap.put(param.getKey(), param.getValue()));
					}
					Map<String, Object> outputMap = Maps.newHashMap();
					outputParams.stream().forEach(outputParam -> {
						outputMap.put(outputParam.getKey(), inputMap.get(outputParam.getKey()));
					});
					nodeResult.setOutput(JsonUtils.toJson(outputMap));
				}
				else {
					nodeResult.setOutput(null);
				}
				nodeResult.setNodeStatus(NodeStatusEnum.EXECUTING.getCode());
				wfContext.getNodeResultMap().put(nodeResult.getNodeId(), nodeResult);
				wfContext.setTaskStatus(NodeStatusEnum.EXECUTING.getCode());
				workflowInnerService.forceRefreshContextCache(wfContext);
			}
			TaskResumeResponse response = new TaskResumeResponse();
			response.setTaskId(request.getTaskId());
			response.setRequestId(IdGenerator.uuid());
			return Result.success(response);
		}
		catch (Exception e) {
			log.error("WorkflowController resumeDebugTask error taskId:{}, costTime:{}", request.getTaskId(),
					System.currentTimeMillis() - start, e);
			return Result.error(ErrorCode.WORKFLOW_DEBUG_FAIL);
		}
	}

	/**
	 * Executes a partial workflow graph for testing. This endpoint allows running a
	 * subset of the workflow graph, useful for testing specific workflow components.
	 * @param request Partial graph execution request containing: - nodes: List of nodes
	 * to execute - edges: List of edges connecting the nodes - inputParams: Input
	 * parameters for the execution - appId: Application identifier
	 * @return TaskPartGraphResponse containing: - taskId: Task identifier - requestId:
	 * Request tracking identifier
	 */
	@PostMapping(value = { "/workflow/debug/part-graph/run-task" })
	public Result<TaskPartGraphResponse> partGraphRunTask(@RequestBody TaskPartGraphRequest request) {
		RequestContext requestContext = RequestContextHolder.getRequestContext();
		if (request == null || CollectionUtils.isEmpty(request.getNodes())) {
			return Result.error(requestContext.getRequestId(), ErrorCode.MISSING_PARAMS.toError("nodes is null"));
		}

		List<Edge> edges = request.getEdges() == null ? Lists.newArrayList() : request.getEdges();
		WorkflowConfig workflowConfig = workflowExecuteManager.constructDebugConfig4Fragment(request.getNodes(), edges);
		Map<String, Object> inputMap = Maps.newHashMap();
		if (CollectionUtils.isNotEmpty(request.getInputParams())) {
			request.getInputParams().stream().forEach(inputParam -> {
				inputMap.put(inputParam.getKey(), VariableUtils.convertValueByType(inputParam.getKey(),
						inputParam.getType(), inputParam.getValue()));
			});
		}
		Map<String, Object> variableMap = workflowExecuteManager.constructDebugVariables(inputMap);
		WorkflowContext context = new WorkflowContext();
		context.setWorkflowConfig(workflowConfig);
		context.getVariablesMap().putAll(variableMap);
		context.setAppId(request.getAppId());
		context.setTaskStatus(NodeStatusEnum.EXECUTING.getCode());
		context.setRequestId(requestContext.getRequestId());
		context.setInvokeSource(InvokeSourceEnum.console.getCode());
		context.setWorkspaceId(requestContext.getWorkspaceId());
		context.setConversationId(IdGenerator.uuid());
		String taskId = workflowExecuteManager.execute(workflowConfig, context);
		TaskPartGraphResponse response = new TaskPartGraphResponse();
		response.setTaskId(taskId);
		response.setRequestId(requestContext.getRequestId());
		return Result.success(response);
	}

	@PostMapping(value = { "/workflow/debug/part-graph/stop-task" })
	public Result<Boolean> stopTask(@RequestBody TaskStopRequest request) {
		RequestContext requestContext = RequestContextHolder.getRequestContext();
		if (request == null || StringUtils.isBlank(request.getTaskId())) {
			return Result.error(requestContext.getRequestId(), ErrorCode.MISSING_PARAMS.toError("taskId is null"));
		}
		return Result.success(workflowExecuteManager.stopTask(request.getTaskId()));
	}

	/**
	 * Streams workflow execution events using Server-Sent Events (SSE). This endpoint
	 * provides real-time updates about workflow execution progress, including node status
	 * changes and execution results.
	 * @param appId Application identifier
	 * @param request Task run request containing: - inputs: Input parameters for the
	 * workflow - conversationId: Session identifier for tracking
	 * @return SseEmitter for streaming events including: - Message events: Node execution
	 * updates - Error events: Execution errors - Pause events: Task pause notifications -
	 * Finish events: Task completion notifications
	 */
	@PostMapping(value = "/workflow/{appId}/run_stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
	public SseEmitter streamEvents(@PathVariable("appId") String appId, @RequestBody ApiTaskRunRequest request) {
		SseEmitter emitter = new SseEmitter(0L);

		// 设置超时和完成回调
		emitter.onTimeout(() -> {
			ApiTaskMsg timeoutMsg = new ApiTaskMsg();
			timeoutMsg.setEvent(ApiTaskMsg.Event.Error.name());
			timeoutMsg.setError_code("TIMEOUT");
			timeoutMsg.setError_message("Stream connection timeout");
			try {
				emitter.send(timeoutMsg);
				emitter.complete();
			}
			catch (IOException e) {
				log.error("Send timeout message error", e);
			}
		});

		emitter.onCompletion(() -> {
			log.info("SSE connection completed for task: {}", request.getConversationId());
		});

		ThreadPoolUtils.DEFAULT_TASK_EXECUTOR.execute(() -> {
			innerRunStream(emitter, appId, request);
		});

		return emitter;
	}

	private void innerRunStream(SseEmitter emitter, String appId, ApiTaskRunRequest request) {
		RequestContext context = RequestContextHolder.getRequestContext();
		long start = System.currentTimeMillis();
		String taskId = null;
		String conversationId = null;

		try {
			// Parameter validation and initialization
			validateAndInitParams(request, appId);
			String version = "lastPublished";
			WorkflowContext workflowContext = new WorkflowContext();
			workflowContext.setInvokeSource(InvokeSourceEnum.console.getCode());
			ApplicationVersion appVersion = appService.getAppVersion(appId, version);

			// Run task and get response
			TaskRunResponse response = workflowExecuteManager.runTask(appVersion, request.getInputs(),
					request.getConversationId(), workflowContext);
			taskId = response.getTaskId();
			conversationId = response.getConversationId();

			// Process task execution
			processTaskExecution(emitter, context, taskId, conversationId, start);

		}
		catch (Exception e) {
			log.error("WorkflowController streamEvents error requestId:{}, costTime:{}", context.getRequestId(),
					System.currentTimeMillis() - start, e);
			sendErrorMessage(emitter, taskId, conversationId, "INTERNAL_ERROR", e.getMessage());
		}
		finally {
			try {
				emitter.complete();
			}
			catch (Exception e) {
				log.error("Error completing emitter", e);
			}
		}
	}

	private void validateAndInitParams(ApiTaskRunRequest request, String appId) {
		if (request == null || StringUtils.isBlank(appId)) {
			throw new BizException(ErrorCode.MISSING_PARAMS.toError("request or appId is null"));
		}
	}

	private void processTaskExecution(SseEmitter emitter, RequestContext context, String taskId, String conversationId,
			long startTime) throws Exception {
		List<NodeResult> lastNodeResults = Lists.newArrayList();
		Map<String, AtomicInteger> recmsgSeqIdMap = Maps.newHashMap();
		while (true) {
			// 检查超时
			if (System.currentTimeMillis() - startTime > InvokeSourceEnum.api.getTimeoutSeconds() * 1000) {
				sendTimeoutMessage(emitter, taskId, conversationId);
				break;
			}

			// 获取最新上下文
			WorkflowContext latestContext = getLatestContext(context, taskId);
			if (latestContext == null) {
				throw new BizException(ErrorCode.WORKFLOW_NODE_DEBUG_FAIL.toError());
			}

			List<NodeResult> currentNodeResults = Lists.newArrayList();
			latestContext.getExecuteOrderList().stream().forEach(nodeId -> {
				NodeResult currentNodeResult = latestContext.getNodeResultMap().get(nodeId);
				if ((NodeTypeEnum.OUTPUT.getCode().equals(currentNodeResult.getNodeType())
						|| NodeTypeEnum.END.getCode().equals(currentNodeResult.getNodeType())
						|| NodeTypeEnum.INPUT.getCode().equals(currentNodeResult.getNodeType()))
						&& (NodeStatusEnum.EXECUTING.getCode().equals(currentNodeResult.getNodeStatus())
								|| NodeStatusEnum.SUCCESS.getCode().equals(currentNodeResult.getNodeStatus())
								|| NodeStatusEnum.PAUSE.getCode().equals(currentNodeResult.getNodeStatus()))) {
					currentNodeResults.add(currentNodeResult);
				}
			});
			boolean handleResult = handleNodeMessage(emitter, taskId, conversationId, recmsgSeqIdMap, lastNodeResults,
					currentNodeResults);
			if (handleResult) {
				lastNodeResults = currentNodeResults;
			}

			// 处理任务状态
			if (handleTaskStatus(emitter, latestContext, taskId, conversationId)) {
				break;
			}
		}
	}

	private boolean handleNodeMessage(SseEmitter emitter, String taskId, String conversationId,
			Map<String, AtomicInteger> recmsgSeqIdMap, List<NodeResult> lastNodeResults,
			List<NodeResult> thisNodeResult) throws Exception {
		if (CollectionUtils.isEmpty(thisNodeResult)) {
			return false;
		}
		boolean diff = false;
		if (CollectionUtils.isEmpty(lastNodeResults)) {
			for (NodeResult nodeResult : thisNodeResult) {
				String incrementalContent = calculateIncrementalContent(nodeResult.getOutput(), "");
				AtomicInteger atomicInteger = recmsgSeqIdMap.get(nodeResult.getNodeId());
				if (atomicInteger == null) {
					atomicInteger = new AtomicInteger(0);
				}
				boolean nodeCompleted = nodeResult.getNodeStatus().equals(NodeStatusEnum.SUCCESS.getCode()) ? true
						: false;
				if (StringUtils.isNotBlank(incrementalContent) || nodeCompleted) {
					sendNodeMessage(emitter, nodeResult, taskId, conversationId, atomicInteger.incrementAndGet(),
							incrementalContent);
					diff = true;
				}
				recmsgSeqIdMap.put(nodeResult.getNodeId(), atomicInteger);
			}
		}
		else {
			Map<String, List<NodeResult>> listMap = lastNodeResults.stream()
				.collect(Collectors.groupingBy(NodeResult::getNodeId));
			for (NodeResult nodeResult : thisNodeResult) {
				List<NodeResult> lastNodeResultList = listMap.get(nodeResult.getNodeId());
				boolean nodeCompleted = (!lastNodeResultList.get(0)
					.getNodeStatus()
					.equals(NodeStatusEnum.SUCCESS.getCode())
						&& nodeResult.getNodeStatus().equals(NodeStatusEnum.SUCCESS.getCode())) ? true : false;
				String incrementalContent;
				if (CollectionUtils.isEmpty(lastNodeResultList)) {
					incrementalContent = calculateIncrementalContent(nodeResult.getOutput(), "");
				}
				else {
					incrementalContent = calculateIncrementalContent(nodeResult.getOutput(),
							lastNodeResultList.get(0).getOutput());
				}

				AtomicInteger atomicInteger = recmsgSeqIdMap.get(nodeResult.getNodeId());
				if (atomicInteger == null) {
					atomicInteger = new AtomicInteger(0);
				}
				if (StringUtils.isNotBlank(incrementalContent) || nodeCompleted) {
					sendNodeMessage(emitter, nodeResult, taskId, conversationId, atomicInteger.incrementAndGet(),
							incrementalContent);
					diff = true;
				}
				recmsgSeqIdMap.put(nodeResult.getNodeId(), atomicInteger);
			}
		}
		return diff;
	}

	private WorkflowContext getLatestContext(RequestContext context, String taskId) {
		return redisManager.get(WORKFLOW_TASK_CONTEXT_PREFIX + context.getWorkspaceId() + "_" + taskId);
	}

	private boolean handleTaskStatus(SseEmitter emitter, WorkflowContext context, String taskId, String conversationId)
			throws IOException {
		String taskStatus = context.getTaskStatus();

		if (NodeStatusEnum.FAIL.getCode().equals(taskStatus) || NodeStatusEnum.STOP.getCode().equals(taskStatus)) {
			sendErrorMessage(emitter, taskId, conversationId, context.getErrorCode(), context.getErrorInfo());
			return true;
		}

		if (NodeStatusEnum.PAUSE.getCode().equals(taskStatus)) {
			sendPauseMessage(emitter, context, taskId, conversationId);
			return true;
		}

		if (NodeStatusEnum.SUCCESS.getCode().equals(taskStatus)) {
			sendFinishMessage(emitter, taskId, conversationId);
			return true;
		}

		return false;
	}

	private String calculateIncrementalContent(String currentOutput, String lastOutput) {
		if (lastOutput != null && !lastOutput.isEmpty() && currentOutput.startsWith(lastOutput)) {
			return currentOutput.substring(lastOutput.length());
		}
		return currentOutput;
	}

	private void sendNodeMessage(SseEmitter emitter, NodeResult nodeResult, String taskId, String conversationId,
			int msgSeqId, String content) throws IOException {
		ApiTaskMsg message = new ApiTaskMsg();
		message.setEvent(ApiTaskMsg.Event.Message.name());
		message.setTaskId(taskId);
		message.setConversationId(conversationId);
		message.setNodeId(nodeResult.getNodeId());
		message.setNodeName(nodeResult.getNodeName());
		message.setNodeType(nodeResult.getNodeType());
		message.setNodeStatus(nodeResult.getNodeStatus());
		if (nodeResult.getNodeStatus().equals(NodeStatusEnum.SUCCESS.getCode())) {
			message.setNodeIsCompleted(true);
		}
		else {
			message.setNodeIsCompleted(false);
		}
		message.setNodeMsgSeqId(msgSeqId);
		message.setTextContent(content);
		emitter.send(message);
	}

	private void sendErrorMessage(SseEmitter emitter, String taskId, String conversationId, String errorCode,
			String errorMessage) {
		try {
			ApiTaskMsg errorMsg = new ApiTaskMsg();
			errorMsg.setEvent(ApiTaskMsg.Event.Error.name());
			errorMsg.setError_code(errorCode);
			errorMsg.setError_message(errorMessage);
			errorMsg.setTaskId(taskId);
			errorMsg.setConversationId(conversationId);
			emitter.send(errorMsg);
		}
		catch (IOException e) {
			log.error("Error sending error message", e);
		}
	}

	private void sendTimeoutMessage(SseEmitter emitter, String taskId, String conversationId) {
		sendErrorMessage(emitter, taskId, conversationId, "TIMEOUT", "Task execution timeout");
	}

	private void sendPauseMessage(SseEmitter emitter, WorkflowContext context, String taskId, String conversationId)
			throws IOException {
		ApiTaskMsg pauseMsg = new ApiTaskMsg();
		pauseMsg.setEvent(ApiTaskMsg.Event.Paused.name());
		pauseMsg.setTaskId(taskId);
		pauseMsg.setConversationId(conversationId);
		pauseMsg.setPauseType(ApiTaskMsg.PauseType.InputNodeInterrupt.name());

		Optional<NodeResult> pauseNodeResult = context.getNodeResultMap()
			.values()
			.stream()
			.filter(result -> NodeStatusEnum.PAUSE.getCode().equals(result.getNodeStatus()))
			.findFirst();

		if (pauseNodeResult.isPresent()) {
			NodeResult nodeResult = pauseNodeResult.get();
			Map<String, Object> pauseData = Maps.newHashMap();
			pauseData.put("node_id", nodeResult.getNodeId());
			pauseData.put("node_name", nodeResult.getNodeName());
			pauseData.put("node_type", nodeResult.getNodeType());
			pauseData.put("input_params", nodeResult.getInput());
			pauseMsg.setPause_data(pauseData);
		}
		emitter.send(pauseMsg);
	}

	private void sendFinishMessage(SseEmitter emitter, String taskId, String conversationId) throws IOException {
		ApiTaskMsg finishMsg = new ApiTaskMsg();
		finishMsg.setEvent(ApiTaskMsg.Event.Finished.name());
		finishMsg.setTaskId(taskId);
		finishMsg.setConversationId(conversationId);
		emitter.send(finishMsg);
	}

}
