ai-agent/internal/ai/runtime/reply_trigger_service.go

package runtime

import (
	"context"
	"fmt"
	"log/slog"
	"strconv"
	"strings"
	"time"

	"code.tczkiot.com/wlw/ai-agent/contract"
	applicationruntime "code.tczkiot.com/wlw/ai-agent/internal/ai/application/runtime"
	"code.tczkiot.com/wlw/ai-agent/internal/models"
	"code.tczkiot.com/wlw/ai-agent/internal/pkg/enums"
	"code.tczkiot.com/wlw/ai-agent/internal/pkg/tracex"
	svc "code.tczkiot.com/wlw/ai-agent/internal/services"

	"github.com/mlogclub/simple/sqls"
)

const aiReplyFailedReply = "已收到您的消息，但本次智能处理没有完成。请稍后重试，或回复“人工客服”继续处理。"

const businessIdentityMenuPrefix = "identity_menu"

const aiReplyInvocationToolCode = "runtime/ai_reply"

const aiReplyInvocationRecoveryGrace = 30 * time.Second

func (s *aiReplyService) resolveReplyTimeout(aiAgent models.AIAgent) time.Duration {
	if aiAgent.ReplyTimeoutSeconds <= 0 {
		return time.Duration(defaultAIReplyAsyncTimeoutSeconds) * time.Second
	}
	if aiAgent.ReplyTimeoutSeconds > maxAIReplyAsyncTimeoutSeconds {
		return time.Duration(maxAIReplyAsyncTimeoutSeconds) * time.Second
	}
	return time.Duration(aiAgent.ReplyTimeoutSeconds) * time.Second
}

func (s *aiReplyService) TriggerReplyAsync(requestContext context.Context, conversation models.Conversation, message models.Message) {
	aiAgent := svc.AIAgentService.Get(conversation.AIAgentID)
	if aiAgent == nil || aiAgent.Status != enums.StatusOk {
		return
	}
	timeout := s.resolveReplyTimeout(*aiAgent)
	invocationKey := fmt.Sprintf("message:%d:revision:%d", message.ID, aiAgent.PublishedRevisionID)
	claim, err := svc.AgentToolInvocationService.ClaimRecoverable(
		conversation.ID,
		aiAgent.ID,
		aiReplyInvocationToolCode,
		invocationKey,
		time.Now().Add(-(timeout + aiReplyInvocationRecoveryGrace)),
	)
	if err != nil {
		slog.Error("failed to claim ai reply run",
			"requestId", message.RequestID,
			"conversation_id", conversation.ID,
			"message_id", message.ID,
			"revision_id", aiAgent.PublishedRevisionID,
			"error", err)
		return
	}
	if claim == nil || claim.Item == nil || !claim.Acquired {
		return
	}
	if committedAIReply(conversation.ID, message.ID) != nil {
		if err := svc.AgentToolInvocationService.Complete(claim.Item, `{}`); err != nil {
			slog.Error("failed to reconcile recovered ai reply claim", "conversation_id", conversation.ID, "message_id", message.ID, "error", err)
		}
		return
	}
	proofContext := contract.BindCustomerAccessProofToMessage(requestContext, conversation.ID, message.ID, message.RequestID)
	proof, hasProof := contract.CustomerAccessProofFromContext(proofContext)
	go func() {
		startedAt := time.Now()
		ctx := tracex.ContextWithRequestID(context.Background(), message.RequestID)
		if hasProof {
			ctx = contract.WithCustomerAccessProof(ctx, proof)
		}
		ctx, cancel := context.WithTimeout(ctx, timeout)
		defer cancel()
		defer func() {
			if recovered := recover(); recovered != nil {
				err := fmt.Errorf("ai reply panic: %v", recovered)
				_ = svc.AgentToolInvocationService.FailRetryable(claim.Item, err)
				slog.Error("panic while triggering ai reply",
					"requestId", message.RequestID,
					"conversation_id", conversation.ID,
					"message_id", message.ID,
					"error", err)
				s.commitFailureReplyIfNeeded(conversation, message, *aiAgent, err)
			}
		}()
		var triggerErr error
		if s.triggerReply != nil {
			triggerErr = s.triggerReply(ctx, conversation, message, *aiAgent)
		} else {
			triggerErr = s.TriggerReply(ctx, conversation, message, *aiAgent)
		}
		if triggerErr != nil {
			_ = svc.AgentToolInvocationService.FailRetryable(claim.Item, triggerErr)
			slog.Error("failed to trigger ai reply",
				"requestId", message.RequestID,
				"message_id", message.ID,
				"timeout_ms", timeout.Milliseconds(),
				"elapsed_ms", time.Since(startedAt).Milliseconds(),
				"error", triggerErr)
			s.commitFailureReplyIfNeeded(conversation, message, *aiAgent, triggerErr)
			return
		}
		if err := svc.AgentToolInvocationService.Complete(claim.Item, `{}`); err != nil {
			slog.Error("failed to complete ai reply run claim",
				"requestId", message.RequestID,
				"conversation_id", conversation.ID,
				"message_id", message.ID,
				"error", err)
		}
	}()
}

