modify: MQTT Bridge 兼容非信封原始报文（盒子心跳）——不再丢弃，整包视为 data 路由，与 worker 运行时容错语义一致；每主题仅告警一次

src/chongming/app/core/mqtt_bridge.py

@@ -12,6 +12,9 @@ logger = get_logger("mqtt")
 MQTT_TOPIC_PREFIX = "chongming"
 MQTT_SUBSCRIBE_TOPIC = f"{MQTT_TOPIC_PREFIX}/+/+"
 
+# 已按"非信封载荷"路由过告警的主题（每主题只告警一次，避免设备 5s 心跳刷屏）
+_warned_legacy_topics: set = set()
+
 
 def _compile_topic(topic: str) -> re.Pattern:
     """
@@ -154,18 +157,26 @@ class MQTTBridge:
         except json.JSONDecodeError:
             return
 
-        # 信封协议（严格）: {"meta": {...}, "data": {...}}
+        # 信封协议: {"meta": {...}, "data": {...}}
-        # 非信封格式载荷直接丢弃并告警，请设备侧按新协议发布
+        # 兼容非信封载荷（如第三方设备原始报文）: 整包视为业务数据、meta 置空，
+        # 与 worker 运行时 runtime.py 的容错语义一致（method 默认 GET）——
+        # 若在此丢弃，盒子心跳等设备报文将永远到不了 Worker。
         if not isinstance(pl, dict) or ("meta" not in pl and "data" not in pl):
-            logger.warning(
+            data = pl if isinstance(pl, dict) else {}
-                "丢弃非信封格式载荷 [%s]: 请使用 {'meta': {...}, 'data': {...}} 格式", t
+            pl = {"meta": {}, "data": data}
-            )
+            if t not in _warned_legacy_topics:
-            return
+                _warned_legacy_topics.add(t)
-        meta = pl.get("meta") or {}
+                logger.warning(
-        data = pl.get("data") or {}
+                    "非信封格式载荷（缺少 meta/data）按原始报文路由 [%s]: "
-        if not isinstance(data, dict):
+                    "整包视为 data，建议设备侧改用 {'meta': {...}, 'data': {...}} 格式",
-            data = {}
+                    t,
-        pl = {"meta": meta, "data": data}
+                )
+        else:
+            meta = pl.get("meta") or {}
+            data = pl.get("data") or {}
+            if not isinstance(data, dict):
+                data = {}
+            pl = {"meta": meta, "data": data}
 
         # 1. 优先匹配显式配置的自定义订阅主题（顺序即优先级）
         for ct, compiled in self._compiled:

test_mqtt_custom_topics.py

@@ -139,16 +139,18 @@ async def test_envelope_meta_fields():
     print("✅ 信封 meta 字段 (method/timeout/cacheable) 透传通过")
 
 
-async def test_flat_payload_dropped():
+async def test_flat_payload_routed():
-    """严格信封协议：扁平载荷（无 meta/data）应被丢弃且不调用 worker"""
+    """非信封（扁平）载荷兼容：整包视为 data、meta 置空照常路由（与 worker 运行时容错一致）"""
     wm = FakeWM()
     bridge, published = make_bridge(
         wm, [{"topic": "devices/+/telemetry", "worker": "device"}]
     )
     await bridge._handle(make_msg("devices/d1/telemetry", {"_method": "POST", "v": 1}))
-    assert wm.calls == []
+    assert wm.calls, "扁平载荷不应被丢弃，应照常路由到 worker"
-    assert published == []
+    assert wm.calls[-1]["worker_type"] == "device"
-    print("✅ 非信封（扁平）载荷被丢弃通过")
+    assert wm.calls[-1]["payload"] == {"meta": {}, "data": {"_method": "POST", "v": 1}}
+    assert published[-1][0] == "devices/d1/telemetry/response"
+    print("✅ 非信封（扁平）载荷按原始报文路由通过")
 
 
 async def test_non_json_dropped():
@@ -182,7 +184,14 @@ async def test_board_ping_topic():
     assert wm.calls[-1]["worker_type"] == "aibox_sophon_adapter"
     assert wm.calls[-1]["subject"] == "aibox_sophon_adapter.board_ping"
     assert published[-1][0] == "/board_ping/response"
-    print("✅ /board_ping 自定义主题路由通过")
+    # 回归：盒子心跳为原始报文（无信封），不得被丢弃，应整包作为 data 路由
+    wm.calls.clear(); published.clear()
+    await bridge._handle(make_msg("/board_ping", {"BoardId": "B001", "Status": 1}))
+    assert wm.calls and wm.calls[-1]["payload"] == {
+        "meta": {}, "data": {"BoardId": "B001", "Status": 1}
+    }, wm.calls
+    assert published[-1][0] == "/board_ping/response"
+    print("✅ /board_ping 自定义主题路由通过（含原始报文兼容）")
 
 
 def test_plugin_config_driven_collection():
@@ -209,7 +218,7 @@ async def main():
     test_compile_topic()
     await test_custom_routing()
     await test_envelope_meta_fields()
-    await test_flat_payload_dropped()
+    await test_flat_payload_routed()
     await test_non_json_dropped()
     await test_board_ping_topic()
     test_plugin_config_driven_collection()