refactor: make AI workflows agent-centric

2026-06-22 10:08:51 +08:00
parent 14f50e064f
commit 32ab43e7c2
18 changed files with 436 additions and 400 deletions
@@ -20,6 +20,7 @@ import (
 	"agent-desk/internal/repositories"

 	"github.com/mlogclub/simple/sqls"
+	"gorm.io/gorm"
 )

 var AIWorkflowService = newAIWorkflowService()
@@ -56,6 +57,49 @@ func (s *aiWorkflowService) FindVersionPageByParams(params *params.QueryParams)
 	return repositories.AIWorkflowVersionRepository.FindPageByParams(sqls.DB(), params)
 }

+func (s *aiWorkflowService) GetByAgentID(agentID int64) *models.AIWorkflow {
+	if agentID <= 0 {
+		return nil
+	}
+	return repositories.AIWorkflowRepository.Take(sqls.DB(), "agent_id = ? AND status <> ?", agentID, enums.StatusDeleted)
+}
+
+func (s *aiWorkflowService) GetOrCreateAgentWorkflow(agentID int64, operator *dto.AuthPrincipal) (*models.AIWorkflow, error) {
+	if operator == nil {
+		return nil, errorsx.UnauthorizedI18n("error.auth.expired")
+	}
+	if agentID <= 0 {
+		return nil, errorsx.InvalidParam("agent id is required")
+	}
+	if agent := AIAgentService.Get(agentID); agent == nil || agent.Status == enums.StatusDeleted {
+		return nil, errorsx.InvalidParamI18n("error.e0002")
+	}
+	if item := s.GetByAgentID(agentID); item != nil {
+		return item, nil
+	}
+	var item *models.AIWorkflow
+	err := sqls.WithTransaction(func(ctx *sqls.TxContext) error {
+		if current := repositories.AIWorkflowRepository.Take(ctx.Tx, "agent_id = ? AND status <> ?", agentID, enums.StatusDeleted); current != nil {
+			item = current
+			return nil
+		}
+		agent := repositories.AIAgentRepository.Get(ctx.Tx, agentID)
+		if agent == nil || agent.Status == enums.StatusDeleted {
+			return errorsx.InvalidParamI18n("error.e0002")
+		}
+		created, err := s.createDefaultAgentWorkflow(ctx.Tx, agent, operator)
+		if err != nil {
+			return err
+		}
+		item = created
+		return nil
+	})
+	if err != nil {
+		return nil, err
+	}
+	return item, nil
+}
+
 func (s *aiWorkflowService) ListNodeSpecs() []workflowregistry.NodeSpec {
 	return s.registry.List()
 }