func committedAIReply(conversationID, messageID int64) *models.Message {
	for _, prefix := range []string{"ai_reply", "identity_prompt", "ai_interrupt", "ai_interrupt_expired", "ai_resume", "ai_interrupt_resume"} {
		clientMsgID := fmt.Sprintf("%s_%d", prefix, messageID)
		if existing := svc.MessageService.FindOne(sqls.NewCnd().Eq("conversation_id", conversationID).Eq("client_msg_id", clientMsgID)); existing != nil {
			return existing
		}
	}
	return nil
}

func (s *aiReplyService) commitFailureReplyIfNeeded(conversation models.Conversation, message models.Message, aiAgent models.AIAgent, cause error) {
	clientMsgID := fmt.Sprintf("ai_error_%d", message.ID)
	if existing := svc.MessageService.FindOne(
		sqls.NewCnd().Eq("conversation_id", conversation.ID).Eq("client_msg_id", clientMsgID),
	); existing != nil {
		return
	}
	if _, err := s.commit.CommitAIReply(replyCommitInput{
		Conversation: conversation,
		Message:      message,
		AIAgent:      aiAgent,
		ReplyText:    aiReplyFailureText(cause),
		ClientPrefix: "ai_error",
	}); err != nil {
		slog.Error("failed to commit ai error reply",
			"requestId", message.RequestID,
			"conversation_id", conversation.ID,
			"message_id", message.ID,
			"error", err)
	}
}

func aiReplyFailureText(cause error) string {
	if cause == nil {
		return aiReplyFailedReply
	}
	message := strings.ToLower(cause.Error())
	switch {
	case strings.Contains(message, "ai 请求标识未设置"),
		strings.Contains(message, "invalid_ai_request_id"):
		return "系统内置 AI 请求失败：AI 请求标识无效。请联系管理员检查网关配置，或回复“人工客服”继续处理。"
	case strings.Contains(message, "insufficient_ai_balance"),
		strings.Contains(message, "ai 额度不足"):
		return "系统内置 AI 额度不足，请充值后重试，或回复“人工客服”继续处理。"
	case strings.Contains(message, "invalid_ai_key"),
		strings.Contains(message, "missing_ai_key"),
		strings.Contains(message, "ai key 格式无效"),
		strings.Contains(message, "ai 授权凭证无效"):
		return "系统内置 AI Key 无效或已撤销，请联系管理员检查客服设置。"
	case strings.Contains(message, "ai_gateway_not_configured"),
		strings.Contains(message, "system built-in llm model is not configured"),
		strings.Contains(message, "系统内置模型尚未配置"):
		return "系统内置 AI 模型尚未配置，请联系管理员完成配置。"
	case strings.Contains(message, "context deadline exceeded"),
		strings.Contains(message, "request timeout"),
		strings.Contains(message, "client.timeout"):
		return "系统内置 AI 请求超时，请稍后重试，或回复“人工客服”继续处理。"
	case strings.Contains(message, "internal_error"),
		strings.Contains(message, "网关内部异常"):
		if requestID := extractAIGatewayRequestID(cause.Error()); requestID != "" {
			return "系统内置 AI 网关内部异常，请稍后重试；排查编号：" + requestID + "。如仍失败，请联系管理员或回复“人工客服”。"
		}
		return "系统内置 AI 网关内部异常，请稍后重试；如仍失败，请联系管理员或回复“人工客服”。"
	case isAIUpstreamError(message):
		return aiUpstreamFailureText(message)
	default:
		return aiReplyFailedReply
	}
}

