wlw/ai-agent
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(workflow): add WorkflowRunID to message handling and related services

Browse Source
This commit is contained in:
mlogclub
2026-06-24 17:49:39 +08:00
parent 05233820fb
commit 0b5a24975c
16 changed files with 513 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/ai/runtime/reply_commit_service.go
+3 -1
View File
@@ -22,6 +22,7 @@ type replyCommitInput struct {
AIAgent models.AIAgent
ReplyText string
ClientPrefix string
WorkflowRunID int64
IncrementRound bool
}
@@ -34,7 +35,7 @@ func (s *replyCommitService) SendAIReply(input replyCommitInput) (*models.Messag
if replyText == "" {
return nil, nil
}
replyMessage, err := svc.MessageService.SendAIMessageWithRequestID(
replyMessage, err := svc.MessageService.SendAIMessageWithRequestIDAndWorkflowRunID(
input.Conversation.ID,
input.AIAgent.ID,
fmt.Sprintf("%s_%d", strings.TrimSpace(input.ClientPrefix), input.Message.ID),
@@ -43,6 +44,7 @@ func (s *replyCommitService) SendAIReply(input replyCommitInput) (*models.Messag
"",
s.buildAIPrincipal(input.AIAgent),
input.Message.RequestID,
input.WorkflowRunID,
)
if err != nil || !input.IncrementRound {
return replyMessage, err
internal/ai/runtime/reply_commit_service_test.go
+120
View File
@@ -0,0 +1,120 @@
package runtime
import (
"strings"
"testing"
"time"
"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 TestReplyCommitStoresWorkflowRunIDOnAIMessage(t *testing.T) {
db := setupReplyCommitTestDB(t)
aiAgent := createReplyCommitTestAIAgent(t, db)
conversation := createReplyCommitTestConversation(t, db, aiAgent.ID)
replyMessage, err := newReplyCommitService().CommitAIReply(replyCommitInput{
Conversation: *conversation,
Message: models.Message{ID: 101, RequestID: "trace-101"},
AIAgent: *aiAgent,
ReplyText: "AI reply",
ClientPrefix: "ai_reply",
WorkflowRunID: 9988,
})
if err != nil {
t.Fatalf("CommitAIReply() error = %v", err)
}
if replyMessage == nil {
t.Fatalf("expected reply message")
}
if replyMessage.WorkflowRunID != 9988 {
t.Fatalf("replyMessage.WorkflowRunID=%d want 9988", replyMessage.WorkflowRunID)
}
var stored models.Message
if err := db.First(&stored, replyMessage.ID).Error; err != nil {
t.Fatalf("find reply message: %v", err)
}
if stored.WorkflowRunID != 9988 {
t.Fatalf("stored.WorkflowRunID=%d want 9988", stored.WorkflowRunID)
}
}
func setupReplyCommitTestDB(t *testing.T) *gorm.DB {
t.Helper()
dbName := "reply_commit_test_" + 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 db: %v", err)
}
sqlDB, err := db.DB()
if err != nil {
t.Fatalf("get sqlite db: %v", err)
}
t.Cleanup(func() {
if err := sqlDB.Close(); err != nil {
t.Fatalf("close sqlite db: %v", err)
}
})
if err := db.AutoMigrate(
&models.AIAgent{},
&models.Channel{},
&models.ChannelMessageOutbox{},
&models.Conversation{},
&models.ConversationReadState{},
&models.ConversationEventLog{},
&models.Message{},
); err != nil {
t.Fatalf("auto migrate: %v", err)
}
sqls.SetDB(db)
return db
}
func createReplyCommitTestAIAgent(t *testing.T, db *gorm.DB) *models.AIAgent {
t.Helper()
now := time.Now()
item := &models.AIAgent{
Name: "reply-agent",
Status: enums.StatusOk,
AuditFields: models.AuditFields{
CreatedAt: now,
UpdatedAt: now,
},
}
if err := db.Create(item).Error; err != nil {
t.Fatalf("create ai agent: %v", err)
}
return item
}
func createReplyCommitTestConversation(t *testing.T, db *gorm.DB, aiAgentID int64) *models.Conversation {
t.Helper()
now := time.Now()
item := &models.Conversation{
CustomerID: 1,
ChannelID: 11,
AIAgentID: aiAgentID,
Status: enums.IMConversationStatusAIServing,
LastActiveAt: now,
AuditFields: models.AuditFields{
CreatedAt: now,
UpdatedAt: now,
},
}
if err := db.Create(item).Error; err != nil {
t.Fatalf("create conversation: %v", err)
}
return item
}
internal/ai/runtime/reply_interrupt_service.go
+24 -20
View File
@@ -32,11 +32,12 @@ func (s *replyInterruptService) ResumePendingInterrupt(ctx context.Context, owne
summary = expiredInterruptSummary()
replyCtx.setSummary(summary)
replyMessage, expireErr := owner.commit.CommitAIReply(replyCommitInput{
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: summary.ReplyText,
ClientPrefix: "ai_interrupt_expired",
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: summary.ReplyText,
ClientPrefix: "ai_interrupt_expired",
WorkflowRunID: summary.WorkflowRunID,
})
if expireErr != nil {
return expireErr
@@ -57,11 +58,12 @@ func (s *replyInterruptService) ResumePendingInterrupt(ctx context.Context, owne
}
if summary != nil && strings.TrimSpace(summary.ReplyText) != "" {
replyMessage, err := owner.commit.CommitAIReply(replyCommitInput{
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: summary.ReplyText,
ClientPrefix: "ai_resume",
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: summary.ReplyText,
ClientPrefix: "ai_resume",
WorkflowRunID: summary.WorkflowRunID,
})
if err != nil {
return err
@@ -86,11 +88,12 @@ func (s *replyInterruptService) HandleInterruptedSummary(owner *aiReplyService,
pending = svc.ConversationInterruptService.GetByCheckPointID(summary.CheckPointID)
replyText := resolveInterruptPrompt(summary)
replyMessage, err := owner.commit.CommitAIReply(replyCommitInput{
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: replyText,
ClientPrefix: "ai_interrupt",
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: replyText,
ClientPrefix: "ai_interrupt",
WorkflowRunID: summary.WorkflowRunID,
})
if err != nil {
return err
@@ -107,11 +110,12 @@ func (s *replyInterruptService) HandleInterruptedResume(owner *aiReplyService, r
}
replyText := resolveInterruptPrompt(summary)
replyMessage, err := owner.commit.CommitAIReply(replyCommitInput{
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: replyText,
ClientPrefix: "ai_interrupt_resume",
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: replyText,
ClientPrefix: "ai_interrupt_resume",
WorkflowRunID: summary.WorkflowRunID,
})
if err != nil {
return err
internal/ai/runtime/reply_trigger_service.go
+6 -5
View File
@@ -84,11 +84,12 @@ func (s *aiReplyService) executeReply(ctx context.Context, replyCtx aiReplyConte
}
if summary != nil && strings.TrimSpace(summary.ReplyText) != "" {
_, err := s.commit.CommitAIReply(replyCommitInput{
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: summary.ReplyText,
ClientPrefix: "ai_reply",
Conversation: replyCtx.Conversation,
Message: replyCtx.Message,
AIAgent: replyCtx.AIAgent,
ReplyText: summary.ReplyText,
ClientPrefix: "ai_reply",
WorkflowRunID: summary.WorkflowRunID,
})
if err != nil {
return err
internal/builders/conversation_builder.go
+1
View File
@@ -150,6 +150,7 @@ func BuildMessageWithReadStatesAndLocale(item *models.Message, agentReadState, c
ID: item.ID,
ConversationID: item.ConversationID,
RequestID: item.RequestID,
WorkflowRunID: item.WorkflowRunID,
ClientMsgID: item.ClientMsgID,
SenderType: item.SenderType,
SenderID: item.SenderID,
internal/builders/conversation_builder_test.go
+17
View File
@@ -3,6 +3,8 @@ package builders
import (
"testing"
"agent-desk/internal/models"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/i18nx"
)
@@ -97,3 +99,18 @@ func TestLocalizeRenderableMessageContent(t *testing.T) {
})
}
}
func TestBuildMessageIncludesWorkflowRunID(t *testing.T) {
resp := BuildMessageWithReadStatesAndLocale(&models.Message{
ID: 1,
ConversationID: 2,
SenderType: enums.IMSenderTypeAI,
MessageType: enums.IMMessageTypeText,
Content: "AI reply",
WorkflowRunID: 9988,
}, nil, nil, nil, nil, nil, i18nx.DefaultLocale)
if resp.WorkflowRunID != 9988 {
t.Fatalf("resp.WorkflowRunID=%d want 9988", resp.WorkflowRunID)
}
}
internal/models/models.go
+1
View File
@@ -390,6 +390,7 @@ type Message struct {
ID int64 `gorm:"primaryKey;autoIncrement"`
ConversationID int64 `gorm:"type:bigint;not null;index;uniqueIndex:uk_conversation_seq;uniqueIndex:uk_conversation_client_msg"`
RequestID string `gorm:"type:varchar(128);not null;default:'';index"`
WorkflowRunID int64 `gorm:"type:bigint;not null;default:0;index"`
ClientMsgID string `gorm:"type:varchar(128);not null;default:'';uniqueIndex:uk_conversation_client_msg"`
SenderType enums.IMSenderType `gorm:"type:varchar(30);not null;default:'';index"`
SenderID int64 `gorm:"type:bigint;not null;default:0;index"`
internal/pkg/dto/response/message_response.go
+1
View File
@@ -6,6 +6,7 @@ type MessageResponse struct {
ID int64 `json:"id"`
ConversationID int64 `json:"conversationId"`
RequestID string `json:"requestId,omitempty"`
WorkflowRunID int64 `json:"workflowRunId,omitempty"`
ClientMsgID string `json:"clientMsgId,omitempty"`
SenderType enums.IMSenderType `json:"senderType"`
SenderID int64 `json:"senderId"`
internal/services/message_service.go
+15 -10
View File
@@ -132,11 +132,11 @@ func (s *messageService) GetConversationReadTarget(conversationID, messageID int
func (s *messageService) SendMessage(conversationID int64, senderType enums.IMSenderType, reqSenderID int64, clientMsgID string, messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, external *openidentity.ExternalUser) (*models.Message, error) {
switch senderType {
case enums.IMSenderTypeAgent:
return s.sendMessage(conversationID, enums.IMSenderTypeAgent, reqSenderID, clientMsgID, messageType, content, payload, operator, nil, "")
return s.sendMessage(conversationID, enums.IMSenderTypeAgent, reqSenderID, clientMsgID, messageType, content, payload, operator, nil, "", 0)
case enums.IMSenderTypeAI:
return s.sendMessage(conversationID, enums.IMSenderTypeAI, reqSenderID, clientMsgID, messageType, content, payload, operator, nil, "")
return s.sendMessage(conversationID, enums.IMSenderTypeAI, reqSenderID, clientMsgID, messageType, content, payload, operator, nil, "", 0)
case enums.IMSenderTypeCustomer:
return s.sendMessage(conversationID, enums.IMSenderTypeCustomer, 0, clientMsgID, messageType, content, payload, nil, external, "")
return s.sendMessage(conversationID, enums.IMSenderTypeCustomer, 0, clientMsgID, messageType, content, payload, nil, external, "", 0)
default:
return nil, errorsx.InvalidParamI18n("error.e0080")
}
@@ -147,7 +147,7 @@ func (s *messageService) SendAgentMessage(conversationID int64, reqSenderID int6
}
func (s *messageService) SendAgentMessageWithRequestID(conversationID int64, reqSenderID int64, clientMsgID string, messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, requestID string) (*models.Message, error) {
return s.sendMessage(conversationID, enums.IMSenderTypeAgent, reqSenderID, clientMsgID, messageType, content, payload, operator, nil, requestID)
return s.sendMessage(conversationID, enums.IMSenderTypeAgent, reqSenderID, clientMsgID, messageType, content, payload, operator, nil, requestID, 0)
}
func (s *messageService) RecallAgentMessage(messageID int64, operator *dto.AuthPrincipal) (*models.Message, error) {
@@ -250,7 +250,11 @@ func (s *messageService) SendAIMessage(conversationID int64, aiAgentID int64, cl
}
func (s *messageService) SendAIMessageWithRequestID(conversationID int64, aiAgentID int64, clientMsgID string, messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, requestID string) (*models.Message, error) {
return s.sendMessage(conversationID, enums.IMSenderTypeAI, aiAgentID, clientMsgID, messageType, content, payload, operator, nil, requestID)
return s.SendAIMessageWithRequestIDAndWorkflowRunID(conversationID, aiAgentID, clientMsgID, messageType, content, payload, operator, requestID, 0)
}
func (s *messageService) SendAIMessageWithRequestIDAndWorkflowRunID(conversationID int64, aiAgentID int64, clientMsgID string, messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, requestID string, workflowRunID int64) (*models.Message, error) {
return s.sendMessage(conversationID, enums.IMSenderTypeAI, aiAgentID, clientMsgID, messageType, content, payload, operator, nil, requestID, workflowRunID)
}
func (s *messageService) SendAIServiceNotice(conversationID int64, aiAgentID int64, content string) (*models.Message, error) {
@@ -269,7 +273,7 @@ func (s *messageService) SendAIServiceNoticeWithRequestID(conversationID int64,
UserID: 0,
Username: "system",
Nickname: "system",
}, nil, requestID)
}, nil, requestID, 0)
}
func (s *messageService) createAIWelcomeMessage(ctx *sqls.TxContext, conversation *models.Conversation, aiAgent *models.AIAgent, now time.Time) (*models.Message, error) {
@@ -370,11 +374,11 @@ func (s *messageService) SendCustomerMessage(conversationID int64, clientMsgID s
func (s *messageService) SendCustomerMessageWithRequestID(conversationID int64, clientMsgID string, messageType enums.IMMessageType, content, payload string, external openidentity.ExternalUser, requestID string) (*models.Message, error) {
ext := external
return s.sendMessage(conversationID, enums.IMSenderTypeCustomer, 0, clientMsgID, messageType, content, payload, nil, &ext, requestID)
return s.sendMessage(conversationID, enums.IMSenderTypeCustomer, 0, clientMsgID, messageType, content, payload, nil, &ext, requestID, 0)
}
func (s *messageService) sendMessage(conversationID int64, senderType enums.IMSenderType, reqSenderID int64, clientMsgID string,
messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, external *openidentity.ExternalUser, requestID string) (*models.Message, error) {
messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, external *openidentity.ExternalUser, requestID string, workflowRunID int64) (*models.Message, error) {
if senderType == enums.IMSenderTypeCustomer {
if external == nil || strings.TrimSpace(external.ExternalID) == "" {
@@ -391,11 +395,11 @@ func (s *messageService) sendMessage(conversationID int64, senderType enums.IMSe
if err != nil {
return nil, err
}
return s.sendValidatedMessage(conversation, senderType, reqSenderID, clientMsgID, messageType, content, payload, operator, external, requestID)
return s.sendValidatedMessage(conversation, senderType, reqSenderID, clientMsgID, messageType, content, payload, operator, external, requestID, workflowRunID)
}
func (s *messageService) sendValidatedMessage(conversation *models.Conversation, senderType enums.IMSenderType, reqSenderID int64, clientMsgID string,
messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, external *openidentity.ExternalUser, requestID string) (*models.Message, error) {
messageType enums.IMMessageType, content, payload string, operator *dto.AuthPrincipal, external *openidentity.ExternalUser, requestID string, workflowRunID int64) (*models.Message, error) {
var err error
var summary string
@@ -432,6 +436,7 @@ func (s *messageService) sendValidatedMessage(conversation *models.Conversation,
message := &models.Message{
ConversationID: conversation.ID,
RequestID: traceID,
WorkflowRunID: workflowRunID,
ClientMsgID: clientMsgID,
SenderType: senderType,
SenderID: reqSenderID,
internal/services/message_service_test.go
+57
View File
@@ -6,6 +6,7 @@ import (
"time"
"agent-desk/internal/models"
"agent-desk/internal/pkg/dto"
"agent-desk/internal/pkg/enums"
"agent-desk/internal/pkg/openidentity"
@@ -103,6 +104,30 @@ func welcomeTestExternalUser(id string) openidentity.ExternalUser {
}
}
func createMessageTestConversation(t *testing.T, db *gorm.DB, aiAgentID int64) *models.Conversation {
t.Helper()
now := time.Now()
conversation := &models.Conversation{
CustomerID: 1,
ChannelID: 11,
AIAgentID: aiAgentID,
Status: enums.IMConversationStatusAIServing,
LastActiveAt: now,
AuditFields: models.AuditFields{
CreatedAt: now,
UpdatedAt: now,
},
}
if err := db.Create(conversation).Error; err != nil {
t.Fatalf("create conversation: %v", err)
}
return conversation
}
func workflowTestAIPrincipal() *dto.AuthPrincipal {
return &dto.AuthPrincipal{UserID: 0, Username: "AI", Nickname: "AI"}
}
func TestConversationCreateCreatesAIWelcomeMessage(t *testing.T) {
db := setupMessageWelcomeTestDB(t)
aiAgent := createWelcomeTestAIAgent(t, db, " 您好，请问有什么可以帮您？ ")
@@ -197,6 +222,38 @@ func TestSendCustomerMessageStoresRequestIDOnMessageAndEvent(t *testing.T) {
}
}
func TestSendAIMessageStoresWorkflowRunID(t *testing.T) {
db := setupMessageWelcomeTestDB(t)
aiAgent := createWelcomeTestAIAgent(t, db, "")
conversation := createMessageTestConversation(t, db, aiAgent.ID)
message, err := MessageService.SendAIMessageWithRequestIDAndWorkflowRunID(
conversation.ID,
aiAgent.ID,
"ai-reply-workflow-1",
enums.IMMessageTypeText,
"AI reply",
"",
workflowTestAIPrincipal(),
"trace-workflow-1",
9988,
)
if err != nil {
t.Fatalf("SendAIMessageWithRequestIDAndWorkflowRunID() error = %v", err)
}
if message.WorkflowRunID != 9988 {
t.Fatalf("message.WorkflowRunID=%d want 9988", message.WorkflowRunID)
}
var stored models.Message
if err := db.First(&stored, message.ID).Error; err != nil {
t.Fatalf("find message: %v", err)
}
if stored.WorkflowRunID != 9988 {
t.Fatalf("stored.WorkflowRunID=%d want 9988", stored.WorkflowRunID)
}
}
func TestConversationCreateDoesNotDuplicateWelcomeMessageForExistingConversation(t *testing.T) {
db := setupMessageWelcomeTestDB(t)
aiAgent := createWelcomeTestAIAgent(t, db, "欢迎咨询")
internal/services/ws_service.go
+1
View File
@@ -327,6 +327,7 @@ func (s *wsService) buildRealtimeMessage(item *models.Message) response.MessageR
ID: item.ID,
ConversationID: item.ConversationID,
RequestID: item.RequestID,
WorkflowRunID: item.WorkflowRunID,
ClientMsgID: item.ClientMsgID,
SenderType: item.SenderType,
SenderID: item.SenderID,