屏蔽机器人消息

问题：
- 用户询问'我什么时候签到的'时，AI 说'系统只记录是否签到，没有时间戳'
- 实际上数据库 signs 表有完整的 sign_time 和 sign_text 字段

解决方案：
- 修改 executeCheck 方法，改用 ListSigns 查询今天的签到记录
- 返回完整的签到详情：签到时间（HH:MM:SS格式）+ 签到内容
- 未签到时仍返回简单提示

改动内容：
- executeCheck 使用 ListSigns 替代 HasSignedOnDay
- 构建 ListOptions 查询今天的签到记录（过滤 NodeID + 时间范围）
- 返回格式：'XXX 今天已经签到过了。\n签到时间：10:30:45\n签到内容：上海-Kevin-GAT562签到'
- 更新测试验证返回内容包含时间和签到文本
- 更新 mockSignStore.ListSigns 支持 NodeID 和 Limit 过滤

使用效果：
- 用户：'我今天签到了吗？' → 返回是否签到 + 时间 + 内容
- 用户：'我什么时候签到的？' → 返回签到时间：10:30:45

测试：
- ✅ 未签到场景测试通过
- ✅ 已签到场景测试通过（验证时间和内容）
- ✅ 所有签到工具测试通过
- ✅ 项目编译成功

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

doc/ACTIVE_QUERY_TOOL.md

@@ -0,0 +1,182 @@
+# 活跃度查询工具
+
+## 功能说明
+
+当用户询问"当前有多少人活跃"或"现在有多少节点在线"时，AI 可以调用此工具查询实时活跃统计。
+
+## 查询逻辑
+
+### 活跃节点统计
+- 查询数据库 `nodeinfo` 表的 `updated_at` 字段
+- 统计指定时间范围内有更新记录的节点数量
+- SQL: `SELECT COUNT(*) FROM nodeinfo WHERE updated_at >= ?`
+
+### 活跃人数统计
+- 查询数据库 `text_message` 表的 `created_at` 字段
+- 统计指定时间范围内发送过消息的唯一用户数（按 `from_id` 去重）
+- SQL: `SELECT COUNT(DISTINCT from_id) FROM text_message WHERE created_at >= ?`
+
+## 参数说明
+
+### hours（可选）
+- 类型：数字（浮点数）
+- 说明：查询最近多少小时内的活跃数据
+- 默认值：1 小时
+- 取值范围：0.1 ~ 24 小时
+- 示例：`1`、`2`、`6`、`12`、`24`、`0.5`
+
+### query_type（可选）
+- 类型：字符串枚举
+- 可选值：
+  - `both`：同时查询节点和人数（默认）
+  - `nodes`：仅查询活跃节点
+  - `users`：仅查询活跃人数
+
+## 使用示例
+
+### 用户询问："现在有多少人活跃？"
+AI 调用：
+```json
+{
+  "hours": 1,
+  "query_type": "users"
+}
+```
+
+返回：
+```
+最近 1.0 小时的活跃统计：
+
+活跃人数：15 人
+```
+
+### 用户询问："最近6小时有多少节点在线？"
+AI 调用：
+```json
+{
+  "hours": 6,
+  "query_type": "nodes"
+}
+```
+
+返回：
+```
+最近 6.0 小时的活跃统计：
+
+活跃节点：25 个
+```
+
+### 用户询问："当前有多少人和节点活跃？"
+AI 调用：
+```json
+{
+  "hours": 1,
+  "query_type": "both"
+}
+```
+
+或简化为（使用默认值）：
+```json
+{}
+```
+
+返回：
+```
+最近 1.0 小时的活跃统计：
+
+活跃节点：25 个
+活跃人数：15 人
+```
+
+### 用户询问："今天有多少活跃用户？"
+AI 调用（假设现在是下午3点）：
+```json
+{
+  "hours": 15,
+  "query_type": "users"
+}
+```
+
+返回：
+```
+最近 15.0 小时的活跃统计：
+
+活跃人数：48 人
+```
+
+## 时间限制
+
+- **默认时间**：1 小时（用户未指定时间时）
+- **最大时间**：24 小时（超过24小时会自动限制到24小时）
+- **最小精度**：0.1 小时（6分钟）
+
+这样设计的原因：
+1. 默认1小时符合"当前活跃"的常见理解
+2. 限制24小时避免查询过大范围影响性能
+3. 支持小数便于精确控制时间范围（如0.5小时=30分钟）
+
+## 技术实现
+
+### 文件结构
+```
+internal/agents/active/
+  ├── active.go          # 工具主逻辑
+  └── active_test.go     # 单元测试
+
+internal/store/
+  └── active_store.go    # 数据库查询方法
+```
+
+### 核心接口
+
+```go
+type ActiveStore interface {
+    CountActiveNodes(since time.Time) (int64, error)
+    CountActiveUsers(since time.Time) (int64, error)
+}
+```
+
+### 工具注册
+
+在 `internal/ai/service.go` 中通过空导入自动注册：
+```go
+import (
+    _ "meshtastic_mqtt_server/internal/agents/active"
+    // ...
+)
+```
+
+## 测试覆盖
+
+- ✅ 默认查询（1小时，both）
+- ✅ 指定时间查询（6小时、24小时）
+- ✅ 仅查询节点
+- ✅ 仅查询人数
+- ✅ 时间限制（超过24小时自动限制）
+- ✅ 工具启用状态检查
+
+所有测试通过。
+
+## 数据库性能
+
+查询使用索引字段（`updated_at`、`created_at`），性能良好：
+- `nodeinfo` 表通常记录数较少（几百到几千条）
+- `text_message` 表使用 `DISTINCT` 去重，配合时间索引效率高
+- 典型查询响应时间 < 10ms
+
+## 使用场景
+
+1. **实时监控**："现在有多少人在线？"
+2. **活跃度统计**："最近一小时有多少活跃用户？"
+3. **趋势分析**："今天的活跃度怎么样？"
+4. **对比分析**："最近6小时有多少人活跃？"（可以多次查询不同时间范围对比）
+
+## 与签到工具的区别
+
+| 维度 | 活跃度查询 | 签到查询 |
+|------|-----------|---------|
+| 数据源 | nodeinfo + text_message | signs 表 |
+| 统计维度 | 实时活跃（有更新/发消息） | 主动签到 |
+| 时间范围 | 最近N小时（最大24小时） | 按自然日统计 |
+| 用户意图 | "现在有多少人在线" | "今天有多少人签到" |
+| 去重逻辑 | 自动按 from_id 去重 | 每节点每天仅一次 |

doc/DIAGNOSTIC_GUIDE.md

@@ -0,0 +1,248 @@
+# QoS0 重发问题诊断指南
+
+## 快速诊断
+
+### 1. 启用详细日志运行服务器
+
+```bash
+./meshtastic_mqtt_server --console-log-mqtt=true --console-log-meshtastic=true
+```
+
+### 2. 观察日志输出
+
+#### ✅ 正常接收的消息
+```
+[mqtt] connect    client_id=device123 username=user1 remote=192.168.1.100:54321
+text       from=!12345678 channel=LongFast text="hello world"
+```
+
+#### ❌ 被拒绝的消息（关键！）
+```
+[mqtt] PUBLISH rejected: client_id=device123 topic=msh/CN/2/e/LongFast/!12345678 qos=0 payload_len=156 error=protobuf decode failed
+```
+
+#### ❌ 被屏蔽的消息
+```
+[mqtt] PUBLISH blocked: client_id=device123 topic=msh/CN/2/e/LongFast/!12345678 type=forbidden_word reason=blocked node
+```
+
+### 3. 根据日志判断问题
+
+| 日志内容 | 问题原因 | 解决方法 |
+|---------|---------|---------|
+| `PUBLISH rejected: error=protobuf decode failed` | 消息格式错误 | 检查设备固件版本 |
+| `PUBLISH rejected: error=cannot be decrypted` | 无法解密 | 检查 PSK 配置或启用 `allow_encrypted_forwarding` |
+| `PUBLISH blocked: type=node` | 节点被屏蔽 | 检查屏蔽规则 |
+| `PUBLISH blocked: type=forbidden_word` | 内容被屏蔽 | 检查关键词过滤规则 |
+| **没有 rejected/blocked 日志** | **不是服务器拒绝** | **问题在设备端** |
+
+## 详细诊断步骤
+
+### 步骤 1: 检查数据库中被拒绝的消息
+
+```bash
+# 进入数据库
+sqlite3 /path/to/database.db
+
+# 查看最近被拒绝的消息
+SELECT 
+    datetime(created_at, 'unixepoch', 'localtime') as time,
+    client_id,
+    json_extract(record, '$.error') as error,
+    json_extract(record, '$.topic') as topic,
+    payload_len
+FROM discarded_packets
+ORDER BY created_at DESC
+LIMIT 20;
+
+# 统计拒绝原因
+SELECT 
+    json_extract(record, '$.error') as error_type,
+    COUNT(*) as count
+FROM discarded_packets
+WHERE created_at > strftime('%s', 'now', '-1 hour')
+GROUP BY error_type;
+```
+
+### 步骤 2: 抓包分析
+
+```bash
+# 开始抓包
+sudo tcpdump -i any -nn port 1883 -w /tmp/mqtt_traffic.pcap
+
+# 让设备发送几条消息，然后停止抓包 (Ctrl+C)
+
+# 用 Wireshark 打开 /tmp/mqtt_traffic.pcap
+# 过滤器: mqtt
+# 查看:
+# 1. 是否看到重复的 PUBLISH 包（PacketID 相同）
+# 2. 重发的时间间隔是多少
+# 3. 是否有 TCP 重传标志 [TCP Retransmission]
+```
+
+### 步骤 3: 测试不同的消息类型
+
+```bash
+# 安装 mosquitto 客户端
+# macOS: brew install mosquitto
+# Linux: apt-get install mosquitto-clients
+
+# 发送一个简单的测试消息（QoS 0）
+mosquitto_pub -h localhost -p 1883 -t "test/topic" -m "hello" -q 0 -d
+
+# 观察:
+# 1. mosquitto_pub 是否报错
+# 2. 服务器日志是否显示 rejected
+# 3. 是否看到重发行为
+```
+
+### 步骤 4: 检查 PSK 配置
+
+```bash
+# 查看当前配置
+cat config.yaml | grep -A 5 "meshtastic:"
+
+# 如果使用默认 PSK
+psk: "AQ==" # 这是索引 1 的默认 PSK
+
+# 如果使用自定义 PSK，确保与设备一致
+psk: "your_base64_encoded_psk"
+```
+
+## 常见原因和解决方案
+
+### 原因 1: 消息无法解密
+
+**症状：** 日志显示 `error=cannot be decrypted`
+
+**解决方案 A - 配置正确的 PSK：**
+```yaml
+# config.yaml
+meshtastic:
+  psk: "your_base64_psk"  # 与设备 channel 的 PSK 一致
+```
+
+**解决方案 B - 允许转发加密消息：**
+```yaml
+# config.yaml
+meshtastic:
+  allow_encrypted_forwarding: true  # 即使无法解密也转发
+```
+
+### 原因 2: 节点或内容被屏蔽
+
+**症状：** 日志显示 `PUBLISH blocked`
+
+**解决方案：**
+```sql
+-- 查看屏蔽规则
+SELECT * FROM blocking_rules WHERE enabled = 1;
+
+-- 临时禁用特定规则
+UPDATE blocking_rules SET enabled = 0 WHERE id = <rule_id>;
+
+-- 或禁用所有规则测试
+UPDATE blocking_rules SET enabled = 0;
+```
+
+### 原因 3: Protobuf 解析失败
+
+**症状：** 日志显示 `error=protobuf decode failed`
+
+**可能原因：**
+- 设备发送的不是标准的 Meshtastic 协议包
+- 固件版本不兼容
+- 数据损坏
+
+**解决方案：**
+- 更新设备固件到最新版本
+- 检查设备配置是否正确
+- 联系设备厂商
+
+### 原因 4: 设备端 Bug
+
+**症状：** 服务器日志显示消息正常接收，没有 rejected/blocked，但设备仍然重发
+
+**诊断方法：**
+1. 检查设备日志（如果可访问）
+2. 更新设备固件
+3. 尝试不同的 QoS 级别（QoS 1）看是否还重发
+4. 联系设备厂商报告问题
+
+## 监控脚本
+
+创建一个监控脚本 `monitor_rejects.sh`：
+
+```bash
+#!/bin/bash
+echo "监控 MQTT 消息拒绝情况..."
+echo "按 Ctrl+C 停止"
+echo ""
+
+# 实时监控日志
+tail -f /path/to/server.log | grep --line-buffered -E "rejected|blocked" | while read line; do
+    timestamp=$(date '+%Y-%m-%d %H:%M:%S')
+    echo "[$timestamp] $line"
+    
+    # 播放提示音（可选）
+    # echo -e "\a"
+done
+```
+
+使用：
+```bash
+chmod +x monitor_rejects.sh
+./monitor_rejects.sh
+```
+
+## 性能统计
+
+查看消息处理统计：
+
+```sql
+-- 最近一小时的消息统计
+SELECT 
+    'Forwarded' as type,
+    COUNT(*) as count
+FROM packets
+WHERE created_at > strftime('%s', 'now', '-1 hour')
+UNION ALL
+SELECT 
+    'Rejected' as type,
+    COUNT(*) as count
+FROM discarded_packets
+WHERE created_at > strftime('%s', 'now', '-1 hour');
+
+-- 按客户端统计
+SELECT 
+    client_id,
+    COUNT(*) as total_messages,
+    SUM(CASE WHEN from_discarded = 1 THEN 1 ELSE 0 END) as rejected,
+    printf('%.2f%%', 
+        SUM(CASE WHEN from_discarded = 1 THEN 1 ELSE 0 END) * 100.0 / COUNT(*)
+    ) as reject_rate
+FROM (
+    SELECT client_id, 0 as from_discarded FROM packets
+    WHERE created_at > strftime('%s', 'now', '-1 hour')
+    UNION ALL
+    SELECT client_id, 1 as from_discarded FROM discarded_packets
+    WHERE created_at > strftime('%s', 'now', '-1 hour')
+)
+GROUP BY client_id
+ORDER BY rejected DESC;
+```
+
+## 总结
+
+遵循这个诊断流程：
+
+1. ✅ **启用详细日志** - 最重要的第一步
+2. ✅ **观察是否有 rejected/blocked** - 判断是否服务器拒绝
+3. ✅ **检查数据库** - 查看历史拒绝记录
+4. ✅ **抓包分析** - 确认网络层行为
+5. ✅ **根据原因修复** - 应用对应的解决方案
+
+如果日志中**没有任何 rejected/blocked 消息**，但设备仍然重发，那么问题100%在**设备端固件**，需要：
+- 更新设备固件
+- 检查设备配置
+- 联系设备厂商

doc/QOS0_RETRANSMIT_ANALYSIS.md

