diff --git a/internal/ai/application/runtime/prepare_service.go b/internal/ai/application/runtime/prepare_service.go index 8881cb6..fb189aa 100644 --- a/internal/ai/application/runtime/prepare_service.go +++ b/internal/ai/application/runtime/prepare_service.go @@ -1,5 +1,7 @@ package runtime +import "cs-agent/internal/ai/runtime/registry" + func newPrepareService(catalog *toolCatalog) *prepareService { return &prepareService{catalog: catalog} } @@ -8,30 +10,16 @@ type prepareService struct { catalog *toolCatalog } -func (s *prepareService) prepareToolsForRun(req *Request) error { - if req == nil || req.ToolSet != nil || s.catalog == nil { - return nil +func (s *prepareService) prepareToolsForRun(req Request) (*registry.ToolSet, error) { + if req.ToolSet != nil { + return req.ToolSet, nil } - toolSet, err := s.catalog.resolveForRun(req) - if err != nil { - return err - } - if toolSet != nil { - req.ToolSet = toolSet - } - return nil + return s.catalog.resolveForRun(req) } -func (s *prepareService) prepareToolsForResume(req *ResumeRequest) error { - if req == nil || req.ToolSet != nil || s.catalog == nil { - return nil +func (s *prepareService) prepareToolsForResume(req ResumeRequest) (*registry.ToolSet, error) { + if req.ToolSet != nil { + return req.ToolSet, nil } - toolSet, err := s.catalog.resolveForResume(req) - if err != nil { - return err - } - if toolSet != nil { - req.ToolSet = toolSet - } - return nil + return s.catalog.resolveForResume(req) } diff --git a/internal/ai/application/runtime/service.go b/internal/ai/application/runtime/service.go index 6de73bd..c67b6e1 100644 --- a/internal/ai/application/runtime/service.go +++ b/internal/ai/application/runtime/service.go @@ -23,13 +23,12 @@ func NewService() *Service { } func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) { - if s == nil || s.runtime == nil || s.prepare == nil { - return nil, nil - } req.UserMessage.Content = utils.BuildRuntimeMessageText(req.UserMessage.MessageType, req.UserMessage.Content) - if err := s.prepare.prepareToolsForRun(&req); err != nil { + toolSet, err := s.prepare.prepareToolsForRun(req) + if err != nil { return nil, err } + req.ToolSet = toolSet summary, err := s.runtime.ExecuteRun(ctx, executor.RunInput{ Conversation: req.Conversation, UserMessage: req.UserMessage, @@ -45,12 +44,11 @@ func (s *Service) Run(ctx context.Context, req Request) (*Summary, error) { } func (s *Service) Resume(ctx context.Context, req ResumeRequest) (*Summary, error) { - if s == nil || s.runtime == nil || s.prepare == nil { - return nil, nil - } - if err := s.prepare.prepareToolsForResume(&req); err != nil { + toolSet, err := s.prepare.prepareToolsForResume(req) + if err != nil { return nil, err } + req.ToolSet = toolSet summary, err := s.runtime.ExecuteResume(ctx, executor.ResumeInput{ Conversation: req.Conversation, AIAgent: req.AIAgent, diff --git a/internal/ai/application/runtime/tool_catalog.go b/internal/ai/application/runtime/tool_catalog.go index c72a906..0a525fa 100644 --- a/internal/ai/application/runtime/tool_catalog.go +++ b/internal/ai/application/runtime/tool_catalog.go @@ -32,10 +32,7 @@ func buildRuntimeStaticTools() []registry.Tool { return ret } -func (c *toolCatalog) resolveForRun(req *Request) (*registry.ToolSet, error) { - if req == nil || req.ToolSet != nil || c == nil || c.registry == nil { - return nil, nil - } +func (c *toolCatalog) resolveForRun(req Request) (*registry.ToolSet, error) { return c.registry.Resolve(registry.Context{ Conversation: req.Conversation, AIAgent: req.AIAgent, @@ -45,10 +42,7 @@ func (c *toolCatalog) resolveForRun(req *Request) (*registry.ToolSet, error) { }) } -func (c *toolCatalog) resolveForResume(req *ResumeRequest) (*registry.ToolSet, error) { - if req == nil || req.ToolSet != nil || c == nil || c.registry == nil { - return nil, nil - } +func (c *toolCatalog) resolveForResume(req ResumeRequest) (*registry.ToolSet, error) { return c.registry.Resolve(registry.Context{ Conversation: req.Conversation, AIAgent: req.AIAgent, diff --git a/internal/ai/runtime/reply_commit_service.go b/internal/ai/runtime/reply_commit_service.go index c312176..cef8cb0 100644 --- a/internal/ai/runtime/reply_commit_service.go +++ b/internal/ai/runtime/reply_commit_service.go @@ -16,46 +16,54 @@ import ( type replyCommitService struct{} +type replyCommitInput struct { + Conversation models.Conversation + Message models.Message + AIAgent models.AIAgent + ReplyText string + Trace *aiReplyTraceData + ClientPrefix string + IncrementRound bool +} + func newReplyCommitService() *replyCommitService { return &replyCommitService{} } -func (s *replyCommitService) SendAIReply(conversation models.Conversation, message models.Message, aiAgent models.AIAgent, - replyText string, trace *aiReplyTraceData, clientPrefix string) (*models.Message, error) { - replyText = strings.TrimSpace(replyText) +func (s *replyCommitService) SendAIReply(input replyCommitInput) (*models.Message, error) { + replyText := strings.TrimSpace(input.ReplyText) if replyText == "" { return nil, nil } commitStartedAt := time.Now() replyMessage, err := svc.MessageService.SendAIMessage( - conversation.ID, - aiAgent.ID, - fmt.Sprintf("%s_%d", strings.TrimSpace(clientPrefix), message.ID), + input.Conversation.ID, + input.AIAgent.ID, + fmt.Sprintf("%s_%d", strings.TrimSpace(input.ClientPrefix), input.Message.ID), enums.IMMessageTypeText, replyText, "", - s.buildAIPrincipal(aiAgent), + s.buildAIPrincipal(input.AIAgent), ) - if trace != nil { - trace.CommitMs = time.Since(commitStartedAt).Milliseconds() - trace.ReplySent = err == nil && replyMessage != nil + if input.Trace != nil { + input.Trace.CommitMs = time.Since(commitStartedAt).Milliseconds() + input.Trace.ReplySent = err == nil && replyMessage != nil if replyMessage != nil { - trace.ReplyMessageID = replyMessage.ID + input.Trace.ReplyMessageID = replyMessage.ID } } + if err != nil || !input.IncrementRound { + return replyMessage, err + } + if err := s.IncrementAIReplyRounds(input.Conversation.ID, input.Conversation.AIReplyRounds+1, input.AIAgent.Name); err != nil { + return nil, err + } return replyMessage, err } -func (s *replyCommitService) CommitAIReply(conversation models.Conversation, message models.Message, aiAgent models.AIAgent, - replyText string, trace *aiReplyTraceData, clientPrefix string) (*models.Message, error) { - replyMessage, err := s.SendAIReply(conversation, message, aiAgent, replyText, trace, clientPrefix) - if err != nil { - return nil, err - } - if err := s.IncrementAIReplyRounds(conversation.ID, conversation.AIReplyRounds+1, aiAgent.Name); err != nil { - return nil, err - } - return replyMessage, nil +func (s *replyCommitService) CommitAIReply(input replyCommitInput) (*models.Message, error) { + input.IncrementRound = true + return s.SendAIReply(input) } func (s *replyCommitService) IncrementAIReplyRounds(conversationID int64, nextRounds int, aiAgentName string) error { diff --git a/internal/ai/runtime/reply_interrupt_service.go b/internal/ai/runtime/reply_interrupt_service.go index eea853e..48543e8 100644 --- a/internal/ai/runtime/reply_interrupt_service.go +++ b/internal/ai/runtime/reply_interrupt_service.go @@ -18,10 +18,16 @@ func newReplyInterruptService() *replyInterruptService { func (s *replyInterruptService) ResumePendingInterrupt(ctx context.Context, owner *aiReplyService, conversation models.Conversation, message models.Message, aiAgent models.AIAgent, pendingInterrupt *models.ConversationInterrupt, trace *aiReplyTraceData, summaryRef **applicationruntime.Summary) error { - if pendingInterrupt == nil || owner == nil || owner.executor == nil { + if pendingInterrupt == nil || owner == nil { return nil } - summary, err := owner.executor.ResumePendingInterrupt(ctx, conversation, message, aiAgent, pendingInterrupt, trace) + summary, err := owner.executor.ResumePendingInterrupt(ctx, runtimeReplyResumeInput{ + Conversation: conversation, + Message: message, + AIAgent: aiAgent, + PendingInterrupt: pendingInterrupt, + Trace: trace, + }) *summaryRef = summary if err != nil { if isCheckpointMissingError(err) { @@ -29,7 +35,14 @@ func (s *replyInterruptService) ResumePendingInterrupt(ctx context.Context, owne *summaryRef = summary trace.Status = "interrupt_expired" trace.FinalAction = "expired" - replyMessage, expireErr := owner.commit.CommitAIReply(conversation, message, aiAgent, summary.ReplyText, trace, "ai_interrupt_expired") + replyMessage, expireErr := owner.commit.CommitAIReply(replyCommitInput{ + Conversation: conversation, + Message: message, + AIAgent: aiAgent, + ReplyText: summary.ReplyText, + Trace: trace, + ClientPrefix: "ai_interrupt_expired", + }) if expireErr != nil { return expireErr } @@ -48,7 +61,14 @@ func (s *replyInterruptService) ResumePendingInterrupt(ctx context.Context, owne return s.HandleInterruptedResume(owner, conversation, message, aiAgent, pendingInterrupt, summary, trace) } if summary != nil && strings.TrimSpace(summary.ReplyText) != "" { - replyMessage, err := owner.commit.CommitAIReply(conversation, message, aiAgent, summary.ReplyText, trace, "ai_resume") + replyMessage, err := owner.commit.CommitAIReply(replyCommitInput{ + Conversation: conversation, + Message: message, + AIAgent: aiAgent, + ReplyText: summary.ReplyText, + Trace: trace, + ClientPrefix: "ai_resume", + }) if err != nil { return err } @@ -75,7 +95,14 @@ func (s *replyInterruptService) HandleInterruptedSummary(owner *aiReplyService, } pending = svc.ConversationInterruptService.GetByCheckPointID(summary.CheckPointID) replyText := resolveInterruptPrompt(summary) - replyMessage, err := owner.commit.CommitAIReply(conversation, message, aiAgent, replyText, trace, "ai_interrupt") + replyMessage, err := owner.commit.CommitAIReply(replyCommitInput{ + Conversation: conversation, + Message: message, + AIAgent: aiAgent, + ReplyText: replyText, + Trace: trace, + ClientPrefix: "ai_interrupt", + }) if err != nil { return err } @@ -91,7 +118,14 @@ func (s *replyInterruptService) HandleInterruptedResume(owner *aiReplyService, c return nil } replyText := resolveInterruptPrompt(summary) - replyMessage, err := owner.commit.CommitAIReply(conversation, message, aiAgent, replyText, trace, "ai_interrupt_resume") + replyMessage, err := owner.commit.CommitAIReply(replyCommitInput{ + Conversation: conversation, + Message: message, + AIAgent: aiAgent, + ReplyText: replyText, + Trace: trace, + ClientPrefix: "ai_interrupt_resume", + }) if err != nil { return err } diff --git a/internal/ai/runtime/reply_runlog_service.go b/internal/ai/runtime/reply_runlog_service.go index 8fd0588..c5e2ebd 100644 --- a/internal/ai/runtime/reply_runlog_service.go +++ b/internal/ai/runtime/reply_runlog_service.go @@ -18,47 +18,57 @@ func newReplyRunLogService() *replyRunLogService { type replyRunLogService struct{} -func (s *replyRunLogService) Write(startedAt time.Time, message models.Message, conversation models.Conversation, aiAgent models.AIAgent, - question string, runErr error, trace *aiReplyTraceData, summary *applicationruntime.Summary) { +type replyRunLogInput struct { + StartedAt time.Time + Message models.Message + Conversation models.Conversation + AIAgent models.AIAgent + Question string + RunErr error + Trace *aiReplyTraceData + Summary *applicationruntime.Summary +} + +func (s *replyRunLogService) Write(input replyRunLogInput) { errorMessage := "" - if runErr != nil { - errorMessage = runErr.Error() - } else if summary != nil && strings.TrimSpace(summary.ErrorMessage) != "" { - errorMessage = strings.TrimSpace(summary.ErrorMessage) + if input.RunErr != nil { + errorMessage = input.RunErr.Error() + } else if input.Summary != nil && strings.TrimSpace(input.Summary.ErrorMessage) != "" { + errorMessage = strings.TrimSpace(input.Summary.ErrorMessage) } - traceData := buildAIReplyTraceData(trace) - plannedAction, plannedToolCode, planReason := buildRunLogPlan(summary) + traceData := buildAIReplyTraceData(input.Trace) + plannedAction, plannedToolCode, planReason := buildRunLogPlan(input.Summary) logItem := &models.AgentRunLog{ - ConversationID: conversation.ID, - MessageID: message.ID, - AIAgentID: aiAgent.ID, - AIConfigID: aiAgent.AIConfigID, - UserMessage: strings.TrimSpace(question), + ConversationID: input.Conversation.ID, + MessageID: input.Message.ID, + AIAgentID: input.AIAgent.ID, + AIConfigID: input.AIAgent.AIConfigID, + UserMessage: strings.TrimSpace(input.Question), PlannedAction: plannedAction, - PlannedSkillCode: strings.TrimSpace(summaryPlannedSkillCode(summary)), - PlannedSkillName: strings.TrimSpace(summaryPlannedSkillName(summary)), - SkillRouteTrace: strings.TrimSpace(summarySkillRouteTrace(summary)), - ToolSearchTrace: extractToolSearchTrace(summary), - GraphToolTrace: extractGraphToolTrace(summary), - GraphToolCode: firstGraphToolCode(summary), - HandoffReason: extractHandoffReason(summary), + PlannedSkillCode: strings.TrimSpace(summaryPlannedSkillCode(input.Summary)), + PlannedSkillName: strings.TrimSpace(summaryPlannedSkillName(input.Summary)), + SkillRouteTrace: strings.TrimSpace(summarySkillRouteTrace(input.Summary)), + ToolSearchTrace: extractToolSearchTrace(input.Summary), + GraphToolTrace: extractGraphToolTrace(input.Summary), + GraphToolCode: firstGraphToolCode(input.Summary), + HandoffReason: extractHandoffReason(input.Summary), PlannedToolCode: plannedToolCode, PlanReason: planReason, - InterruptType: firstInterruptType(summary), - ResumeSource: runLogResumeSource(trace), - FinalAction: toRunLogFinalAction(summary), - FinalStatus: runLogFinalStatus(summary), - ReplyText: buildRunLogReplyText(summary), + InterruptType: firstInterruptType(input.Summary), + ResumeSource: runLogResumeSource(input.Trace), + FinalAction: toRunLogFinalAction(input.Summary), + FinalStatus: runLogFinalStatus(input.Summary), + ReplyText: buildRunLogReplyText(input.Summary), ErrorMessage: errorMessage, - LatencyMs: time.Since(startedAt).Milliseconds(), + LatencyMs: time.Since(input.StartedAt).Milliseconds(), TraceData: traceData, CreatedAt: time.Now(), } if err := svc.AgentRunLogService.Create(logItem); err != nil { slog.Warn("create agent run log failed", - "message_id", message.ID, + "message_id", input.Message.ID, "conversation_id", logItem.ConversationID, - "ai_agent_id", aiAgent.ID, + "ai_agent_id", input.AIAgent.ID, "error", err) } } diff --git a/internal/ai/runtime/reply_trigger_service.go b/internal/ai/runtime/reply_trigger_service.go index 74e8362..9c92286 100644 --- a/internal/ai/runtime/reply_trigger_service.go +++ b/internal/ai/runtime/reply_trigger_service.go @@ -53,7 +53,16 @@ func (s *aiReplyService) TriggerReply(ctx context.Context, conversation models.C return nil } defer func() { - s.runlog.Write(startedAt, message, conversation, aiAgent, message.Content, retErr, trace, summary) + s.runlog.Write(replyRunLogInput{ + StartedAt: startedAt, + Message: message, + Conversation: conversation, + AIAgent: aiAgent, + Question: message.Content, + RunErr: retErr, + Trace: trace, + Summary: summary, + }) }() if pendingInterrupt := svc.ConversationInterruptService.FindLatestPendingByConversationID(conversation.ID); pendingInterrupt != nil { return s.resumePendingInterrupt(ctx, conversation, message, aiAgent, pendingInterrupt, trace, &summary) @@ -63,18 +72,17 @@ func (s *aiReplyService) TriggerReply(ctx context.Context, conversation models.C func (s *aiReplyService) resumePendingInterrupt(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent, pendingInterrupt *models.ConversationInterrupt, trace *aiReplyTraceData, summaryRef **applicationruntime.Summary) error { - if s == nil || s.interrupts == nil { - return nil - } return s.interrupts.ResumePendingInterrupt(ctx, s, conversation, message, aiAgent, pendingInterrupt, trace, summaryRef) } func (s *aiReplyService) executeReply(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent, trace *aiReplyTraceData, summaryRef **applicationruntime.Summary) error { - if s == nil || s.executor == nil { - return nil - } - summary, err := s.executor.Run(ctx, conversation, message, aiAgent, trace) + summary, err := s.executor.Run(ctx, runtimeReplyRunInput{ + Conversation: conversation, + Message: message, + AIAgent: aiAgent, + Trace: trace, + }) if summaryRef != nil { *summaryRef = summary } @@ -85,7 +93,14 @@ func (s *aiReplyService) executeReply(ctx context.Context, conversation models.C return s.interrupts.HandleInterruptedSummary(s, conversation, message, aiAgent, summary, trace) } if summary != nil && strings.TrimSpace(summary.ReplyText) != "" { - replyMessage, err := s.commit.CommitAIReply(conversation, message, aiAgent, summary.ReplyText, trace, "ai_reply") + replyMessage, err := s.commit.CommitAIReply(replyCommitInput{ + Conversation: conversation, + Message: message, + AIAgent: aiAgent, + ReplyText: summary.ReplyText, + Trace: trace, + ClientPrefix: "ai_reply", + }) if err != nil { return err } diff --git a/internal/ai/runtime/runtime_reply_executor.go b/internal/ai/runtime/runtime_reply_executor.go index af91553..9143812 100644 --- a/internal/ai/runtime/runtime_reply_executor.go +++ b/internal/ai/runtime/runtime_reply_executor.go @@ -15,53 +15,68 @@ import ( type runtimeReplyExecutor struct{} +type runtimeReplyRunInput struct { + Conversation models.Conversation + Message models.Message + AIAgent models.AIAgent + Trace *aiReplyTraceData +} + +type runtimeReplyResumeInput struct { + Conversation models.Conversation + Message models.Message + AIAgent models.AIAgent + PendingInterrupt *models.ConversationInterrupt + Trace *aiReplyTraceData +} + func newRuntimeReplyExecutor() *runtimeReplyExecutor { return &runtimeReplyExecutor{} } -func (e *runtimeReplyExecutor) Run(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent, trace *aiReplyTraceData) (*applicationruntime.Summary, error) { - aiConfig := svc.AIConfigService.Get(aiAgent.AIConfigID) +func (e *runtimeReplyExecutor) Run(ctx context.Context, input runtimeReplyRunInput) (*applicationruntime.Summary, error) { + aiConfig := svc.AIConfigService.Get(input.AIAgent.AIConfigID) if aiConfig == nil { return nil, fmt.Errorf("ai config is nil") } runtimeStartedAt := time.Now() summary, err := Service.Run(ctx, applicationruntime.Request{ - Conversation: conversation, - UserMessage: message, - AIAgent: aiAgent, + Conversation: input.Conversation, + UserMessage: input.Message, + AIAgent: input.AIAgent, AIConfig: *aiConfig, }) - if trace != nil { - trace.RuntimeLatencyMs = time.Since(runtimeStartedAt).Milliseconds() - e.fillTraceFromSummary(trace, summary, err) + if input.Trace != nil { + input.Trace.RuntimeLatencyMs = time.Since(runtimeStartedAt).Milliseconds() + e.fillTraceFromSummary(input.Trace, summary, err) } return summary, err } -func (e *runtimeReplyExecutor) ResumePendingInterrupt(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent, pendingInterrupt *models.ConversationInterrupt, trace *aiReplyTraceData) (*applicationruntime.Summary, error) { - if pendingInterrupt == nil { +func (e *runtimeReplyExecutor) ResumePendingInterrupt(ctx context.Context, input runtimeReplyResumeInput) (*applicationruntime.Summary, error) { + if input.PendingInterrupt == nil { return nil, nil } - aiConfig := svc.AIConfigService.Get(aiAgent.AIConfigID) + aiConfig := svc.AIConfigService.Get(input.AIAgent.AIConfigID) if aiConfig == nil { return nil, fmt.Errorf("ai config is nil") } runtimeStartedAt := time.Now() - if trace != nil { - trace.ResumeSource = "pending_interrupt" + if input.Trace != nil { + input.Trace.ResumeSource = "pending_interrupt" } summary, err := Service.Resume(ctx, applicationruntime.ResumeRequest{ - Conversation: conversation, - AIAgent: aiAgent, + Conversation: input.Conversation, + AIAgent: input.AIAgent, AIConfig: *aiConfig, - CheckPointID: strings.TrimSpace(pendingInterrupt.CheckPointID), + CheckPointID: strings.TrimSpace(input.PendingInterrupt.CheckPointID), ResumeData: map[string]string{ - strings.TrimSpace(pendingInterrupt.InterruptID): strings.TrimSpace(message.Content), + strings.TrimSpace(input.PendingInterrupt.InterruptID): strings.TrimSpace(input.Message.Content), }, }) - if trace != nil { - trace.RuntimeLatencyMs = time.Since(runtimeStartedAt).Milliseconds() - e.fillTraceFromSummary(trace, summary, err) + if input.Trace != nil { + input.Trace.RuntimeLatencyMs = time.Since(runtimeStartedAt).Milliseconds() + e.fillTraceFromSummary(input.Trace, summary, err) } return summary, err } diff --git a/internal/ai/runtime/service.go b/internal/ai/runtime/service.go index d27fd24..a302c7b 100644 --- a/internal/ai/runtime/service.go +++ b/internal/ai/runtime/service.go @@ -19,15 +19,9 @@ type service struct { } func (s *service) Run(ctx context.Context, req applicationruntime.Request) (*applicationruntime.Summary, error) { - if s == nil || s.app == nil { - return nil, nil - } return s.app.Run(ctx, req) } func (s *service) Resume(ctx context.Context, req applicationruntime.ResumeRequest) (*applicationruntime.Summary, error) { - if s == nil || s.app == nil { - return nil, nil - } return s.app.Resume(ctx, req) }