Add files via upload

internal/handler/agent.go

@@ -1529,6 +1529,10 @@ func (h *AgentHandler) SubscribeAgentTaskEvents(c *gin.Context) {
 
 	flusher, _ := c.Writer.(http.Flusher)
 	ctx := c.Request.Context()
+	var writeMu sync.Mutex
+	stopKeepalive := make(chan struct{})
+	go sseKeepalive(c, stopKeepalive, &writeMu)
+	defer close(stopKeepalive)
 
 	for {
 		select {
@@ -1538,12 +1542,15 @@ func (h *AgentHandler) SubscribeAgentTaskEvents(c *gin.Context) {
 			if !ok {
 				return
 			}
+			writeMu.Lock()
 			if _, err := c.Writer.Write(chunk); err != nil {
+				writeMu.Unlock()
 				return
 			}
 			if flusher != nil {
 				flusher.Flush()
 			}
+			writeMu.Unlock()
 		}
 	}
 }

internal/handler/workflow.go

@@ -41,6 +41,12 @@ type workflowSaveRequest struct {
 	GraphJSON   json.RawMessage `json:"graph_json,omitempty"`
 }
 
+type workflowDryRunRequest struct {
+	Graph     json.RawMessage        `json:"graph,omitempty"`
+	GraphJSON json.RawMessage        `json:"graph_json,omitempty"`
+	Inputs    map[string]interface{} `json:"inputs,omitempty"`
+}
+
 func (h *WorkflowHandler) List(c *gin.Context) {
 	includeDisabled := strings.EqualFold(c.Query("includeDisabled"), "true") || c.Query("include_disabled") == "1"
 	items, err := h.db.ListWorkflowDefinitions(includeDisabled)
@@ -69,6 +75,57 @@ func (h *WorkflowHandler) Create(c *gin.Context) {
 	h.save(c, "")
 }
 
+func (h *WorkflowHandler) Validate(c *gin.Context) {
+	var req workflowSaveRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"ok": false, "error": "无效的请求参数: " + err.Error()})
+		return
+	}
+	graph := req.Graph
+	if len(graph) == 0 {
+		graph = req.GraphJSON
+	}
+	if len(graph) == 0 {
+		c.JSON(http.StatusBadRequest, gin.H{"ok": false, "error": "graph 不能为空"})
+		return
+	}
+	if !json.Valid(graph) {
+		c.JSON(http.StatusBadRequest, gin.H{"ok": false, "error": "graph 必须是合法 JSON"})
+		return
+	}
+	if err := workflowrunner.ValidateGraphJSON(c.Request.Context(), string(graph)); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"ok": false, "error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"ok": true})
+}
+
+func (h *WorkflowHandler) DryRun(c *gin.Context) {
+	var req workflowDryRunRequest
+	if err := c.ShouldBindJSON(&req); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "无效的请求参数: " + err.Error()})
+		return
+	}
+	graph := req.Graph
+	if len(graph) == 0 {
+		graph = req.GraphJSON
+	}
+	if len(graph) == 0 || !json.Valid(graph) {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "graph 必须是合法 JSON"})
+		return
+	}
+	inputs := make(map[string]any, len(req.Inputs))
+	for k, v := range req.Inputs {
+		inputs[k] = v
+	}
+	result, err := workflowrunner.DryRunGraphJSON(c.Request.Context(), string(graph), inputs)
+	if err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"result": result})
+}
+
 func (h *WorkflowHandler) Update(c *gin.Context) {
 	h.save(c, c.Param("id"))
 }

internal/handler/workflow_run.go

@@ -1,6 +1,7 @@
 package handler
 
 import (
+	"encoding/json"
 	"net/http"
 	"strings"
 	"time"
@@ -28,7 +29,40 @@ func (h *WorkflowHandler) GetRun(c *gin.Context) {
 		c.JSON(http.StatusNotFound, gin.H{"error": "工作流运行不存在"})
 		return
 	}
-	c.JSON(http.StatusOK, gin.H{"run": run})
+	nodeRuns, err := h.db.ListWorkflowNodeRuns(runID)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	c.JSON(http.StatusOK, gin.H{"run": run, "nodeRuns": nodeRuns})
+}
+
+func (h *WorkflowHandler) ReplayRun(c *gin.Context) {
+	runID := strings.TrimSpace(c.Param("runId"))
+	nodeRuns, err := h.db.ListWorkflowNodeRuns(runID)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+		return
+	}
+	steps := make([]gin.H, 0, len(nodeRuns))
+	for i, nodeRun := range nodeRuns {
+		var input any
+		var output any
+		_ = json.Unmarshal([]byte(nodeRun.InputJSON), &input)
+		_ = json.Unmarshal([]byte(nodeRun.OutputJSON), &output)
+		steps = append(steps, gin.H{
+			"step":       i + 1,
+			"nodeRunId":  nodeRun.ID,
+			"nodeId":     nodeRun.NodeID,
+			"status":     nodeRun.Status,
+			"input":      input,
+			"output":     output,
+			"error":      nodeRun.Error,
+			"startedAt":  nodeRun.StartedAt,
+			"finishedAt": nodeRun.FinishedAt,
+		})
+	}
+	c.JSON(http.StatusOK, gin.H{"workflowRunId": runID, "steps": steps})
 }
 
 func (h *WorkflowHandler) ListPendingRuns(c *gin.Context) {
@@ -99,10 +133,10 @@ func (h *WorkflowHandler) ResumeRun(c *gin.Context) {
 	}
 	if delegated {
 		c.JSON(http.StatusOK, gin.H{
-			"workflowRunId":    runID,
-			"status":           "delegated",
-			"streamResuming":   true,
-			"approved":         req.Approved,
+			"workflowRunId":  runID,
+			"status":         "delegated",
+			"streamResuming": true,
+			"approved":       req.Approved,
 		})
 		return
 	}