根本原因：PKI 私聊需要接收方的公钥，但 !72d17be4 这个节点从未广播过节点信息，nodeinfo 表中没有它的公钥。这傻x豆包ai突然发什么神经，蠢死了

2026-06-18 00:49:42 +08:00
parent ce07792c4b
commit dfb3fdd3e5
16 changed files with 757 additions and 31 deletions
@@ -6,6 +6,7 @@ import (
 	"strings"
 	"sync"
 	"time"
+	"unicode/utf8"

 	"meshtastic_mqtt_server/completion"
 	"meshtastic_mqtt_server/conversation"
@@ -164,13 +165,32 @@ func (s *Service) processQueue(ctx context.Context) {
 	}
 }

+// printJSON outputs a structured log message (imported from main package pattern)
+func printJSON(v any) {
+	fmt.Printf("%+v\n", v)
+}
+
 // processMessage processes a single queued message
 func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	// Mark message as processing
 	if err := s.msgQueue.MarkAsProcessing(msg.ID); err != nil {
+		printJSON(map[string]any{
+			"event": "llm_process_failed",
+			"msg_id": msg.ID,
+			"step":  "mark_as_processing",
+			"error": err.Error(),
+		})
 		return
 	}

+	printJSON(map[string]any{
+		"event":        "llm_process_start",
+		"msg_id":       msg.ID,
+		"bot_id":       msg.BotID,
+		"from_node_id": msg.FromNodeID,
+		"text":         msg.Text,
+	})
+
 	// Create processing context with timeout
 	procCtx, cancel := context.WithTimeout(ctx, MaxProcessingTime)
 	defer cancel()
@@ -178,7 +198,9 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	// Get or create conversation for this bot
 	conv, err := s.convStore.GetOrCreateForBot(msg.BotID, msg.BotNodeID, msg.FromNodeID)
 	if err != nil {
-		_ = s.msgQueue.MarkAsFailed(msg.ID, fmt.Sprintf("failed to get conversation: %v", err))
+		errMsg := fmt.Sprintf("failed to get conversation: %v", err)
+		printJSON(map[string]any{"event": "llm_process_failed", "msg_id": msg.ID, "step": "get_conversation", "error": errMsg})
+		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}

@@ -188,21 +210,34 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 		Content: s.formatUserMessage(msg),
 	}
 	if err := s.convStore.AddMessage(conv.ID, userMsg); err != nil {
-		_ = s.msgQueue.MarkAsFailed(msg.ID, fmt.Sprintf("failed to add message: %v", err))
+		errMsg := fmt.Sprintf("failed to add message: %v", err)
+		printJSON(map[string]any{"event": "llm_process_failed", "msg_id": msg.ID, "step": "add_message", "error": errMsg})
+		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}

 	// Get the LLM profile
 	profile := s.llmState.ActiveProfile()
 	if profile == nil {
-		_ = s.msgQueue.MarkAsFailed(msg.ID, "no active LLM profile")
+		errMsg := "no active LLM profile - check if LLM providers are configured"
+		printJSON(map[string]any{"event": "llm_process_failed", "msg_id": msg.ID, "step": "get_profile", "error": errMsg})
+		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}

+	printJSON(map[string]any{
+		"event":       "llm_process_profile",
+		"msg_id":      msg.ID,
+		"model":       profile.Config.Model,
+		"base_url":    profile.Config.BaseURL,
+	})
+
 	// Reload conversation to get updated messages
 	conv, err = s.convStore.Get(conv.ID)
 	if err != nil {
-		_ = s.msgQueue.MarkAsFailed(msg.ID, fmt.Sprintf("failed to reload conversation: %v", err))
+		errMsg := fmt.Sprintf("failed to reload conversation: %v", err)
+		printJSON(map[string]any{"event": "llm_process_failed", "msg_id": msg.ID, "step": "reload_conversation", "error": errMsg})
+		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}

@@ -211,17 +246,55 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	_ = augmentedMessages // We'll use this in the future with proper tool support

 	// For now, use simple completion since we don't have tools registered yet
+	printJSON(map[string]any{"event": "llm_process_completion_start", "msg_id": msg.ID})
 	reply, err := completion.CompleteText(procCtx, profile, conv.Messages, 512)
 	if err != nil {
-		_ = s.msgQueue.MarkAsFailed(msg.ID, fmt.Sprintf("LLM completion failed: %v", err))
+		errMsg := fmt.Sprintf("LLM completion failed: %v", err)
+		printJSON(map[string]any{"event": "llm_process_failed", "msg_id": msg.ID, "step": "llm_completion", "error": errMsg})
+		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}