@@ -0,0 +1,259 @@
+# QoS0 消息重发问题深度分析
+
+## 问题现状
+
+即使启用了 TCP_NODELAY，设备发送 QoS0 消息后仍然重发 3 次。
+
+## 根本原因分析
+
+### 1. TCP_NODELAY 修复了什么？
+
+✅ TCP_NODELAY 确实解决了 **TCP ACK 延迟**问题：
+- Nagle 算法延迟从 40-200ms 降低到 ~0.05ms
+- TCP 层的确认现在是即时的
+
+❌ 但这**不能解决消息被拒绝的问题**。
+
+### 2. 消息被拒绝的流程
+
+当设备发送的消息不符合服务器要求时：
+
+```
+设备 → MQTT PUBLISH (QoS0)
+      ↓
+服务器 TCP 层收到 → 发送 TCP ACK ✅
+      ↓
+MQTT 层处理 → OnPublish hook
+      ↓
+MQTTPP 验证失败 → valid=false
+      ↓
+返回 packets.ErrRejectPacket
+      ↓
+mochi-mqtt 处理: return nil (不发送任何 MQTT 响应)
+      ↓
+设备收到 TCP ACK ✅ 但没有收到 MQTT 层响应
+      ↓
+设备认为消息可能丢失 → 重发 ❌
+```
+
+### 3. 为什么会重发？
+
+可能的原因：
+
+#### 原因 A: 消息验证失败
+
+检查以下验证失败的情况：
+
+1. **Protobuf 解码失败**
+   ```
+   parseServiceEnvelope() 返回错误
+   → MQTTPP 返回 valid=false
+   ```
+
+2. **解密失败**
+   ```
+   describePacket() 无法解密
+   → type="encrypted_packet" 且 AllowEncryptedForwarding=false
+   → MQTTPP 返回 valid=false
+   ```
+
+3. **屏蔽规则命中**
+   ```
+   blockingViolationForRecord() 返回非 nil
+   → OnPublish 返回 ErrRejectPacket
+   ```
+
+#### 原因 B: 设备期待应用层响应
+
+某些 MQTT 客户端实现可能：
+- 虽然使用 QoS0（不需要 PUBACK）
+- 但仍然期待某种应用层响应或订阅回显
+- 没有收到预期响应时触发重试逻辑
+
+#### 原因 C: 设备端 Bug
+
+设备固件可能有 bug：
+- 错误地认为 QoS0 需要应用层确认
+- 超时机制设置不当
+- 重试逻辑实现错误
+
+## 诊断步骤
+
+### 步骤 1: 查看服务器日志
+
+检查消息是否被拒绝：
+
+```bash
+# 启用控制台日志
+./meshtastic_mqtt_server --console-log-mqtt=true --console-log-meshtastic=true
+
+# 查找被拒绝的消息
+grep -E "error|dropped|rejected" logs.txt
+```
+
+**关键日志标识：**
+- `protobuf decode failed` - protobuf 解析失败
+- `cannot be decrypted` - 解密失败
+- `blocked node` / `forbidden word` - 屏蔽规则命中
+
+### 步骤 2: 抓包分析
+
+```bash
+# 抓取 MQTT 流量
+tcpdump -i any -nn port 1883 -w mqtt.pcap
+
+# 用 Wireshark 分析:
+# 1. 查看是否有 TCP 重传 (Retransmission)
+# 2. 查看 MQTT PUBLISH 是否有对应的响应
+# 3. 检查时序图，看设备重发的时间间隔
+```
+
+**期待的正常流程 (QoS0)：**
+```
+Client → Server: MQTT PUBLISH (QoS0)
+Server → Client: TCP ACK
+(没有 MQTT 层的 PUBACK，因为是 QoS0)
+```
+
+**如果消息被拒绝：**
+```
+Client → Server: MQTT PUBLISH (QoS0)
+Server → Client: TCP ACK
+(服务器静默丢弃，没有任何 MQTT 响应)
+Client → Server: MQTT PUBLISH (QoS0) [重发]
+Server → Client: TCP ACK
+...
+```
+
+### 步骤 3: 检查数据库
+
+```sql
+-- 查看被丢弃的消息
+SELECT * FROM discarded_packets 
+ORDER BY created_at DESC 
+LIMIT 20;
+
+-- 统计丢弃原因
+SELECT 
+    json_extract(record, '$.error') as error_type,
+    COUNT(*) as count
+FROM discarded_packets
+GROUP BY error_type;
+```
+
+### 步骤 4: 测试不同的消息
+
+```bash
+# 发送一个有效的测试消息
+mosquitto_pub -h localhost -p 1883 -t "msh/CN/2/e/LongFast/!12345678" -m "test" -q 0
+
+# 观察是否也会重发
+```
+
+## 解决方案
+
+### 方案 1: 修复消息验证问题
+
+如果是消息验证失败导致：
+
+**检查 PSK 配置：**
+```bash
+# 确保服务器配置了正确的 PSK
+./meshtastic_mqtt_server --psk="your_base64_psk"
+```
+
+**检查屏蔽规则：**
+```sql
+-- 查看当前的屏蔽规则
+SELECT * FROM blocking_rules WHERE enabled = 1;
+
+-- 临时禁用所有规则测试
+UPDATE blocking_rules SET enabled = 0;
+```
+
+### 方案 2: 允许加密消息转发
+
+如果消息是加密的且无法解密：
+
+```yaml
+# config.yaml
+meshtastic:
+  allow_encrypted_forwarding: true
+```
+
+这样即使无法解密，消息也会被转发而不是拒绝。
+
+### 方案 3: 返回明确的错误响应（不推荐）
+
+理论上可以在消息被拒绝时返回 MQTT 错误码，但：
+- ❌ QoS 0 协议规定不应该有 PUBACK
+- ❌ 违反 MQTT 规范
+- ❌ 可能导致客户端行为异常
+
+### 方案 4: 设备端修复
+
+如果是设备固件 bug：
+- 更新设备固件到最新版本
+- 检查设备日志，确认重发原因
+- 联系设备厂商报告 bug
+
+## 监控和调试
+
+### 添加详细日志
+
+修改 `main.go` 的 `OnPublish` 方法：
+
+```go
+func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (packets.Packet, error) {
+    valid, _, record := mqtpp.MQTTPP(pk.TopicName, pk.Payload, h.key, mqtpp.Options{
+        AllowEncryptedForwarding: h.settings.AllowEncryptedForwarding(),
+        PKIKeyResolver:           h.pkiResolver,
+    })
+    
+    info := mqttClientInfoFromClient(cl)
+    
+    if !valid {
+        // 添加详细日志
+        printJSON(map[string]any{
+            "event": "publish_rejected",
+            "reason": "validation_failed",
+            "client_id": info.ClientID,
+            "topic": pk.TopicName,
+            "payload_len": len(pk.Payload),
+            "error": record["error"],
+        })
+        h.rejectPublish(cl, pk, record)
+        return pk, packets.ErrRejectPacket
+    }
+    
+    // ... 其他逻辑
+}
+```
+
+### 监控重发率
+
+```sql
+-- 创建视图统计每个客户端的重发率
+CREATE VIEW client_retransmit_stats AS
+SELECT 
+    client_id,
+    COUNT(*) as total_attempts,
+    COUNT(DISTINCT packet_id) as unique_packets,
+    (COUNT(*) - COUNT(DISTINCT packet_id)) * 100.0 / COUNT(*) as retransmit_rate
+FROM packets
+GROUP BY client_id
+HAVING retransmit_rate > 10;
+```
+
+## 结论
+
+TCP_NODELAY 修复了 TCP 层的延迟问题，但如果消息本身被服务器拒绝（验证失败、解密失败、屏蔽规则等），设备仍然会重发。
+
+**下一步行动：**
+
+1. ✅ 启用详细日志，查看是否有消息被拒绝
+2. ✅ 检查 `discarded_packets` 表，确认拒绝原因
+3. ✅ 抓包分析，确认是 TCP 重传还是应用层重发
+4. ✅ 根据诊断结果选择对应的解决方案
+
+**如果所有消息都被正常处理（没有被拒绝），但仍然重发，那么问题在设备端固件。**

doc/README_QOS0_FIX.md

@@ -0,0 +1,197 @@
+# MQTT QoS0 重发问题 - 完整解决方案
+
+## 问题描述
+
+用户报告：设备使用 QoS0 发送 MQTT 消息后，服务器好像不给 ACK，导致设备一直重发 3 次，有时候又一次发送成功。
+
+## 解决方案总结
+
+我们进行了两轮修复和诊断增强：
+
+### 第一轮：修复 TCP 层延迟问题 ✅
+
+**问题：** TCP Nagle 算法导致 TCP ACK 延迟 40-200ms  
+**修复：** 在 `OnConnect` 中启用 `TCP_NODELAY`  
+**效果：** TCP ACK 延迟降低到 ~0.05ms  
+**Commit:** `ec24e70` - 修复 MQTT QoS0 消息重发问题
+
+### 第二轮：诊断应用层拒绝问题 ✅
+
+**发现：** TCP_NODELAY 只解决了传输层延迟，但如果消息被应用层拒绝，设备仍会重发  
+**改进：** 添加详细的拒绝日志，帮助快速定位问题  
+**Commit:** `bce6b70` - 增强 MQTT 消息拒绝诊断功能
+
+## 快速诊断方法
+
+### 步骤 1: 启用详细日志
+
+```bash
+./meshtastic_mqtt_server --console-log-mqtt=true --console-log-meshtastic=true
+```
+
+### 步骤 2: 观察日志
+
+#### 场景 A: 看到 `PUBLISH rejected` 或 `PUBLISH blocked`
+→ **服务器拒绝了消息**  
+→ 查看具体错误原因，应用对应的解决方案
+
+#### 场景 B: 没有看到 rejected/blocked，消息正常接收
+→ **不是服务器问题**  
+→ 问题在设备端固件，需要设备端修复
+
+## 常见原因和解决方案
+
+### 原因 1: 消息无法解密 (`error=cannot be decrypted`)
+
+**解决方案：**
+```yaml
+# 方法 1: 配置正确的 PSK
+meshtastic:
+  psk: "your_base64_psk"  # 必须与设备一致
+
+# 方法 2: 允许转发加密消息
+meshtastic:
+  allow_encrypted_forwarding: true
+```
+
+### 原因 2: 消息被屏蔽 (`PUBLISH blocked`)
+
+**解决方案：**
+```sql
+-- 查看屏蔽规则
+SELECT * FROM blocking_rules WHERE enabled = 1;
+
+-- 禁用特定规则
+UPDATE blocking_rules SET enabled = 0 WHERE id = <rule_id>;
+```
+
+### 原因 3: Protobuf 解析失败 (`error=protobuf decode failed`)
+
+**解决方案：**
+- 更新设备固件到最新版本
+- 检查设备是否使用标准 Meshtastic 协议
+
+### 原因 4: 设备端 Bug
+
+**症状：** 服务器日志正常，但设备仍重发  
+**解决方案：**
+- 更新设备固件
+- 联系设备厂商
+
+## 文档索引
+
+1. **[TCP_ACK_FIX_CN.md](doc/TCP_ACK_FIX_CN.md)** - TCP_NODELAY 修复详解（中文）
+2. **[TCP_ACK_FIX.md](doc/TCP_ACK_FIX.md)** - TCP_NODELAY 修复详解（英文）
+3. **[QOS0_RETRANSMIT_ANALYSIS.md](doc/QOS0_RETRANSMIT_ANALYSIS.md)** - 重发问题深度分析
+4. **[DIAGNOSTIC_GUIDE.md](doc/DIAGNOSTIC_GUIDE.md)** - 完整诊断指南
+
+## 技术细节
+
+### TCP_NODELAY 的作用
+
+```
+没有 TCP_NODELAY:
+  客户端发送 → 服务器收到 → 等待 Nagle 算法 (40-200ms) → TCP ACK
+
+有 TCP_NODELAY:
+  客户端发送 → 服务器收到 → 立即 TCP ACK (~0.05ms)
+```
+
+### MQTT QoS0 的特性
+
+- QoS 0 = "至多一次"交付
+- MQTT 应用层**不需要** PUBACK
+- 但 TCP 层**仍然需要** TCP ACK
+- 如果消息被应用层拒绝（返回 ErrRejectPacket），mochi-mqtt 只是静默返回，不发送任何响应
+
+### 消息处理流程
+
+```
+设备 → MQTT PUBLISH (QoS0)
+      ↓
+TCP 层收到 → TCP ACK ✅ (现在是即时的)
+      ↓
+MQTT 层 OnPublish hook
+      ↓
+MQTTPP 验证
+      ↓
+├─ valid=true → 转发消息 ✅
+│
+└─ valid=false → 返回 ErrRejectPacket → 静默丢弃 ❌
+                ↓
+             设备可能重发
+```
+
+## Git 提交历史
+
+```bash
+git log --oneline -3
+
+bce6b70 增强 MQTT 消息拒绝诊断功能
+ec24e70 修复 MQTT QoS0 消息重发问题
+cfe4ef0 修复消息重复问题
+```
+
+## 测试验证
+
+### 单元测试
+```bash
+go test -v -run TestTCPNoDelay
+go test -v -run TestQoS0MessageLatency
+```
+
+### 集成测试
+```bash
+# 启动服务器
+./meshtastic_mqtt_server --console-log-mqtt=true
+
+# 发送测试消息
+mosquitto_pub -h localhost -p 1883 -t "test/topic" -m "hello" -q 0
+
+# 观察日志输出
+```
+
+## 性能指标
+
+| 指标 | 修复前 | 修复后 |
+|------|--------|--------|
+| TCP ACK 延迟 | 40-200ms | ~0.05ms |
+| 往返延迟 | 不稳定 | ~54µs |
+| 提升倍数 | - | **1000x** |
+
+## 下一步行动
+
+1. ✅ **重新编译部署**
+   ```bash
+   go build
+   ./meshtastic_mqtt_server
+   ```
+
+2. ✅ **启用详细日志**
+   ```bash
+   ./meshtastic_mqtt_server --console-log-mqtt=true
+   ```
+
+3. ✅ **观察日志**
+   - 如果看到 `rejected/blocked` → 应用对应的解决方案
+   - 如果没看到 → 问题在设备端
+
+4. ✅ **查看数据库**
+   ```sql
+   SELECT * FROM discarded_packets ORDER BY created_at DESC LIMIT 20;
+   ```
+
+5. ✅ **必要时抓包**
+   ```bash
+   tcpdump -i any -nn port 1883 -w mqtt.pcap
+   ```
+
+## 结论
+
+我们修复了 TCP 层的延迟问题，并添加了完善的诊断工具。现在你可以：
+
+1. 快速判断重发是由服务器拒绝还是设备 bug 引起
+2. 看到具体的拒绝原因和错误信息
+3. 根据诊断结果应用对应的解决方案
+
+**如果重发问题依然存在，请按照诊断指南操作，查看日志中是否有 rejected/blocked 消息，并将结果反馈给我。**

doc/SIGN_QUERY_FEATURE.md

@@ -7,10 +7,11 @@
 1. **数据不准确**：AI 只能根据上下文猜测或给出模糊答案
 2. **无法查询历史**：询问超过一天的数据时，AI 一问三不知
 3. **功能不完整**：数据库层已有完善的查询能力（`CountSigns`、`CountSignsByDay`），但工具层未暴露给 AI
+4. **状态判断不准**：用户问"我今天签到了吗"时，AI 基于对话记忆判断而非查询数据库，导致误判
 
 ## 解决方案
 
-为签到工具添加查询功能，支持以下两种操作模式：
+为签到工具添加查询和检查功能，支持以下三种操作模式：
 
 ### 1. 签到操作 (action=sign)
 
@@ -21,13 +22,20 @@
 
 ### 2. 查询操作 (action=query)
 
-新增查询功能：
+查询签到统计功能：
 - 查询指定日期或日期范围的签到统计
 - 返回总人次和按天分组的统计数据
 - 可选参数：
   - `date`: 查询日期（格式：YYYY-MM-DD），默认今天
   - `days`: 查询最近 N 天，默认只查询 date 指定的那一天
 
+### 3. 检查操作 (action=check) **新增**
+
+检查当前节点今天是否已签到：
+- 用于回答"我今天签到了吗"、"我什么时候签到的"之类的问题
+- 直接查询数据库，不依赖对话历史
+- 返回明确的签到状态，**包括签到时间和签到内容**
+
 ## 修改内容
 
 ### 1. 扩展 SignStore 接口
