fix: validate workflows and preserve execution outcomes

internal/handler/finalization_helpers.go

@@ -149,6 +149,9 @@ func finalizationCheckMessage(d agentfinalizer.Decision) string {
 }
 
 func finalizationBlockedMessage(d agentfinalizer.Decision) string {
+	if d.CompletionReason == "workflow_rejected" && strings.TrimSpace(d.FinalText) != "" {
+		return d.FinalText
+	}
 	parts := []string{"任务尚未达到最终回复条件，暂不生成成功结论。"}
 	if d.CompletionReason != "" {
 		parts = append(parts, "原因: "+d.CompletionReason)

internal/handler/workflow.go

@@ -1,10 +1,12 @@
 package handler
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"net/http"
 	"strings"
+	"unicode"
 
 	"cyberstrike-ai/internal/agent"
 	"cyberstrike-ai/internal/audit"
@@ -16,6 +18,33 @@ import (
 	"go.uber.org/zap"
 )
 
+func validWorkflowID(id string) bool {
+	if len(id) == 0 || len(id) > 128 {
+		return false
+	}
+	for i, r := range id {
+		if !unicode.IsLetter(r) && !unicode.IsDigit(r) && !(i > 0 && (r == '_' || r == '-')) {
+			return false
+		}
+	}
+	return true
+}
+
+type workflowDefinitionResponse struct {
+	*database.WorkflowDefinition
+	ValidationError string `json:"validation_error,omitempty"`
+}
+
+func workflowResponse(ctx context.Context, wf *database.WorkflowDefinition) workflowDefinitionResponse {
+	r := workflowDefinitionResponse{WorkflowDefinition: wf}
+	if wf != nil {
+		if err := workflowrunner.ValidateGraphJSON(ctx, wf.GraphJSON); err != nil {
+			r.ValidationError = err.Error()
+		}
+	}
+	return r
+}
+
 type WorkflowHandler struct {
 	db     *database.DB
 	logger *zap.Logger
@@ -61,7 +90,11 @@ func (h *WorkflowHandler) List(c *gin.Context) {
 		c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
 		return
 	}
-	c.JSON(http.StatusOK, gin.H{"workflows": items})
+	responses := make([]workflowDefinitionResponse, 0, len(items))
+	for _, item := range items {
+		responses = append(responses, workflowResponse(c.Request.Context(), item))
+	}
+	c.JSON(http.StatusOK, gin.H{"workflows": responses})
 }
 
 func (h *WorkflowHandler) Get(c *gin.Context) {
@@ -75,7 +108,7 @@ func (h *WorkflowHandler) Get(c *gin.Context) {
 		c.JSON(http.StatusNotFound, gin.H{"error": "工作流不存在"})
 		return
 	}
-	c.JSON(http.StatusOK, gin.H{"workflow": wf})
+	c.JSON(http.StatusOK, gin.H{"workflow": workflowResponse(c.Request.Context(), wf)})
 }
 
 func (h *WorkflowHandler) Create(c *gin.Context) {
@@ -190,6 +223,25 @@ func (h *WorkflowHandler) save(c *gin.Context, pathID string) {
 		c.JSON(http.StatusBadRequest, gin.H{"error": "工作流 id 和 name 不能为空"})
 		return
 	}
+	if pathID == "" && !validWorkflowID(id) {
+		c.JSON(http.StatusBadRequest, gin.H{"error": "工作流 ID 必须以字母或数字开头，仅含字母、数字、下划线、连字符，且不超过 128 字节"})
+		return
+	}
+	if pathID != "" {
+		if req.ID != "" && strings.TrimSpace(req.ID) != id {
+			c.JSON(http.StatusBadRequest, gin.H{"error": "工作流 ID 不允许通过编辑更改"})
+			return
+		}
+		existing, err := h.db.GetWorkflowDefinition(id)
+		if err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": err.Error()})
+			return
+		}
+		if existing == nil {
+			c.JSON(http.StatusNotFound, gin.H{"error": "工作流不存在"})
+			return
+		}
+	}
 	graph := req.Graph
 	if len(graph) == 0 {
 		graph = req.GraphJSON
@@ -222,7 +274,17 @@ func (h *WorkflowHandler) save(c *gin.Context, pathID string) {
 		GraphJSON:   string(graph),
 		Enabled:     enabled,
 	}
-	if err := h.db.UpsertWorkflowDefinition(wf); err != nil {
+	var saveErr error
+	if pathID == "" {
+		saveErr = h.db.CreateWorkflowDefinition(wf)
+	} else {
+		saveErr = h.db.UpsertWorkflowDefinition(wf)
+	}
+	if err := saveErr; err != nil {
+		if errors.Is(err, database.ErrWorkflowAlreadyExists) {
+			c.JSON(http.StatusConflict, gin.H{"error": err.Error()})
+			return
+		}
 		if h.logger != nil {
 			h.logger.Warn("保存工作流失败", zap.String("id", id), zap.Error(err))
 		}

internal/handler/workflow_bug_sheet_test.go

@@ -0,0 +1,197 @@
+package handler
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	agentpkg "cyberstrike-ai/internal/agent"
+	"cyberstrike-ai/internal/agentfinalizer"
+	"cyberstrike-ai/internal/config"
+	"cyberstrike-ai/internal/database"
+	"cyberstrike-ai/internal/mcp"
+	workflowrunner "cyberstrike-ai/internal/workflow"
+	"github.com/gin-gonic/gin"
+	"go.uber.org/zap"
+)
+
+const bugSheetLinearGraph = `{"nodes":[{"id":"start-1","type":"start","config":{}},{"id":"out","type":"output","config":{"output_key":"result","source_binding":{"from":"inputs","field":"message"}}}],"edges":[{"source":"start-1","target":"out"}],"config":{"schema_version":1}}`
+
+func workflowBugDB(t *testing.T) *database.DB {
+	t.Helper()
+	db, err := database.NewDB(filepath.Join(t.TempDir(), "workflow.db"), zap.NewNop())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = db.Close() })
+	workflowrunner.SetCheckpointDir(filepath.Join(t.TempDir(), "checkpoints"))
+	return db
+}
+
+func TestWorkflowCreateValidatesIDAndNeverReplacesExisting(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	db := workflowBugDB(t)
+	h := NewWorkflowHandler(db, zap.NewNop())
+	r := gin.New()
+	r.POST("/workflows", h.Create)
+	r.PUT("/workflows/:id", h.Update)
+	call := func(method, url, id, name string) *httptest.ResponseRecorder {
+		body, _ := json.Marshal(map[string]any{"id": id, "name": name, "graph": json.RawMessage(bugSheetLinearGraph)})
+		req := httptest.NewRequest(method, url, bytes.NewReader(body))
+		req.Header.Set("Content-Type", "application/json")
+		rec := httptest.NewRecorder()
+		r.ServeHTTP(rec, req)
+		return rec
+	}
+	for _, id := range []string{"../wf", "a/b", "a\\b", "a b", "_wf", strings.Repeat("w", 129)} {
+		rec := call(http.MethodPost, "/workflows", id, "name")
+		if rec.Code != http.StatusBadRequest {
+			t.Fatalf("accepted id %q: %d %s", id, rec.Code, rec.Body)
+		}
+	}
+	rec := call(http.MethodPost, "/workflows", "wf-test", "original")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("create: %d %s", rec.Code, rec.Body)
+	}
+	rec = call(http.MethodPost, "/workflows", "wf-test", "replacement")
+	if rec.Code != http.StatusConflict {
+		t.Fatalf("duplicate: %d %s", rec.Code, rec.Body)
+	}
+	saved, _ := db.GetWorkflowDefinition("wf-test")
+	if saved.Name != "original" || saved.Version != 1 {
+		t.Fatalf("duplicate replaced: %#v", saved)
+	}
+	rec = call(http.MethodPut, "/workflows/missing", "missing", "new")
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("missing update: %d %s", rec.Code, rec.Body)
+	}
+	rec = call(http.MethodPut, "/workflows/wf-test", "wf-test", "updated")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("update: %d %s", rec.Code, rec.Body)
+	}
+	rec = call(http.MethodPut, "/workflows/wf-test", "other-id", "renamed")
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("ID change accepted: %d %s", rec.Code, rec.Body)
+	}
+	// Existing IDs from older saves/imports can still be edited in place.
+	if err := db.UpsertWorkflowDefinition(&database.WorkflowDefinition{ID: "legacy.id", Name: "legacy", GraphJSON: bugSheetLinearGraph}); err != nil {
+		t.Fatal(err)
+	}
+	rec = call(http.MethodPut, "/workflows/legacy.id", "legacy.id", "updated legacy")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("legacy edit rejected: %d %s", rec.Code, rec.Body)
+	}
+}
+
+func TestWorkflowListAndDetailExposeValidationFailure(t *testing.T) {
+	db := workflowBugDB(t)
+	for _, wf := range []*database.WorkflowDefinition{
+		{ID: "valid", Name: "valid", GraphJSON: bugSheetLinearGraph, Enabled: true},
+		{ID: "invalid", Name: "invalid", GraphJSON: `{"nodes":[],"edges":[]}`, Enabled: true},
+	} {
+		if err := db.UpsertWorkflowDefinition(wf); err != nil {
+			t.Fatal(err)
+		}
+	}
+	h := NewWorkflowHandler(db, zap.NewNop())
+	r := gin.New()
+	r.GET("/workflows", h.List)
+	r.GET("/workflows/:id", h.Get)
+	for _, url := range []string{"/workflows", "/workflows/invalid"} {
+		rec := httptest.NewRecorder()
+		r.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, url, nil))
+		if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), `"validation_error"`) {
+			t.Fatalf("validation missing: %d %s", rec.Code, rec.Body)
+		}
+	}
+	rec := httptest.NewRecorder()
+	r.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/workflows/valid", nil))
+	if strings.Contains(rec.Body.String(), `"validation_error"`) {
+		t.Fatalf("valid graph mislabeled: %s", rec.Body)
+	}
+}
+
+func TestWorkflowDeliveryHandlesTextOnlyAndRejectedRuns(t *testing.T) {
+	db := workflowBugDB(t)
+	conv, err := db.CreateConversation("workflow", database.ConversationCreateMeta{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	message, err := db.AddMessage(conv.ID, "assistant", "pending", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	h := &AgentHandler{db: db, logger: zap.NewNop()}
+	wf := &database.WorkflowDefinition{ID: "wf-delivery", Name: "text", GraphJSON: bugSheetLinearGraph, Enabled: true}
+	if err := db.UpsertWorkflowDefinition(wf); err != nil {
+		t.Fatal(err)
+	}
+	args := workflowrunner.RunArgs{DB: db, Logger: zap.NewNop(), Role: config.RoleConfig{Name: "test", Enabled: true, WorkflowID: wf.ID, WorkflowPolicy: "auto"}, UserMessage: "OK", ConversationID: conv.ID}
+	result, err := workflowrunner.RunRoleBoundWorkflow(context.Background(), args)
+	if err != nil {
+		t.Fatal(err)
+	}
+	decision := h.finalizeWorkflowRunForDelivery(conv.ID, message.ID, result)
+	if !decision.Finalized || decision.Status != agentfinalizer.StatusCompleted {
+		t.Fatalf("text-only run rejected: %+v", decision)
+	}
+	// A real approval node pauses, then rejection remains cancellation rather than
+	// becoming a successful answer or a missing-evidence error.
+	wf.GraphJSON = `{"nodes":[{"id":"start-1","type":"start","config":{}},{"id":"review","type":"hitl","config":{"prompt":"approve?"}},{"id":"out","type":"output","config":{"output_key":"result"}}],"edges":[{"source":"start-1","target":"review"},{"source":"review","target":"out"}],"config":{"schema_version":1}}`
+	if err := db.UpsertWorkflowDefinition(wf); err != nil {
+		t.Fatal(err)
+	}
+	workflowrunner.InvalidateCompiledCache(wf.ID)
+	paused, err := workflowrunner.RunRoleBoundWorkflow(context.Background(), args)
+	if err != nil {
+		t.Fatal(err)
+	}
+	decision = h.finalizeWorkflowRunForDelivery(conv.ID, message.ID, paused)
+	if decision.Status != agentfinalizer.StatusAwaitingHITL || decision.Finalized {
+		t.Fatalf("pause finalized: %+v", decision)
+	}
+	rejected, err := workflowrunner.ResumeWorkflowRun(context.Background(), args, paused.RunID, false, "test refusal")
+	if err != nil {
+		t.Fatal(err)
+	}
+	decision = h.finalizeWorkflowRunForDelivery(conv.ID, message.ID, rejected)
+	if decision.Status != agentfinalizer.StatusCancelled || decision.CompletionReason != "workflow_rejected" || decision.Finalized {
+		t.Fatalf("rejection status lost: %+v", decision)
+	}
+	var text string
+	if err := db.QueryRow("SELECT content FROM messages WHERE id=?", message.ID).Scan(&text); err != nil {
+		t.Fatal(err)
+	}
+	if text != rejected.Response || strings.Contains(text, "missing_execution_evidence") {
+		t.Fatalf("rejection reason lost: %s", text)
+	}
+}
+
+func TestWorkflowToolReceivesExpandedParameters(t *testing.T) {
+	db := workflowBugDB(t)
+	var received string
+	server := mcp.NewServerWithStorage(zap.NewNop(), db)
+	server.RegisterTool(mcp.Tool{Name: "test_echo", InputSchema: map[string]interface{}{"type": "object"}}, func(ctx context.Context, args map[string]interface{}) (*mcp.ToolResult, error) {
+		received, _ = args["target"].(string)
+		return &mcp.ToolResult{Content: []mcp.Content{{Type: "text", Text: "received " + received}}}, nil
+	})
+	agent := agentpkg.NewAgent(&config.OpenAIConfig{}, &config.AgentConfig{}, server, nil, zap.NewNop(), 10)
+	graph := `{"nodes":[{"id":"start-1","type":"start","config":{}},{"id":"tool-1","type":"tool","config":{"tool_name":"test_echo","arguments":"{\"target\":\"{{inputs.message}}\"}"}},{"id":"out-1","type":"output","config":{"output_key":"result"}}],"edges":[{"source":"start-1","target":"tool-1"},{"source":"tool-1","target":"out-1"}],"config":{"schema_version":1}}`
+	wf := &database.WorkflowDefinition{ID: "wf-params", Name: "parameters", GraphJSON: graph, Enabled: true}
+	if err := db.UpsertWorkflowDefinition(wf); err != nil {
+		t.Fatal(err)
+	}
+	message := "test-only-target \"quoted\""
+	result, err := workflowrunner.RunRoleBoundWorkflow(context.Background(), workflowrunner.RunArgs{DB: db, Agent: agent, AppCfg: &config.Config{}, Logger: zap.NewNop(), Role: config.RoleConfig{Name: "test", Enabled: true, WorkflowID: wf.ID, WorkflowPolicy: "auto"}, UserMessage: message})
+	if err != nil || result.Status != "completed" {
+		t.Fatalf("workflow failed: %#v %v", result, err)
+	}
+	if received != message {
+		t.Fatalf("tool received %q, want %q", received, message)
+	}
+}

internal/handler/workflow_integration.go

@@ -7,6 +7,7 @@ import (
 	"strings"
 	"time"
 
+	"cyberstrike-ai/internal/agentfinalizer"
 	"cyberstrike-ai/internal/config"
 	workflowrunner "cyberstrike-ai/internal/workflow"
 
@@ -14,6 +15,25 @@ import (
 	"go.uber.org/zap"
 )
 
+func (h *AgentHandler) finalizeWorkflowRunForDelivery(conversationID, messageID string, result *workflowrunner.RunResult) agentfinalizer.Decision {
+	in := agentfinalizer.Input{ConversationID: conversationID, AssistantMessageID: messageID, AgentMode: "workflow"}
+	if result == nil {
+		in.Status = agentfinalizer.StatusFailed
+		in.Response = "工作流未返回执行结果。"
+	} else {
+		in.Response = result.Response
+		in.AwaitingHITL = result.AwaitingHITL
+		in.Status = result.Status
+		if result.Status == "rejected" {
+			in.Status = agentfinalizer.StatusCancelled
+			in.CompletionReason = "workflow_rejected"
+		}
+	}
+	decision := agentfinalizer.Decide(h.db, in)
+	h.persistFinalizationDecision(conversationID, messageID, "workflow", nil, "", decision)
+	return decision
+}
+
 func (h *AgentHandler) roleForWorkflow(req *ChatRequest) (config.RoleConfig, bool) {
 	if h == nil || h.config == nil || h.config.Roles == nil || req == nil {
 		return config.RoleConfig{}, false
@@ -158,16 +178,7 @@ func (h *AgentHandler) runRoleWorkflowStreamIfBound(
 		sendEvent("done", "", map[string]interface{}{"conversationId": conversationID})
 		return true
 	}
-	decision := h.finalizeCandidateForDeliveryWithPolicy(
-		prep.ConversationID,
-		prep.AssistantMessageID,
-		"workflow",
-		result.Response,
-		nil,
-		result.AwaitingHITL,
-		"",
-		true,
-	)
+	decision := h.finalizeWorkflowRunForDelivery(prep.ConversationID, prep.AssistantMessageID, result)
 	responseText := decision.FinalText
 	if !decision.Finalizable {
 		responseText = finalizationBlockedMessage(decision)
@@ -280,16 +291,7 @@ func (h *AgentHandler) runRoleWorkflowJSONIfBound(c *gin.Context, req *ChatReque
 		respond(http.StatusInternalServerError, gin.H{"error": errMsg, "conversationId": conversationID})
 		return true
 	}
-	decision := h.finalizeCandidateForDeliveryWithPolicy(
-		prep.ConversationID,
-		prep.AssistantMessageID,
-		"workflow",
-		result.Response,
-		nil,
-		result.AwaitingHITL,
-		"",
-		true,
-	)
+	decision := h.finalizeWorkflowRunForDelivery(prep.ConversationID, prep.AssistantMessageID, result)
 	responseText := decision.FinalText
 	if !decision.Finalizable {
 		responseText = finalizationBlockedMessage(decision)