From 0127455be5d3b86c997394ddecb682fc8e622720 Mon Sep 17 00:00:00 2001 From: mlogclub Date: Mon, 22 Jun 2026 00:11:26 +0800 Subject: [PATCH] feat: add AI workflow service foundation --- internal/builders/ai_workflow_builder.go | 94 ++++++++ internal/models/models.go | 64 +++++ .../pkg/dto/request/ai_workflow_request.go | 33 +++ .../pkg/dto/response/ai_workflow_response.go | 52 +++++ .../ai_workflow_node_run_repository.go | 37 +++ .../repositories/ai_workflow_repository.go | 61 +++++ .../ai_workflow_run_repository.go | 37 +++ .../ai_workflow_version_repository.go | 58 +++++ internal/services/ai_workflow_service.go | 219 ++++++++++++++++++ internal/services/ai_workflow_service_test.go | 152 ++++++++++++ 10 files changed, 807 insertions(+) create mode 100644 internal/builders/ai_workflow_builder.go create mode 100644 internal/pkg/dto/request/ai_workflow_request.go create mode 100644 internal/pkg/dto/response/ai_workflow_response.go create mode 100644 internal/repositories/ai_workflow_node_run_repository.go create mode 100644 internal/repositories/ai_workflow_repository.go create mode 100644 internal/repositories/ai_workflow_run_repository.go create mode 100644 internal/repositories/ai_workflow_version_repository.go create mode 100644 internal/services/ai_workflow_service.go create mode 100644 internal/services/ai_workflow_service_test.go diff --git a/internal/builders/ai_workflow_builder.go b/internal/builders/ai_workflow_builder.go new file mode 100644 index 0000000..6a78dd7 --- /dev/null +++ b/internal/builders/ai_workflow_builder.go @@ -0,0 +1,94 @@ +package builders + +import ( + "encoding/json" + + "agent-desk/internal/ai/workflow/dsl" + workflowregistry "agent-desk/internal/ai/workflow/registry" + "agent-desk/internal/models" + "agent-desk/internal/pkg/dto/response" +) + +func BuildAIWorkflow(item *models.AIWorkflow) response.AIWorkflowResponse { + if item == nil { + return response.AIWorkflowResponse{} + } + return response.AIWorkflowResponse{ + ID: item.ID, + Name: item.Name, + Description: item.Description, + OwnerType: item.OwnerType, + OwnerID: item.OwnerID, + Status: item.Status, + DraftDefinition: parseWorkflowDefinition(item.DraftDefinition), + PublishedVersionID: item.PublishedVersionID, + SortNo: item.SortNo, + CreatedAt: item.CreatedAt.Format("2006-01-02 15:04:05"), + UpdatedAt: item.UpdatedAt.Format("2006-01-02 15:04:05"), + CreateUserName: item.CreateUserName, + UpdateUserName: item.UpdateUserName, + } +} + +func BuildAIWorkflowList(list []models.AIWorkflow) []response.AIWorkflowResponse { + ret := make([]response.AIWorkflowResponse, 0, len(list)) + for i := range list { + ret = append(ret, BuildAIWorkflow(&list[i])) + } + return ret +} + +func BuildAIWorkflowVersion(item *models.AIWorkflowVersion) response.AIWorkflowVersionResponse { + if item == nil { + return response.AIWorkflowVersionResponse{} + } + publishedAt := "" + if item.PublishedAt != nil { + publishedAt = item.PublishedAt.Format("2006-01-02 15:04:05") + } + return response.AIWorkflowVersionResponse{ + ID: item.ID, + WorkflowID: item.WorkflowID, + Version: item.Version, + Status: item.Status, + Definition: parseWorkflowDefinition(item.Definition), + DefinitionHash: item.DefinitionHash, + PublishedAt: publishedAt, + PublishedByID: item.PublishedByID, + PublishedByName: item.PublishedByName, + CreatedAt: item.CreatedAt.Format("2006-01-02 15:04:05"), + UpdatedAt: item.UpdatedAt.Format("2006-01-02 15:04:05"), + } +} + +func BuildAIWorkflowVersionList(list []models.AIWorkflowVersion) []response.AIWorkflowVersionResponse { + ret := make([]response.AIWorkflowVersionResponse, 0, len(list)) + for i := range list { + ret = append(ret, BuildAIWorkflowVersion(&list[i])) + } + return ret +} + +func BuildAIWorkflowNodeSpecs(list []workflowregistry.NodeSpec) []response.AIWorkflowNodeSpecResponse { + ret := make([]response.AIWorkflowNodeSpecResponse, 0, len(list)) + for _, item := range list { + ret = append(ret, response.AIWorkflowNodeSpecResponse{ + Type: item.Type, + Title: item.Title, + Description: item.Description, + RiskLevel: item.RiskLevel, + Interruptible: item.Interruptible, + RequiresConfirmationPredecessor: item.RequiresConfirmationPredecessor, + }) + } + return ret +} + +func parseWorkflowDefinition(raw string) dsl.Definition { + var ret dsl.Definition + if raw == "" { + return ret + } + _ = json.Unmarshal([]byte(raw), &ret) + return ret +} diff --git a/internal/models/models.go b/internal/models/models.go index fe9bcd8..ccefa63 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -58,6 +58,10 @@ var Models = []any{ &SkillDefinition{}, &SkillRunLog{}, &AgentRunLog{}, + &AIWorkflow{}, + &AIWorkflowVersion{}, + &AIWorkflowRun{}, + &AIWorkflowNodeRun{}, &ConversationInterrupt{}, &SystemConfig{}, } @@ -527,6 +531,66 @@ type AIAgent struct { AuditFields } +// AIWorkflow 表示客服 AI Agent 可编辑会话流程主表。 +type AIWorkflow struct { + ID int64 `gorm:"primaryKey;autoIncrement"` + Name string `gorm:"type:varchar(100);not null;default:'';index"` + Description string `gorm:"type:text"` + OwnerType string `gorm:"type:varchar(30);not null;default:'';index"` + OwnerID int64 `gorm:"type:bigint;not null;default:0;index"` + Status enums.Status `gorm:"type:int;not null;default:0;index"` + DraftDefinition string `gorm:"type:longtext"` + PublishedVersionID int64 `gorm:"type:bigint;not null;default:0;index"` + SortNo int `gorm:"type:int;not null;default:0;index"` + AuditFields +} + +// AIWorkflowVersion 表示 AI 会话流程的不可变发布版本。 +type AIWorkflowVersion struct { + ID int64 `gorm:"primaryKey;autoIncrement"` + WorkflowID int64 `gorm:"type:bigint;not null;default:0;index"` + Version int `gorm:"type:int;not null;default:0;index"` + Status enums.Status `gorm:"type:int;not null;default:0;index"` + Definition string `gorm:"type:longtext"` + DefinitionHash string `gorm:"type:varchar(64);not null;default:'';index"` + PublishedAt *time.Time `gorm:"type:datetime;index"` + PublishedByID int64 `gorm:"type:bigint;not null;default:0;index"` + PublishedByName string `gorm:"type:varchar(100);not null;default:''"` + AuditFields +} + +// AIWorkflowRun 表示一次会话 workflow 执行记录。 +type AIWorkflowRun struct { + ID int64 `gorm:"primaryKey;autoIncrement"` + WorkflowID int64 `gorm:"type:bigint;not null;default:0;index"` + WorkflowVersionID int64 `gorm:"type:bigint;not null;default:0;index"` + ConversationID int64 `gorm:"type:bigint;not null;default:0;index"` + AIAgentID int64 `gorm:"type:bigint;not null;default:0;index"` + MessageID int64 `gorm:"type:bigint;not null;default:0;index"` + Status int `gorm:"type:int;not null;default:0;index"` + StartedAt time.Time `gorm:"type:datetime;not null;index"` + EndedAt *time.Time `gorm:"type:datetime;index"` + InterruptType string `gorm:"type:varchar(50);not null;default:'';index"` + InterruptNodeID string `gorm:"type:varchar(100);not null;default:'';index"` + ErrorMessage string `gorm:"type:text"` + AuditFields +} + +// AIWorkflowNodeRun 表示 workflow 执行中的单节点审计记录。 +type AIWorkflowNodeRun struct { + ID int64 `gorm:"primaryKey;autoIncrement"` + WorkflowRunID int64 `gorm:"type:bigint;not null;default:0;index"` + NodeID string `gorm:"type:varchar(100);not null;default:'';index"` + NodeType string `gorm:"type:varchar(50);not null;default:'';index"` + Status int `gorm:"type:int;not null;default:0;index"` + InputPreview string `gorm:"type:text"` + OutputPreview string `gorm:"type:text"` + ErrorMessage string `gorm:"type:text"` + StartedAt time.Time `gorm:"type:datetime;not null;index"` + EndedAt *time.Time `gorm:"type:datetime;index"` + DurationMS int `gorm:"type:int;not null;default:0"` +} + // Channel 接入渠道配置。 // // 用于统一描述系统的外部接入入口。不同渠道类型共享统一的接入配置骨架， diff --git a/internal/pkg/dto/request/ai_workflow_request.go b/internal/pkg/dto/request/ai_workflow_request.go new file mode 100644 index 0000000..3c4d8c6 --- /dev/null +++ b/internal/pkg/dto/request/ai_workflow_request.go @@ -0,0 +1,33 @@ +package request + +import "agent-desk/internal/ai/workflow/dsl" + +type CreateAIWorkflowRequest struct { + Name string `json:"name"` + Description string `json:"description"` + OwnerType string `json:"ownerType"` + OwnerID int64 `json:"ownerId"` + Definition dsl.Definition `json:"definition"` +} + +type UpdateAIWorkflowRequest struct { + ID int64 `json:"id"` + CreateAIWorkflowRequest +} + +type DeleteAIWorkflowRequest struct { + ID int64 `json:"id"` +} + +type ValidateAIWorkflowRequest struct { + Definition dsl.Definition `json:"definition"` +} + +type PublishAIWorkflowRequest struct { + WorkflowID int64 `json:"workflowId"` + Definition dsl.Definition `json:"definition"` +} + +type AIWorkflowVersionListRequest struct { + WorkflowID int64 `json:"workflowId"` +} diff --git a/internal/pkg/dto/response/ai_workflow_response.go b/internal/pkg/dto/response/ai_workflow_response.go new file mode 100644 index 0000000..d4077f9 --- /dev/null +++ b/internal/pkg/dto/response/ai_workflow_response.go @@ -0,0 +1,52 @@ +package response + +import ( + "agent-desk/internal/ai/workflow/dsl" + workflowregistry "agent-desk/internal/ai/workflow/registry" + workflowvalidator "agent-desk/internal/ai/workflow/validator" + "agent-desk/internal/pkg/enums" +) + +type AIWorkflowResponse struct { + ID int64 `json:"id"` + Name string `json:"name"` + Description string `json:"description"` + OwnerType string `json:"ownerType"` + OwnerID int64 `json:"ownerId"` + Status enums.Status `json:"status"` + DraftDefinition dsl.Definition `json:"draftDefinition"` + PublishedVersionID int64 `json:"publishedVersionId"` + SortNo int `json:"sortNo"` + CreatedAt string `json:"createdAt"` + UpdatedAt string `json:"updatedAt"` + CreateUserName string `json:"createUserName"` + UpdateUserName string `json:"updateUserName"` +} + +type AIWorkflowVersionResponse struct { + ID int64 `json:"id"` + WorkflowID int64 `json:"workflowId"` + Version int `json:"version"` + Status enums.Status `json:"status"` + Definition dsl.Definition `json:"definition"` + DefinitionHash string `json:"definitionHash"` + PublishedAt string `json:"publishedAt"` + PublishedByID int64 `json:"publishedById"` + PublishedByName string `json:"publishedByName"` + CreatedAt string `json:"createdAt"` + UpdatedAt string `json:"updatedAt"` +} + +type AIWorkflowValidationResponse struct { + Valid bool `json:"valid"` + Errors []workflowvalidator.Error `json:"errors"` +} + +type AIWorkflowNodeSpecResponse struct { + Type string `json:"type"` + Title string `json:"title"` + Description string `json:"description"` + RiskLevel workflowregistry.NodeRiskLevel `json:"riskLevel"` + Interruptible bool `json:"interruptible"` + RequiresConfirmationPredecessor bool `json:"requiresConfirmationPredecessor"` +} diff --git a/internal/repositories/ai_workflow_node_run_repository.go b/internal/repositories/ai_workflow_node_run_repository.go new file mode 100644 index 0000000..f63c4b5 --- /dev/null +++ b/internal/repositories/ai_workflow_node_run_repository.go @@ -0,0 +1,37 @@ +package repositories + +import ( + "agent-desk/internal/models" + + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" +) + +var AIWorkflowNodeRunRepository = newAIWorkflowNodeRunRepository() + +func newAIWorkflowNodeRunRepository() *aiWorkflowNodeRunRepository { + return &aiWorkflowNodeRunRepository{} +} + +type aiWorkflowNodeRunRepository struct{} + +func (r *aiWorkflowNodeRunRepository) Get(db *gorm.DB, id int64) *models.AIWorkflowNodeRun { + ret := &models.AIWorkflowNodeRun{} + if err := db.First(ret, "id = ?", id).Error; err != nil { + return nil + } + return ret +} + +func (r *aiWorkflowNodeRunRepository) Find(db *gorm.DB, cnd *sqls.Cnd) (list []models.AIWorkflowNodeRun) { + cnd.Find(db, &list) + return +} + +func (r *aiWorkflowNodeRunRepository) Create(db *gorm.DB, t *models.AIWorkflowNodeRun) error { + return db.Create(t).Error +} + +func (r *aiWorkflowNodeRunRepository) Updates(db *gorm.DB, id int64, columns map[string]interface{}) error { + return db.Model(&models.AIWorkflowNodeRun{}).Where("id = ?", id).Updates(columns).Error +} diff --git a/internal/repositories/ai_workflow_repository.go b/internal/repositories/ai_workflow_repository.go new file mode 100644 index 0000000..877a87c --- /dev/null +++ b/internal/repositories/ai_workflow_repository.go @@ -0,0 +1,61 @@ +package repositories + +import ( + "agent-desk/internal/models" + "agent-desk/internal/pkg/httpx/params" + + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" +) + +var AIWorkflowRepository = newAIWorkflowRepository() + +func newAIWorkflowRepository() *aiWorkflowRepository { + return &aiWorkflowRepository{} +} + +type aiWorkflowRepository struct{} + +func (r *aiWorkflowRepository) Get(db *gorm.DB, id int64) *models.AIWorkflow { + ret := &models.AIWorkflow{} + if err := db.First(ret, "id = ?", id).Error; err != nil { + return nil + } + return ret +} + +func (r *aiWorkflowRepository) Take(db *gorm.DB, where ...interface{}) *models.AIWorkflow { + ret := &models.AIWorkflow{} + if err := db.Take(ret, where...).Error; err != nil { + return nil + } + return ret +} + +func (r *aiWorkflowRepository) Find(db *gorm.DB, cnd *sqls.Cnd) (list []models.AIWorkflow) { + cnd.Find(db, &list) + return +} + +func (r *aiWorkflowRepository) FindPageByParams(db *gorm.DB, params *params.QueryParams) (list []models.AIWorkflow, paging *sqls.Paging) { + return r.FindPageByCnd(db, ¶ms.Cnd) +} + +func (r *aiWorkflowRepository) FindPageByCnd(db *gorm.DB, cnd *sqls.Cnd) (list []models.AIWorkflow, paging *sqls.Paging) { + cnd.Find(db, &list) + count := cnd.Count(db, &models.AIWorkflow{}) + paging = &sqls.Paging{ + Page: cnd.Paging.Page, + Limit: cnd.Paging.Limit, + Total: count, + } + return +} + +func (r *aiWorkflowRepository) Create(db *gorm.DB, t *models.AIWorkflow) error { + return db.Create(t).Error +} + +func (r *aiWorkflowRepository) Updates(db *gorm.DB, id int64, columns map[string]interface{}) error { + return db.Model(&models.AIWorkflow{}).Where("id = ?", id).Updates(columns).Error +} diff --git a/internal/repositories/ai_workflow_run_repository.go b/internal/repositories/ai_workflow_run_repository.go new file mode 100644 index 0000000..1d8dc7f --- /dev/null +++ b/internal/repositories/ai_workflow_run_repository.go @@ -0,0 +1,37 @@ +package repositories + +import ( + "agent-desk/internal/models" + + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" +) + +var AIWorkflowRunRepository = newAIWorkflowRunRepository() + +func newAIWorkflowRunRepository() *aiWorkflowRunRepository { + return &aiWorkflowRunRepository{} +} + +type aiWorkflowRunRepository struct{} + +func (r *aiWorkflowRunRepository) Get(db *gorm.DB, id int64) *models.AIWorkflowRun { + ret := &models.AIWorkflowRun{} + if err := db.First(ret, "id = ?", id).Error; err != nil { + return nil + } + return ret +} + +func (r *aiWorkflowRunRepository) Find(db *gorm.DB, cnd *sqls.Cnd) (list []models.AIWorkflowRun) { + cnd.Find(db, &list) + return +} + +func (r *aiWorkflowRunRepository) Create(db *gorm.DB, t *models.AIWorkflowRun) error { + return db.Create(t).Error +} + +func (r *aiWorkflowRunRepository) Updates(db *gorm.DB, id int64, columns map[string]interface{}) error { + return db.Model(&models.AIWorkflowRun{}).Where("id = ?", id).Updates(columns).Error +} diff --git a/internal/repositories/ai_workflow_version_repository.go b/internal/repositories/ai_workflow_version_repository.go new file mode 100644 index 0000000..a997e4e --- /dev/null +++ b/internal/repositories/ai_workflow_version_repository.go @@ -0,0 +1,58 @@ +package repositories + +import ( + "agent-desk/internal/models" + "agent-desk/internal/pkg/httpx/params" + + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" +) + +var AIWorkflowVersionRepository = newAIWorkflowVersionRepository() + +func newAIWorkflowVersionRepository() *aiWorkflowVersionRepository { + return &aiWorkflowVersionRepository{} +} + +type aiWorkflowVersionRepository struct{} + +func (r *aiWorkflowVersionRepository) Get(db *gorm.DB, id int64) *models.AIWorkflowVersion { + ret := &models.AIWorkflowVersion{} + if err := db.First(ret, "id = ?", id).Error; err != nil { + return nil + } + return ret +} + +func (r *aiWorkflowVersionRepository) Find(db *gorm.DB, cnd *sqls.Cnd) (list []models.AIWorkflowVersion) { + cnd.Find(db, &list) + return +} + +func (r *aiWorkflowVersionRepository) FindPageByParams(db *gorm.DB, params *params.QueryParams) (list []models.AIWorkflowVersion, paging *sqls.Paging) { + return r.FindPageByCnd(db, ¶ms.Cnd) +} + +func (r *aiWorkflowVersionRepository) FindPageByCnd(db *gorm.DB, cnd *sqls.Cnd) (list []models.AIWorkflowVersion, paging *sqls.Paging) { + cnd.Find(db, &list) + count := cnd.Count(db, &models.AIWorkflowVersion{}) + paging = &sqls.Paging{ + Page: cnd.Paging.Page, + Limit: cnd.Paging.Limit, + Total: count, + } + return +} + +func (r *aiWorkflowVersionRepository) Create(db *gorm.DB, t *models.AIWorkflowVersion) error { + return db.Create(t).Error +} + +func (r *aiWorkflowVersionRepository) MaxVersionByWorkflowID(db *gorm.DB, workflowID int64) int { + var maxVersion int + db.Model(&models.AIWorkflowVersion{}). + Where("workflow_id = ?", workflowID). + Select("COALESCE(MAX(version), 0)"). + Scan(&maxVersion) + return maxVersion +} diff --git a/internal/services/ai_workflow_service.go b/internal/services/ai_workflow_service.go new file mode 100644 index 0000000..8f1f89c --- /dev/null +++ b/internal/services/ai_workflow_service.go @@ -0,0 +1,219 @@ +package services + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "strings" + "time" + + "agent-desk/internal/ai/workflow/dsl" + workflowregistry "agent-desk/internal/ai/workflow/registry" + workflowvalidator "agent-desk/internal/ai/workflow/validator" + "agent-desk/internal/models" + "agent-desk/internal/pkg/dto" + "agent-desk/internal/pkg/dto/request" + "agent-desk/internal/pkg/enums" + "agent-desk/internal/pkg/errorsx" + "agent-desk/internal/pkg/httpx/params" + "agent-desk/internal/pkg/utils" + "agent-desk/internal/repositories" + + "github.com/mlogclub/simple/sqls" +) + +var AIWorkflowService = newAIWorkflowService() + +func newAIWorkflowService() *aiWorkflowService { + return &aiWorkflowService{ + registry: workflowregistry.DefaultRegistry(), + } +} + +type aiWorkflowService struct { + registry *workflowregistry.Registry +} + +func (s *aiWorkflowService) Get(id int64) *models.AIWorkflow { + if id <= 0 { + return nil + } + return repositories.AIWorkflowRepository.Get(sqls.DB(), id) +} + +func (s *aiWorkflowService) GetVersion(id int64) *models.AIWorkflowVersion { + if id <= 0 { + return nil + } + return repositories.AIWorkflowVersionRepository.Get(sqls.DB(), id) +} + +func (s *aiWorkflowService) FindPageByCnd(cnd *sqls.Cnd) (list []models.AIWorkflow, paging *sqls.Paging) { + return repositories.AIWorkflowRepository.FindPageByCnd(sqls.DB(), cnd) +} + +func (s *aiWorkflowService) FindVersionPageByParams(params *params.QueryParams) (list []models.AIWorkflowVersion, paging *sqls.Paging) { + return repositories.AIWorkflowVersionRepository.FindPageByParams(sqls.DB(), params) +} + +func (s *aiWorkflowService) ListNodeSpecs() []workflowregistry.NodeSpec { + return s.registry.List() +} + +func (s *aiWorkflowService) ValidateDefinition(def dsl.Definition) workflowvalidator.Result { + return workflowvalidator.ValidateDefinition(def, s.registry) +} + +func (s *aiWorkflowService) CreateWorkflow(req request.CreateAIWorkflowRequest, operator *dto.AuthPrincipal) (*models.AIWorkflow, error) { + if operator == nil { + return nil, errorsx.UnauthorizedI18n("error.auth.expired") + } + 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") + } + 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), + } + if err := repositories.AIWorkflowRepository.Create(sqls.DB(), item); err != nil { + return nil, err + } + return item, nil +} + +func (s *aiWorkflowService) UpdateWorkflow(req request.UpdateAIWorkflowRequest, operator *dto.AuthPrincipal) error { + if operator == nil { + return errorsx.UnauthorizedI18n("error.auth.expired") + } + if s.Get(req.ID) == nil { + return errorsx.InvalidParamI18n("error.e0002") + } + name := strings.TrimSpace(req.Name) + 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") + } + definition, err := marshalDefinition(req.Definition) + if err != nil { + return err + } + 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, + "draft_definition": definition, + "update_user_id": operator.UserID, + "update_user_name": operator.Username, + "updated_at": time.Now(), + }) +} + +func (s *aiWorkflowService) DeleteWorkflow(id int64, operator *dto.AuthPrincipal) error { + if operator == nil { + return errorsx.UnauthorizedI18n("error.auth.expired") + } + if s.Get(id) == nil { + return errorsx.InvalidParamI18n("error.e0002") + } + return repositories.AIWorkflowRepository.Updates(sqls.DB(), id, map[string]interface{}{ + "status": enums.StatusDeleted, + "update_user_id": operator.UserID, + "update_user_name": operator.Username, + "updated_at": time.Now(), + }) +} + +func (s *aiWorkflowService) PublishWorkflow(req request.PublishAIWorkflowRequest, operator *dto.AuthPrincipal) (*models.AIWorkflowVersion, error) { + if operator == nil { + return nil, errorsx.UnauthorizedI18n("error.auth.expired") + } + workflow := s.Get(req.WorkflowID) + if workflow == nil || workflow.Status == enums.StatusDeleted { + return nil, errorsx.InvalidParamI18n("error.e0002") + } + 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 { + nextVersion := repositories.AIWorkflowVersionRepository.MaxVersionByWorkflowID(ctx.Tx, req.WorkflowID) + 1 + version = &models.AIWorkflowVersion{ + WorkflowID: req.WorkflowID, + 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 + } + return repositories.AIWorkflowRepository.Updates(ctx.Tx, req.WorkflowID, map[string]interface{}{ + "draft_definition": definition, + "published_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 marshalDefinition(def dsl.Definition) (string, error) { + buf, err := json.Marshal(def) + if err != nil { + return "", errorsx.InvalidParam("invalid workflow definition") + } + return string(buf), nil +} + +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 "" + } +} diff --git a/internal/services/ai_workflow_service_test.go b/internal/services/ai_workflow_service_test.go new file mode 100644 index 0000000..497466b --- /dev/null +++ b/internal/services/ai_workflow_service_test.go @@ -0,0 +1,152 @@ +package services + +import ( + "encoding/json" + "testing" + + "agent-desk/internal/ai/workflow/dsl" + "agent-desk/internal/models" + "agent-desk/internal/pkg/dto" + "agent-desk/internal/pkg/dto/request" + + "github.com/glebarez/sqlite" + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" +) + +func TestAIWorkflowServiceValidateDefinitionReportsErrors(t *testing.T) { + setupAIWorkflowTestDB(t) + result := AIWorkflowService.ValidateDefinition(dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + Nodes: []dsl.Node{ + {ID: "start_1", Type: "start"}, + {ID: "create_1", Type: "create_ticket"}, + {ID: "end_1", Type: "end"}, + }, + Edges: []dsl.Edge{ + {ID: "e1", Source: "start_1", Target: "create_1"}, + {ID: "e2", Source: "create_1", Target: "end_1"}, + }, + }) + + if result.Valid { + t.Fatalf("expected invalid workflow definition") + } + if len(result.Errors) == 0 { + t.Fatalf("expected validation errors") + } +} + +func TestAIWorkflowServicePublishCreatesImmutableVersion(t *testing.T) { + setupAIWorkflowTestDB(t) + operator := aiWorkflowTestOperator() + workflow, err := AIWorkflowService.CreateWorkflow(request.CreateAIWorkflowRequest{ + Name: "support flow", + Description: "customer service flow", + OwnerType: "ai_agent", + OwnerID: 12, + Definition: validAIWorkflowDefinition(), + }, operator) + if err != nil { + t.Fatalf("CreateWorkflow() error = %v", err) + } + + version, err := AIWorkflowService.PublishWorkflow(request.PublishAIWorkflowRequest{ + WorkflowID: workflow.ID, + Definition: validAIWorkflowDefinition(), + }, operator) + if err != nil { + t.Fatalf("PublishWorkflow() error = %v", err) + } + + if version.WorkflowID != workflow.ID { + t.Fatalf("expected workflow id %d, got %d", workflow.ID, version.WorkflowID) + } + if version.Version != 1 { + t.Fatalf("expected first version to be 1, got %d", version.Version) + } + if version.DefinitionHash == "" { + t.Fatalf("expected definition hash") + } + if version.PublishedAt == nil { + t.Fatalf("expected published timestamp") + } + + var stored dsl.Definition + if err := json.Unmarshal([]byte(version.Definition), &stored); err != nil { + t.Fatalf("unmarshal stored definition: %v", err) + } + if stored.EntryNodeID != "start_1" { + t.Fatalf("unexpected stored definition: %+v", stored) + } +} + +func TestAIWorkflowServicePublishIncrementsVersion(t *testing.T) { + setupAIWorkflowTestDB(t) + operator := aiWorkflowTestOperator() + workflow, err := AIWorkflowService.CreateWorkflow(request.CreateAIWorkflowRequest{ + Name: "support flow versions", + OwnerType: "ai_agent", + OwnerID: 99, + Definition: validAIWorkflowDefinition(), + }, operator) + if err != nil { + t.Fatalf("CreateWorkflow() error = %v", err) + } + + first, err := AIWorkflowService.PublishWorkflow(request.PublishAIWorkflowRequest{ + WorkflowID: workflow.ID, + Definition: validAIWorkflowDefinition(), + }, operator) + if err != nil { + t.Fatalf("PublishWorkflow() first error = %v", err) + } + second, err := AIWorkflowService.PublishWorkflow(request.PublishAIWorkflowRequest{ + WorkflowID: workflow.ID, + Definition: validAIWorkflowDefinition(), + }, operator) + if err != nil { + t.Fatalf("PublishWorkflow() second error = %v", err) + } + + if first.Version != 1 || second.Version != 2 { + t.Fatalf("expected versions 1 and 2, got %d and %d", first.Version, second.Version) + } +} + +func setupAIWorkflowTestDB(t *testing.T) { + t.Helper() + db, err := gorm.Open(sqlite.Open("file:"+t.Name()+"?mode=memory&cache=shared"), &gorm.Config{}) + if err != nil { + t.Fatalf("open sqlite db: %v", err) + } + if err := db.AutoMigrate(&models.AIWorkflow{}, &models.AIWorkflowVersion{}, &models.AIWorkflowRun{}, &models.AIWorkflowNodeRun{}); err != nil { + t.Fatalf("auto migrate: %v", err) + } + sqls.SetDB(db) +} + +func validAIWorkflowDefinition() dsl.Definition { + return dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + Nodes: []dsl.Node{ + {ID: "start_1", Type: "start"}, + {ID: "reply_1", Type: "send_reply", Config: json.RawMessage(`{"text":"hello"}`)}, + {ID: "end_1", Type: "end"}, + }, + Edges: []dsl.Edge{ + {ID: "e1", Source: "start_1", Target: "reply_1"}, + {ID: "e2", Source: "reply_1", Target: "end_1"}, + }, + } +} + +func aiWorkflowTestOperator() *dto.AuthPrincipal { + return &dto.AuthPrincipal{ + UserID: 1, + Username: "workflow-tester", + Nickname: "workflow-tester", + } +}