@@ -65,37 +109,52 @@ func (s *aiWorkflowService) ValidateDefinition(def dsl.Definition) workflowvalid
 }

 func (s *aiWorkflowService) CreateWorkflow(req request.CreateAIWorkflowRequest, operator *dto.AuthPrincipal) (*models.AIWorkflow, error) {
+	return s.SaveAgentWorkflow(req, operator)
+}
+
+func (s *aiWorkflowService) SaveAgentWorkflow(req request.SaveAIWorkflowRequest, operator *dto.AuthPrincipal) (*models.AIWorkflow, error) {
 	if operator == nil {
 		return nil, errorsx.UnauthorizedI18n("error.auth.expired")
 	}
+	agent := AIAgentService.Get(req.AgentID)
+	if agent == nil || agent.Status == enums.StatusDeleted {
+		return nil, errorsx.InvalidParamI18n("error.e0002")
+	}
 	name := strings.TrimSpace(req.Name)
 	if name == "" {
-		return nil, errorsx.InvalidParam("workflow name is required")
-	}
-	ownerType := normalizeWorkflowOwnerType(req.OwnerType)
-	if ownerType == "" {
-		return nil, errorsx.InvalidParam("workflow owner type is required")
-	}
-	if req.OwnerID <= 0 {
-		return nil, errorsx.InvalidParam("workflow owner id is required")
+		name = defaultAgentWorkflowName(agent.Name)
 	}
 	definition, err := marshalDefinition(req.Definition)
 	if err != nil {
 		return nil, err
 	}
-	item := &models.AIWorkflow{
-		Name:            name,
-		Description:     strings.TrimSpace(req.Description),
-		OwnerType:       ownerType,
-		OwnerID:         req.OwnerID,
-		Status:          enums.StatusOk,
-		DraftDefinition: definition,
-		AuditFields:     utils.BuildAuditFields(operator),
+	current := s.GetByAgentID(req.AgentID)
+	if current == nil {
+		item := &models.AIWorkflow{
+			Name:            name,
+			Description:     strings.TrimSpace(req.Description),
+			AgentID:         req.AgentID,
+			Status:          enums.StatusOk,
+			DraftDefinition: definition,
+			AuditFields:     utils.BuildAuditFields(operator),
+		}
+		if err := repositories.AIWorkflowRepository.Create(sqls.DB(), item); err != nil {
+			return nil, err
+		}
+		return item, nil
 	}
-	if err := repositories.AIWorkflowRepository.Create(sqls.DB(), item); err != nil {
+	if err := repositories.AIWorkflowRepository.Updates(sqls.DB(), current.ID, map[string]interface{}{
+		"name":             name,
+		"description":      strings.TrimSpace(req.Description),
+		"agent_id":         req.AgentID,
+		"draft_definition": definition,
+		"update_user_id":   operator.UserID,
+		"update_user_name": operator.Username,
+		"updated_at":       time.Now(),
+	}); err != nil {
 		return nil, err
 	}
-	return item, nil
+	return s.Get(current.ID), nil
 }

 func (s *aiWorkflowService) UpdateWorkflow(req request.UpdateAIWorkflowRequest, operator *dto.AuthPrincipal) error {
@@ -109,12 +168,8 @@ func (s *aiWorkflowService) UpdateWorkflow(req request.UpdateAIWorkflowRequest,
 	if name == "" {
 		return errorsx.InvalidParam("workflow name is required")
 	}
-	ownerType := normalizeWorkflowOwnerType(req.OwnerType)
-	if ownerType == "" {
-		return errorsx.InvalidParam("workflow owner type is required")
-	}
-	if req.OwnerID <= 0 {
-		return errorsx.InvalidParam("workflow owner id is required")
+	if req.AgentID <= 0 {
+		return errorsx.InvalidParam("agent id is required")
 	}
 	definition, err := marshalDefinition(req.Definition)
 	if err != nil {
@@ -123,8 +178,7 @@ func (s *aiWorkflowService) UpdateWorkflow(req request.UpdateAIWorkflowRequest,
 	return repositories.AIWorkflowRepository.Updates(sqls.DB(), req.ID, map[string]interface{}{
 		"name":             name,
 		"description":      strings.TrimSpace(req.Description),
-		"owner_type":       ownerType,
-		"owner_id":         req.OwnerID,
+		"agent_id":         req.AgentID,
 		"draft_definition": definition,
 		"update_user_id":   operator.UserID,
 		"update_user_name": operator.Username,
@@ -148,6 +202,9 @@ func (s *aiWorkflowService) DeleteWorkflow(id int64, operator *dto.AuthPrincipal
 }

 func (s *aiWorkflowService) PublishWorkflow(req request.PublishAIWorkflowRequest, operator *dto.AuthPrincipal) (*models.AIWorkflowVersion, error) {
+	if req.AgentID > 0 {
+		return s.PublishAgentWorkflow(req, operator)
+	}
 	if operator == nil {
 		return nil, errorsx.UnauthorizedI18n("error.auth.expired")
 	}
@@ -195,6 +252,106 @@ func (s *aiWorkflowService) PublishWorkflow(req request.PublishAIWorkflowRequest
 	return version, nil
 }

+func (s *aiWorkflowService) PublishAgentWorkflow(req request.PublishAIWorkflowRequest, operator *dto.AuthPrincipal) (*models.AIWorkflowVersion, error) {
+	if operator == nil {
+		return nil, errorsx.UnauthorizedI18n("error.auth.expired")
+	}
+	workflow, err := s.GetOrCreateAgentWorkflow(req.AgentID, operator)
+	if err != nil {
+		return nil, err
+	}
+	req.WorkflowID = workflow.ID
+	result := s.ValidateDefinition(req.Definition)
+	if !result.Valid {
+		return nil, errorsx.InvalidParam("workflow definition is invalid")
+	}
+	definition, err := marshalDefinition(req.Definition)
+	if err != nil {
+		return nil, err
+	}
+	now := time.Now()
+	var version *models.AIWorkflowVersion
+	err = sqls.WithTransaction(func(ctx *sqls.TxContext) error {
+		current := repositories.AIWorkflowRepository.Get(ctx.Tx, workflow.ID)
+		if current == nil || current.AgentID != req.AgentID || current.Status == enums.StatusDeleted {
+			return errorsx.InvalidParamI18n("error.e0002")
+		}
+		nextVersion := repositories.AIWorkflowVersionRepository.MaxVersionByWorkflowID(ctx.Tx, current.ID) + 1
+		version = &models.AIWorkflowVersion{
+			WorkflowID:      current.ID,
+			Version:         nextVersion,
+			Status:          enums.StatusOk,
+			Definition:      definition,
+			DefinitionHash:  hashDefinition(definition),
+			PublishedAt:     &now,
+			PublishedByID:   operator.UserID,
+			PublishedByName: operator.Username,
+			AuditFields:     utils.BuildAuditFields(operator),
+		}
+		if err := repositories.AIWorkflowVersionRepository.Create(ctx.Tx, version); err != nil {
+			return err
+		}
+		if err := repositories.AIWorkflowRepository.Updates(ctx.Tx, current.ID, map[string]interface{}{
+			"draft_definition":     definition,
+			"published_version_id": version.ID,
+			"update_user_id":       operator.UserID,
+			"update_user_name":     operator.Username,
+			"updated_at":           now,
+		}); err != nil {
+			return err
+		}
+		return repositories.AIAgentRepository.Updates(ctx.Tx, req.AgentID, map[string]any{
+			"runtime_mode":        enums.AIAgentRuntimeModeWorkflow,
+			"workflow_version_id": version.ID,
+			"update_user_id":      operator.UserID,
+			"update_user_name":    operator.Username,
+			"updated_at":          now,
+		})
+	})
+	if err != nil {
+		return nil, err
+	}
+	return version, nil
+}
+
+func (s *aiWorkflowService) createDefaultAgentWorkflow(db *gorm.DB, agent *models.AIAgent, operator *dto.AuthPrincipal) (*models.AIWorkflow, error) {
+	definition, err := marshalDefinition(defaultAgentWorkflowDefinition())
+	if err != nil {
+		return nil, err
+	}
+	item := &models.AIWorkflow{
+		Name:            defaultAgentWorkflowName(agent.Name),
+		AgentID:         agent.ID,
+		Status:          enums.StatusOk,
+		DraftDefinition: definition,
+		AuditFields:     utils.BuildAuditFields(operator),
+	}
+	if err := repositories.AIWorkflowRepository.Create(db, item); err != nil {
+		return nil, err
+	}
+	return item, nil
+}
+
+func defaultAgentWorkflowDefinition() dsl.Definition {
+	return dsl.Definition{
+		SchemaVersion: 1,
+		EntryNodeID:   "start_1",
+		Nodes: []dsl.Node{
+			{ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start", Position: dsl.Position{X: 0, Y: 80}},
+			{ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End", Position: dsl.Position{X: 360, Y: 80}},
+		},
+		Edges: []dsl.Edge{{ID: "edge_start_end", Source: "start_1", Target: "end_1"}},
+	}
+}
+
+func defaultAgentWorkflowName(agentName string) string {
+	agentName = strings.TrimSpace(agentName)
+	if agentName == "" {
+		return "会话流程"
+	}
+	return agentName + " 会话流程"
+}
+
 func marshalDefinition(def dsl.Definition) (string, error) {
 	buf, err := json.Marshal(def)
 	if err != nil {
@@ -207,13 +364,3 @@ func hashDefinition(definition string) string {
 	sum := sha256.Sum256([]byte(definition))
 	return hex.EncodeToString(sum[:])
 }
-
-func normalizeWorkflowOwnerType(ownerType string) string {
-	ownerType = strings.TrimSpace(ownerType)
-	switch ownerType {
-	case "ai_agent", "workspace":
-		return ownerType
-	default:
-		return ""
-	}
-}