func extractAIGatewayRequestID(message string) string {
	lowerMessage := strings.ToLower(message)
	for _, marker := range []string{"request_id:", "request_id="} {
		start := strings.Index(lowerMessage, marker)
		if start < 0 {
			continue
		}
		value := strings.TrimSpace(message[start+len(marker):])
		value = strings.TrimLeft(value, "(\"'")
		end := 0
		for end < len(value) {
			char := value[end]
			if (char >= 'a' && char <= 'z') ||
				(char >= 'A' && char <= 'Z') ||
				(char >= '0' && char <= '9') || char == '-' || char == '_' {
				end++
				continue
			}
			break
		}
		if end > 0 {
			return value[:end]
		}
	}
	return ""
}

func isAIUpstreamError(message string) bool {
	if strings.Contains(message, "ai_upstream_failed") || strings.Contains(message, "ai_request_failed") {
		return true
	}
	providerMentioned := strings.Contains(message, "qwen") ||
		strings.Contains(message, "千问") ||
		strings.Contains(message, "deepseek") ||
		strings.Contains(message, "dashscope")
	if !providerMentioned {
		return false
	}
	for _, marker := range []string{
		"status code", "bad request", "unauthorized", "forbidden", "too many requests",
		"returned 4", "returned 5", "返回 4", "返回 5", "invalidparameter",
		"invalid_parameter", "throttling", "arrearage", "accessdenied", "error",
	} {
		if strings.Contains(message, marker) {
			return true
		}
	}
	return false
}

func aiUpstreamFailureText(message string) string {
	provider := "上游模型"
	if strings.Contains(message, "qwen") || strings.Contains(message, "千问") || strings.Contains(message, "dashscope") {
		provider = "千问"
	} else if strings.Contains(message, "deepseek") {
		provider = "DeepSeek"
	}
	switch {
	case strings.Contains(message, "model not exist"),
		strings.Contains(message, "model_not_found"),
		strings.Contains(message, "invalid model"),
		strings.Contains(message, "model.accessdenied"),
		strings.Contains(message, "model access denied"),
		strings.Contains(message, "模型不存在"):
		return "系统内置 AI 调用失败：" + provider + "模型不存在或暂不可用，也可能尚未开通，请联系管理员检查模型名称和开通状态。"
	case strings.Contains(message, "authentication"),
		strings.Contains(message, "invalid api key"),
		strings.Contains(message, "invalid_api_key"),
		strings.Contains(message, "invalidapikey"),
		strings.Contains(message, "unauthorized"):
		return "系统内置 AI 调用失败：" + provider + " API Key 无效或无权限，请联系管理员检查官网模型配置。"
	case strings.Contains(message, "rate limit"),
		strings.Contains(message, "rate_limit"),
		strings.Contains(message, "too many requests"),
		strings.Contains(message, "throttling"):
		return "系统内置 AI 调用失败：" + provider + "请求过于频繁，请稍后重试。"
	case strings.Contains(message, "insufficient balance"),
		strings.Contains(message, "insufficient quota"),
		strings.Contains(message, "arrearage"),
		strings.Contains(message, "quota"):
		return "系统内置 AI 调用失败：" + provider + "账户额度不足或已欠费，请联系管理员处理。"
	case strings.Contains(message, "maximum context length"),
		strings.Contains(message, "context_length"),
		strings.Contains(message, "input length"),
		strings.Contains(message, "tokens exceed"),
		strings.Contains(message, "too many tokens"):
		return "系统内置 AI 调用失败：" + provider + "请求内容超出模型上下文长度，请缩短消息后重试。"
	case strings.Contains(message, "does not support tools"),
		strings.Contains(message, "tool calling is not supported"),
		strings.Contains(message, "function calling is not supported"),
		strings.Contains(message, "unsupported tool"),
		strings.Contains(message, "unsupported function"):
		return "系统内置 AI 调用失败：当前" + provider + "模型不支持客服工具调用，请联系管理员更换可用模型。"
	case strings.Contains(message, "data_inspection_failed"),
		strings.Contains(message, "content_filter"),
		strings.Contains(message, "inappropriate content"):
		return "系统内置 AI 调用失败：" + provider + "拒绝了本次内容，请调整表述后重试。"
	case strings.Contains(message, "invalidparameter"),
		strings.Contains(message, "invalid_parameter"),
		strings.Contains(message, "bad request"),
		strings.Contains(message, "返回 400"),
		strings.Contains(message, "returned 400"):
		return "系统内置 AI 调用失败：" + provider + "请求参数不兼容，请联系管理员检查模型与客服工具配置。"
	default:
		return "系统内置 AI 调用失败：" + provider + "服务返回错误，请联系管理员在 AI 回复记录中查看详细原因。"
	}
}