-	// Truncate reply for Meshtastic
+	printJSON(map[string]any{
+		"event":      "llm_process_completion_success",
+		"msg_id":     msg.ID,
+		"reply_len":  len(reply),
+	})
+
+	// Clean and validate reply text
+	reply = cleanReplyText(reply)
+	printJSON(map[string]any{
+		"event":       "llm_process_text_cleaned",
+		"msg_id":      msg.ID,
+		"cleaned_len": len(reply),
+	})
+
+	// Truncate reply for Meshtastic (UTF-8 safe truncation)
 	if len([]byte(reply)) > MaxReplyLength {
-		reply = string([]byte(reply)[:MaxReplyLength-3]) + "..."
+		reply = truncateUTF8(reply, MaxReplyLength-3) + "..."
+		printJSON(map[string]any{
+			"event":       "llm_process_text_truncated",
+			"msg_id":      msg.ID,
+			"truncated_len": len(reply),
+		})
 	}

+	// Final UTF-8 validation before sending
+	if !utf8.ValidString(reply) {
+		printJSON(map[string]any{
+			"event":   "llm_process_utf8_warning",
+			"msg_id":  msg.ID,
+			"message": "final text still invalid, using fallback",
+		})
+		reply = "抱歉，我暂时无法回复。请稍后再试。"
+	}
+	printJSON(map[string]any{
+		"event":    "llm_process_final_check",
+		"msg_id":   msg.ID,
+		"valid_utf8": utf8.ValidString(reply),
+		"final_len":  len(reply),
+	})
+
 	// Add assistant reply to conversation
 	assistantMsg := message.ChatMessage{
 		Role:    "assistant",
@@ -232,13 +305,21 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	}

 	// Send the reply via the bot
+	printJSON(map[string]any{"event": "llm_process_send_start", "msg_id": msg.ID, "to_node_num": msg.FromNodeNum})
 	if err := s.botSender.SendText(procCtx, msg.BotID, msg.FromNodeNum, reply); err != nil {
-		_ = s.msgQueue.MarkAsFailed(msg.ID, fmt.Sprintf("failed to send reply: %v", err))
+		errMsg := fmt.Sprintf("failed to send reply: %v", err)
+		printJSON(map[string]any{"event": "llm_process_failed", "msg_id": msg.ID, "step": "send_reply", "error": errMsg})
+		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}

 	// Mark message as processed
 	_ = s.msgQueue.MarkAsProcessed(msg.ID, reply)
+	printJSON(map[string]any{
+		"event":    "llm_process_success",
+		"msg_id":   msg.ID,
+		"reply":    reply,
+	})
 }

 // formatUserMessage formats the incoming message for the LLM
@@ -256,3 +337,71 @@ func (s *Service) formatUserMessage(msg QueuedMessage) string {
 	sb.WriteString(msg.Text)
 	return sb.String()
 }
+
+// cleanReplyText cleans LLM reply text to ensure it's valid for Meshtastic
+func cleanReplyText(text string) string {
+	// Force convert to valid UTF-8 using rune-by-rune processing
+	v := make([]rune, 0, len(text))
+	for i, r := range text {
+		if r == utf8.RuneError {
+			_, size := utf8.DecodeRuneInString(text[i:])
+			if size == 1 {
+				continue // Skip invalid runes
+			}
+		}
+		// Skip all problematic characters
+		if r < 32 && r != '\n' && r != '\r' && r != '\t' {
+			continue
+		}
+		if r == 65533 || r == 0xfffd { // Unicode replacement character
+			continue
+		}
+		v = append(v, r)
+	}
+	text = string(v)
+
+	// Additional cleanup - use only explicitly allowed ASCII printable + CJK
+	var sb strings.Builder
+	for _, r := range text {
+		switch {
+		case r == '\n':
+			sb.WriteRune(' ') // Replace newlines with space for Meshtastic
+		case r == '\r' || r == '\t':
+			continue
+		case r >= 32 && r <= 126: // Printable ASCII
+			sb.WriteRune(r)
+		case r >= 0x4E00 && r <= 0x9FFF: // CJK Unified Ideographs
+			sb.WriteRune(r)
+		case r >= 0x3400 && r <= 0x4DBF: // CJK Extension A
+			sb.WriteRune(r)
+		case r >= 0x20000 && r <= 0x2A6DF: // CJK Extension B
+			sb.WriteRune(r)
+		case r >= 0xFF01 && r <= 0xFF5E: // Fullwidth ASCII variants
+			sb.WriteRune(r)
+		case r == 0x3002 || r == 0xFF1F || r == 0xFF01 || r == 0xFF0C || r == 0xFF1A: // Fullwidth punctuation
+			sb.WriteRune(r)
+		default:
+			continue // Skip all other characters
+		}
+	}
+
+	result := strings.TrimSpace(sb.String())
+	if result == "" {
+		result = "抱歉，我无法回复此消息。"
+	}
+	return result
+}
+
+// truncateUTF8 safely truncates a UTF-8 string to max bytes without breaking in the middle of a rune
+func truncateUTF8(s string, maxBytes int) string {
+	if len([]byte(s)) <= maxBytes {
+		return s
+	}
+	bytes := []byte(s)
+	for i := maxBytes; i > 0; i-- {
+		if utf8.RuneStart(bytes[i]) {
+			return string(bytes[:i])
+		}
+	}
+	return ""
+}