feat(llm): console_log.llm 控制可读单行流程日志，并追踪路由 AI 决策

autoreply.Service 增加 consoleLog 字段和 logf 辅助；processMessage 中
原本的 printJSON map dump 全部替换为 [llm] 前缀的单行 key=value 输出。
关闭开关时整条链路静默。

把原本传 nil 给 toolrouter.RunAgentToolLoop 的 emit 接进来，将
stream.Frame trace 帧渲染为可读行，区分三类来源：

- 主 AI：main_model / completion / 最终 reply
- 路由 AI：router=<model> prepare/decide iter=N → call_tools=[...] / no_tool
- 工具调用：tool=<name> args=... / result(Nms)=... / ERROR ...

ai.Config 增加 ConsoleLog 字段，由 main.go 传入 cfg.ConsoleLog.LLM，
再透传给 autoreply.NewService。

Co-Authored-By: Claude <noreply@anthropic.com>

install.sh

@@ -89,6 +89,7 @@ console_log:
   mqtt: true
   llm: true
   sql: true
+  meshtastic: true
 EOF
   chown "${SERVICE_USER}:${SERVICE_USER}" "${CONFIG_DIR}/config.yaml"
   chmod 0640 "${CONFIG_DIR}/config.yaml"

internal/ai/service.go

@@ -29,6 +29,7 @@ type Config struct {
 	LLMProviders     []llm.ProviderConfig
 	DataDir          string
 	Enabled          bool
+	ConsoleLog       bool
 	ToolConfigStore  ToolConfigStore
 }
 
@@ -99,6 +100,7 @@ func NewService(cfg Config, db *gorm.DB, botSender autoreply.BotSender) (*Servic
 		msgQueue,
 		botSender,
 		cfg.ToolConfigStore,
+		cfg.ConsoleLog,
 	)
 
 	return &Service{

internal/autoreply/service.go

@@ -3,6 +3,7 @@ package autoreply
 import (
 	"context"
 	"fmt"
+	"os"
 	"strings"
 	"sync"
 	"time"
@@ -12,6 +13,7 @@ import (
 	"meshtastic_mqtt_server/internal/conversation"
 	"meshtastic_mqtt_server/internal/llm"
 	"meshtastic_mqtt_server/internal/message"
+	"meshtastic_mqtt_server/internal/stream"
 	"meshtastic_mqtt_server/internal/toolmanager"
 	"meshtastic_mqtt_server/internal/toolrouter"
 
@@ -73,13 +75,14 @@ type ToolConfigStore interface {
 
 // Service manages automatic AI replies for bots
 type Service struct {
-	llmState         *llm.State
-	toolRouter       *toolrouter.State
-	toolMgr          *toolmanager.Manager
-	convStore        *conversation.Store
-	msgQueue         MessageQueue
-	botSender        BotSender
-	toolConfigStore  ToolConfigStore
+	llmState        *llm.State
+	toolRouter      *toolrouter.State
+	toolMgr         *toolmanager.Manager
+	convStore       *conversation.Store
+	msgQueue        MessageQueue
+	botSender       BotSender
+	toolConfigStore ToolConfigStore
+	consoleLog      bool
 
 	running bool
 	mu      sync.Mutex
@@ -96,15 +99,17 @@ func NewService(
 	msgQueue MessageQueue,
 	botSender BotSender,
 	toolConfigStore ToolConfigStore,
+	consoleLog bool,
 ) *Service {
 	return &Service{
-		llmState:         llmState,
-		toolRouter:       toolRouter,
-		toolMgr:          toolMgr,
-		convStore:        convStore,
-		msgQueue:         msgQueue,
-		botSender:        botSender,
-		toolConfigStore:  toolConfigStore,
+		llmState:        llmState,
+		toolRouter:      toolRouter,
+		toolMgr:         toolMgr,
+		convStore:       convStore,
+		msgQueue:        msgQueue,
+		botSender:       botSender,
+		toolConfigStore: toolConfigStore,
+		consoleLog:      consoleLog,
 	}
 }
 
@@ -180,31 +185,73 @@ 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)
+// logf 仅在 console_log.llm 开启时输出一行可读日志（带 [llm] 前缀）。
+func (s *Service) logf(format string, args ...any) {
+	if !s.consoleLog {
+		return
+	}
+	fmt.Fprintf(os.Stderr, "[llm] "+format+"\n", args...)
+}
+
+// emit 把 toolrouter.Frame 转成单行日志，区分主 AI / 路由 AI / 工具调用。
+func (s *Service) emit(msgID uint64, routerModel string) stream.EmitFunc {
+	if !s.consoleLog {
+		return nil
+	}
+	return func(f stream.Frame) {
+		switch f.Stage {
+		case "prepare":
+			tools, _ := f.Data["tools"].([]string)
+			s.logf("msg=%d router=%s prepare tools=%v", msgID, routerModel, tools)
+		case "request":
+			if f.Status == "success" {
+				// 模型未请求工具
+				s.logf("msg=%d router=%s decide → no_tool（直接生成回答）", msgID, routerModel)
+				return
+			}
+			iter, _ := f.Data["iteration"].(int)
+			s.logf("msg=%d router=%s decide iter=%d ...", msgID, routerModel, iter)
+		case "tool_calls":
+			calls, _ := f.Data["tools"].([]string)
+			iter, _ := f.Data["iteration"].(int)
+			s.logf("msg=%d router=%s decide iter=%d → call_tools=%v", msgID, routerModel, iter, calls)
+		case "arguments":
+			args, _ := f.Data["arguments"].(string)
+			s.logf("msg=%d tool=%s args=%s", msgID, f.Tool, truncate(args, 200))
+		case "result":
+			dur, _ := f.Data["duration_ms"].(int64)
+			preview, _ := f.Data["result_preview"].(string)
+			s.logf("msg=%d tool=%s result(%dms)=%s", msgID, f.Tool, dur, truncate(preview, 200))
+		case "execute":
+			if f.Status == "error" {
+				errStr, _ := f.Data["error"].(string)
+				if errStr == "" {
+					errStr = f.Message
+				}
+				s.logf("msg=%d tool=%s ERROR: %s", msgID, f.Tool, errStr)
+			}
+		case "decision":
+			// 中间帧，已被 tool_calls / request(success) 覆盖，跳过
+		}
+	}
+}
+
+func truncate(s string, n int) string {
+	if len(s) <= n {
+		return s
+	}
+	return s[:n] + "..."
 }
 
 // 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(),
-		})
+		s.logf("msg=%d FAIL step=mark_as_processing err=%v", msg.ID, err)
 		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,
-	})
+	s.logf("msg=%d from=%s start text=%q", msg.ID, msg.FromNodeID, msg.Text)
 
 	// Create processing context with timeout
 	procCtx, cancel := context.WithTimeout(ctx, MaxProcessingTime)
