diff --git a/docs b/docs index 0b30092..d75c928 160000 --- a/docs +++ b/docs @@ -1 +1 @@ -Subproject commit 0b30092a2d46a70f883a7b41666f61e75d33da21 +Subproject commit d75c9282f9270b116d23bd1eab5e5fa1a4a17247 diff --git a/internal/ai/application/runtime/service.go b/internal/ai/application/runtime/service.go index d989f6e..b68a406 100644 --- a/internal/ai/application/runtime/service.go +++ b/internal/ai/application/runtime/service.go @@ -3,6 +3,7 @@ package runtime import ( "context" "encoding/json" + "strings" "time" "agent-desk/internal/ai/runtime/executor" @@ -53,11 +54,23 @@ func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) { } func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, error) { - aiAgent, _, err := prepareWorkflowAgent(req.AIAgent) + aiAgent, workflow, err := prepareWorkflowAgent(req.AIAgent) if err != nil { 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 { + return nil, err + } + return toWorkflowSummary(workflowResult, req.AIConfig.ModelName), nil + } toolSet, err := s.prepare.prepareToolsForResume(req) if err != nil { return nil, err @@ -77,6 +90,15 @@ func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, erro return toSummary(summary), nil } +func firstWorkflowResumeText(data map[string]string) string { + for _, value := range data { + if strings.TrimSpace(value) != "" { + return strings.TrimSpace(value) + } + } + return "" +} + func toWorkflowSummary(result *workflowexecutor.Result, modelName string) *Summary { if result == nil { return nil @@ -94,9 +116,28 @@ func toWorkflowSummary(result *workflowexecutor.Result, modelName string) *Summa CompletionTokens: result.CompletionTokens, RetrieverCount: result.RetrieverCount, TraceData: string(traceData), + CheckPointID: result.CheckPointID, + CheckPointData: result.CheckPointData, + Interrupted: result.Interrupted, + Interrupts: toWorkflowInterruptSummaries(result.Interrupts), } } +func toWorkflowInterruptSummaries(items []workflowexecutor.InterruptSummary) []InterruptContextSummary { + if len(items) == 0 { + return nil + } + ret := make([]InterruptContextSummary, 0, len(items)) + for _, item := range items { + ret = append(ret, InterruptContextSummary{ + Type: item.Type, + ID: item.ID, + InfoPreview: item.InfoPreview, + }) + } + return ret +} + func writeWorkflowRun(req Request, workflow resolvedWorkflow, result *workflowexecutor.Result, errorMessage string) error { if result == nil { return nil diff --git a/internal/ai/application/runtime/types.go b/internal/ai/application/runtime/types.go index f2beb1c..0ec1832 100644 --- a/internal/ai/application/runtime/types.go +++ b/internal/ai/application/runtime/types.go @@ -47,6 +47,7 @@ type Summary struct { ToolCodes []string InvokedToolCodes []string CheckPointID string + CheckPointData string Interrupted bool Interrupts []InterruptContextSummary TraceData string diff --git a/internal/ai/application/runtime/workflow_summary_test.go b/internal/ai/application/runtime/workflow_summary_test.go new file mode 100644 index 0000000..f83cac9 --- /dev/null +++ b/internal/ai/application/runtime/workflow_summary_test.go @@ -0,0 +1,162 @@ +package runtime + +import ( + "context" + "encoding/json" + "strings" + "testing" + + workflowexecutor "agent-desk/internal/ai/runtime/workflow" + "agent-desk/internal/ai/workflow/dsl" + workflowregistry "agent-desk/internal/ai/workflow/registry" + "agent-desk/internal/models" + "agent-desk/internal/pkg/enums" + + "github.com/glebarez/sqlite" + "github.com/mlogclub/simple/sqls" + "gorm.io/gorm" + "gorm.io/gorm/schema" +) + +func TestToWorkflowSummaryPreservesInterruptCheckpoint(t *testing.T) { + summary := toWorkflowSummary(&workflowexecutor.Result{ + Status: "interrupted", + CheckPointID: "workflow:1:2:confirm_1", + CheckPointData: `{"confirmNodeId":"confirm_1"}`, + Interrupted: true, + Interrupts: []workflowexecutor.InterruptSummary{ + {Type: "human_confirm", ID: "confirm_1", InfoPreview: `{"message":"请确认"}`}, + }, + }, "test-model") + + if summary == nil || !summary.Interrupted { + t.Fatalf("expected interrupted summary, got %#v", summary) + } + if summary.CheckPointID != "workflow:1:2:confirm_1" { + t.Fatalf("unexpected checkpoint id: %q", summary.CheckPointID) + } + if summary.CheckPointData == "" { + t.Fatalf("expected checkpoint data") + } + if len(summary.Interrupts) != 1 || summary.Interrupts[0].ID != "confirm_1" { + t.Fatalf("unexpected interrupts: %#v", summary.Interrupts) + } +} + +func TestServiceResumeUsesWorkflowCheckpointData(t *testing.T) { + db := setupWorkflowResumeTestDB(t) + def := runtimeHumanConfirmDefinition() + definitionJSON := mustMarshalDefinition(t, def) + version := models.AIWorkflowVersion{ + WorkflowID: 1, + Version: 1, + Status: enums.StatusOk, + Definition: definitionJSON, + } + if err := db.Create(&version).Error; err != nil { + t.Fatalf("create workflow version: %v", err) + } + checkpointData := mustMarshalWorkflowCheckpoint(t, def) + if err := db.Create(&models.ConversationInterrupt{ + ConversationID: 1, + AIAgentID: 1, + CheckPointID: "workflow:1:2:confirm_1", + RequestData: checkpointData, + Status: "pending", + }).Error; err != nil { + t.Fatalf("create interrupt: %v", err) + } + + summary, err := NewService().Resume(context.Background(), ResumeRequest{ + Conversation: models.Conversation{ID: 1}, + 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 == nil || summary.Status != "completed" || summary.Interrupted { + t.Fatalf("unexpected summary: %#v", summary) + } +} + +func setupWorkflowResumeTestDB(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.AIWorkflowVersion{}, &models.ConversationInterrupt{}); err != nil { + t.Fatalf("auto migrate: %v", err) + } + sqls.SetDB(db) + return db +} + +func runtimeHumanConfirmDefinition() 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: "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_end", Source: "confirm_1", Target: "end_1"}, + }, + } +} + +func mustMarshalDefinition(t *testing.T, def dsl.Definition) string { + t.Helper() + buf, err := json.Marshal(def) + if err != nil { + t.Fatalf("marshal definition: %v", err) + } + return string(buf) +} + +func mustMarshalWorkflowCheckpoint(t *testing.T, def dsl.Definition) string { + t.Helper() + buf, err := json.Marshal(struct { + Definition dsl.Definition `json:"definition"` + ConfirmNodeID string `json:"confirmNodeId"` + Vars map[string]map[string]any `json:"vars"` + }{ + Definition: def, + ConfirmNodeID: "confirm_1", + Vars: map[string]map[string]any{ + "start_1": {"userMessage": "创建工单"}, + "prompt_1": {"replyText": "请确认"}, + }, + }) + if err != nil { + t.Fatalf("marshal checkpoint: %v", err) + } + return string(buf) +} diff --git a/internal/ai/runtime/reply_helpers_test.go b/internal/ai/runtime/reply_helpers_test.go index 994bef2..e398f31 100644 --- a/internal/ai/runtime/reply_helpers_test.go +++ b/internal/ai/runtime/reply_helpers_test.go @@ -5,6 +5,7 @@ import ( "testing" applicationruntime "agent-desk/internal/ai/application/runtime" + "agent-desk/internal/models" "agent-desk/internal/pkg/toolx" ) @@ -66,6 +67,22 @@ func TestExtractInterruptMessageAndCheckpointError(t *testing.T) { } } +func TestBuildConversationInterruptStoresWorkflowCheckpointData(t *testing.T) { + item := buildConversationInterrupt(testConversation(1), testMessage(2), testAIAgent(3), &applicationruntime.Summary{ + CheckPointData: `{"confirmNodeId":"confirm_1"}`, + Interrupted: true, + Interrupts: []applicationruntime.InterruptContextSummary{ + {Type: "human_confirm", ID: "confirm_1", InfoPreview: `{"message":"请确认"}`}, + }, + }) + if item == nil { + t.Fatalf("expected interrupt item") + } + if item.RequestData != `{"confirmNodeId":"confirm_1"}` { + t.Fatalf("unexpected request data: %q", item.RequestData) + } +} + func TestGraphPlanReason(t *testing.T) { summary := &applicationruntime.Summary{ TraceData: `{ @@ -109,3 +126,15 @@ type fakeErr string func (e fakeErr) Error() string { return string(e) } + +func testConversation(id int64) models.Conversation { + return models.Conversation{ID: id} +} + +func testMessage(id int64) models.Message { + return models.Message{ID: id} +} + +func testAIAgent(id int64) models.AIAgent { + return models.AIAgent{ID: id} +} diff --git a/internal/ai/runtime/reply_interrupt_helpers.go b/internal/ai/runtime/reply_interrupt_helpers.go index c77df1d..b6c6774 100644 --- a/internal/ai/runtime/reply_interrupt_helpers.go +++ b/internal/ai/runtime/reply_interrupt_helpers.go @@ -34,6 +34,7 @@ func buildConversationInterrupt(conversation models.Conversation, message models item.InterruptType = firstInterruptType(summary) item.Status = "pending" item.PromptText = resolveInterruptPrompt(summary) + item.RequestData = strings.TrimSpace(summary.CheckPointData) item.UpdatedAt = now return item } diff --git a/internal/ai/runtime/workflow/executor.go b/internal/ai/runtime/workflow/executor.go index 7e6a492..1ea50e9 100644 --- a/internal/ai/runtime/workflow/executor.go +++ b/internal/ai/runtime/workflow/executor.go @@ -14,6 +14,8 @@ import ( "agent-desk/internal/ai/workflow/dsl" workflowregistry "agent-desk/internal/ai/workflow/registry" "agent-desk/internal/models" + "agent-desk/internal/pkg/dto" + "agent-desk/internal/pkg/dto/request" "agent-desk/internal/pkg/utils" "agent-desk/internal/services" ) @@ -36,6 +38,16 @@ type Result struct { CompletionTokens int RetrieverCount int TraceData string + CheckPointID string + CheckPointData string + Interrupted bool + Interrupts []InterruptSummary +} + +type InterruptSummary struct { + Type string + ID string + InfoPreview string } type Executor struct{} @@ -52,12 +64,66 @@ type runState struct { result Result } +type workflowCheckPoint struct { + Definition dsl.Definition `json:"definition"` + ConfirmNodeID string `json:"confirmNodeId"` + Vars map[string]map[string]any `json:"vars"` +} + func (e *Executor) Execute(ctx context.Context, input Input) (*Result, error) { state := newRunState(input) currentID := strings.TrimSpace(input.Definition.EntryNodeID) if currentID == "" { return nil, fmt.Errorf("workflow entry node is required") } + return e.executeFrom(ctx, state, currentID) +} + +func (e *Executor) Resume(ctx context.Context, input Input, checkPointData string, resumeText string) (*Result, error) { + var checkpoint workflowCheckPoint + if err := json.Unmarshal([]byte(strings.TrimSpace(checkPointData)), &checkpoint); err != nil { + return nil, fmt.Errorf("invalid workflow checkpoint: %w", err) + } + if len(checkpoint.Definition.Nodes) > 0 { + input.Definition = checkpoint.Definition + } + state := newRunState(input) + state.vars = checkpoint.Vars + if state.vars == nil { + state.vars = make(map[string]map[string]any) + } + confirmNodeID := strings.TrimSpace(checkpoint.ConfirmNodeID) + if confirmNodeID == "" { + return nil, fmt.Errorf("workflow checkpoint confirm node is required") + } + decision := graphs.ParseConfirmationDecision(resumeText) + if decision == "" { + node, ok := state.nodesByID[confirmNodeID] + if !ok { + return nil, fmt.Errorf("workflow node does not exist: %s", confirmNodeID) + } + if err := e.executeHumanConfirm(state, node); err != nil { + return nil, err + } + state.result.Status = "interrupted" + return &state.result, nil + } + state.setNodeVars(confirmNodeID, map[string]any{ + "confirmed": decision == graphs.ConfirmationDecisionConfirm, + "responseText": strings.TrimSpace(resumeText), + }) + nextID, ok, err := state.nextNodeID(confirmNodeID) + if err != nil { + return nil, err + } + if !ok { + state.result.Status = "completed" + return &state.result, nil + } + return e.executeFrom(ctx, state, nextID) +} + +func (e *Executor) executeFrom(ctx context.Context, state *runState, currentID string) (*Result, error) { for step := 0; step < maxWorkflowSteps; step++ { node, ok := state.nodesByID[currentID] if !ok { @@ -67,6 +133,10 @@ func (e *Executor) Execute(ctx context.Context, input Input) (*Result, error) { if err := e.executeNode(ctx, state, node); err != nil { return nil, err } + if state.result.Interrupted { + state.result.Status = "interrupted" + return &state.result, nil + } if node.Type == workflowregistry.NodeTypeEnd { state.result.Status = "completed" return &state.result, nil @@ -127,6 +197,12 @@ func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.No state.setNodeVars(node.ID, map[string]any{"matched": true}) case workflowregistry.NodeTypeAnalyzeConversation: return e.executeAnalyzeConversation(ctx, state, node) + case workflowregistry.NodeTypePrepareTicketDraft: + return e.executePrepareTicketDraft(ctx, state, node) + case workflowregistry.NodeTypeHumanConfirm: + return e.executeHumanConfirm(state, node) + case workflowregistry.NodeTypeCreateTicket: + return e.executeCreateTicket(state, node) case workflowregistry.NodeTypeLLMReply: return e.executeLLMReply(ctx, state, node) case workflowregistry.NodeTypeSendReply: @@ -146,6 +222,125 @@ func (e *Executor) executeNode(ctx context.Context, state *runState, node dsl.No return nil } +func (e *Executor) executeCreateTicket(state *runState, node dsl.Node) error { + confirmed := truthy(state.resolveInput(node, "confirmed")) + if !confirmed { + state.setNodeVars(node.ID, map[string]any{ + "ticketId": int64(0), + "created": false, + }) + return nil + } + draft := asMap(state.resolveInput(node, "ticketDraft")) + title := strings.TrimSpace(toString(draft["title"])) + description := strings.TrimSpace(toString(draft["description"])) + item, err := services.TicketService.CreateFromConversation(request.CreateTicketFromConversationRequest{ + ConversationID: state.input.Conversation.ID, + Title: title, + Description: description, + }, workflowAIPrincipal(state.input.AIAgent)) + if err != nil { + return err + } + state.setNodeVars(node.ID, map[string]any{ + "ticketId": item.ID, + "ticketNo": item.TicketNo, + "created": true, + }) + return nil +} + +func workflowAIPrincipal(aiAgent models.AIAgent) *dto.AuthPrincipal { + username := strings.TrimSpace(aiAgent.Name) + if username == "" { + username = "AI" + } + return &dto.AuthPrincipal{ + UserID: 0, + Username: username, + Nickname: username, + } +} + +func (e *Executor) executeHumanConfirm(state *runState, node dsl.Node) error { + prompt := strings.TrimSpace(toString(state.resolveInput(node, "prompt"))) + if prompt == "" { + prompt = "请确认是否继续。" + } + infoPreview, err := json.Marshal(map[string]string{"message": prompt}) + if err != nil { + return err + } + state.result.Interrupted = true + state.result.CheckPointID = buildWorkflowCheckPointID(state.input, node.ID) + checkpoint, err := json.Marshal(workflowCheckPoint{ + Definition: state.input.Definition, + ConfirmNodeID: node.ID, + Vars: state.vars, + }) + if err != nil { + return err + } + state.result.CheckPointData = string(checkpoint) + state.result.Interrupts = []InterruptSummary{ + { + Type: workflowregistry.NodeTypeHumanConfirm, + ID: node.ID, + InfoPreview: string(infoPreview), + }, + } + return nil +} + +func buildWorkflowCheckPointID(input Input, nodeID string) string { + return fmt.Sprintf("workflow:%d:%d:%s", input.Conversation.ID, input.UserMessage.ID, strings.TrimSpace(nodeID)) +} + +func (e *Executor) executePrepareTicketDraft(ctx context.Context, state *runState, node dsl.Node) error { + issue := strings.TrimSpace(toString(state.resolveInput(node, "issue"))) + input := graphs.PrepareTicketDraftInput{ + Issue: issue, + } + if title := strings.TrimSpace(readStringConfig(node.Config, "title")); title != "" { + input.Title = title + } + if description := strings.TrimSpace(readStringConfig(node.Config, "description")); description != "" { + input.Description = description + } + if impact := strings.TrimSpace(readStringConfig(node.Config, "impact")); impact != "" { + input.Impact = impact + } + if expectedOutcome := strings.TrimSpace(readStringConfig(node.Config, "expectedOutcome")); expectedOutcome != "" { + input.ExpectedOutcome = expectedOutcome + } + if currentAttempt := strings.TrimSpace(readStringConfig(node.Config, "currentAttempt")); currentAttempt != "" { + input.CurrentAttempt = currentAttempt + } + args, err := json.Marshal(input) + if err != nil { + return err + } + raw, err := graphs.NewPrepareTicketDraftGraph(state.input.Conversation).Run(ctx, string(args)) + if err != nil { + return err + } + var result graphs.PrepareTicketDraftResult + if err := json.Unmarshal([]byte(raw), &result); err != nil { + return err + } + state.setNodeVars(node.ID, map[string]any{ + "ticketDraft": map[string]any{ + "ready": result.Ready, + "title": strings.TrimSpace(result.Title), + "description": strings.TrimSpace(result.Description), + "missingFields": result.MissingFields, + "followUpQuestions": result.FollowUpQuestions, + "conversationFacts": result.ConversationFacts, + }, + }) + return nil +} + func (e *Executor) executeAnalyzeConversation(ctx context.Context, state *runState, node dsl.Node) error { userMessage := strings.TrimSpace(toString(state.resolveInput(node, "userMessage"))) input := graphs.AnalyzeConversationInput{ @@ -445,6 +640,25 @@ func toFloat(value any) float64 { } } +func asMap(value any) map[string]any { + switch v := value.(type) { + case map[string]any: + return v + case map[string]string: + ret := make(map[string]any, len(v)) + for key, item := range v { + ret[key] = item + } + return ret + case string: + var ret map[string]any + if err := json.Unmarshal([]byte(strings.TrimSpace(v)), &ret); err == nil { + return ret + } + } + return map[string]any{} +} + func truthy(value any) bool { switch v := value.(type) { case nil: diff --git a/internal/ai/runtime/workflow/executor_test.go b/internal/ai/runtime/workflow/executor_test.go index 8d6f1bd..27f2ddf 100644 --- a/internal/ai/runtime/workflow/executor_test.go +++ b/internal/ai/runtime/workflow/executor_test.go @@ -110,6 +110,131 @@ func TestExecutorAnalyzeConversationOutputsBranchVariables(t *testing.T) { assertPath(t, result.NodePath, []string{"start_1", "analyze_1", "handoff_end"}) } +func TestExecutorPrepareTicketDraftOutputsDraftVariable(t *testing.T) { + db := setupWorkflowExecutorHandoffDB(t) + aiAgent := createWorkflowExecutorHandoffAIAgent(t, db, "1") + conversation := createWorkflowExecutorHandoffConversation(t, db, aiAgent.ID) + userMessage := createWorkflowExecutorCustomerMessage(t, db, conversation.ID, "订单支付失败，请帮我登记工单") + + result, err := NewExecutor().Execute(context.Background(), Input{ + Definition: prepareTicketDraftWorkflowDefinition(), + Conversation: conversation, + UserMessage: userMessage, + AIAgent: aiAgent, + }) + if err != nil { + t.Fatalf("execute workflow: %v", err) + } + assertPath(t, result.NodePath, []string{"start_1", "draft_1", "ready_end"}) +} + +func TestExecutorHumanConfirmInterruptsWithCheckpoint(t *testing.T) { + result, err := NewExecutor().Execute(context.Background(), Input{ + Definition: humanConfirmWorkflowDefinition(), + Conversation: models.Conversation{ + ID: 11, + }, + UserMessage: models.Message{ + ID: 22, + Content: "创建工单", + }, + AIAgent: models.AIAgent{ + ID: 33, + }, + }) + if err != nil { + t.Fatalf("execute workflow: %v", err) + } + if !result.Interrupted { + t.Fatalf("expected workflow to interrupt") + } + if result.CheckPointID == "" { + t.Fatalf("expected checkpoint id") + } + if len(result.Interrupts) != 1 { + t.Fatalf("expected one interrupt, got %#v", result.Interrupts) + } + if result.Interrupts[0].Type != "human_confirm" || result.Interrupts[0].ID != "confirm_1" { + t.Fatalf("unexpected interrupt summary: %#v", result.Interrupts[0]) + } + if !strings.Contains(result.Interrupts[0].InfoPreview, "请确认创建工单") { + t.Fatalf("expected confirmation prompt, got %q", result.Interrupts[0].InfoPreview) + } + assertPath(t, result.NodePath, []string{"start_1", "prompt_1", "confirm_1"}) +} + +func TestExecutorResumeHumanConfirmContinuesWithConfirmedVariable(t *testing.T) { + executor := NewExecutor() + input := Input{ + Definition: humanConfirmWorkflowDefinition(), + Conversation: models.Conversation{ + ID: 11, + }, + UserMessage: models.Message{ + ID: 22, + Content: "创建工单", + }, + AIAgent: models.AIAgent{ + ID: 33, + }, + } + interrupted, err := executor.Execute(context.Background(), input) + if err != nil { + t.Fatalf("execute workflow: %v", err) + } + result, err := executor.Resume(context.Background(), input, interrupted.CheckPointData, "确认") + if err != nil { + t.Fatalf("resume workflow: %v", err) + } + if result.Interrupted { + t.Fatalf("expected workflow resume to complete") + } + assertPath(t, result.NodePath, []string{"end_1"}) +} + +func TestExecutorResumeCreatesTicketAfterHumanConfirmation(t *testing.T) { + db := setupWorkflowExecutorHandoffDB(t) + aiAgent := createWorkflowExecutorHandoffAIAgent(t, db, "1") + conversation := createWorkflowExecutorHandoffConversation(t, db, aiAgent.ID) + userMessage := createWorkflowExecutorCustomerMessage(t, db, conversation.ID, "订单支付失败，请帮我登记工单") + executor := NewExecutor() + + interrupted, err := executor.Execute(context.Background(), Input{ + Definition: createTicketWorkflowDefinition(), + Conversation: conversation, + UserMessage: userMessage, + AIAgent: aiAgent, + }) + if err != nil { + t.Fatalf("execute workflow: %v", err) + } + if !interrupted.Interrupted { + t.Fatalf("expected workflow to interrupt before creating ticket") + } + + result, err := executor.Resume(context.Background(), Input{ + Definition: createTicketWorkflowDefinition(), + Conversation: conversation, + UserMessage: userMessage, + AIAgent: aiAgent, + }, interrupted.CheckPointData, "确认") + if err != nil { + t.Fatalf("resume workflow: %v", err) + } + if result.Interrupted { + t.Fatalf("expected workflow to complete") + } + assertPath(t, result.NodePath, []string{"create_ticket_1", "end_1"}) + + var ticket models.Ticket + if err := db.First(&ticket, "conversation_id = ?", conversation.ID).Error; err != nil { + t.Fatalf("expected created ticket: %v", err) + } + if ticket.Title == "" || !strings.Contains(ticket.Description, "订单支付失败") { + t.Fatalf("unexpected ticket: %+v", ticket) + } +} + func conditionalReplyDefinition() dsl.Definition { return dsl.Definition{ SchemaVersion: 1, @@ -148,6 +273,103 @@ func conditionalReplyDefinition() dsl.Definition { } } +func createTicketWorkflowDefinition() dsl.Definition { + return dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + Nodes: []dsl.Node{ + {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, + {ID: "draft_1", Type: workflowregistry.NodeTypePrepareTicketDraft, Name: "Draft", Inputs: map[string]dsl.VariableSelector{ + "issue": {NodeID: "start_1", Field: "userMessage"}, + }}, + {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: "create_ticket_1", Type: workflowregistry.NodeTypeCreateTicket, Name: "Create Ticket", Inputs: map[string]dsl.VariableSelector{ + "ticketDraft": {NodeID: "draft_1", Field: "ticketDraft"}, + "confirmed": {NodeID: "confirm_1", Field: "confirmed"}, + }}, + {ID: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"}, + {ID: "cancel_end", Type: workflowregistry.NodeTypeEnd, Name: "Cancel"}, + }, + Edges: []dsl.Edge{ + {ID: "edge_start_draft", Source: "start_1", Target: "draft_1"}, + {ID: "edge_draft_prompt", Source: "draft_1", Target: "prompt_1"}, + {ID: "edge_prompt_confirm", Source: "prompt_1", Target: "confirm_1"}, + { + ID: "edge_confirm_create", + Source: "confirm_1", + Target: "create_ticket_1", + Condition: &dsl.Condition{ + Left: &dsl.VariableSelector{NodeID: "confirm_1", Field: "confirmed"}, + Operator: "is_true", + }, + }, + {ID: "edge_confirm_cancel", Source: "confirm_1", Target: "cancel_end"}, + {ID: "edge_create_end", Source: "create_ticket_1", Target: "end_1"}, + }, + } +} + +func humanConfirmWorkflowDefinition() 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: "end_1", Type: workflowregistry.NodeTypeEnd, Name: "End"}, + {ID: "cancel_end", Type: workflowregistry.NodeTypeEnd, Name: "Cancel"}, + }, + 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_yes", + Source: "confirm_1", + Target: "end_1", + Condition: &dsl.Condition{ + Left: &dsl.VariableSelector{NodeID: "confirm_1", Field: "confirmed"}, + Operator: "is_true", + }, + }, + {ID: "edge_confirm_cancel", Source: "confirm_1", Target: "cancel_end"}, + }, + } +} + +func prepareTicketDraftWorkflowDefinition() dsl.Definition { + return dsl.Definition{ + SchemaVersion: 1, + EntryNodeID: "start_1", + Nodes: []dsl.Node{ + {ID: "start_1", Type: workflowregistry.NodeTypeStart, Name: "Start"}, + {ID: "draft_1", Type: workflowregistry.NodeTypePrepareTicketDraft, Name: "Draft", Inputs: map[string]dsl.VariableSelector{ + "issue": {NodeID: "start_1", Field: "userMessage"}, + }}, + {ID: "ready_end", Type: workflowregistry.NodeTypeEnd, Name: "Ready"}, + {ID: "default_end", Type: workflowregistry.NodeTypeEnd, Name: "Default"}, + }, + Edges: []dsl.Edge{ + {ID: "edge_start_draft", Source: "start_1", Target: "draft_1"}, + { + ID: "edge_draft_ready", + Source: "draft_1", + Target: "ready_end", + Condition: &dsl.Condition{ + Left: &dsl.VariableSelector{NodeID: "draft_1", Field: "ticketDraft"}, + Operator: "exists", + }, + }, + {ID: "edge_draft_default", Source: "draft_1", Target: "default_end"}, + }, + } +} + func analyzeConversationWorkflowDefinition() dsl.Definition { return dsl.Definition{ SchemaVersion: 1, @@ -225,16 +447,23 @@ func setupWorkflowExecutorHandoffDB(t *testing.T) *gorm.DB { }) if err := db.AutoMigrate( &models.User{}, + &models.Customer{}, + &models.CustomerIdentity{}, &models.AIAgent{}, &models.AgentTeam{}, &models.AgentTeamSchedule{}, &models.AgentProfile{}, + &models.Channel{}, &models.Conversation{}, &models.ConversationAssignment{}, &models.ConversationEventLog{}, &models.ConversationReadState{}, &models.Message{}, &models.ChannelMessageOutbox{}, + &models.Ticket{}, + &models.TicketNoSequence{}, + &models.TicketTag{}, + &models.TicketProgress{}, ); err != nil { t.Fatalf("auto migrate error = %v", err) } @@ -303,6 +532,13 @@ func createWorkflowExecutorHandoffAgentProfile(t *testing.T, db *gorm.DB, userID func createWorkflowExecutorHandoffConversation(t *testing.T, db *gorm.DB, aiAgentID int64) models.Conversation { t.Helper() now := time.Now() + if err := db.FirstOrCreate(&models.Customer{ + ID: 1, + Name: "测试访客", + Status: enums.StatusOk, + }).Error; err != nil { + t.Fatalf("create customer error = %v", err) + } item := models.Conversation{ AIAgentID: aiAgentID, ChannelID: 1,