diff --git a/internal/handler/agent.go b/internal/handler/agent.go index 54972373..2c2f357a 100644 --- a/internal/handler/agent.go +++ b/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() } } } diff --git a/internal/handler/workflow.go b/internal/handler/workflow.go index 4ea7166d..acbcf186 100644 --- a/internal/handler/workflow.go +++ b/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")) } diff --git a/internal/handler/workflow_run.go b/internal/handler/workflow_run.go index 0754912f..f9e4e6d7 100644 --- a/internal/handler/workflow_run.go +++ b/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 }