@@ -214,7 +261,7 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	conv, err := s.convStore.GetOrCreateForBot(msg.BotID, msg.BotNodeID, msg.FromNodeID)
 	if err != nil {
 		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.logf("msg=%d FAIL step=get_conversation err=%s", msg.ID, errMsg)
 		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}
@@ -226,7 +273,7 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	}
 	if err := s.convStore.AddMessage(conv.ID, userMsg); err != nil {
 		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.logf("msg=%d FAIL step=add_message err=%s", msg.ID, errMsg)
 		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}
@@ -235,23 +282,18 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	profile := s.llmState.ActiveProfile()
 	if profile == nil {
 		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.logf("msg=%d FAIL step=get_profile err=%s", msg.ID, 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,
-	})
+	s.logf("msg=%d main_model=%s base=%s", msg.ID, profile.Config.Model, profile.Config.BaseURL)
 
 	// Reload conversation to get updated messages
 	conv, err = s.convStore.Get(conv.ID)
 	if err != nil {
 		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.logf("msg=%d FAIL step=reload_conversation err=%s", msg.ID, errMsg)
 		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}
@@ -262,11 +304,11 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	if s.toolConfigStore != nil {
 		systemPrompt, err = s.toolConfigStore.GetLLMPrimaryConfigSystemPrompt()
 		if err != nil {
-			printJSON(map[string]any{"event": "llm_system_prompt_warning", "msg_id": msg.ID, "error": err.Error()})
+			s.logf("msg=%d WARN system_prompt err=%v", msg.ID, err)
 		}
 		enableTool, err = s.toolConfigStore.GetLLMPrimaryConfigEnableTool()
 		if err != nil {
-			printJSON(map[string]any{"event": "llm_enable_tool_warning", "msg_id": msg.ID, "error": err.Error()})
+			s.logf("msg=%d WARN enable_tool err=%v", msg.ID, err)
 		}
 	}
 
