diff --git a/internal/ai/application/runtime/service.go b/internal/ai/application/runtime/service.go index f56db47..db19195 100644 --- a/internal/ai/application/runtime/service.go +++ b/internal/ai/application/runtime/service.go @@ -9,6 +9,7 @@ import ( "agent-desk/internal/ai/runtime/executor" workflowexecutor "agent-desk/internal/ai/runtime/workflow" "agent-desk/internal/models" + "agent-desk/internal/pkg/errorsx" "agent-desk/internal/pkg/utils" "agent-desk/internal/repositories" @@ -69,34 +70,40 @@ func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, erro return nil, err } req.AIAgent = aiAgent - if interrupt := repositories.ConversationInterruptRepository.GetByCheckPointID(sqls.DB(), req.CheckPointID); interrupt != nil && strings.TrimSpace(interrupt.RequestData) != "" { - workflowResult, err := workflowexecutor.NewExecutor().Resume(ctx, workflowexecutor.Input{ - Definition: workflow.Definition, - Conversation: req.Conversation, - AIAgent: req.AIAgent, - AIConfig: req.AIConfig, - }, interrupt.RequestData, firstWorkflowResumeText(req.ResumeData)) - if err != nil { - if workflowResult != nil { - _, _ = writeWorkflowRun(Request{ - Conversation: req.Conversation, - UserMessage: req.UserMessage, - AIAgent: req.AIAgent, - AIConfig: req.AIConfig, - }, workflow, workflowResult, err.Error()) + if interrupt := repositories.ConversationInterruptRepository.GetByCheckPointID(sqls.DB(), req.CheckPointID); interrupt != nil { + if strings.TrimSpace(interrupt.RequestData) == "" { + if interrupt.WorkflowRunID > 0 || strings.HasPrefix(strings.TrimSpace(req.CheckPointID), "workflow:") { + return nil, errorsx.InvalidParam("workflow checkpoint data is required") } - return nil, err + } else { + workflowResult, err := workflowexecutor.NewExecutor().Resume(ctx, workflowexecutor.Input{ + Definition: workflow.Definition, + Conversation: req.Conversation, + AIAgent: req.AIAgent, + AIConfig: req.AIConfig, + }, interrupt.RequestData, firstWorkflowResumeText(req.ResumeData)) + if err != nil { + if workflowResult != nil { + _, _ = writeWorkflowRunWithExistingID(Request{ + Conversation: req.Conversation, + UserMessage: req.UserMessage, + AIAgent: req.AIAgent, + AIConfig: req.AIConfig, + }, workflow, workflowResult, err.Error(), interrupt.WorkflowRunID) + } + return nil, err + } + workflowRunID, err := writeWorkflowRunWithExistingID(Request{ + Conversation: req.Conversation, + UserMessage: req.UserMessage, + AIAgent: req.AIAgent, + AIConfig: req.AIConfig, + }, workflow, workflowResult, "", interrupt.WorkflowRunID) + if err != nil { + return nil, err + } + return toWorkflowSummary(workflowResult, req.AIConfig.ModelName, workflow, workflowRunID), nil } - workflowRunID, err := writeWorkflowRun(Request{ - Conversation: req.Conversation, - UserMessage: req.UserMessage, - AIAgent: req.AIAgent, - AIConfig: req.AIConfig, - }, workflow, workflowResult, "") - if err != nil { - return nil, err - } - return toWorkflowSummary(workflowResult, req.AIConfig.ModelName, workflow, workflowRunID), nil } toolSet, err := s.prepare.prepareToolsForResume(req) if err != nil { @@ -173,6 +180,10 @@ func toWorkflowInterruptSummaries(items []workflowexecutor.InterruptSummary) []I } func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowexecutor.Result, errorMessage string) (int64, error) { + return writeWorkflowRunWithExistingID(req, workflow, result, errorMessage, 0) +} + +func writeWorkflowRunWithExistingID(req Request, workflow resolvedWorkflow, result *workflowexecutor.Result, errorMessage string, existingRunID int64) (int64, error) { if result == nil { return 0, nil } @@ -185,20 +196,32 @@ func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowex runStatus := workflowRunStatus(result.Status, errorMessage) var runID int64 err := sqls.WithTransaction(func(ctx *sqls.TxContext) error { - run := &models.AIWorkflowRun{ - WorkflowID: workflow.WorkflowID, - WorkflowVersionID: workflow.VersionID, - ConversationID: req.Conversation.ID, - AIAgentID: req.AIAgent.ID, - MessageID: req.UserMessage.ID, - Status: runStatus, - StartedAt: now, - EndedAt: &endedAt, - InterruptType: firstWorkflowInterruptType(result), - InterruptNodeID: firstWorkflowInterruptNodeID(result), - ErrorMessage: errorMessage, - } - if err := repositories.AIWorkflowRunRepository.Create(ctx.Tx, run); err != nil { + run := repositories.AIWorkflowRunRepository.Get(ctx.Tx, existingRunID) + if run == nil { + run = &models.AIWorkflowRun{ + WorkflowID: workflow.WorkflowID, + WorkflowVersionID: workflow.VersionID, + ConversationID: req.Conversation.ID, + AIAgentID: req.AIAgent.ID, + MessageID: req.UserMessage.ID, + Status: runStatus, + StartedAt: now, + EndedAt: &endedAt, + InterruptType: firstWorkflowInterruptType(result), + InterruptNodeID: firstWorkflowInterruptNodeID(result), + ErrorMessage: errorMessage, + } + if err := repositories.AIWorkflowRunRepository.Create(ctx.Tx, run); err != nil { + return err + } + } else if err := repositories.AIWorkflowRunRepository.Updates(ctx.Tx, run.ID, map[string]any{ + "status": runStatus, + "ended_at": &endedAt, + "interrupt_type": firstWorkflowInterruptType(result), + "interrupt_node_id": firstWorkflowInterruptNodeID(result), + "error_message": errorMessage, + "updated_at": now, + }); err != nil { return err } runID = run.ID diff --git a/internal/ai/application/runtime/workflow_summary_test.go b/internal/ai/application/runtime/workflow_summary_test.go index 166b943..dbe8ec3 100644 --- a/internal/ai/application/runtime/workflow_summary_test.go +++ b/internal/ai/application/runtime/workflow_summary_test.go @@ -101,6 +101,87 @@ func TestServiceResumeUsesWorkflowCheckpointData(t *testing.T) { } } +func TestServiceResumeReusesInterruptedWorkflowRun(t *testing.T) { + db := setupWorkflowResumeTestDB(t) + def := runtimeHumanConfirmDefinition() + version := models.AIWorkflowVersion{ + WorkflowID: 1, + Version: 1, + Status: enums.StatusOk, + Definition: mustMarshalDefinition(t, def), + } + if err := db.Create(&version).Error; err != nil { + t.Fatalf("create workflow version: %v", err) + } + interruptedRun := models.AIWorkflowRun{ + WorkflowID: version.WorkflowID, + WorkflowVersionID: version.ID, + ConversationID: 1, + AIAgentID: 1, + MessageID: 2, + Status: workflowRunStatusInterrupted, + InterruptType: "human_confirm", + InterruptNodeID: "confirm_1", + } + if err := db.Create(&interruptedRun).Error; err != nil { + t.Fatalf("create interrupted workflow run: %v", err) + } + if err := db.Create(&models.ConversationInterrupt{ + ConversationID: 1, + AIAgentID: 1, + CheckPointID: "workflow:1:2:confirm_1", + InterruptID: "confirm_1", + InterruptType: "human_confirm", + WorkflowRunID: interruptedRun.ID, + WorkflowNodeID: "confirm_1", + RequestData: mustMarshalWorkflowCheckpoint(t, def), + Status: "pending", + }).Error; err != nil { + t.Fatalf("create interrupt: %v", err) + } + + summary, err := NewService().Resume(context.Background(), ResumeRequest{ + Conversation: models.Conversation{ID: 1}, + UserMessage: models.Message{ID: 3, Content: "确认"}, + AIAgent: models.AIAgent{ + ID: 1, + WorkflowVersionID: version.ID, + }, + AIConfig: models.AIConfig{ModelName: "test-model"}, + CheckPointID: "workflow:1:2:confirm_1", + ResumeData: map[string]string{ + "confirm_1": "确认", + }, + }) + if err != nil { + t.Fatalf("resume workflow: %v", err) + } + if summary.WorkflowRunID != interruptedRun.ID { + t.Fatalf("expected resume to reuse workflow run %d, got %d", interruptedRun.ID, summary.WorkflowRunID) + } + var runCount int64 + if err := db.Model(&models.AIWorkflowRun{}).Count(&runCount).Error; err != nil { + t.Fatalf("count workflow runs: %v", err) + } + if runCount != 1 { + t.Fatalf("expected one workflow run after resume, got %d", runCount) + } + var updated models.AIWorkflowRun + if err := db.First(&updated, interruptedRun.ID).Error; err != nil { + t.Fatalf("find updated workflow run: %v", err) + } + if updated.Status != workflowRunStatusCompleted || updated.ErrorMessage != "" { + t.Fatalf("unexpected updated workflow run: %#v", updated) + } + var nodeCount int64 + if err := db.Model(&models.AIWorkflowNodeRun{}).Where("workflow_run_id = ?", interruptedRun.ID).Count(&nodeCount).Error; err != nil { + t.Fatalf("count node runs: %v", err) + } + if nodeCount == 0 { + t.Fatalf("expected resumed node traces to be appended to original workflow run") + } +} + func TestServiceRunWritesFailedWorkflowRun(t *testing.T) { db := setupWorkflowResumeTestDB(t) def := dsl.Definition{ diff --git a/internal/ai/runtime/reply_helpers_test.go b/internal/ai/runtime/reply_helpers_test.go index 36ad2db..e9d4d6a 100644 --- a/internal/ai/runtime/reply_helpers_test.go +++ b/internal/ai/runtime/reply_helpers_test.go @@ -47,6 +47,7 @@ func TestBuildConversationInterruptStoresWorkflowCheckpointData(t *testing.T) { item := buildConversationInterrupt(testConversation(1), testMessage(2), testAIAgent(3), &applicationruntime.Summary{ CheckPointData: `{"confirmNodeId":"confirm_1"}`, Interrupted: true, + WorkflowRunID: 99, Interrupts: []applicationruntime.InterruptContextSummary{ {Type: "human_confirm", ID: "confirm_1", InfoPreview: `{"message":"请确认"}`}, }, @@ -57,6 +58,9 @@ func TestBuildConversationInterruptStoresWorkflowCheckpointData(t *testing.T) { if item.RequestData != `{"confirmNodeId":"confirm_1"}` { t.Fatalf("unexpected request data: %q", item.RequestData) } + if item.WorkflowRunID != 99 || item.WorkflowNodeID != "confirm_1" { + t.Fatalf("unexpected workflow interrupt identity: run=%d node=%q", item.WorkflowRunID, item.WorkflowNodeID) + } } type fakeErr string diff --git a/internal/ai/runtime/reply_interrupt_helpers.go b/internal/ai/runtime/reply_interrupt_helpers.go index b6c6774..1dc5571 100644 --- a/internal/ai/runtime/reply_interrupt_helpers.go +++ b/internal/ai/runtime/reply_interrupt_helpers.go @@ -32,6 +32,8 @@ func buildConversationInterrupt(conversation models.Conversation, message models item.SourceMessageID = message.ID item.InterruptID = firstInterruptID(summary) item.InterruptType = firstInterruptType(summary) + item.WorkflowRunID = summary.WorkflowRunID + item.WorkflowNodeID = firstInterruptID(summary) item.Status = "pending" item.PromptText = resolveInterruptPrompt(summary) item.RequestData = strings.TrimSpace(summary.CheckPointData) diff --git a/internal/ai/runtime/workflow/executor.go b/internal/ai/runtime/workflow/executor.go index bf07177..17dc5f0 100644 --- a/internal/ai/runtime/workflow/executor.go +++ b/internal/ai/runtime/workflow/executor.go @@ -420,6 +420,18 @@ func (e *Executor) executeAnalyzeConversation(ctx context.Context, state *runSta } func (e *Executor) executeHandoffToHuman(state *runState, node dsl.Node) error { + if _, hasConfirmedInput := node.Inputs["confirmed"]; hasConfirmedInput && !truthy(state.resolveInput(node, "confirmed")) { + state.setNodeVars(node.ID, map[string]any{ + "handoffId": int64(0), + "reason": strings.TrimSpace(toString(state.resolveInput(node, "reason"))), + "decision": "cancelled", + "teamId": int64(0), + "assigneeId": int64(0), + "message": "", + "skipped": true, + }) + return nil + } reason := strings.TrimSpace(toString(state.resolveInput(node, "reason"))) result, err := services.ConversationHumanDispatchService.HandoffByAIWithRequestID( state.input.Conversation.ID, diff --git a/internal/ai/runtime/workflow/executor_test.go b/internal/ai/runtime/workflow/executor_test.go index 27f2ddf..91fc8cb 100644 --- a/internal/ai/runtime/workflow/executor_test.go +++ b/internal/ai/runtime/workflow/executor_test.go @@ -92,6 +92,50 @@ func TestExecutorHandoffToHumanRunsRealDispatchAction(t *testing.T) { } } +func TestExecutorResumeSkipsHandoffWhenConfirmationCancelled(t *testing.T) { + db := setupWorkflowExecutorHandoffDB(t) + aiAgent := createWorkflowExecutorHandoffAIAgent(t, db, "1") + createWorkflowExecutorHandoffTeam(t, db, 1, "售后支持组") + createWorkflowExecutorHandoffActiveSchedule(t, db, 1) + createWorkflowExecutorHandoffAgentProfile(t, db, 101, 1) + conversation := createWorkflowExecutorHandoffConversation(t, db, aiAgent.ID) + userMessage := createWorkflowExecutorCustomerMessage(t, db, conversation.ID, "需要人工处理") + input := Input{ + Definition: handoffAfterConfirmationWorkflowDefinition(), + Conversation: conversation, + UserMessage: userMessage, + AIAgent: aiAgent, + } + + interrupted, err := NewExecutor().Execute(context.Background(), input) + if err != nil { + t.Fatalf("execute workflow: %v", err) + } + if !interrupted.Interrupted { + t.Fatalf("expected workflow to interrupt before handoff") + } + + result, err := NewExecutor().Resume(context.Background(), input, interrupted.CheckPointData, "取消") + if err != nil { + t.Fatalf("resume workflow: %v", err) + } + if result.Interrupted { + t.Fatalf("expected cancelled resume to complete") + } + assertPath(t, result.NodePath, []string{"handoff_1", "end_1"}) + + current := services.ConversationService.Get(conversation.ID) + if current.Status != enums.IMConversationStatusAIServing { + t.Fatalf("expected conversation to remain ai serving, got status=%d", current.Status) + } + if current.CurrentAssigneeID != 0 || current.CurrentTeamID != 0 || current.HandoffAt != nil { + t.Fatalf("expected no handoff side effect, got assignee=%d team=%d handoffAt=%v", current.CurrentAssigneeID, current.CurrentTeamID, current.HandoffAt) + } + if count := services.MessageService.Count(sqls.NewCnd().Eq("conversation_id", conversation.ID).Eq("sender_type", enums.IMSenderTypeAI)); count != 0 { + t.Fatalf("expected no handoff notice message, got %d", count) + } +} + func TestExecutorAnalyzeConversationOutputsBranchVariables(t *testing.T) { db := setupWorkflowExecutorHandoffDB(t) aiAgent := createWorkflowExecutorHandoffAIAgent(t, db, "1") @@ -427,6 +471,31 @@ func handoffWorkflowDefinition() dsl.Definition { } } +func handoffAfterConfirmationWorkflowDefinition() dsl.Definition { + return dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + Nodes: []dsl.Node{ + {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, + {ID: "prompt_1", Type: workflowregistry.NodeTypeLLMReply, Name: "Prompt", Config: []byte(`{"staticReply":"请确认转人工"}`)}, + {ID: "confirm_1", Type: workflowregistry.NodeTypeHumanConfirm, Name: "Confirm", Inputs: map[string]dsl.VariableSelector{ + "prompt": {NodeID: "prompt_1", Field: "replyText"}, + }}, + {ID: "handoff_1", Type: workflowregistry.NodeTypeHandoffToHuman, Name: "Handoff", Inputs: map[string]dsl.VariableSelector{ + "reason": {NodeID: "start_1", Field: "userMessage"}, + "confirmed": {NodeID: "confirm_1", Field: "confirmed"}, + }}, + {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"}, + }, + Edges: []dsl.Edge{ + {ID: "edge_start_prompt", Source: "start_1", Target: "prompt_1"}, + {ID: "edge_prompt_confirm", Source: "prompt_1", Target: "confirm_1"}, + {ID: "edge_confirm_handoff", Source: "confirm_1", Target: "handoff_1"}, + {ID: "edge_handoff_end", Source: "handoff_1", Target: "end_1"}, + }, + } +} + func setupWorkflowExecutorHandoffDB(t *testing.T) *gorm.DB { t.Helper() dbName := strings.NewReplacer("/", "_", " ", "_").Replace(t.Name()) diff --git a/internal/models/models.go b/internal/models/models.go index c169806..e9e582a 100644 --- a/internal/models/models.go +++ b/internal/models/models.go @@ -918,6 +918,8 @@ type ConversationInterrupt struct { AIAgentID int64 `gorm:"type:bigint;not null;default:0;index"` SourceMessageID int64 `gorm:"type:bigint;not null;default:0;index"` LastResumeMessageID int64 `gorm:"type:bigint;not null;default:0;index"` + WorkflowRunID int64 `gorm:"type:bigint;not null;default:0;index"` + WorkflowNodeID string `gorm:"type:varchar(100);not null;default:'';index"` CheckPointID string `gorm:"type:varchar(128);not null;default:'';uniqueIndex"` InterruptID string `gorm:"type:varchar(255);not null;default:'';index"` InterruptType string `gorm:"type:varchar(50);not null;default:'';index"` diff --git a/internal/repositories/conversation_interrupt_repository.go b/internal/repositories/conversation_interrupt_repository.go index 59359c4..88dede3 100644 --- a/internal/repositories/conversation_interrupt_repository.go +++ b/internal/repositories/conversation_interrupt_repository.go @@ -74,6 +74,8 @@ func (r *conversationInterruptRepository) UpsertByCheckPointID(db *gorm.DB, item "ai_agent_id": item.AIAgentID, "source_message_id": item.SourceMessageID, "last_resume_message_id": item.LastResumeMessageID, + "workflow_run_id": item.WorkflowRunID, + "workflow_node_id": item.WorkflowNodeID, "interrupt_id": item.InterruptID, "interrupt_type": item.InterruptType, "status": item.Status, diff --git a/internal/services/channel_service.go b/internal/services/channel_service.go index 100723b..630eac5 100644 --- a/internal/services/channel_service.go +++ b/internal/services/channel_service.go @@ -398,6 +398,9 @@ func (s *channelService) buildChannelModel(id int64, req request.CreateChannelRe if aiAgent == nil || aiAgent.Status != enums.StatusOk { return nil, errorsx.InvalidParamI18n("error.e0004") } + if aiAgent.WorkflowVersionID <= 0 { + return nil, errorsx.InvalidParam("ai agent workflow must be published before binding channel") + } status := enums.Status(req.Status) if req.Status == 0 { status = enums.StatusOk diff --git a/internal/services/channel_service_test.go b/internal/services/channel_service_test.go new file mode 100644 index 0000000..2ade90d --- /dev/null +++ b/internal/services/channel_service_test.go @@ -0,0 +1,91 @@ +package services + +import ( + "strings" + "testing" + + "agent-desk/internal/models" + "agent-desk/internal/pkg/dto" + "agent-desk/internal/pkg/dto/request" + "agent-desk/internal/pkg/enums" + + "github.com/glebarez/sqlite" + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" + "gorm.io/gorm/schema" +) + +func TestChannelServiceRejectsAgentWithoutPublishedWorkflow(t *testing.T) { + db := setupChannelServiceTestDB(t) + agent := createChannelServiceTestAgent(t, db, 0) + + _, err := ChannelService.CreateChannel(request.CreateChannelRequest{ + ChannelType: enums.ChannelTypeWeb, + AIAgentID: agent.ID, + Name: "官网客服", + Status: int(enums.StatusOk), + }, channelServiceTestOperator()) + if err == nil { + t.Fatalf("expected channel creation to reject unpublished ai agent") + } +} + +func TestChannelServiceAllowsAgentWithPublishedWorkflow(t *testing.T) { + db := setupChannelServiceTestDB(t) + agent := createChannelServiceTestAgent(t, db, 1001) + + item, err := ChannelService.CreateChannel(request.CreateChannelRequest{ + ChannelType: enums.ChannelTypeWeb, + AIAgentID: agent.ID, + Name: "官网客服", + Status: int(enums.StatusOk), + }, channelServiceTestOperator()) + if err != nil { + t.Fatalf("create channel: %v", err) + } + if item == nil || item.AIAgentID != agent.ID { + t.Fatalf("unexpected channel: %#v", item) + } +} + +func setupChannelServiceTestDB(t *testing.T) *gorm.DB { + t.Helper() + dbName := strings.NewReplacer("/", "_", " ", "_").Replace(t.Name()) + db, err := gorm.Open(sqlite.Open("file:"+dbName+"?mode=memory&cache=shared"), &gorm.Config{ + NamingStrategy: schema.NamingStrategy{ + TablePrefix: "t_", + SingularTable: true, + }, + }) + if err != nil { + t.Fatalf("open sqlite: %v", err) + } + t.Cleanup(func() { + sqlDB, err := db.DB() + if err == nil { + _ = sqlDB.Close() + } + }) + if err := db.AutoMigrate(&models.AIAgent{}, &models.Channel{}); err != nil { + t.Fatalf("auto migrate: %v", err) + } + sqls.SetDB(db) + return db +} + +func createChannelServiceTestAgent(t *testing.T, db *gorm.DB, workflowVersionID int64) models.AIAgent { + t.Helper() + item := models.AIAgent{ + Name: "测试 AI", + Status: enums.StatusOk, + WorkflowVersionID: workflowVersionID, + } + if err := db.Create(&item).Error; err != nil { + t.Fatalf("create ai agent: %v", err) + } + return item +} + +func channelServiceTestOperator() *dto.AuthPrincipal { + return &dto.AuthPrincipal{UserID: 1, Username: "admin"} +} diff --git a/internal/services/conversation_interrupt_service.go b/internal/services/conversation_interrupt_service.go index 58a3f19..6c806d3 100644 --- a/internal/services/conversation_interrupt_service.go +++ b/internal/services/conversation_interrupt_service.go @@ -96,6 +96,8 @@ func (s *conversationInterruptService) mergeForCheckpointUpdate(current, next *m merged.AIAgentID = current.AIAgentID merged.SourceMessageID = current.SourceMessageID merged.LastResumeMessageID = current.LastResumeMessageID + merged.WorkflowRunID = current.WorkflowRunID + merged.WorkflowNodeID = current.WorkflowNodeID merged.InterruptID = current.InterruptID merged.InterruptType = current.InterruptType merged.Status = current.Status @@ -119,6 +121,8 @@ func (s *conversationInterruptService) mergeForPendingUpdate(current, next *mode merged.ConversationID = next.ConversationID merged.AIAgentID = next.AIAgentID merged.SourceMessageID = next.SourceMessageID + merged.WorkflowRunID = next.WorkflowRunID + merged.WorkflowNodeID = next.WorkflowNodeID merged.InterruptID = next.InterruptID merged.InterruptType = next.InterruptType merged.Status = next.Status