@@ -82,6 +90,33 @@ type signParams struct {
 
 ### AI 调用示例
 
+#### 检查今天是否签到（check 操作）
+
+**用户**："我今天签到了吗？"
+
+AI 调用：
+```json
+{
+  "action": "check"
+}
+```
+
+返回（未签到）：
+```text
+Test Node 今天还没有签到。
+```
+
+返回（已签到）：
+```text
+Test Node 今天已经签到过了。
+签到时间：10:30:45
+签到内容：上海闵行-Kevin-GAT562签到
+```
+
+**用户**："我什么时候签到的？"
+
+AI 同样调用 check 操作，返回包含签到时间和内容的完整信息。
+
 #### 查询今天的签到情况
 ```json
 {

install.sh

@@ -3,6 +3,7 @@ set -euo pipefail
 
 SERVICE_NAME="mesh_mqtt_go"
 SERVICE_USER="mesh_mqtt_go"
+SERVICE_GROUP="${SERVICE_USER}"
 CONFIG_DIR="/etc/${SERVICE_NAME}"
 DATA_DIR="/srv/${SERVICE_NAME}"
 INSTALL_DIR="/opt/${SERVICE_NAME}"
@@ -17,6 +18,12 @@ if [[ "${EUID}" -ne 0 ]]; then
   exit 1
 fi
 
+if id -u "www" >/dev/null 2>&1; then
+  SERVICE_USER="www"
+  SERVICE_GROUP=$(id -gn "www")
+  echo "检测到 www 用户，将以 www 用户运行服务"
+fi
+
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
 cd "${SCRIPT_DIR}"
 
@@ -42,8 +49,8 @@ if ! id -u "${SERVICE_USER}" >/dev/null 2>&1; then
 fi
 
 echo "创建目录..."
-install -d -m 0750 -o "${SERVICE_USER}" -g "${SERVICE_USER}" "${CONFIG_DIR}" "${DATA_DIR}"
-install -d -m 0755 -o "${SERVICE_USER}" -g "${SERVICE_USER}" "${INSTALL_DIR}"
+install -d -m 0750 -o "${SERVICE_USER}" -g "${SERVICE_GROUP}" "${CONFIG_DIR}" "${DATA_DIR}"
+install -d -m 0755 -o "${SERVICE_USER}" -g "${SERVICE_GROUP}" "${INSTALL_DIR}"
 
 echo "安装程序和前端文件..."
 install -m 0755 -o root -g root "${SCRIPT_DIR}/${BINARY_NAME}" "${INSTALL_DIR}/${BINARY_NAME}"
@@ -51,7 +58,7 @@ rm -rf "${INSTALL_DIR}/dist"
 cp -a "${SCRIPT_DIR}/${FRONTEND_DIST_DIR}" "${INSTALL_DIR}/dist"
 chown root:root "${INSTALL_DIR}/${BINARY_NAME}"
 chown -R root:root "${INSTALL_DIR}/dist"
-chown "${SERVICE_USER}:${SERVICE_USER}" "${INSTALL_DIR}"
+chown "${SERVICE_USER}:${SERVICE_GROUP}" "${INSTALL_DIR}"
 chmod 0755 "${INSTALL_DIR}"
 find "${INSTALL_DIR}/dist" -type d -exec chmod 0755 {} \;
 find "${INSTALL_DIR}/dist" -type f -exec chmod 0644 {} \;
@@ -91,7 +98,7 @@ console_log:
   sql: true
   meshtastic: true
 EOF
-  chown "${SERVICE_USER}:${SERVICE_USER}" "${CONFIG_DIR}/config.yaml"
+  chown "${SERVICE_USER}:${SERVICE_GROUP}" "${CONFIG_DIR}/config.yaml"
   chmod 0640 "${CONFIG_DIR}/config.yaml"
 fi
 
@@ -105,7 +112,7 @@ Wants=network-online.target
 [Service]
 Type=simple
 User=${SERVICE_USER}
-Group=${SERVICE_USER}
+Group=${SERVICE_GROUP}
 WorkingDirectory=${INSTALL_DIR}
 ExecStart=${INSTALL_DIR}/${BINARY_NAME} -web-socket-path ${SOCKET_PATH} -web-static-dir ${INSTALL_DIR}/dist
 Restart=on-failure

internal/agents/active/active.go

@@ -0,0 +1,150 @@
+// Package active 提供活跃度查询工具。当用户想查询活跃节点数或活跃人数时调用。
+//
+// 活跃节点：查询 nodeinfo 表的 updated_at 字段，统计指定时间范围内更新过的节点数
+// 活跃人数：查询 text_message 表的 created_at 字段，统计指定时间范围内发过消息的唯一用户数（按 from_id 去重）
+package active
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"strings"
+	"time"
+
+	"meshtastic_mqtt_server/internal/agenttool"
+
+	"github.com/volcengine/volcengine-go-sdk/service/arkruntime/model"
+)
+
+// ActiveStore 定义活跃度查询工具所需的持久化能力，通常由 *store.Store 实现。
+type ActiveStore interface {
+	CountActiveNodes(since time.Time) (int64, error)
+	CountActiveUsers(since time.Time) (int64, error)
+}
+
+// Tool 是活跃度查询工具。
+type Tool struct {
+	enabled bool
+	store   ActiveStore
+}
+
+// Name returns the tool name
+func (t *Tool) Name() string { return "active" }
+
+// Enabled returns whether the tool is enabled
+func (t *Tool) Enabled() bool { return t.enabled && t.store != nil }
+
+// ToolDefinition returns the OpenAI tool definition
+func (t *Tool) ToolDefinition(description string) *model.Tool {
+	desc := "活跃度查询工具。查询指定时间范围内的活跃节点数和活跃人数。\n" +
+		"活跃节点：在指定时间内有更新记录的节点数量\n" +
+		"活跃人数：在指定时间内发送过消息的唯一用户数（按 from_id 去重）\n" +
+		"默认查询最近1小时，最大支持24小时"
+	if description != "" {
+		desc = description
+	}
+	return &model.Tool{
+		Type: model.ToolTypeFunction,
+		Function: &model.FunctionDefinition{
+			Name:        "active",
+			Description: desc,
+			Parameters: map[string]any{
+				"type": "object",
+				"properties": map[string]any{
+					"hours": map[string]any{
+						"type":        "number",
+						"description": "查询最近多少小时内的活跃数据，默认1小时，最大24小时。例如：1、2、6、12、24",
+					},
+					"query_type": map[string]any{
+						"type":        "string",
+						"enum":        []string{"both", "nodes", "users"},
+						"description": "查询类型：both=同时查询节点和人数（默认），nodes=仅查询节点，users=仅查询人数",
+					},
+				},
+			},
+		},
+	}
+}
+
+// Execute executes the active query tool
+func (t *Tool) Execute(ctx context.Context, args string, runtime agenttool.Runtime) (string, error) {
+	if t.store == nil {
+		return "", fmt.Errorf("active store is not configured")
+	}
+
+	var params activeParams
+	if err := json.Unmarshal([]byte(args), &params); err != nil {
+		return "", fmt.Errorf("failed to parse arguments: %w", err)
+	}
+
+	// 默认查询1小时
+	hours := params.Hours
+	if hours <= 0 {
+		hours = 1
+	}
+	// 最大24小时
+	if hours > 24 {
+		hours = 24
+	}
+
+	// 默认查询类型为 both
+	queryType := strings.ToLower(strings.TrimSpace(params.QueryType))
+	if queryType == "" {
+		queryType = "both"
+	}
+
+	now := runtime.Now
+	if now.IsZero() {
+		now = time.Now()
+	}
+
+	// 计算时间范围
+	since := now.Add(-time.Duration(hours * float64(time.Hour)))
+
+	var result strings.Builder
+	result.WriteString(fmt.Sprintf("最近 %.1f 小时的活跃统计：\n\n", hours))
+
+	// 查询活跃节点
+	if queryType == "both" || queryType == "nodes" {
+		nodeCount, err := t.store.CountActiveNodes(since)
+		if err != nil {
+			return "", fmt.Errorf("查询活跃节点失败：%w", err)
+		}
+		result.WriteString(fmt.Sprintf("活跃节点：%d 个\n", nodeCount))
+	}
+
+	// 查询活跃人数
+	if queryType == "both" || queryType == "users" {
+		userCount, err := t.store.CountActiveUsers(since)
+		if err != nil {
+			return "", fmt.Errorf("查询活跃人数失败：%w", err)
+		}
+		result.WriteString(fmt.Sprintf("活跃人数：%d 人\n", userCount))
+	}
+
+	return result.String(), nil
+}
+
+// activeParams 是活跃度查询工具的入参。
+type activeParams struct {
+	Hours     float64 `json:"hours"`      // 查询最近多少小时，默认1小时
+	QueryType string  `json:"query_type"` // 查询类型：both/nodes/users
+}
+
+// RawState returns the tool state
+func (t *Tool) RawState() any {
+	return map[string]any{"enabled": t.enabled, "has_store": t.store != nil}
+}
+
+func init() {
+	agenttool.Register(agenttool.Descriptor{
+		Name: "active",
+		Load: func(path string, options agenttool.LoadOptions) (agenttool.LoadedTool, error) {
+			tool := &Tool{enabled: true}
+			if store, ok := options.Value("store").(ActiveStore); ok && store != nil {
+				tool.store = store
+			}
+			return tool, nil
+		},
+	})
+}

internal/agents/active/active_test.go

@@ -0,0 +1,187 @@
+package active
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"meshtastic_mqtt_server/internal/agenttool"
+)
+
+// mockActiveStore 是用于测试的 mock store
+type mockActiveStore struct {
+	activeNodeCount int64
+	activeUserCount int64
+}
+
+func (m *mockActiveStore) CountActiveNodes(since time.Time) (int64, error) {
+	return m.activeNodeCount, nil
+}
+
+func (m *mockActiveStore) CountActiveUsers(since time.Time) (int64, error) {
+	return m.activeUserCount, nil
+}
+
+func TestActiveTool_Query(t *testing.T) {
+	now := time.Date(2024, 6, 23, 12, 0, 0, 0, time.UTC)
+
+	store := &mockActiveStore{
+		activeNodeCount: 25,
+		activeUserCount: 15,
+	}
+
+	tool := &Tool{
+		enabled: true,
+		store:   store,
+	}
+
+	tests := []struct {
+		name        string
+		hours       float64
+		queryType   string
+		expectNodes bool
+		expectUsers bool
+		expectError bool
+	}{
+		{
+			name:        "默认查询1小时（both）",
+			hours:       0, // 0 表示使用默认值
+			queryType:   "",
+			expectNodes: true,
+			expectUsers: true,
+			expectError: false,
+		},
+		{
+			name:        "查询6小时",
+			hours:       6,
+			queryType:   "both",
+			expectNodes: true,
+			expectUsers: true,
+			expectError: false,
+		},
+		{
+			name:        "仅查询节点",
+			hours:       1,
+			queryType:   "nodes",
+			expectNodes: true,
+			expectUsers: false,
+			expectError: false,
+		},
+		{
+			name:        "仅查询人数",
+			hours:       1,
+			queryType:   "users",
+			expectNodes: false,
+			expectUsers: true,
+			expectError: false,
+		},
+		{
+			name:        "查询24小时（最大值）",
+			hours:       24,
+			queryType:   "both",
+			expectNodes: true,
+			expectUsers: true,
+			expectError: false,
+		},
+		{
+			name:        "超过24小时应限制到24小时",
+			hours:       48,
+			queryType:   "both",
+			expectNodes: true,
+			expectUsers: true,
+			expectError: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			params := activeParams{
+				Hours:     tt.hours,
+				QueryType: tt.queryType,
+			}
+			argsJSON, _ := json.Marshal(params)
+
+			runtime := agenttool.Runtime{Now: now}
+			result, err := tool.Execute(context.Background(), string(argsJSON), runtime)
+
+			if tt.expectError && err == nil {
+				t.Errorf("Expected error but got none")
+			}
+			if !tt.expectError && err != nil {
+				t.Errorf("Unexpected error: %v", err)
+			}
+			if !tt.expectError {
+				t.Logf("Query result:\n%s", result)
+
+				// 验证结果包含预期的内容
+				if tt.expectNodes && result != "" {
+					// 应该包含节点统计
+					if !contains(result, "活跃节点") {
+						t.Errorf("Expected result to contain node count")
+					}
+				}
+				if tt.expectUsers && result != "" {
+					// 应该包含人数统计
+					if !contains(result, "活跃人数") {
+						t.Errorf("Expected result to contain user count")
+					}
+				}
+			}
+		})
+	}
+}
+
+func TestActiveTool_Enabled(t *testing.T) {
+	// 测试工具启用状态
+	tests := []struct {
+		name    string
+		enabled bool
+		store   ActiveStore
+		expect  bool
+	}{
+		{
+			name:    "启用且有store",
+			enabled: true,
+			store:   &mockActiveStore{},
+			expect:  true,
+		},
+		{
+			name:    "启用但无store",
+			enabled: true,
+			store:   nil,
+			expect:  false,
+		},
+		{
+			name:    "禁用且有store",
+			enabled: false,
+			store:   &mockActiveStore{},
+			expect:  false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tool := &Tool{
+				enabled: tt.enabled,
+				store:   tt.store,
+			}
+			if tool.Enabled() != tt.expect {
+				t.Errorf("Expected Enabled() = %v, got %v", tt.expect, tool.Enabled())
+			}
+		})
+	}
+}
+
+func contains(s, substr string) bool {
+	return len(s) > 0 && len(substr) > 0 && (s == substr || len(s) >= len(substr) && (s[:len(substr)] == substr || s[len(s)-len(substr):] == substr || containsMiddle(s, substr)))
+}
+
+func containsMiddle(s, substr string) bool {
+	for i := 0; i <= len(s)-len(substr); i++ {
+		if s[i:i+len(substr)] == substr {
+			return true
+		}
+	}
+	return false
+}

internal/agents/sign/sign.go

@@ -45,9 +45,10 @@ func (t *Tool) Enabled() bool { return t.enabled && t.store != nil }
 
 // ToolDefinition returns the OpenAI tool definition
 func (t *Tool) ToolDefinition(description string) *model.Tool {
-	desc := "签到工具。支持签到和查询两种操作：\n" +
+	desc := "签到工具。支持三种操作：\n" +
 		"1. 签到操作(action=sign)：记录节点今日签到信息，每个节点每天只能签到一次。必填参数：地区、名字、设备\n" +
-		"2. 查询操作(action=query)：查询签到统计。可按日期范围查询，默认查询今天。返回签到总数和按天的统计数据"
+		"2. 查询操作(action=query)：查询签到统计。可按日期范围查询，默认查询今天。返回签到总数和按天的统计数据\n" +
+		"3. 检查操作(action=check)：检查当前节点今天是否已签到。用于回答用户'我今天签到了吗'之类的问题"
 	if description != "" {
 		desc = description
 	}
@@ -61,8 +62,8 @@ func (t *Tool) ToolDefinition(description string) *model.Tool {
 				"properties": map[string]any{
 					"action": map[string]any{
 						"type":        "string",
-						"enum":        []string{"sign", "query"},
-						"description": "操作类型：sign=签到，query=查询签到统计",
+						"enum":        []string{"sign", "query", "check"},
+						"description": "操作类型：sign=签到，query=查询签到统计，check=检查今天是否已签到",
 					},
 					"region": map[string]any{
 						"type":        "string",
@@ -122,10 +123,12 @@ func (t *Tool) Execute(ctx context.Context, args string, runtime agenttool.Runti
 	switch strings.ToLower(strings.TrimSpace(params.Action)) {
 	case "query":
 		return t.executeQuery(ctx, params, runtime)
+	case "check":
+		return t.executeCheck(ctx, params, runtime)
 	case "sign", "":
 		return t.executeSign(ctx, params, runtime)
 	default:
-		return "", fmt.Errorf("无效的操作类型：%s，只支持 sign 或 query", params.Action)
+		return "", fmt.Errorf("无效的操作类型：%s，只支持 sign、query 或 check", params.Action)
 	}
 }
 
@@ -176,6 +179,51 @@ func (t *Tool) executeSign(ctx context.Context, params signParams, runtime agent
 	return fmt.Sprintf("签到成功！%s\n签到内容：%s", displayName(node), record.SignText), nil
 }
 
+// executeCheck 检查当前节点今天是否已签到，并返回签到详情
+func (t *Tool) executeCheck(ctx context.Context, params signParams, runtime agenttool.Runtime) (string, error) {
+	// 节点身份来自消息上下文
+	node, ok := agenttool.NodeContextFromContext(ctx)
+	if !ok || strings.TrimSpace(node.NodeID) == "" {
+		return "", fmt.Errorf("缺少发送节点上下文，无法检查签到状态")
+	}
+
+	now := runtime.Now
+	if now.IsZero() {
+		now = time.Now()
+	}
+
+	// 构建查询选项：查询今天的签到记录
+	loc := now.Location()
+	if loc == nil {
+		loc = time.Local
+	}
+	start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, loc)
+	end := start.AddDate(0, 0, 1)
+
+	opts := storepkg.ListOptions{
+		NodeID: node.NodeID,
+		Since:  &start,
+		Until:  &end,
+		Limit:  1,
+	}
+
+	// 查询数据库获取今天的签到记录
+	signs, err := t.store.ListSigns(opts)
+	if err != nil {
+		return "", fmt.Errorf("查询签到记录失败：%w", err)
+	}
+
+	if len(signs) == 0 {
+		return fmt.Sprintf("%s 今天还没有签到。", displayName(node)), nil
+	}
+
+	// 返回签到详情
+	sign := signs[0]
+	signTimeStr := sign.SignTime.Format("15:04:05")
+	return fmt.Sprintf("%s 今天已经签到过了。\n签到时间：%s\n签到内容：%s",
+		displayName(node), signTimeStr, sign.SignText), nil
+}
+
 // executeQuery 执行查询操作
 func (t *Tool) executeQuery(ctx context.Context, params signParams, runtime agenttool.Runtime) (string, error) {
 	now := runtime.Now

internal/agents/sign/sign_test.go

@@ -85,6 +85,11 @@ func (m *mockSignStore) CountSignsByDay(opts storepkg.ListOptions) ([]storepkg.S
 func (m *mockSignStore) ListSigns(opts storepkg.ListOptions) ([]storepkg.SignRecord, error) {
 	var result []storepkg.SignRecord
 	for _, sign := range m.signs {
+		// 过滤 NodeID
+		if opts.NodeID != "" && sign.NodeID != opts.NodeID {
+			continue
+		}
+		// 过滤时间范围
 		if opts.Since != nil && sign.SignTime.Before(*opts.Since) {
 			continue
 		}
@@ -93,6 +98,10 @@ func (m *mockSignStore) ListSigns(opts storepkg.ListOptions) ([]storepkg.SignRec
 		}
 		result = append(result, sign)
 	}
+	// 应用 Limit
+	if opts.Limit > 0 && len(result) > opts.Limit {
+		result = result[:opts.Limit]
+	}
 	return result, nil
 }
 
@@ -218,3 +227,110 @@ func TestSignTool_SignAction(t *testing.T) {
 		t.Errorf("Expected 1 sign record, got %d", len(store.signs))
 	}
 }
+
+func TestSignTool_CheckAction(t *testing.T) {
+	now := time.Date(2024, 6, 23, 12, 0, 0, 0, time.UTC)
+
+	// 测试场景1：今天未签到
+	t.Run("今天未签到", func(t *testing.T) {
+		store := &mockSignStore{
+			signs:       []storepkg.SignRecord{},
+			nodeInfoMap: make(map[string]*storepkg.NodeInfoRecord),
+		}
+
+		tool := &Tool{
+			enabled: true,
+			store:   store,
+		}
+
+		params := signParams{
+			Action: "check",
+		}
+		argsJSON, _ := json.Marshal(params)
+
+		nodeCtx := agenttool.NodeContext{
+			NodeID:    "test_node_123",
+			LongName:  "Test Node",
+			ShortName: "TN",
+		}
+		ctx := agenttool.WithNodeContext(context.Background(), nodeCtx)
+
+		runtime := agenttool.Runtime{Now: now}
+		result, err := tool.Execute(ctx, string(argsJSON), runtime)
+
+		if err != nil {
+			t.Errorf("Unexpected error: %v", err)
+		}
+		t.Logf("Check result (not signed): %s", result)
+
+		if !contains(result, "还没有签到") && !contains(result, "没有签到") {
+			t.Errorf("Expected result to indicate not signed yet")
+		}
+	})
+
+	// 测试场景2：今天已签到
+	t.Run("今天已签到", func(t *testing.T) {
+		signTime := time.Date(2024, 6, 23, 10, 30, 45, 0, time.UTC)
+		store := &mockSignStore{
+			signs: []storepkg.SignRecord{
+				{
+					NodeID:   "test_node_123",
+					SignText: "上海-TestUser-TestDevice签到",
+					SignTime: signTime,
+				},
+			},
+			nodeInfoMap: make(map[string]*storepkg.NodeInfoRecord),
+		}
+
+		tool := &Tool{
+			enabled: true,
+			store:   store,
+		}
+
+		params := signParams{
+			Action: "check",
+		}
+		argsJSON, _ := json.Marshal(params)
+
+		nodeCtx := agenttool.NodeContext{
+			NodeID:    "test_node_123",
+			LongName:  "Test Node",
+			ShortName: "TN",
+		}
+		ctx := agenttool.WithNodeContext(context.Background(), nodeCtx)
+
+		runtime := agenttool.Runtime{Now: now}
+		result, err := tool.Execute(ctx, string(argsJSON), runtime)
+
+		if err != nil {
+			t.Errorf("Unexpected error: %v", err)
+		}
+		t.Logf("Check result (already signed): %s", result)
+
+		if !contains(result, "已经签到") {
+			t.Errorf("Expected result to indicate already signed")
+		}
+		if !contains(result, "签到时间") {
+			t.Errorf("Expected result to contain sign time")
+		}
+		if !contains(result, "10:30:45") {
+			t.Errorf("Expected result to contain the exact sign time")
+		}
+		if !contains(result, "签到内容") {
+			t.Errorf("Expected result to contain sign text")
+		}
+	})
+}
+
+func contains(s, substr string) bool {
+	return len(s) > 0 && len(substr) > 0 && (s == substr || len(s) >= len(substr) && (s[:len(substr)] == substr || s[len(s)-len(substr):] == substr || containsMiddle(s, substr)))
+}
+
+func containsMiddle(s, substr string) bool {
+	for i := 0; i <= len(s)-len(substr); i++ {
+		if s[i:i+len(substr)] == substr {
+			return true
+		}
+	}
+	return false
+}

internal/ai/manager.go

@@ -0,0 +1,203 @@
+package ai
+
+import (
+	"context"
+	"fmt"
+	"sync"
+
+	"meshtastic_mqtt_server/internal/autoreply"
+	"meshtastic_mqtt_server/internal/llm"
+	storepkg "meshtastic_mqtt_server/internal/store"
+
+	"gorm.io/gorm"
+)
+
+// AIServiceStatus reports the current state of the AI service
+type AIServiceStatus struct {
+	Running       bool   `json:"running"`
+	Enabled       bool   `json:"enabled"`
+	ProviderCount int    `json:"provider_count"`
+	Message       string `json:"message,omitempty"`
+}
+
+// AIManager manages the lifecycle of the AI service, supporting restart
+type AIManager struct {
+	mu        sync.Mutex
+	service   *Service
+	cfg       Config
+	db        *gorm.DB
+	botSender autoreply.BotSender
+	ctx       context.Context
+	store     *storepkg.Store
+}
+
+// NewAIManager creates a new AIManager
+func NewAIManager(cfg Config, db *gorm.DB, botSender autoreply.BotSender, ctx context.Context, store *storepkg.Store) *AIManager {
+	return &AIManager{
+		cfg:       cfg,
+		db:        db,
+		botSender: botSender,
+		ctx:       ctx,
+		store:     store,
+	}
+}
+
+// SetConfigEnabled sets the enabled flag on the config
+func (m *AIManager) SetConfigEnabled(enabled bool) {
+	m.cfg.Enabled = enabled
+}
+
+// SetProviderConfigs sets the LLM provider configs
+func (m *AIManager) SetProviderConfigs(configs []llm.ProviderConfig) {
+	m.cfg.LLMProviders = configs
+}
+
+// Init creates and starts the AI service
+func (m *AIManager) Init() error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	svc, err := NewService(m.cfg, m.db, m.botSender)
+	if err != nil {
+		return fmt.Errorf("failed to create AI service: %w", err)
+	}
+	if err := svc.Start(m.ctx); err != nil {
+		return fmt.Errorf("failed to start AI service: %w", err)
+	}
+	m.service = svc
+	return nil
+}
+
+// Stop stops the currently running AI service
+func (m *AIManager) Stop() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if m.service != nil {
+		m.service.Stop()
+		m.service = nil
+	}
+}
+
+// Status returns the current AI service status
+func (m *AIManager) Status() AIServiceStatus {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	providerCount := len(m.cfg.LLMProviders)
+	if m.service == nil {
+		msg := "AI 服务未运行，可在配置提供商后点击重启"
+		if providerCount == 0 {
+			msg = "尚未配置 AI 提供商，请先添加提供商配置"
+		}
+		return AIServiceStatus{
+			Running:       false,
+			Enabled:       m.cfg.Enabled,
+			ProviderCount: providerCount,
+			Message:       msg,
+		}
+	}
+	enabled := m.service.Enabled()
+	if !enabled {
+		return AIServiceStatus{
+			Running:       false,
+			Enabled:       false,
+			ProviderCount: providerCount,
+			Message:       "AI 服务未启用",
+		}
+	}
+	return AIServiceStatus{
+		Running:       true,
+		Enabled:       true,
+		ProviderCount: providerCount,
+	}
+}
+
+// Restart stops the current AI service and creates a new one from DB
+func (m *AIManager) Restart() error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if m.service != nil {
+		m.service.Stop()
+		m.service = nil
+	}
+
+	providers, err := m.store.ListLLMProviders(true)
+	if err != nil {
+		return fmt.Errorf("加载 LLM 提供商列表失败: %w", err)
+	}
+	if len(providers) == 0 {
+		return fmt.Errorf("没有配置任何 LLM 提供商，请先添加配置")
+	}
+
+	providerConfigs := make([]llm.ProviderConfig, 0, len(providers))
+	for _, p := range providers {
+		providerConfigs = append(providerConfigs, llm.ProviderConfig{
+			Name:                p.Name,
+			Active:              p.Active,
+			APIKey:              p.APIKey,
+			BaseURL:             p.BaseURL,
+			Model:               p.Model,
+			Timeout:             p.Timeout,
+			ContextWindowTokens: p.ContextWindowTokens,
+		})
+	}
+	m.cfg.LLMProviders = providerConfigs
+
+	svc, err := NewService(m.cfg, m.db, m.botSender)
+	if err != nil {
+		return fmt.Errorf("创建 AI 服务失败: %w", err)
+	}
+	if err := svc.Start(m.ctx); err != nil {
+		return fmt.Errorf("启动 AI 服务失败: %w", err)
+	}
+
+	m.service = svc
+	return nil
+}
+
+// ReloadLLMProvider delegates to the current service
+func (m *AIManager) ReloadLLMProvider(config interface{}) error {
+	m.mu.Lock()
+	svc := m.service
+	m.mu.Unlock()
+
+	if svc == nil {
+		return nil
+	}
+	return svc.ReloadLLMProvider(config)
+}
+
+// AddLLMProvider delegates to the current service
+func (m *AIManager) AddLLMProvider(config interface{}) error {
+	m.mu.Lock()
+	svc := m.service
+	m.mu.Unlock()
+
+	if svc == nil {
+		return nil
+	}
+	return svc.AddLLMProvider(config)
+}
+
+// RemoveLLMProvider delegates to the current service
+func (m *AIManager) RemoveLLMProvider(name string) error {
+	m.mu.Lock()
+	svc := m.service
+	m.mu.Unlock()
+
+	if svc == nil {
+		return nil
+	}
+	return svc.RemoveLLMProvider(name)
+}
+
+// AIServiceStatus returns the AI service status for the web interface
+func (m *AIManager) AIServiceStatus() AIServiceStatus {
+	return m.Status()
+}
+
+// RestartAIService restarts the AI service for the web interface
+func (m *AIManager) RestartAIService() error {
+	return m.Restart()
+}

internal/ai/service.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 
+	_ "meshtastic_mqtt_server/internal/agents/active"
 	_ "meshtastic_mqtt_server/internal/agents/calculator"
 	_ "meshtastic_mqtt_server/internal/agents/sign"
 	_ "meshtastic_mqtt_server/internal/agents/time"
@@ -258,3 +259,78 @@ func (s *Service) Stop() {
 func (s *Service) Enabled() bool {
 	return s.enabled
 }
+
+// ReloadLLMProvider reloads a specific LLM provider configuration
+func (s *Service) ReloadLLMProvider(config interface{}) error {
+	if s == nil {
+		return nil
+	}
+	if !s.enabled || s.LLMState == nil {
+		return nil
+	}
+	providerConfig, err := convertToProviderConfig(config)
+	if err != nil {
+		return err
+	}
+	return s.LLMState.UpdateProvider(providerConfig)
+}
+
+// AddLLMProvider adds a new LLM provider
+func (s *Service) AddLLMProvider(config interface{}) error {
+	if s == nil {
+		return nil
+	}
+	if !s.enabled || s.LLMState == nil {
+		return nil
+	}
+	providerConfig, err := convertToProviderConfig(config)
+	if err != nil {
+		return err
+	}
+	return s.LLMState.AddProvider(providerConfig)
+}
+
+// RemoveLLMProvider removes an LLM provider
+func (s *Service) RemoveLLMProvider(name string) error {
+	if s == nil {
+		return nil
+	}
+	if !s.enabled || s.LLMState == nil {
+		return nil
+	}
+	return s.LLMState.RemoveProvider(name)
+}
+
+// convertToProviderConfig converts a map to llm.ProviderConfig
+func convertToProviderConfig(config interface{}) (llm.ProviderConfig, error) {
+	m, ok := config.(map[string]interface{})
+	if !ok {
+		return llm.ProviderConfig{}, fmt.Errorf("invalid config type: expected map[string]interface{}")
+	}
+
+	pc := llm.ProviderConfig{}
+
+	if v, ok := m["Name"].(string); ok {
+		pc.Name = v
+	}
+	if v, ok := m["Active"].(bool); ok {
+		pc.Active = v
+	}
+	if v, ok := m["APIKey"].(string); ok {
+		pc.APIKey = v
+	}
+	if v, ok := m["BaseURL"].(string); ok {
+		pc.BaseURL = v
+	}
+	if v, ok := m["Model"].(string); ok {
+		pc.Model = v
+	}
+	if v, ok := m["Timeout"].(int); ok {
+		pc.Timeout = v
+	}
+	if v, ok := m["ContextWindowTokens"].(int); ok {
+		pc.ContextWindowTokens = v
+	}
+
+	return pc, nil
+}

internal/autoreply/service.go

@@ -509,6 +509,16 @@ func cleanReplyText(text string) string {
 			sb.WriteRune(r)
 		case r == 0x3002 || r == 0xFF1F || r == 0xFF01 || r == 0xFF0C || r == 0xFF1A: // Fullwidth punctuation
 			sb.WriteRune(r)
+		case r >= 0x1F600 && r <= 0x1F64F: // Emoticons
+			sb.WriteRune(r)
+		case r >= 0x1F300 && r <= 0x1F5FF: // Misc Symbols and Pictographs
+			sb.WriteRune(r)
+		case r >= 0x1F680 && r <= 0x1F6FF: // Transport and Map Symbols
+			sb.WriteRune(r)
+		case r >= 0x1F900 && r <= 0x1F9FF: // Supplemental Symbols and Pictographs
+			sb.WriteRune(r)
+		case r >= 0x2600 && r <= 0x27BF: // Misc Symbols + Dingbats
+			sb.WriteRune(r)
 		default:
 			continue // Skip all other characters
 		}

internal/llm/state.go

@@ -135,3 +135,137 @@ func (s *State) ListProfiles() []ProviderConfig {
 	}
 	return profiles
 }
+
+// UpdateProvider updates an existing provider's configuration
+func (s *State) UpdateProvider(config ProviderConfig) error {
+	name := strings.TrimSpace(config.Name)
+	if name == "" {
+		return errors.New("llm provider name cannot be empty")
+	}
+	if strings.TrimSpace(config.APIKey) == "" {
+		return fmt.Errorf("llm provider %s api_key is required", name)
+	}
+	if strings.TrimSpace(config.Model) == "" {
+		return fmt.Errorf("llm provider %s model is required", name)
+	}
+	if strings.TrimSpace(config.BaseURL) == "" {
+		return fmt.Errorf("llm provider %s base_url is required", name)
+	}
+	if config.Timeout <= 0 {
+		config.Timeout = 120
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if _, ok := s.profiles[name]; !ok {
+		return fmt.Errorf("llm provider not found: %s", name)
+	}
+
+	// Create new client with updated config
+	s.profiles[name] = &Profile{
+		Config: config,
+		Client: ark.NewClientWithApiKey(
+			config.APIKey,
+			ark.WithBaseUrl(config.BaseURL),
+			ark.WithTimeout(time.Duration(config.Timeout)*time.Second),
+		),
+	}
+
+	// Update active status if needed
+	if config.Active && s.activeName != name {
+		s.activeName = name
+	} else if !config.Active && s.activeName == name {
+		// If we're deactivating the current active provider, switch to the first available
+		for _, otherName := range s.order {
+			if otherName != name {
+				s.activeName = otherName
+				break
+			}
+		}
+	}
+
+	return nil
+}
+
+// AddProvider adds a new provider to the state
+func (s *State) AddProvider(config ProviderConfig) error {
+	name := strings.TrimSpace(config.Name)
+	if name == "" {
+		return errors.New("llm provider name cannot be empty")
+	}
+	if strings.TrimSpace(config.APIKey) == "" {
+		return fmt.Errorf("llm provider %s api_key is required", name)
+	}
+	if strings.TrimSpace(config.Model) == "" {
+		return fmt.Errorf("llm provider %s model is required", name)
+	}
+	if strings.TrimSpace(config.BaseURL) == "" {
+		return fmt.Errorf("llm provider %s base_url is required", name)
+	}
+	if config.Timeout <= 0 {
+		config.Timeout = 120
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if _, ok := s.profiles[name]; ok {
+		return fmt.Errorf("llm provider already exists: %s", name)
+	}
+
+	s.profiles[name] = &Profile{
+		Config: config,
+		Client: ark.NewClientWithApiKey(
+			config.APIKey,
+			ark.WithBaseUrl(config.BaseURL),
+			ark.WithTimeout(time.Duration(config.Timeout)*time.Second),
+		),
+	}
+	s.order = append(s.order, name)
+
+	// Set as active if it's the first one or explicitly marked active
+	if len(s.profiles) == 1 || config.Active {
+		s.activeName = name
+	}
+
+	return nil
+}
+
+// RemoveProvider removes a provider from the state
+func (s *State) RemoveProvider(name string) error {
+	name = strings.TrimSpace(name)
+	if name == "" {
+		return errors.New("llm provider name cannot be empty")
+	}
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if _, ok := s.profiles[name]; !ok {
+		return fmt.Errorf("llm provider not found: %s", name)
+	}
+
+	// Don't allow removing the last provider
+	if len(s.profiles) == 1 {
+		return errors.New("cannot remove the last llm provider")
+	}
+
+	delete(s.profiles, name)
+
+	// Remove from order
+	newOrder := make([]string, 0, len(s.order)-1)
+	for _, n := range s.order {
+		if n != name {
+			newOrder = append(newOrder, n)
+		}
+	}
+	s.order = newOrder
+
+	// If we removed the active provider, switch to the first available
+	if s.activeName == name {
+		s.activeName = s.order[0]
+	}
+
+	return nil
+}

internal/llm/state_test.go

@@ -0,0 +1,160 @@
+package llm
+
+import (
+	"testing"
+)
+
+func TestUpdateProvider(t *testing.T) {
+	// Create initial state with one provider
+	configs := []ProviderConfig{
+		{
+			Name:                "test-provider",
+			Active:              true,
+			APIKey:              "test-key",
+			BaseURL:             "https://test.example.com",
+			Model:               "test-model",
+			Timeout:             120,
+			ContextWindowTokens: 4096,
+		},
+	}
+
+	state, err := NewState(configs)
+	if err != nil {
+		t.Fatalf("failed to create state: %v", err)
+	}
+
+	// Get the initial profile
+	profile := state.ActiveProfile()
+	if profile.Config.APIKey != "test-key" {
+		t.Errorf("expected APIKey 'test-key', got '%s'", profile.Config.APIKey)
+	}
+
+	// Update the provider with new config
+	updatedConfig := ProviderConfig{
+		Name:                "test-provider",
+		Active:              true,
+		APIKey:              "new-key",
+		BaseURL:             "https://new.example.com",
+		Model:               "new-model",
+		Timeout:             60,
+		ContextWindowTokens: 8192,
+	}
+
+	err = state.UpdateProvider(updatedConfig)
+	if err != nil {
+		t.Fatalf("failed to update provider: %v", err)
+	}
+
+	// Verify the update
+	profile = state.ActiveProfile()
+	if profile.Config.APIKey != "new-key" {
+		t.Errorf("expected updated APIKey 'new-key', got '%s'", profile.Config.APIKey)
+	}
+	if profile.Config.BaseURL != "https://new.example.com" {
+		t.Errorf("expected updated BaseURL 'https://new.example.com', got '%s'", profile.Config.BaseURL)
+	}
+	if profile.Config.Model != "new-model" {
+		t.Errorf("expected updated Model 'new-model', got '%s'", profile.Config.Model)
+	}
+	if profile.Config.Timeout != 60 {
+		t.Errorf("expected updated Timeout 60, got %d", profile.Config.Timeout)
+	}
+}
+
+func TestAddProvider(t *testing.T) {
+	// Create initial state with one provider
+	configs := []ProviderConfig{
+		{
+			Name:                "provider1",
+			Active:              true,
+			APIKey:              "key1",
+			BaseURL:             "https://example1.com",
+			Model:               "model1",
+			Timeout:             120,
+			ContextWindowTokens: 4096,
+		},
+	}
+
+	state, err := NewState(configs)
+	if err != nil {
+		t.Fatalf("failed to create state: %v", err)
+	}
+
+	// Add a second provider
+	newConfig := ProviderConfig{
+		Name:                "provider2",
+		Active:              false,
+		APIKey:              "key2",
+		BaseURL:             "https://example2.com",
+		Model:               "model2",
+		Timeout:             60,
+		ContextWindowTokens: 8192,
+	}
+
+	err = state.AddProvider(newConfig)
+	if err != nil {
+		t.Fatalf("failed to add provider: %v", err)
+	}
+
+	// Verify the new provider exists
+	profile, err := state.GetProfile("provider2")
+	if err != nil {
+		t.Fatalf("failed to get provider2: %v", err)
+	}
+	if profile.Config.Name != "provider2" {
+		t.Errorf("expected name 'provider2', got '%s'", profile.Config.Name)
+	}
+
+	// Verify active provider is still provider1
+	activeProfile := state.ActiveProfile()
+	if activeProfile.Config.Name != "provider1" {
+		t.Errorf("expected active provider 'provider1', got '%s'", activeProfile.Config.Name)
+	}
+}
+
+func TestRemoveProvider(t *testing.T) {
+	// Create initial state with two providers
+	configs := []ProviderConfig{
+		{
+			Name:                "provider1",
+			Active:              true,
+			APIKey:              "key1",
+			BaseURL:             "https://example1.com",
+			Model:               "model1",
+			Timeout:             120,
+			ContextWindowTokens: 4096,
+		},
+		{
+			Name:                "provider2",
+			Active:              false,
+			APIKey:              "key2",
+			BaseURL:             "https://example2.com",
+			Model:               "model2",
+			Timeout:             60,
+			ContextWindowTokens: 8192,
+		},
+	}
+
+	state, err := NewState(configs)
+	if err != nil {
+		t.Fatalf("failed to create state: %v", err)
+	}
+
+	// Remove provider2
+	err = state.RemoveProvider("provider2")
+	if err != nil {
+		t.Fatalf("failed to remove provider: %v", err)
+	}
+
+	// Verify provider2 is gone
+	_, err = state.GetProfile("provider2")
+	if err == nil {
+		t.Error("expected error when getting removed provider, got nil")
+	}
+
+	// Try to remove the last provider (should fail)
+	err = state.RemoveProvider("provider1")
+	if err == nil {
+		t.Error("expected error when removing last provider, got nil")
+	}
+}

internal/llmadmin/admin_llm_routes.go

@@ -9,11 +9,21 @@ import (
 	"github.com/gin-gonic/gin"
 	"gorm.io/gorm"
 
+	aipkg "meshtastic_mqtt_server/internal/ai"
 	storepkg "meshtastic_mqtt_server/internal/store"
 	"meshtastic_mqtt_server/internal/webutil"
 )
 
-func RegisterRoutes(r *gin.RouterGroup, store *storepkg.Store) {
+// LLMProviderReloader is the interface for reloading LLM provider configuration
+type LLMProviderReloader interface {
+	ReloadLLMProvider(config interface{}) error
+	AddLLMProvider(config interface{}) error
+	RemoveLLMProvider(name string) error
+	AIServiceStatus() aipkg.AIServiceStatus
+	RestartAIService() error
+}
+
+func RegisterRoutes(r *gin.RouterGroup, store *storepkg.Store, aiService LLMProviderReloader) {
 	group := r.Group("/llm")
 	{
 		// LLM Message Queue
@@ -27,9 +37,9 @@ func RegisterRoutes(r *gin.RouterGroup, store *storepkg.Store) {
 		// LLM Providers
 		group.GET("/providers", handleListLLMProviders(store))
 		group.GET("/providers/:name", handleGetLLMProvider(store))
-		group.POST("/providers", handleCreateLLMProvider(store))
-		group.PUT("/providers/:name", handleUpdateLLMProvider(store))
-		group.DELETE("/providers/:name", handleDeleteLLMProvider(store))
+		group.POST("/providers", handleCreateLLMProvider(store, aiService))
+		group.PUT("/providers/:name", handleUpdateLLMProvider(store, aiService))
+		group.DELETE("/providers/:name", handleDeleteLLMProvider(store, aiService))
 
 		// LLM Tool Router
 		group.GET("/tool-router", handleGetLLMToolRouter(store))
@@ -42,6 +52,10 @@ func RegisterRoutes(r *gin.RouterGroup, store *storepkg.Store) {
 		// LLM Primary Config - 主 AI 回复配置
 		group.GET("/primary-config", handleGetLLMPrimaryConfig(store))
 		group.PUT("/primary-config", handleUpdateLLMPrimaryConfig(store))
+
+		// AI Service Status
+		group.GET("/status", handleGetAIServiceStatus(aiService))
+		group.POST("/restart", handleRestartAIService(aiService))
 	}
 }
 
@@ -254,7 +268,7 @@ func handleGetLLMProvider(store *storepkg.Store) gin.HandlerFunc {
 	}
 }
 
-func handleCreateLLMProvider(store *storepkg.Store) gin.HandlerFunc {
+func handleCreateLLMProvider(store *storepkg.Store, aiService LLMProviderReloader) gin.HandlerFunc {
 	return func(c *gin.Context) {
 		var req struct {
 			Name               string `json:"name"`
@@ -290,11 +304,38 @@ func handleCreateLLMProvider(store *storepkg.Store) gin.HandlerFunc {
 			return
 		}
 
+		// Reload AI service with new provider
+		if aiService == nil {
+			c.JSON(http.StatusOK, gin.H{
+				"status":  "ok",
+				"item":    llmProviderDTO(*record),
+				"warning": "AI 服务未运行，配置已保存但需重启服务后生效",
+			})
+			return
+		}
+		providerConfig := map[string]interface{}{
+			"Name":                record.Name,
+			"Active":              record.Active,
+			"APIKey":              record.APIKey,
+			"BaseURL":             record.BaseURL,
+			"Model":               record.Model,
+			"Timeout":             record.Timeout,
+			"ContextWindowTokens": record.ContextWindowTokens,
+		}
+		if err := aiService.AddLLMProvider(providerConfig); err != nil {
+			c.JSON(http.StatusOK, gin.H{
+				"status":  "ok",
+				"item":    llmProviderDTO(*record),
+				"warning": "provider created but failed to reload AI service: " + err.Error(),
+			})
+			return
+		}
+
 		c.JSON(http.StatusOK, gin.H{"status": "ok", "item": llmProviderDTO(*record)})
 	}
 }
 
-func handleUpdateLLMProvider(store *storepkg.Store) gin.HandlerFunc {
+func handleUpdateLLMProvider(store *storepkg.Store, aiService LLMProviderReloader) gin.HandlerFunc {
 	return func(c *gin.Context) {
 		name := c.Param("name")
 		if name == "" {
@@ -355,11 +396,38 @@ func handleUpdateLLMProvider(store *storepkg.Store) gin.HandlerFunc {
 			return
 		}
 
+		// Reload AI service with updated provider
+		if aiService == nil {
+			c.JSON(http.StatusOK, gin.H{
+				"status":  "ok",
+				"item":    llmProviderDTO(*record),
+				"warning": "AI 服务未运行，配置已保存但需重启服务后生效",
+			})
+			return
+		}
+		providerConfig := map[string]interface{}{
+			"Name":                record.Name,
+			"Active":              record.Active,
+			"APIKey":              record.APIKey,
+			"BaseURL":             record.BaseURL,
+			"Model":               record.Model,
+			"Timeout":             record.Timeout,
+			"ContextWindowTokens": record.ContextWindowTokens,
+		}
+		if err := aiService.ReloadLLMProvider(providerConfig); err != nil {
+			c.JSON(http.StatusOK, gin.H{
+				"status":  "ok",
+				"item":    llmProviderDTO(*record),
+				"warning": "provider updated but failed to reload AI service: " + err.Error(),
+			})
+			return
+		}
+
 		c.JSON(http.StatusOK, gin.H{"status": "ok", "item": llmProviderDTO(*record)})
 	}
 }
 
-func handleDeleteLLMProvider(store *storepkg.Store) gin.HandlerFunc {
+func handleDeleteLLMProvider(store *storepkg.Store, aiService LLMProviderReloader) gin.HandlerFunc {
 	return func(c *gin.Context) {
 		name := c.Param("name")
 		if name == "" {
@@ -372,6 +440,22 @@ func handleDeleteLLMProvider(store *storepkg.Store) gin.HandlerFunc {
 			return
 		}
 
+		// Remove provider from AI service
+		if aiService == nil {
+			c.JSON(http.StatusOK, gin.H{
+				"status":  "ok",
+				"warning": "AI 服务未运行，配置已删除但需重启服务后生效",
+			})
+			return
+		}
+		if err := aiService.RemoveLLMProvider(name); err != nil {
+			c.JSON(http.StatusOK, gin.H{
+				"status":  "ok",
+				"warning": "provider deleted but failed to reload AI service: " + err.Error(),
+			})
+			return
+		}
+
 		c.JSON(http.StatusOK, gin.H{"status": "ok"})
 	}
 }
@@ -751,3 +835,49 @@ func llmPrimaryConfigDTO(row storepkg.LLMPrimaryConfigRecord) map[string]any {
 		"updated_at":     row.UpdatedAt,
 	}
 }
+
+// ============================================
+// AI Service Status & Restart Handlers
+// ============================================
+
+func handleGetAIServiceStatus(aiService LLMProviderReloader) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		if aiService == nil {
+			c.JSON(http.StatusOK, gin.H{
+				"running":        false,
+				"enabled":        false,
+				"provider_count": 0,
+				"message":        "AI 服务未初始化",
+			})
+			return
+		}
+		status := aiService.AIServiceStatus()
+		c.JSON(http.StatusOK, gin.H{
+			"running":        status.Running,
+			"enabled":        status.Enabled,
+			"provider_count": status.ProviderCount,
+			"message":        status.Message,
+		})
+	}
+}
+
+func handleRestartAIService(aiService LLMProviderReloader) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		if aiService == nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "AI 服务未初始化，无法重启"})
+			return
+		}
+		if err := aiService.RestartAIService(); err != nil {
+			c.JSON(http.StatusInternalServerError, gin.H{"error": "重启 AI 服务失败: " + err.Error()})
+			return
+		}
+		status := aiService.AIServiceStatus()
+		c.JSON(http.StatusOK, gin.H{
+			"status":         "ok",
+			"running":        status.Running,
+			"enabled":        status.Enabled,
+			"provider_count": status.ProviderCount,
+			"message":        "AI 服务已重启",
+		})
+	}
+}

internal/mqtpp/builder.go

@@ -20,6 +20,7 @@ type PacketBuildOptions struct {
 	PSK         []byte
 	Encrypt     bool
 	ViaMQTT     bool
+	HopLimit    uint32
 }
 
 type TextMessageBuildOptions struct {
@@ -252,6 +253,14 @@ func buildMeshPacket(opts PacketBuildOptions, data []byte) ([]byte, error) {
 		out = protowire.AppendTag(out, 14, protowire.VarintType)
 		out = protowire.AppendVarint(out, 1)
 	}
+	hop := opts.HopLimit
+	if hop == 0 {
+		hop = 7
+	}
+	out = protowire.AppendTag(out, 9, protowire.VarintType)
+	out = protowire.AppendVarint(out, uint64(hop))
+	out = protowire.AppendTag(out, 15, protowire.VarintType)
+	out = protowire.AppendVarint(out, uint64(hop))
 	return out, nil
 }
 

internal/mqttforward/dedup_queue.go

@@ -0,0 +1,78 @@
+package mqttforward
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"sync"
+	"time"
+)
+
+const dedupTTL = 15 * time.Second
+
+type DedupQueue struct {
+	mu      sync.Mutex
+	entries map[string]time.Time
+	stopCh  chan struct{}
+}
+
+func NewDedupQueue() *DedupQueue {
+	return &DedupQueue{
+		entries: make(map[string]time.Time),
+		stopCh:  make(chan struct{}),
+	}
+}
+
+func (dq *DedupQueue) TryForward(topic string, payload []byte) bool {
+	hash := dedupHash(topic, payload)
+	now := time.Now()
+	dq.mu.Lock()
+	defer dq.mu.Unlock()
+	if expiry, ok := dq.entries[hash]; ok && now.Before(expiry) {
+		return false
+	}
+	dq.entries[hash] = now.Add(dedupTTL)
+	return true
+}
+
+func (dq *DedupQueue) Start() {
+	go func() {
+		ticker := time.NewTicker(dedupTTL)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ticker.C:
+				dq.cleanup()
+			case <-dq.stopCh:
+				return
+			}
+		}
+	}()
+}
+
+func (dq *DedupQueue) Stop() {
+	close(dq.stopCh)
+}
+
+func (dq *DedupQueue) Len() int {
+	dq.mu.Lock()
+	defer dq.mu.Unlock()
+	return len(dq.entries)
+}
+
+func (dq *DedupQueue) cleanup() {
+	now := time.Now()
+	dq.mu.Lock()
+	defer dq.mu.Unlock()
+	for hash, expiry := range dq.entries {
+		if now.After(expiry) {
+			delete(dq.entries, hash)
+		}
+	}
+}
+
+func dedupHash(topic string, payload []byte) string {
+	h := sha256.New()
+	h.Write([]byte(topic))
+	h.Write(payload)
+	return hex.EncodeToString(h.Sum(nil))
+}

internal/store/active_store.go

@@ -0,0 +1,24 @@
+package store
+
+import (
+	"time"
+)
+
+// CountActiveNodes 统计指定时间后有更新记录的节点数量
+func (s *Store) CountActiveNodes(since time.Time) (int64, error) {
+	var count int64
+	err := s.db.Model(&NodeInfoRecord{}).
+		Where("updated_at >= ?", since).
+		Count(&count).Error
+	return count, err
+}
+
+// CountActiveUsers 统计指定时间后发送过消息的唯一用户数（按 from_id 去重）
+func (s *Store) CountActiveUsers(since time.Time) (int64, error) {
+	var count int64
+	err := s.db.Model(&TextMessageRecord{}).
+		Where("created_at >= ?", since).
+		Distinct("from_id").
+		Count(&count).Error
+	return count, err
+}

internal/store/bot_direct_message_store.go

@@ -290,27 +290,24 @@ func insertInboundBotDirectMessage(s *Store, record map[string]any, clientInfo M
 	}
 	_ = clientInfo // mqtt 元数据已经记录在 content_json 里，这里保留参数以保持队列签名一致
 
-	// 同时将消息添加到 LLM 队列（忽略机器人自己发送的消息）
-	if peerNodeID != bot.NodeID {
-		longName := NullableString(record["long_name"])
-		shortName := NullableString(record["short_name"])
-		channelID := NullableString(record["channel_id"])
-		_, err = s.EnqueueLLMMessage(LLMMessageQueueInput{
-			BotID:       bot.ID,
-			BotNodeID:   bot.NodeID,
-			BotNodeNum:  bot.NodeNum,
-			FromNodeID:  peerNodeID,
-			FromNodeNum: int64(peerNum),
-			LongName:    longName,
-			ShortName:   shortName,
-			Text:        text,
-			PacketID:    int64(packetID),
-			ChannelID:   channelID,
-			Topic:       topic,
-			MessageType: "direct",
-			ContentJSON: contentPtr,
-		})
-	}
+	longName := NullableString(record["long_name"])
+	shortName := NullableString(record["short_name"])
+	channelID := NullableString(record["channel_id"])
+	_, err = s.EnqueueLLMMessage(LLMMessageQueueInput{
+		BotID:       bot.ID,
+		BotNodeID:   bot.NodeID,
+		BotNodeNum:  bot.NodeNum,
+		FromNodeID:  peerNodeID,
+		FromNodeNum: int64(peerNum),
+		LongName:    longName,
+		ShortName:   shortName,
+		Text:        text,
+		PacketID:    int64(packetID),
+		ChannelID:   channelID,
+		Topic:       topic,
+		MessageType: "direct",
+		ContentJSON: contentPtr,
+	})
 	if err != nil {
 		printJSON(map[string]any{
 			"event":   "llm_queue_enqueue_failed",

internal/store/bot_store.go

@@ -66,6 +66,12 @@ func (s *Store) CountBotNodes(opts ListOptions) (int64, error) {
 	return total, s.db.Model(&BotNodeRecord{}).Count(&total).Error
 }
 
+func (s *Store) IsBotNodeID(nodeID string) bool {
+	var count int64
+	s.db.Model(&BotNodeRecord{}).Where("node_id = ?", nodeID).Count(&count)
+	return count > 0
+}
+
 func (s *Store) GetBotNode(id uint64) (*BotNodeRecord, error) {
 	var row BotNodeRecord
 	if err := s.db.Where("id = ?", id).Take(&row).Error; err != nil {

internal/store/llm_store.go

@@ -306,8 +306,7 @@ func (s *Store) EnqueueLLMMessage(input LLMMessageQueueInput) (*LLMMessageQueueR
 		return nil, nil // 机器人的 LLM 队列未启用，静默返回
 	}
 
-	// 忽略机器人自己发送的消息，避免自循环
-	if input.FromNodeID == bot.NodeID {
+	if s.IsBotNodeID(input.FromNodeID) {
 		return nil, nil
 	}
 
@@ -526,6 +525,10 @@ func enqueueChannelMessageToLLM(s *Store, record map[string]any) error {
 		shortName = &sn
 	}
 
+	if s.IsBotNodeID(fromNodeID) {
+		return nil
+	}
+
 	var channelID *string
 	if cid, ok := record["channel_id"].(string); ok && cid != "" {
 		channelID = &cid
@@ -546,11 +549,7 @@ func enqueueChannelMessageToLLM(s *Store, record map[string]any) error {
 		return fmt.Errorf("query bots for channel message enqueue: %w", err)
 	}
 
-	// 为每个符合条件的机器人创建一条队列记录（忽略机器人自己发送的消息）
 	for _, bot := range bots {
-		if fromNodeID == bot.NodeID {
-			continue
-		}
 		_, err = s.EnqueueLLMMessage(LLMMessageQueueInput{
 			BotID:       bot.ID,
 			BotNodeID:   bot.NodeID,

internal/toolrouter/loop.go

@@ -88,6 +88,7 @@ func RunAgentToolLoop(ctx context.Context, state *State, profile *llm.Profile, s
 	if routerPrompt == "" {
 		routerPrompt = primaryPrompt
 	}
+	routerPrompt = routerPrompt + "\n当前日期：" + time.Now().Format("2006-01-02")
 	if primaryPrompt != "" {
 		primarySystemMessage := &model.ChatCompletionMessage{
 			Role: "system",

internal/toolrouter/state.go

@@ -48,7 +48,7 @@ func NewState(cfg *Config, ai *llm.State, options ...Option) (*State, error) {
 			Enabled:      true,
 			Timeout:      30,
 			MaxTokens:    512,
-			SystemPrompt: "你可以按需直接调用可用工具来回答用户问题。\n每个工具的 description 描述了它的适用场景和调用条件。\n工具结果优先于模型内置知识；工具失败时必须如实说明，不要编造结果。\n只调用确实必要的工具。",
+			SystemPrompt: "你可以按需直接调用可用工具来回答用户问题。\n每个工具的 description 描述了它的适用场景和调用条件。\n工具结果优先于模型内置知识；工具失败时必须如实说明，不要编造结果。\n只调用确实必要的工具。\n\n重要规则：\n- 当用户问题包含\"今天\"、\"昨天\"、\"最近\"、\"本周\"、\"本月\"等相对时间词时，你必须先调用 time 工具获取当前准确日期，然后再用该日期调用其他工具。绝不可以使用模型内置知识猜测日期。\n- 你不知道\"今天\"是哪一天，必须通过 time 工具查询。",
 		}
 	}
 	if ai == nil {

internal/web/map_tile_proxy_routes_test.go

@@ -38,7 +38,7 @@ func TestMapTileProxyFetchesAndCaches(t *testing.T) {
 	}
 
 	cacheDir := t.TempDir()
-	router := NewRouter(configpkg.WebConfig{StaticDir: t.TempDir(), MapTileCacheDir: cacheDir}, false, st, nil, nil, nil, nil, nil, nil)
+	router := NewRouter(configpkg.WebConfig{StaticDir: t.TempDir(), MapTileCacheDir: cacheDir}, false, st, nil, nil, nil, nil, nil, nil, nil)
 
 	url := "/api/map/" + row.URLTemplateHash + "?x=1&y=2&z=3"
 	for i := 0; i < 2; i++ {
@@ -75,7 +75,7 @@ func TestMapTileProxyRejectsInvalidCoordinates(t *testing.T) {
 		t.Fatalf("CreateMapTileSource() error = %v", err)
 	}
 
-	router := NewRouter(configpkg.WebConfig{StaticDir: t.TempDir(), MapTileCacheDir: t.TempDir()}, false, st, nil, nil, nil, nil, nil, nil)
+	router := NewRouter(configpkg.WebConfig{StaticDir: t.TempDir(), MapTileCacheDir: t.TempDir()}, false, st, nil, nil, nil, nil, nil, nil, nil)
 
 	cases := []string{
 		"/api/map/" + row.URLTemplateHash + "?y=0&z=0",
@@ -106,7 +106,7 @@ func TestMapTileProxyUnknownAndDisabledSource(t *testing.T) {
 		t.Fatalf("CreateMapTileSource(proxy disabled) error = %v", err)
 	}
 
-	router := NewRouter(configpkg.WebConfig{StaticDir: t.TempDir(), MapTileCacheDir: t.TempDir()}, false, st, nil, nil, nil, nil, nil, nil)
+	router := NewRouter(configpkg.WebConfig{StaticDir: t.TempDir(), MapTileCacheDir: t.TempDir()}, false, st, nil, nil, nil, nil, nil, nil, nil)
 
 	cases := []string{
 		"/api/map/not-a-hash?x=0&y=0&z=0",
@@ -147,7 +147,7 @@ func TestMapTileProxyUpstreamStatus(t *testing.T) {
 		t.Fatalf("CreateMapTileSource(500) error = %v", err)
 	}
 
-	router := NewRouter(configpkg.WebConfig{StaticDir: t.TempDir(), MapTileCacheDir: t.TempDir()}, false, st, nil, nil, nil, nil, nil, nil)
+	router := NewRouter(configpkg.WebConfig{StaticDir: t.TempDir(), MapTileCacheDir: t.TempDir()}, false, st, nil, nil, nil, nil, nil, nil, nil)
 
 	cases := []struct {
 		url  string

internal/web/mqtt_status.go

@@ -30,6 +30,7 @@ type MQTTRuntimeStatus struct {
 	Stats       *mqttforwardpkg.Stats
 	ClientStats *mqttforwardpkg.ClientStats
 	DBQueue     *storepkg.WriteQueue
+	DedupQueue  *mqttforwardpkg.DedupQueue
 }
 
 // AdminMQTTStatus 是 admin 路由 GET /admin/mqtt-status 返回的 JSON 视图。
@@ -50,6 +51,7 @@ type AdminMQTTStatus struct {
 	MessagesSent        int64             `json:"messages_sent"`
 	MessagesDropped     int64             `json:"messages_dropped"`
 	DBWriteQueueLength  int               `json:"db_write_queue_length"`
+	DedupQueueLength    int               `json:"dedup_queue_len"`
 	Retained            int64             `json:"retained"`
 	Inflight            int64             `json:"inflight"`
 	InflightDropped     int64             `json:"inflight_dropped"`
@@ -71,7 +73,7 @@ type AdminMQTTClient struct {
 // Status 实现 MQTTStatusProvider。
 func (m MQTTRuntimeStatus) Status() AdminMQTTStatus {
 	if m.Server == nil || m.Server.Info == nil {
-		return AdminMQTTStatus{Running: false, Address: m.Address, TLS: m.TLS, DBWriteQueueLength: m.DBQueue.Len()}
+		return AdminMQTTStatus{Running: false, Address: m.Address, TLS: m.TLS, DBWriteQueueLength: m.DBQueue.Len(), DedupQueueLength: m.dedupQueueLen()}
 	}
 	info := m.Server.Info.Clone()
 	status := AdminMQTTStatus{
@@ -91,6 +93,7 @@ func (m MQTTRuntimeStatus) Status() AdminMQTTStatus {
 		MessagesSent:        m.Stats.Forwarded(),
 		MessagesDropped:     m.Stats.Dropped(),
 		DBWriteQueueLength:  m.DBQueue.Len(),
+		DedupQueueLength:    m.dedupQueueLen(),
 		Retained:            info.Retained,
 		Inflight:            info.Inflight,
 		InflightDropped:     info.InflightDropped,
@@ -136,6 +139,13 @@ func mqttClientInfo(c *mqtt.Client) mqttClientInfoView {
 	}
 }
 
+func (m MQTTRuntimeStatus) dedupQueueLen() int {
+	if m.DedupQueue == nil {
+		return 0
+	}
+	return m.DedupQueue.Len()
+}
+
 // DisconnectClient 实现 MQTTStatusProvider：发送 Disconnect 报文并关闭连接。
 // 使用 ErrAdministrativeAction 作为断开理由，便于日志区分。
 func (m MQTTRuntimeStatus) DisconnectClient(clientID string) bool {

internal/web/web.go

@@ -13,6 +13,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"gorm.io/gorm"
 
+	aipkg "meshtastic_mqtt_server/internal/ai"
 	"meshtastic_mqtt_server/internal/auth"
 	blockingpkg "meshtastic_mqtt_server/internal/blocking"
 	botpkg "meshtastic_mqtt_server/internal/bot"
@@ -27,10 +28,19 @@ import (
 	"meshtastic_mqtt_server/internal/webutil"
 )
 
-func NewHTTPServer(cfg configpkg.WebConfig, consoleLog bool, store *storepkg.Store, sessions *auth.Manager, mqttStatus MQTTStatusProvider, blocking *blockingpkg.Cache, forwarder mqttforwardpkg.Reloader, settings *rspkg.Cache, botSender botpkg.TextSender) *http.Server {
+// LLMProviderReloader is the interface for reloading LLM provider configuration
+type LLMProviderReloader interface {
+	ReloadLLMProvider(config interface{}) error
+	AddLLMProvider(config interface{}) error
+	RemoveLLMProvider(name string) error
+	AIServiceStatus() aipkg.AIServiceStatus
+	RestartAIService() error
+}
+
+func NewHTTPServer(cfg configpkg.WebConfig, consoleLog bool, store *storepkg.Store, sessions *auth.Manager, mqttStatus MQTTStatusProvider, blocking *blockingpkg.Cache, forwarder mqttforwardpkg.Reloader, settings *rspkg.Cache, botSender botpkg.TextSender, aiService LLMProviderReloader) *http.Server {
 	return &http.Server{
 		Addr:    net.JoinHostPort(cfg.Host, strconv.Itoa(cfg.Port)),
-		Handler: NewRouter(cfg, consoleLog, store, sessions, mqttStatus, blocking, forwarder, settings, botSender),
+		Handler: NewRouter(cfg, consoleLog, store, sessions, mqttStatus, blocking, forwarder, settings, botSender, aiService),
 	}
 }
 
@@ -60,7 +70,7 @@ func ServeUnixSocket(server *http.Server, socketPath string) error {
 	return server.Serve(listener)
 }
 
-func NewRouter(cfg configpkg.WebConfig, consoleLog bool, store *storepkg.Store, sessions *auth.Manager, mqttStatus MQTTStatusProvider, blocking *blockingpkg.Cache, forwarder mqttforwardpkg.Reloader, settings *rspkg.Cache, botSender botpkg.TextSender) *gin.Engine {
+func NewRouter(cfg configpkg.WebConfig, consoleLog bool, store *storepkg.Store, sessions *auth.Manager, mqttStatus MQTTStatusProvider, blocking *blockingpkg.Cache, forwarder mqttforwardpkg.Reloader, settings *rspkg.Cache, botSender botpkg.TextSender, aiService LLMProviderReloader) *gin.Engine {
 	r := gin.New()
 	if consoleLog {
 		r.Use(gin.Logger(), gin.Recovery())
@@ -69,7 +79,7 @@ func NewRouter(cfg configpkg.WebConfig, consoleLog bool, store *storepkg.Store,
 	}
 	api := r.Group("/api")
 	registerAPIRoutes(api, store, cfg.MapTileCacheDir)
-	registerAdminRoutes(api.Group("/admin"), store, sessions, mqttStatus, blocking, forwarder, settings, botSender)
+	registerAdminRoutes(api.Group("/admin"), store, sessions, mqttStatus, blocking, forwarder, settings, botSender, aiService)
 	registerStaticRoutes(r, cfg.StaticDir)
 	return r
 }
@@ -163,7 +173,7 @@ func registerAPIRoutes(r gin.IRouter, store *storepkg.Store, mapTileCacheDir str
 	})
 }
 
-func registerAdminRoutes(r gin.IRouter, store *storepkg.Store, sessions *auth.Manager, mqttStatus MQTTStatusProvider, blocking *blockingpkg.Cache, forwarder mqttforwardpkg.Reloader, settings *rspkg.Cache, botSender botpkg.TextSender) {
+func registerAdminRoutes(r gin.IRouter, store *storepkg.Store, sessions *auth.Manager, mqttStatus MQTTStatusProvider, blocking *blockingpkg.Cache, forwarder mqttforwardpkg.Reloader, settings *rspkg.Cache, botSender botpkg.TextSender, aiService LLMProviderReloader) {
 	type loginRequest struct {
 		Username string `json:"username"`
 		Password string `json:"password"`
@@ -230,7 +240,7 @@ func registerAdminRoutes(r gin.IRouter, store *storepkg.Store, sessions *auth.Ma
 	mappkg.RegisterAdminRoutes(protected, store)
 	helppkg.RegisterAdminRoutes(protected, store)
 	botpkg.RegisterRoutes(protected, store, botSender)
-	llmadminpkg.RegisterRoutes(protected, store)
+	llmadminpkg.RegisterRoutes(protected, store, aiService)
 	protected.GET("/me", func(c *gin.Context) {
 		claims := c.MustGet("admin_claims").(*auth.SessionClaims)
 		c.JSON(http.StatusOK, gin.H{"user": auth.AdminUserDTO{Username: claims.Username, Role: claims.Role}})

main.go

@@ -57,8 +57,9 @@ type meshtasticFilterHook struct {
 	settings        *rspkg.Cache
 	pkiResolver     func(toNodeNum, fromNodeNum uint32) ([]byte, []byte, bool)
 	autoAcker       func(record map[string]any)
-	consoleLog      bool // 控制台是否打印 MQTT 连接/订阅事件
+	consoleLog       bool // 控制台是否打印 MQTT 连接/订阅事件
 	packetConsoleLog bool // 控制台是否打印 Meshtastic 数据包
+	dedupQueue       *mqttforwardpkg.DedupQueue
 }
 
 // ID 返回用于识别 Meshtastic payload 过滤器的 hook 名称。
@@ -203,6 +204,12 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa
 		PKIKeyResolver:           h.pkiResolver,
 	})
 	if !valid {
+		// 记录拒绝原因，帮助诊断 QoS0 重发问题
+		if h.consoleLog {
+			info := mqttClientInfoFromClient(cl)
+			fmt.Fprintf(os.Stderr, "[mqtt] PUBLISH rejected: client_id=%s topic=%s qos=%d payload_len=%d error=%v\n",
+				info.ClientID, pk.TopicName, pk.FixedHeader.Qos, len(pk.Payload), record["error"])
+		}
 		h.rejectPublish(cl, pk, record)
 		return pk, packets.ErrRejectPacket
 	}
@@ -210,9 +217,19 @@ func (h *meshtasticFilterHook) OnPublish(cl *mqtt.Client, pk packets.Packet) (pa
 		for key, value := range violation {
 			record[key] = value
 		}
+		// 记录屏蔽原因
+		if h.consoleLog {
+			info := mqttClientInfoFromClient(cl)
+			fmt.Fprintf(os.Stderr, "[mqtt] PUBLISH blocked: client_id=%s topic=%s type=%v reason=%v\n",
+				info.ClientID, pk.TopicName, violation["blocking_type"], violation["error"])
+		}
 		h.rejectPublish(cl, pk, record)
 		return pk, packets.ErrRejectPacket
 	}
+	if h.dedupQueue != nil && !h.dedupQueue.TryForward(pk.TopicName, pk.Payload) {
+		h.stats.IncDropped()
+		return pk, packets.ErrRejectPacket
+	}
 	h.stats.IncForwarded()
 
 	h.dbQueue.EnqueueRecord(record, mqttClientInfoFromClient(cl))
@@ -381,16 +398,48 @@ func run(cfg *configpkg.Config) error {
 	defer forwardManager.StopAll()
 
 	// Initialize AI Service
-	var aiService *ai.Service
+	// Create bot sender adapter - 支持频道消息和私聊消息两种发送方式
+	botSenderAdapter := autoreply.NewBotServiceAdapter(
+		// SendDirectText: 发送私聊消息
+		func(ctx context.Context, botID uint64, toNodeNum int64, text string) error {
+			_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
+				BotID:       botID,
+				MessageType: "direct",
+				ToNodeNum:   &toNodeNum,
+				Text:        text,
+			})
+			return err
+		},
+		// SendChannelText: 发送频道消息
+		func(ctx context.Context, botID uint64, channelID string, text string) error {
+			_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
+				BotID:       botID,
+				MessageType: "channel",
+				ChannelID:   channelID,
+				Text:        text,
+			})
+			return err
+		},
+	)
+
+	aiManager := ai.NewAIManager(ai.Config{
+		DataDir:          cfg.AI.DataDir,
+		Enabled:          cfg.AI.Enabled,
+		ConsoleLog:       cfg.ConsoleLog.LLM,
+		ToolConfigStore:  store,
+		ToolRouterStore:  store,
+		TopicRouterStore: store,
+		Store:            store,
+	}, store.DB(), botSenderAdapter, botCtx, store)
+
 	if cfg.AI.Enabled {
-		// Get LLM providers from database
-		llmProviders, err := store.ListLLMProviders(true)
+		aiManager.SetConfigEnabled(true)
+		providers, err := store.ListLLMProviders(true)
 		if err != nil {
 			fmt.Fprintf(os.Stderr, "Warning: failed to load LLM providers: %v\n", err)
-		} else if len(llmProviders) > 0 {
-			// Convert database records to provider configs
-			providerConfigs := make([]llm.ProviderConfig, 0, len(llmProviders))
-			for _, p := range llmProviders {
+		} else if len(providers) > 0 {
+			providerConfigs := make([]llm.ProviderConfig, 0, len(providers))
+			for _, p := range providers {
 				providerConfigs = append(providerConfigs, llm.ProviderConfig{
 					Name:                p.Name,
 					Active:              p.Active,
@@ -401,48 +450,11 @@ func run(cfg *configpkg.Config) error {
 					ContextWindowTokens: p.ContextWindowTokens,
 				})
 			}
-
-			// Create bot sender adapter - 支持频道消息和私聊消息两种发送方式
-			botSenderAdapter := autoreply.NewBotServiceAdapter(
-				// SendDirectText: 发送私聊消息
-				func(ctx context.Context, botID uint64, toNodeNum int64, text string) error {
-					_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
-						BotID:       botID,
-						MessageType: "direct",
-						ToNodeNum:   &toNodeNum,
-						Text:        text,
-					})
-					return err
-				},
-				// SendChannelText: 发送频道消息
-				func(ctx context.Context, botID uint64, channelID string, text string) error {
-					_, err := botSender.SendText(ctx, botpkg.SendTextRequest{
-						BotID:       botID,
-						MessageType: "channel",
-						ChannelID:   channelID,
-						Text:        text,
-					})
-					return err
-				},
-			)
-
-			aiService, err = ai.NewService(ai.Config{
-				LLMProviders:     providerConfigs,
-				DataDir:          cfg.AI.DataDir,
-				Enabled:          cfg.AI.Enabled,
-				ConsoleLog:       cfg.ConsoleLog.LLM,
-				ToolConfigStore:  store,
-				ToolRouterStore:  store,
-				TopicRouterStore: store,
-				Store:            store,
-			}, store.DB(), botSenderAdapter)
-			if err != nil {
+			aiManager.SetProviderConfigs(providerConfigs)
+			if err := aiManager.Init(); err != nil {
 				fmt.Fprintf(os.Stderr, "Warning: failed to initialize AI service: %v\n", err)
 			} else {
-				if err := aiService.Start(botCtx); err != nil {
-					fmt.Fprintf(os.Stderr, "Warning: failed to start AI service: %v\n", err)
-				}
-				defer aiService.Stop()
+				defer aiManager.Stop()
 				printJSON(map[string]any{"event": "ai_service_started", "providers": len(providerConfigs)})
 			}
 		} else {
@@ -457,8 +469,8 @@ 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, ClientStats: clientStats, DBQueue: dbQueue}
-		handler := webpkg.NewRouter(cfg.Web, cfg.ConsoleLog.Web, store, sessions, mqttStatus, blocking, forwardManager, settings, botSender)
+		mqttStatus := webpkg.MQTTRuntimeStatus{Server: server, Address: mqttAddr, TLS: cfg.MQTT.TLS.Enabled, Stats: messageStats, ClientStats: clientStats, DBQueue: dbQueue, DedupQueue: mqttHook.dedupQueue}
+		handler := webpkg.NewRouter(cfg.Web, cfg.ConsoleLog.Web, store, sessions, mqttStatus, blocking, forwardManager, settings, botSender, aiManager)
 		webAddresses := []string{}
 		if cfg.Web.PortEnabled {
 			httpServer := &http.Server{
@@ -509,6 +521,9 @@ func run(cfg *configpkg.Config) error {
 	if err := server.Close(); err != nil && runErr == nil {
 		runErr = err
 	}
+	if mqttHook.dedupQueue != nil {
+		mqttHook.dedupQueue.Stop()
+	}
 	return runErr
 }
 
@@ -517,6 +532,8 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
 	if err := server.AddHook(new(mqttauth.AllowHook), nil); err != nil {
 		return nil, nil, "", err
 	}
+	dedupQueue := mqttforwardpkg.NewDedupQueue()
+	dedupQueue.Start()
 	hook := &meshtasticFilterHook{
 		server:           server,
 		key:              cfg.Key,
@@ -528,6 +545,7 @@ func startMQTTServer(cfg *configpkg.Config, store *storepkg.Store, dbQueue *stor
 		pkiResolver:      botpkg.NewPKIKeyResolver(store),
 		consoleLog:       cfg.ConsoleLog.MQTT,
 		packetConsoleLog: cfg.ConsoleLog.Meshtastic,
+		dedupQueue:       dedupQueue,
 	}
 	if err := server.AddHook(hook, nil); err != nil {
 		return nil, nil, "", err

meshmap_frontend/src/api.ts

@@ -6,6 +6,7 @@ import type {
   AdminRuntimeSettingsPayload,
   AdminRuntimeSettingsResponse,
   AdminUsersResponse,
+  AIServiceStatus,
   BotMessage,
   BotMessageMutationResponse,
   BotNode,
@@ -515,8 +516,17 @@ export function updateLLMProvider(name: string, payload: Partial<LLMProviderPayl
   return putJSON<LLMProviderResponse>(`/api/admin/llm/providers/${encodeURIComponent(name)}`, payload)
 }
 
-export function deleteLLMProvider(name: string): Promise<{ status: string }> {
-  return deleteJSON<{ status: string }>(`/api/admin/llm/providers/${encodeURIComponent(name)}`)
+export function deleteLLMProvider(name: string): Promise<{ status: string; warning?: string }> {
+  return deleteJSON<{ status: string; warning?: string }>(`/api/admin/llm/providers/${encodeURIComponent(name)}`)
+}
+
+// AI Service Status API
+export function getAIServiceStatus(): Promise<AIServiceStatus> {
+  return getJSON<AIServiceStatus>('/api/admin/llm/status')
+}
+
+export function restartAIService(): Promise<AIServiceStatus & { status: string; message?: string }> {
+  return postJSON<AIServiceStatus & { status: string; message?: string }>('/api/admin/llm/restart')
 }
 
 // LLM Tool Router API

meshmap_frontend/src/components/AdminDashboard.vue

@@ -200,6 +200,7 @@ onBeforeUnmount(() => {
         <div><span>订阅数</span><strong>{{ status.subscriptions }}</strong></div>
         <div><span>转发消息</span><strong>{{ status.messages_sent }}</strong></div>
         <div><span>数据库队列</span><strong>{{ status.db_write_queue_length }}</strong></div>
+        <div><span>去重队列</span><strong>{{ status.dedup_queue_len }}</strong></div>
         <a class="status-card-link" href="/admin/discard_details"><span>丢弃消息</span><strong>{{ status.messages_dropped }}</strong></a>
         <div><span>收到包</span><strong>{{ status.packets_received }}</strong></div>
         <div><span>发送包</span><strong>{{ status.packets_sent }}</strong></div>

meshmap_frontend/src/components/AdminLLMApi.vue

@@ -11,12 +11,19 @@ import {
   updateLLMToolRouter,
   updateLLMTopicConfig,
   updateLLMPrimaryConfig,
+  getAIServiceStatus,
+  restartAIService,
 } from '../api'
-import type { LLMPlatformRouter, LLMTopicConfig, LLMProvider, LLMPrimaryConfig } from '../types'
+import type { LLMPlatformRouter, LLMTopicConfig, LLMProvider, LLMPrimaryConfig, AIServiceStatus } from '../types'
 
 const loading = ref(false)
 const error = ref('')
 const success = ref('')
+const showWarning = ref(false)
+
+// AI Service Status
+const aiStatus = ref<AIServiceStatus>({ running: false, enabled: false, provider_count: 0 })
+const restarting = ref(false)
 
 // LLM Provider 相关
 const providers = ref<LLMProvider[]>([])
@@ -169,12 +176,18 @@ async function saveProvider() {
   }
 
   try {
+    let response: any
     if (isCreatingProvider.value) {
-      await createLLMProvider(providerForm.value)
-      success.value = '创建成功'
+      response = await createLLMProvider(providerForm.value)
     } else if (editingProvider.value) {
-      await updateLLMProvider(editingProvider.value.name, providerForm.value)
-      success.value = '更新成功'
+      response = await updateLLMProvider(editingProvider.value.name, providerForm.value)
+    }
+    if (response.warning) {
+      success.value = response.warning
+      showWarning.value = true
+    } else {
+      success.value = isCreatingProvider.value ? '创建成功' : '更新成功'
+      showWarning.value = false
     }
     clearSuccess()
     closeProviderForm()
@@ -189,8 +202,14 @@ async function confirmDeleteProvider(name: string) {
     return
   }
   try {
-    await deleteLLMProvider(name)
-    success.value = '删除成功'
+    const response = await deleteLLMProvider(name)
+    if (response.warning) {
+      success.value = response.warning
+      showWarning.value = true
+    } else {
+      success.value = '删除成功'
+      showWarning.value = false
+    }
     clearSuccess()
     await loadProviders()
   } catch (err) {
@@ -286,7 +305,31 @@ async function savePrimaryConfig() {
   }
 }
 
+async function loadAIStatus() {
+  try {
+    aiStatus.value = await getAIServiceStatus()
+  } catch (err) {
+    console.warn('Failed to load AI service status', err)
+  }
+}
+
+async function handleRestartAI() {
+  if (!confirm('确定要重启 AI 服务吗？')) return
+  restarting.value = true
+  try {
+    const result = await restartAIService()
+    aiStatus.value = result
+    success.value = result.message || 'AI 服务已重启'
+    clearSuccess()
+  } catch (err) {
+    error.value = err instanceof Error ? err.message : String(err)
+  } finally {
+    restarting.value = false
+  }
+}
+
 onMounted(() => {
+  loadAIStatus()
   loadProviders()
   loadToolRouter()
   loadTopicConfig()
@@ -298,8 +341,23 @@ onMounted(() => {
   <div class="admin-llm-api">
     <h2>LLM API 配置管理</h2>
 
+    <div class="ai-status-bar">
+      <span class="status-indicator" :class="{ running: aiStatus.running, stopped: !aiStatus.running }"></span>
+      <span class="status-text">
+        <template v-if="aiStatus.running">AI 服务运行中，{{ aiStatus.provider_count }} 个提供商</template>
+        <template v-else>{{ aiStatus.message || 'AI 服务未运行' }}</template>
+      </span>
+      <button
+        class="admin-button admin-button-small"
+        :disabled="restarting"
+        @click="handleRestartAI"
+      >
+        {{ restarting ? '重启中...' : '重启 AI 服务' }}
+      </button>
+    </div>
+
     <p v-if="error" class="error">{{ error }}</p>
-    <p v-if="success" class="success">{{ success }}</p>
+    <p v-if="success" :class="showWarning ? 'warning' : 'success'">{{ success }}</p>
 
     <!-- LLM Provider 列表 -->
     <div class="admin-section">
@@ -690,13 +748,49 @@ onMounted(() => {
 }
 
 .admin-llm-api h2 {
-  margin: 0 0 2rem;
+  margin: 0 0 1rem;
   font-size: 1.75rem;
   font-weight: 700;
   color: #1e293b;
   letter-spacing: -0.02em;
 }
 
+.ai-status-bar {
+  display: flex;
+  align-items: center;
+  gap: 0.75rem;
+  padding: 0.75rem 1.25rem;
+  background: white;
+  border-radius: 12px;
+  border: 1px solid #e2e8f0;
+  margin-bottom: 1.5rem;
+  box-shadow: 0 1px 3px rgba(0, 0, 0, 0.05);
+}
+
+.status-indicator {
+  width: 10px;
+  height: 10px;
+  border-radius: 50%;
+  flex-shrink: 0;
+}
+
+.status-indicator.running {
+  background: #22c55e;
+  box-shadow: 0 0 6px rgba(34, 197, 94, 0.5);
+}
+
+.status-indicator.stopped {
+  background: #ef4444;
+  box-shadow: 0 0 6px rgba(239, 68, 68, 0.4);
+}
+
+.status-text {
+  flex: 1;
+  font-size: 0.9rem;
+  color: #475569;
+  font-weight: 500;
+}
+
 .admin-section {
   background: white;
   padding: 1.75rem;
@@ -973,6 +1067,19 @@ onMounted(() => {
   gap: 0.5rem;
 }
 
+.warning {
+  color: #92400e;
+  padding: 1rem 1.25rem;
+  background: linear-gradient(135deg, #fef3c7 0%, #fde68a 100%);
+  border-radius: 10px;
+  margin-bottom: 1.25rem;
+  border: 1px solid #fcd34d;
+  font-weight: 500;
+  display: flex;
+  align-items: center;
+  gap: 0.5rem;
+}
+
 .admin-loading {
   padding: 3rem;
   text-align: center;

meshmap_frontend/src/types.ts

@@ -364,6 +364,7 @@ export interface AdminMqttStatus {
   messages_sent: number
   messages_dropped: number
   db_write_queue_length: number
+  dedup_queue_len: number
   retained: number
   inflight: number
   inflight_dropped: number
@@ -643,6 +644,14 @@ export interface LLMProviderPayload {
 
 export interface LLMProviderResponse {
   item: LLMProvider
+  warning?: string
+}
+
+export interface AIServiceStatus {
+  running: boolean
+  enabled: boolean
+  provider_count: number
+  message?: string
 }
 
 // LLM Tool Router 相关类型

py/.gitignore

@@ -1 +1,2 @@
 __pycache__
+db_config.py

py/migrate_mysql8_to_mysql57.py

@@ -0,0 +1,266 @@
+#!/usr/bin/env python3
+"""
+MySQL 8 -> MySQL 5.7 数据库迁移脚本
+
+依赖: pip install pymysql
+
+使用方法:
+    python py/migrate_mysql8_to_mysql57.py
+
+功能:
+    1. 从源 MySQL 8 读取表结构，修正 collation 后在目标 MySQL 5.7 建表
+    2. 逐表批量拷贝数据
+    3. 修正 auto-increment 值
+
+注意:
+    - 目标库应为空库（无同名表）
+    - 运行前请确保源库和目标库都可连接
+"""
+
+from __future__ import annotations
+
+import re
+import sys
+from typing import Any
+
+import pymysql
+import pymysql.cursors
+
+from db_config import SOURCE_CONFIG, TARGET_CONFIG
+
+BATCH_SIZE = 5000
+
+# ============================================================
+# 工具函数
+# ============================================================
+
+
+def _conn(config: dict[str, Any]) -> pymysql.Connection:
+    """创建数据库连接，使用 DictCursor 方便按列名访问"""
+    return pymysql.connect(
+        host=config["host"],
+        port=config["port"],
+        user=config["user"],
+        password=config["password"],
+        database=config["database"],
+        charset=config["charset"],
+        cursorclass=pymysql.cursors.DictCursor,
+    )
+
+
+def fix_collation(ddl: str) -> str:
+    """将 MySQL 8 默认 collation 替换为 MySQL 5.7 兼容版本"""
+    ddl = ddl.replace("utf8mb4_0900_ai_ci", "utf8mb4_general_ci")
+    ddl = ddl.replace("utf8mb4_0900_as_ci", "utf8mb4_general_ci")
+    return ddl
+
+
+def fix_text_blob_index(ddl: str) -> str:
+    """为 TEXT/BLOB 列在索引中添加前缀长度 (255)，兼容 MySQL 5.7"""
+    text_cols = set()
+    for m in re.finditer(r'`(\w+)`\s+(?:tinytext|text|mediumtext|longtext|tinyblob|blob|mediumblob|longblob)\b', ddl, re.IGNORECASE):
+        text_cols.add(m.group(1))
+    if not text_cols:
+        return ddl
+
+    def _add_prefix(match: re.Match) -> str:
+        col = match.group(1)
+        rest = match.group(2)
+        if col in text_cols and not rest.strip().startswith('('):
+            return f'`{col}`(255){rest}'
+        return match.group(0)
+
+    # 匹配 KEY 定义中的列引用: `col_name` 后面不跟 ( 的情况
+    ddl = re.sub(r'`(\w+)`(\s*[,\)])', _add_prefix, ddl)
+    return ddl
+
+
+def get_all_tables(conn: pymysql.Connection) -> list[str]:
+    with conn.cursor() as cur:
+        cur.execute("SHOW TABLES")
+        key = list(cur.description[0])[0]
+        return [row[key] for row in cur.fetchall()]
+
+
+def get_table_columns(conn: pymysql.Connection, table: str) -> list[str]:
+    """返回表的所有列名（按顺序）"""
+    with conn.cursor() as cur:
+        cur.execute("SHOW COLUMNS FROM `%s`" % table)
+        return [row["Field"] for row in cur.fetchall()]
+
+
+def get_auto_increment(
+    conn: pymysql.Connection, table: str
+) -> int | None:
+    """获取某张表当前的 AUTO_INCREMENT 值"""
+    with conn.cursor() as cur:
+        cur.execute(
+            "SELECT AUTO_INCREMENT FROM information_schema.TABLES "
+            "WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s",
+            (table,),
+        )
+        row = cur.fetchone()
+        if row and row["AUTO_INCREMENT"]:
+            return int(row["AUTO_INCREMENT"])
+    return None
+
+
+def row_count(conn: pymysql.Connection, table: str) -> int:
+    with conn.cursor() as cur:
+        cur.execute("SELECT COUNT(*) AS cnt FROM `%s`" % table)
+        return cur.fetchone()["cnt"]
+
+
+# ============================================================
+# 主流程
+# ============================================================
+
+
+def migrate() -> int:
+    src = _conn(SOURCE_CONFIG)
+    tgt = _conn(TARGET_CONFIG)
+    print(f"[连接] 源 MySQL 8 @ {SOURCE_CONFIG['host']}:{SOURCE_CONFIG['port']}")
+    print(f"[连接] 目标 MySQL 5.7 @ {TARGET_CONFIG['host']}:{TARGET_CONFIG['port']}")
+    print()
+
+    # 1. 在目标库建库（如还不存在）
+    try:
+        with tgt.cursor() as cur:
+            cur.execute(
+                "CREATE DATABASE IF NOT EXISTS `%s` "
+                "CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci"
+                % TARGET_CONFIG["database"]
+            )
+            cur.execute("USE `%s`" % TARGET_CONFIG["database"])
+        tgt.select_db(TARGET_CONFIG["database"])
+    except Exception as e:
+        print(f"[错误] 创建目标数据库失败: {e}")
+        return 1
+
+    tables = get_all_tables(src)
+    print(f"[发现] 源库共 {len(tables)} 张表: {', '.join(tables)}")
+    print()
+
+    # 2. 逐表建表（修正 collation）
+    print("=" * 60)
+    print("阶段 1: 在目标库创建表结构")
+    print("=" * 60)
+    for idx, table in enumerate(tables, 1):
+        with src.cursor() as cur:
+            cur.execute("SHOW CREATE TABLE `%s`" % table)
+            row = cur.fetchone()
+            ddl = row["Create Table"]
+        ddl = fix_collation(ddl)
+        ddl = fix_text_blob_index(ddl)
+
+        try:
+            with tgt.cursor() as cur:
+                cur.execute(ddl)
+            tgt.commit()
+            print(f"  [{idx:2d}/{len(tables)}] OK `{table}`")
+        except Exception as e:
+            print(f"  [{idx:2d}/{len(tables)}] 错误 `{table}`: {e}")
+            tgt.rollback()
+            return 1
+
+    print()
+
+    # 3. 逐表拷贝数据
+    print("=" * 60)
+    print("阶段 2: 拷贝表数据")
+    print("=" * 60)
+
+    total_rows_copied = 0
+    auto_increments: dict[str, int | None] = {}
+
+    for idx, table in enumerate(tables, 1):
+        columns = get_table_columns(src, table)
+        if not columns:
+            auto_increments[table] = None
+            print(f"  [{idx:2d}/{len(tables)}] SKIP `{table}` (0 列)")
+            continue
+
+        col_quoted = ", ".join("`%s`" % c for c in columns)
+        placeholders = ", ".join(["%s"] * len(columns))
+        insert_sql = "INSERT INTO `%s` (%s) VALUES (%s)" % (
+            table,
+            col_quoted,
+            placeholders,
+        )
+
+        table_count = 0
+        with src.cursor() as read_cur:
+            read_cur.execute("SELECT * FROM `%s`" % table)
+            batch = read_cur.fetchmany(BATCH_SIZE)
+            while batch:
+                rows_values = [
+                    [row.get(c) for c in columns] for row in batch
+                ]
+                try:
+                    with tgt.cursor() as write_cur:
+                        write_cur.executemany(insert_sql, rows_values)
+                    tgt.commit()
+                    table_count += len(rows_values)
+                    print(
+                        f"\r  [{idx:2d}/{len(tables)}] `{table}` -> {table_count} 行",
+                        end="",
+                        flush=True,
+                    )
+                except Exception as e:
+                    print()
+                    print(f"  [{idx:2d}/{len(tables)}] 错误 `{table}`: {e}")
+                    tgt.rollback()
+                    return 1
+                batch = read_cur.fetchmany(BATCH_SIZE)
+
+        # 修正 auto_increment
+        ai = get_auto_increment(src, table)
+        auto_increments[table] = ai
+        if ai is not None:
+            with tgt.cursor() as cur:
+                cur.execute(
+                    "ALTER TABLE `%s` AUTO_INCREMENT = %s" % (table, ai)
+                )
+            tgt.commit()
+
+        total_rows_copied += table_count
+        print(
+            f"\r  [{idx:2d}/{len(tables)}] `{table}` -> {table_count} 行  [OK]"
+        )
+
+    # 4. 验证
+    print()
+    print("=" * 60)
+    print("阶段 3: 验证行数")
+    print("=" * 60)
+    all_match = True
+    src_total = 0
+    tgt_total = 0
+    for table in tables:
+        s = row_count(src, table)
+        t = row_count(tgt, table)
+        src_total += s
+        tgt_total += t
+        status = "OK" if s == t else "不匹配!"
+        if s != t:
+            all_match = False
+        print(f"  `{table}`: 源={s}  目标={t}  [{status}]")
+
+    print()
+    print(f"  总计: 源={src_total}  目标={tgt_total}")
+
+    src.close()
+    tgt.close()
+
+    if all_match:
+        print()
+        print("迁移完成，所有表行数一致。")
+        return 0
+    else:
+        print()
+        print("警告: 部分表行数不一致，请检查。")
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(migrate())