func (s *aiReplyService) TriggerReply(ctx context.Context, conversation models.Conversation, message models.Message, aiAgent models.AIAgent) (retErr error) {
	var summary *applicationruntime.RunResult
	replyCtx := aiReplyContext{
		Conversation: conversation,
		Message:      message,
		AIAgent:      aiAgent,
		SummaryRef:   &summary,
	}
	if err := ctx.Err(); err != nil {
		return err
	}
	if s.eligibility != nil && !s.eligibility.CanReply(conversation, message, aiAgent) {
		return nil
	}
	if !IsAIAgentRolloutEligible(conversation, aiAgent, svc.ChannelService.Get(conversation.ChannelID)) {
		return nil
	}
	identityResolution, identityErr := resolveGuestBusinessIdentity(ctx, conversation, message)
	if identityErr != nil || identityResolution.NeedsPrompt {
		_, err := s.commit.CommitAIReply(replyCommitInput{
			Conversation: conversation,
			Message:      message,
			AIAgent:      aiAgent,
			ReplyText:    guestBusinessIdentityPrompt(identityResolution, identityErr),
			ClientPrefix: "identity_prompt",
		})
		return err
	}
	replyCtx.Conversation = identityResolution.Conversation
	if pendingInterrupt := svc.ConversationInterruptService.FindLatestPendingByConversationID(conversation.ID); pendingInterrupt != nil {
		replyCtx.PendingInterrupt = pendingInterrupt
		return s.resumePendingInterrupt(ctx, replyCtx)
	}
	if identityResolution.CandidateProvided && isBusinessIdentityOnlyMessage(message) {
		handled, err := s.sendBusinessIdentityMenu(ctx, replyCtx)
		if handled || err != nil {
			return err
		}
	}
	if selection, ok := businessIdentityMenuSelection(message); ok {
		latest := latestAIMessage(conversation.ID)
		if latest != nil && strings.HasPrefix(latest.ClientMsgID, businessIdentityMenuPrefix+"_") {
			matched, aiMessage, err := svc.CustomerQuickActionService.ExecuteSelectedReply(
				ctx, &replyCtx.Conversation, selection, message.RequestID, message.ID,
			)
			if matched || err != nil {
				return s.finishQuickActionReply(ctx, replyCtx, matched, aiMessage, err)
			}
		}
		if actionCode, matched := legacyCardMenuActionCode(latest, selection); matched {
			actionMatched, aiMessage, err := svc.CustomerQuickActionService.ExecuteActionReply(
				ctx, &replyCtx.Conversation, actionCode, message.RequestID, message.ID,
			)
			if actionMatched || err != nil {
				return s.finishQuickActionReply(ctx, replyCtx, actionMatched, aiMessage, err)
			}
		}
	}
	if matched, err := svc.CustomerQuickActionService.ExecuteMatchedReply(
		ctx,
		&replyCtx.Conversation,
		message.Content,
		message.RequestID,
		message.ID,
	); matched || err != nil {
		return err
	}
	return s.executeReply(ctx, replyCtx)
}

func (s *aiReplyService) sendBusinessIdentityMenu(ctx context.Context, replyCtx aiReplyContext) (bool, error) {
	actions, err := svc.CustomerQuickActionService.ListForConversation(ctx, &replyCtx.Conversation)
	if err != nil || len(actions) == 0 {
		return false, err
	}
	objectLabel := "业务对象"
	switch replyCtx.Conversation.CustomerType {
	case "card":
		objectLabel = "卡号"
	case "device":
		objectLabel = "设备号"
	case "mall_user":
		objectLabel = "商城用户"
	}
	var builder strings.Builder
	builder.WriteString("已识别")
	builder.WriteString(objectLabel)
	if identifier := strings.TrimSpace(replyCtx.Conversation.CustomerExternalID); identifier != "" {
		builder.WriteString("：")
		builder.WriteString(identifier)
	}
	builder.WriteString("。\n\n请回复序号选择需要的服务：")
	for index, action := range actions {
		builder.WriteString(fmt.Sprintf("\n%d. %s", index+1, action.Title))
	}
	builder.WriteString("\n\n也可以直接输入要咨询的问题。")
	_, err = svc.MessageService.SendAutomaticServiceMessageWithRequestID(
		replyCtx.Conversation.ID,
		fmt.Sprintf("%s_%d", businessIdentityMenuPrefix, replyCtx.Message.ID),
		builder.String(),
		replyCtx.Message.RequestID,
	)
	return true, err
}