@@ -279,27 +321,27 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 		for _, t := range tools {
 			toolNames = append(toolNames, t.Name())
 		}
-		printJSON(map[string]any{
-			"event":      "llm_tool_manager_status",
-			"msg_id":     msg.ID,
-			"tool_count":  toolCount,
-			"tool_names":  toolNames,
-			"enable_tool": enableTool,
-		})
+		s.logf("msg=%d tools_loaded=%v enable_tool=%t", msg.ID, toolNames, enableTool)
 	}
 
 	// Run the tool loop to get augmented messages - pass system prompt to tool router
 	// Tool loop will handle system prompt and tool calling
 	var augmentedMessages []*model.ChatCompletionMessage
 	if enableTool && toolCount > 0 {
-		augmentedMessages, err = toolrouter.RunAgentToolLoop(procCtx, s.toolRouter, profile, systemPrompt, conv.Messages, s.toolMgr, nil)
+		routerProfile := s.toolRouter.RouterProfile(profile)
+		routerModel := profile.Config.Model
+		if routerProfile != nil {
+			routerModel = routerProfile.Config.Model
+		}
+		s.logf("msg=%d router_model=%s tool_loop start", msg.ID, routerModel)
+		augmentedMessages, err = toolrouter.RunAgentToolLoop(procCtx, s.toolRouter, profile, systemPrompt, conv.Messages, s.toolMgr, s.emit(msg.ID, routerModel))
 		if err != nil {
-			printJSON(map[string]any{"event": "llm_tool_loop_warning", "msg_id": msg.ID, "error": err.Error()})
+			s.logf("msg=%d WARN tool_loop err=%v", msg.ID, err)
 			// Continue with original messages if tool loop fails
 		}
 	}
 
-	printJSON(map[string]any{"event": "llm_process_completion_start", "msg_id": msg.ID, "has_system_prompt": systemPrompt != "", "augmented_messages": len(augmentedMessages)})
+	s.logf("msg=%d completion start has_system_prompt=%t augmented=%d", msg.ID, systemPrompt != "", len(augmentedMessages))
 
 	// Use augmented messages from tool loop (already includes system prompt and tool results)
 	// If augmented messages is empty or nil, fallback to original messages with system prompt
@@ -313,50 +355,27 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	}
 	if err != nil {
 		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.logf("msg=%d FAIL step=llm_completion err=%s", msg.ID, errMsg)
 		_ = s.msgQueue.MarkAsFailed(msg.ID, errMsg)
 		return
 	}
 
-	printJSON(map[string]any{
-		"event":      "llm_process_completion_success",
-		"msg_id":     msg.ID,
-		"reply_len":  len(reply),
-	})
+	s.logf("msg=%d main=%s reply_len=%d reply=%q", msg.ID, profile.Config.Model, len(reply), truncate(reply, 200))
 
 	// 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 = truncateUTF8(reply, MaxReplyLength-3) + "..."
-		printJSON(map[string]any{
-			"event":       "llm_process_text_truncated",
-			"msg_id":      msg.ID,
-			"truncated_len": len(reply),
-		})
+		s.logf("msg=%d reply truncated to %d bytes", msg.ID, 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",
-		})
+		s.logf("msg=%d WARN final text invalid utf8, using fallback", msg.ID)
 		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{
@@ -372,27 +391,23 @@ func (s *Service) processMessage(ctx context.Context, msg QueuedMessage) {
 	var sendErr error
 	if msg.MessageType == "channel" && msg.ChannelID != nil && *msg.ChannelID != "" {
 		// 频道消息 - 回复到原频道
-		printJSON(map[string]any{"event": "llm_process_send_start", "msg_id": msg.ID, "channel_id": *msg.ChannelID, "message_type": "channel"})
+		s.logf("msg=%d send → channel=%s", msg.ID, *msg.ChannelID)
 		sendErr = s.botSender.SendChannelText(procCtx, msg.BotID, *msg.ChannelID, reply)
 	} else {
 		// 私聊消息 - 回复给发送节点
-		printJSON(map[string]any{"event": "llm_process_send_start", "msg_id": msg.ID, "to_node_num": msg.FromNodeNum, "message_type": "direct"})
+		s.logf("msg=%d send → direct to_node_num=%d", msg.ID, msg.FromNodeNum)
 		sendErr = s.botSender.SendDirectText(procCtx, msg.BotID, msg.FromNodeNum, reply)
 	}
 	if sendErr != nil {
 		errMsg := fmt.Sprintf("failed to send reply: %v", sendErr)
-		printJSON(map[string]any{"event": "llm_process_failed", "msg_id": msg.ID, "step": "send_reply", "error": errMsg})
+		s.logf("msg=%d FAIL step=send_reply err=%s", msg.ID, 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,
-	})
+	s.logf("msg=%d done", msg.ID)
 }
 
 // formatUserMessage formats the incoming message for the LLM

internal/config/config.go

@@ -83,10 +83,11 @@ type AIConfig struct {
 
 // ConsoleLogConfig 控制各模块是否在控制台打印日志。后续若新增模块，按需扩展。
 type ConsoleLogConfig struct {
-	Web  bool `yaml:"web"`
-	MQTT bool `yaml:"mqtt"`
-	LLM  bool `yaml:"llm"`
-	SQL  bool `yaml:"sql"`
+	Web        bool `yaml:"web"`
+	MQTT       bool `yaml:"mqtt"`
+	LLM        bool `yaml:"llm"`
+	SQL        bool `yaml:"sql"`
+	Meshtastic bool `yaml:"meshtastic"`
 }
 
 type rawConfig struct {
@@ -99,10 +100,11 @@ type rawConfig struct {
 }
 
 type rawConsoleLogConfig struct {
-	Web  *bool `yaml:"web"`
-	MQTT *bool `yaml:"mqtt"`
-	LLM  *bool `yaml:"llm"`
-	SQL  *bool `yaml:"sql"`
+	Web        *bool `yaml:"web"`
+	MQTT       *bool `yaml:"mqtt"`
+	LLM        *bool `yaml:"llm"`
+	SQL        *bool `yaml:"sql"`
+	Meshtastic *bool `yaml:"meshtastic"`
 }
 
 type rawAIConfig struct {
@@ -200,10 +202,11 @@ func Default() *Config {
 			DataDir: defaultDataDir(),
 		},
 		ConsoleLog: ConsoleLogConfig{
-			Web:  true,
-			MQTT: true,
-			LLM:  true,
-			SQL:  true,
+			Web:        true,
+			MQTT:       true,
+			LLM:        true,
+			SQL:        true,
+			Meshtastic: true,
 		},
 	}
 }
@@ -515,6 +518,11 @@ func normalize(raw rawConfig) (*Config, bool) {
 		} else {
 			cfg.ConsoleLog.SQL = *raw.ConsoleLog.SQL
 		}
+		if raw.ConsoleLog.Meshtastic == nil {
+			changed = true
+		} else {
+			cfg.ConsoleLog.Meshtastic = *raw.ConsoleLog.Meshtastic
+		}
 	}
 
 	return cfg, changed

internal/mqttforward/client_stats.go

@@ -0,0 +1,74 @@
+package mqttforward
+
+import "sync"
+
+// ClientStats 在内存中维护每个 MQTT 客户端的收/发包数量。
+// key 取自 mqtt.Client.ID（broker 内部唯一标识），客户端断开时由调用方调用 Delete 清除，
+// 重新连接同一 client_id 会拿到一份新的零值计数，符合"断链就清空"。
+type ClientStats struct {
+	mu  sync.RWMutex
+	all map[string]*clientCounter
+}
+
+type clientCounter struct {
+	In  int64 // 客户端 → 服务器（broker 收到的报文数）
+	Out int64 // 服务器 → 客户端（broker 发出的报文数）
+}
+
+// NewClientStats 返回一个空的统计器。
+func NewClientStats() *ClientStats {
+	return &ClientStats{all: make(map[string]*clientCounter)}
+}
+
+// IncIn 在 broker 收到客户端报文时调用。clientID 为空直接忽略。
+func (s *ClientStats) IncIn(clientID string) {
+	if s == nil || clientID == "" {
+		return
+	}
+	s.mu.Lock()
+	c, ok := s.all[clientID]
+	if !ok {
+		c = &clientCounter{}
+		s.all[clientID] = c
+	}
+	c.In++
+	s.mu.Unlock()
+}
+
+// IncOut 在 broker 向客户端发出报文时调用。
+func (s *ClientStats) IncOut(clientID string) {
+	if s == nil || clientID == "" {
+		return
+	}
+	s.mu.Lock()
+	c, ok := s.all[clientID]
+	if !ok {
+		c = &clientCounter{}
+		s.all[clientID] = c
+	}
+	c.Out++
+	s.mu.Unlock()
+}
+
+// Get 返回指定 clientID 当前的收/发包数量；不存在时返回 0,0。
+func (s *ClientStats) Get(clientID string) (in, out int64) {
+	if s == nil || clientID == "" {
+		return 0, 0
+	}
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	if c, ok := s.all[clientID]; ok {
+		return c.In, c.Out
+	}
+	return 0, 0
+}
+
+// Delete 在客户端断开连接时清除其计数。重新连接同一 clientID 会从 0 重新计起。
+func (s *ClientStats) Delete(clientID string) {
+	if s == nil || clientID == "" {
+		return
+	}
+	s.mu.Lock()
+	delete(s.all, clientID)
+	s.mu.Unlock()
+}

internal/web/mqtt_status.go

@@ -16,11 +16,12 @@ type MQTTStatusProvider interface {
 // MQTTRuntimeStatus 把 mqtt.Server / 写队列 / 转发统计三个上下文打包成
 // 实现 MQTTStatusProvider 的具体类型。供 main 包构造后注入 newRouter。
 type MQTTRuntimeStatus struct {
-	Server  *mqtt.Server
-	Address string
-	TLS     bool
-	Stats   *mqttforwardpkg.Stats
-	DBQueue *storepkg.WriteQueue
+	Server      *mqtt.Server
+	Address     string
+	TLS         bool
+	Stats       *mqttforwardpkg.Stats
+	ClientStats *mqttforwardpkg.ClientStats
+	DBQueue     *storepkg.WriteQueue
 }
 
 // AdminMQTTStatus 是 admin 路由 GET /admin/mqtt-status 返回的 JSON 视图。
@@ -51,12 +52,12 @@ type AdminMQTTStatus struct {
 }
 
 type AdminMQTTClient struct {
-	ClientID   string `json:"client_id"`
-	Username   string `json:"username"`
-	Listener   string `json:"listener"`
-	RemoteAddr string `json:"remote_addr"`
-	RemoteHost string `json:"remote_host"`
-	RemotePort string `json:"remote_port"`
+	ClientID     string `json:"client_id"`
+	Username     string `json:"username"`
+	Listener     string `json:"listener"`
+	RemoteAddr   string `json:"remote_addr"`
+	PacketsIn    int64  `json:"packets_in"`  // 客户端 → 服务器
+	PacketsOut   int64  `json:"packets_out"` // 服务器 → 客户端
 }
 
 // Status 实现 MQTTStatusProvider。
@@ -94,13 +95,14 @@ func (m MQTTRuntimeStatus) Status() AdminMQTTStatus {
 			continue
 		}
 		info := mqttClientInfo(client)
+		in, out := m.ClientStats.Get(info.ClientID)
 		status.Clients = append(status.Clients, AdminMQTTClient{
 			ClientID:   info.ClientID,
 			Username:   info.Username,
 			Listener:   info.Listener,
 			RemoteAddr: info.RemoteAddr,
-			RemoteHost: info.RemoteHost,
-			RemotePort: info.RemotePort,
+			PacketsIn:  in,
+			PacketsOut: out,
 		})
 	}
 	return status
@@ -112,39 +114,16 @@ type mqttClientInfoView struct {
 	Username   string
 	Listener   string
 	RemoteAddr string
-	RemoteHost string
-	RemotePort string
 }
 
 func mqttClientInfo(c *mqtt.Client) mqttClientInfoView {
 	if c == nil {
 		return mqttClientInfoView{}
 	}
-	info := mqttClientInfoView{
+	return mqttClientInfoView{
 		ClientID:   c.ID,
 		Username:   string(c.Properties.Username),
 		Listener:   c.Net.Listener,
 		RemoteAddr: c.Net.Remote,
 	}
-	host, port := splitHostPort(c.Net.Remote)
-	info.RemoteHost = host
-	info.RemotePort = port
-	return info
-}
-
-func splitHostPort(addr string) (string, string) {
-	if addr == "" {
-		return "", ""
-	}
-	// 复用 net.SplitHostPort，但要兼容 "host" 这种没端口的情况。
-	for i := len(addr) - 1; i >= 0; i-- {
-		if addr[i] == ':' {
-			host := addr[:i]
-			if len(host) >= 2 && host[0] == '[' && host[len(host)-1] == ']' {
-				host = host[1 : len(host)-1]
-			}
-			return host, addr[i+1:]
-		}
-	}
-	return addr, ""
 }

main.go

@@ -20,17 +20,17 @@ import (
 	"github.com/mochi-mqtt/server/v2/packets"
 
 	"meshtastic_mqtt_server/internal/ai"
-	"meshtastic_mqtt_server/internal/autoreply"
 	"meshtastic_mqtt_server/internal/auth"
+	"meshtastic_mqtt_server/internal/autoreply"
 	blockingpkg "meshtastic_mqtt_server/internal/blocking"
 	botpkg "meshtastic_mqtt_server/internal/bot"
 	configpkg "meshtastic_mqtt_server/internal/config"
+	"meshtastic_mqtt_server/internal/llm"
+	"meshtastic_mqtt_server/internal/mqtpp"
 	mqttforwardpkg "meshtastic_mqtt_server/internal/mqttforward"
 	rspkg "meshtastic_mqtt_server/internal/runtimesettings"
 	storepkg "meshtastic_mqtt_server/internal/store"
 	webpkg "meshtastic_mqtt_server/internal/web"
-	"meshtastic_mqtt_server/internal/llm"
-	"meshtastic_mqtt_server/internal/mqtpp"
 )
 
 const (
@@ -46,14 +46,16 @@ const (
 
 type meshtasticFilterHook struct {
 	mqtt.HookBase
-	key         []byte
-	dbQueue     *storepkg.WriteQueue
-	stats       *mqttforwardpkg.Stats
-	blocking    *blockingpkg.Cache
-	settings    *rspkg.Cache
-	pkiResolver func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool)
-	autoAcker   func(record map[string]any)
-	consoleLog  bool // 控制台是否打印 MQTT 连接/订阅事件
+	key             []byte
+	dbQueue         *storepkg.WriteQueue
+	stats           *mqttforwardpkg.Stats
+	clientStats     *mqttforwardpkg.ClientStats
+	blocking        *blockingpkg.Cache
+	settings        *rspkg.Cache
+	pkiResolver     func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool)
+	autoAcker       func(record map[string]any)
+	consoleLog      bool // 控制台是否打印 MQTT 连接/订阅事件
+	packetConsoleLog bool // 控制台是否打印 Meshtastic 数据包
 }
 
 // ID 返回用于识别 Meshtastic payload 过滤器的 hook 名称。
@@ -68,7 +70,9 @@ func (h *meshtasticFilterHook) Provides(b byte) bool {
 		b == mqtt.OnSessionEstablished ||
 		b == mqtt.OnDisconnect ||
 		b == mqtt.OnSubscribed ||
-		b == mqtt.OnUnsubscribed
+		b == mqtt.OnUnsubscribed ||
+		b == mqtt.OnPacketRead ||
+		b == mqtt.OnPacketSent
 }
 
 // OnConnect 在 MQTT 会话建立前拒绝命中 IP 屏蔽表的客户端。
@@ -93,6 +97,9 @@ func (h *meshtasticFilterHook) OnSessionEstablished(cl *mqtt.Client, pk packets.
 
 // OnDisconnect 在客户端断开时打印日志，含触发原因。
 func (h *meshtasticFilterHook) OnDisconnect(cl *mqtt.Client, err error, expire bool) {
+	if cl != nil {
+		h.clientStats.Delete(cl.ID)
+	}
 	if !h.consoleLog {
 		return
 	}
@@ -105,6 +112,22 @@ func (h *meshtasticFilterHook) OnDisconnect(cl *mqtt.Client, err error, expire b
 		info.ClientID, info.Username, info.RemoteHost, info.RemotePort, expire, reason)
 }
 
+// OnPacketRead 在 broker 收到客户端报文时累计入站计数（客户端 → 服务器）。
+// 返回原始 packet 不做修改；该 hook 在 packet 校验前触发。
+func (h *meshtasticFilterHook) OnPacketRead(cl *mqtt.Client, pk packets.Packet) (packets.Packet, error) {
+	if cl != nil {
+		h.clientStats.IncIn(cl.ID)
+	}
+	return pk, nil
+}
+
+// OnPacketSent 在 broker 把报文写出后累计出站计数（服务器 → 客户端）。
+func (h *meshtasticFilterHook) OnPacketSent(cl *mqtt.Client, pk packets.Packet, b []byte) {
+	if cl != nil {
+		h.clientStats.IncOut(cl.ID)
+	}
+}
+
 // OnSubscribed 客户端订阅成功后打印订阅的 topic filter 列表。
 func (h *meshtasticFilterHook) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte) {
 	if !h.consoleLog {
@@ -152,8 +175,8 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa
 	if h.autoAcker != nil {
 		h.autoAcker(record)
 	}
-	if record["type"] != "empty_packet" {
-		printJSON(record)
+	if h.packetConsoleLog && record["type"] != "empty_packet" {
+		printMeshtasticRecord(record)
 	}
 	return pk, nil
 }
@@ -296,7 +319,8 @@ func run(cfg *configpkg.Config) error {
 	}
 
 	messageStats := &mqttforwardpkg.Stats{}
-	server, mqttHook, mqttAddr, err := startMQTTServer(cfg, store, dbQueue, messageStats, blocking, settings)
+	clientStats := mqttforwardpkg.NewClientStats()
+	server, mqttHook, mqttAddr, err := startMQTTServer(cfg, store, dbQueue, messageStats, clientStats, blocking, settings)
 	if err != nil {
 		return err
 	}
@@ -362,6 +386,7 @@ func run(cfg *configpkg.Config) error {
 				LLMProviders:    providerConfigs,
 				DataDir:         cfg.AI.DataDir,
 				Enabled:         cfg.AI.Enabled,
+				ConsoleLog:      cfg.ConsoleLog.LLM,
 				ToolConfigStore: store,
 			}, store.DB(), botSenderAdapter)
 			if err != nil {
@@ -385,7 +410,7 @@ func run(cfg *configpkg.Config) error {
 		if err != nil {
 			return err
 		}
-		mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, DBQueue: dbQueue}
+		mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, ClientStats: clientStats, DBQueue: dbQueue}
 		handler := webpkg.NewRouter(cfg.Web, cfg.ConsoleLog.Web, store, sessions, mqttStatus, blocking, forwardManager, settings, botSender)
 		webAddresses := []string{}
 		if cfg.Web.PortEnabled {
@@ -440,19 +465,21 @@ func run(cfg *configpkg.Config) error {
 	return runErr
 }
 
-func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *storepkg.WriteQueue, stats *mqttforwardpkg.Stats, blocking *blockingpkg.Cache, settings *rspkg.Cache) (*mqtt.Server, *meshtasticFilterHook, string, error) {
+func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *storepkg.WriteQueue, stats *mqttforwardpkg.Stats, clientStats *mqttforwardpkg.ClientStats, blocking *blockingpkg.Cache, settings *rspkg.Cache) (*mqtt.Server, *meshtasticFilterHook, string, error) {
 	server := mqtt.New(&mqtt.Options{InlineClient: true})
 	if err := server.AddHook(new(mqttauth.AllowHook), nil); err != nil {
 		return nil, nil, "", err
 	}
 	hook := &meshtasticFilterHook{
-		key:         cfg.Key,
-		dbQueue:     dbQueue,
-		stats:       stats,
-		blocking:    blocking,
-		settings:    settings,
-		pkiResolver: botpkg.NewPKIKeyResolver(store),
-		consoleLog:  cfg.ConsoleLog.MQTT,
+		key:              cfg.Key,
+		dbQueue:          dbQueue,
+		stats:            stats,
+		clientStats:      clientStats,
+		blocking:         blocking,
+		settings:         settings,
+		pkiResolver:      botpkg.NewPKIKeyResolver(store),
+		consoleLog:       cfg.ConsoleLog.MQTT,
+		packetConsoleLog: cfg.ConsoleLog.Meshtastic,
 	}
 	if err := server.AddHook(hook, nil); err != nil {
 		return nil, nil, "", err
@@ -476,7 +503,76 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
 
 // printJSON 将记录编码为 JSON 后按数据包类型着色输出。
 func printJSON(record map[string]any) {
-	//printJSONBytes(record, mqtpp.MustJSON(record))
+	printJSONBytes(record, mqtpp.MustJSON(record))
+}
+
+// printMeshtasticRecord 把 Meshtastic 解码后的 record 按 type 拼成可读的彩色单行，
+// 不输出原始 JSON。保留与 printJSONBytes 一致的色码方案。
+func printMeshtasticRecord(record map[string]any) {
+	if record == nil {
+		return
+	}
+	typ, _ := record["type"].(string)
+	from := stringField(record, "from")
+	channel := stringField(record, "channel_id")
+	gateway := stringField(record, "gateway_id")
+
+	var color string
+	var body string
+	switch typ {
+	case "nodeinfo":
+		color = ansiGreenBGWhiteText
+		body = fmt.Sprintf("nodeinfo   from=%s long=%q short=%q hw=%s role=%s",
+			from, stringField(record, "long_name"), stringField(record, "short_name"),
+			stringField(record, "hw_model"), stringField(record, "role"))
+	case "map_report":
+		color = ansiBlueBGWhiteText
+		body = fmt.Sprintf("map_report from=%s long=%q lat=%v lon=%v alt=%v fw=%s region=%s",
+			from, stringField(record, "long_name"),
+			record["latitude"], record["longitude"], record["altitude"],
+			stringField(record, "firmware_version"), stringField(record, "region"))
+	case "text_message":
+		color = ansiPurpleBGWhiteText
+		body = fmt.Sprintf("text       from=%s channel=%s text=%q",
+			from, channel, stringField(record, "text"))
+	case "position":
+		color = ansiCyanBGBlackText
+		body = fmt.Sprintf("position   from=%s lat=%v lon=%v alt=%v",
+			from, record["latitude"], record["longitude"], record["altitude"])
+	case "telemetry":
+		color = ansiYellowBGBlackText
+		body = fmt.Sprintf("telemetry  from=%s tt=%v metrics=%v",
+			from, record["telemetry_type"], record["metrics"])
+	case "routing":
+		color = ansiGrayBGWhiteText
+		body = fmt.Sprintf("routing    from=%s pkt_id=%v", from, record["packet_id"])
+	case "traceroute":
+		color = ansiGrayBGWhiteText
+		body = fmt.Sprintf("traceroute from=%s pkt_id=%v", from, record["packet_id"])
+	default:
+		if record["error"] != nil {
+			color = ansiRedBGWhiteText
+			body = fmt.Sprintf("%-10s from=%s error=%v topic=%s", typ, from,
+				record["error"], stringField(record, "topic"))
+		} else {
+			body = fmt.Sprintf("%-10s from=%s", typ, from)
+		}
+	}
+	if gateway != "" {
+		body += " gw=" + gateway
+	}
+	if color != "" {
+		fmt.Printf("%s%s%s\n", color, body, ansiReset)
+		return
+	}
+	fmt.Println(body)
+}
+
+func stringField(record map[string]any, key string) string {
+	if v, ok := record[key].(string); ok {
+		return v
+	}
+	return ""
 }
 
 // printJSONBytes 使用已编码好的 JSON 文本，并根据记录 type 选择控制台颜色。

meshmap_frontend/src/components/AdminDashboard.vue

@@ -162,8 +162,8 @@ onBeforeUnmount(() => {
               <th>Username</th>
               <th>Listener</th>
               <th>Remote Addr</th>
-              <th>Remote Host</th>
-              <th>Remote Port</th>
+              <th>客户端→服务器</th>
+              <th>服务器→客户端</th>
             </tr>
           </thead>
           <tbody>
@@ -172,8 +172,8 @@ onBeforeUnmount(() => {
               <td>{{ client.username || '-' }}</td>
               <td>{{ client.listener || '-' }}</td>
               <td>{{ client.remote_addr || '-' }}</td>
-              <td>{{ client.remote_host || '-' }}</td>
-              <td>{{ client.remote_port || '-' }}</td>
+              <td>{{ client.packets_in ?? 0 }}</td>
+              <td>{{ client.packets_out ?? 0 }}</td>
             </tr>
           </tbody>
         </table>

meshmap_frontend/src/types.ts

@@ -327,8 +327,8 @@ export interface AdminMqttClient {
   username: string
   listener: string
   remote_addr: string
-  remote_host: string
-  remote_port: string
+  packets_in: number
+  packets_out: number
 }
 
 export interface AdminRuntimeSettings {