func isBusinessIdentityOnlyMessage(message models.Message) bool {
	content := businessIdentityMessageContent(message)
	candidates, _ := businessIdentityCandidates(content)
	for _, candidate := range candidates {
		content = strings.ReplaceAll(content, candidate, "")
	}
	for _, marker := range []string{"卡号", "卡板", "设备号", "设备", "iccid", "imei"} {
		content = strings.ReplaceAll(strings.ToLower(content), marker, "")
	}
	content = strings.Map(func(r rune) rune {
		if r == ' ' || r == '　' || r == '\n' || r == '\r' || r == '\t' || r == ' ' {
			return -1
		}
		switch r {
		case '：', ':', '，', ',', '。', '.', '；', ';', '-', '_':
			return -1
		default:
			return r
		}
	}, content)
	return content == ""
}

func businessIdentityMenuSelection(message models.Message) (int, bool) {
	content := strings.TrimSpace(businessIdentityMessageContent(message))
	selection, err := strconv.Atoi(content)
	return selection, err == nil && selection > 0
}

func latestAIMessage(conversationID int64) *models.Message {
	return svc.MessageService.FindOne(sqls.NewCnd().
		Eq("conversation_id", conversationID).
		Eq("sender_type", enums.IMSenderTypeAI).
		Desc("id"))
}

func legacyCardMenuActionCode(message *models.Message, selection int) (string, bool) {
	if message == nil || selection <= 0 {
		return "", false
	}
	content := businessIdentityMessageContent(*message)
	if strings.Contains(content, "卡片提示停机") &&
		strings.Contains(content, "无法上网") &&
		strings.Contains(content, "无信号") &&
		strings.Contains(content, "已充值但未恢复") {
		if selection >= 1 && selection <= 4 {
			return "card/network_diagnosis", true
		}
		return "", false
	}
	if strings.Contains(content, "卡片状态") &&
		strings.Contains(content, "网络连接") &&
		strings.Contains(content, "套餐") &&
		strings.Contains(content, "其他问题") {
		actions := map[int]string{
			1: "card/status",
			2: "card/network_diagnosis",
			3: "card/package",
		}
		code, ok := actions[selection]
		return code, ok
	}
	return "", false
}

func (s *aiReplyService) finishQuickActionReply(
	ctx context.Context,
	replyCtx aiReplyContext,
	matched bool,
	aiMessage string,
	err error,
) error {
	if err != nil || !matched {
		return err
	}
	if strings.TrimSpace(aiMessage) == "" {
		return nil
	}
	replyCtx.Message.Content = aiMessage
	replyCtx.Message.MessageType = enums.IMMessageTypeText
	return s.executeReply(ctx, replyCtx)
}

func (s *aiReplyService) resumePendingInterrupt(ctx context.Context, replyCtx aiReplyContext) error {
	return s.interrupts.ResumePendingInterrupt(ctx, s, replyCtx)
}

func (s *aiReplyService) executeReply(ctx context.Context, replyCtx aiReplyContext) error {
	summary, err := s.executor.Run(ctx, runtimeReplyRunInput{
		Conversation: replyCtx.Conversation,
		Message:      replyCtx.Message,
		AIAgent:      replyCtx.AIAgent,
	})
	replyCtx.setSummary(summary)
	if err != nil {
		return err
	}
	if summary != nil && summary.Interrupted {
		return s.interrupts.HandleInterruptedSummary(s, replyCtx, summary)
	}
	if summary != nil && summary.HandoffRequested {
		if _, err := svc.ConversationHumanDispatchService.HandoffByAIWithRequestID(
			replyCtx.Conversation.ID,
			replyCtx.AIAgent,
			summary.HandoffReason,
			replyCtx.Message.RequestID,
		); err != nil {
			return err
		}
		return nil
	}
	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",
		})
		if err != nil {
			return err
		}
	}
	return nil
}