plugin/hermes_meshtastic/adapter.py

@@ -214,13 +214,21 @@ def get_adapter_class():
                 """Handle incoming text packet from Meshtastic pubsub (worker thread)."""
                 try:
                     ch = packet.get("channel", 0)
-                    if self.channel_index is not None and ch != self.channel_index:
+                    my_info = getattr(interface, "myInfo", None)
+                    my_node_num = getattr(my_info, "my_node_num", 0)
+                    is_direct_to_me = packet.get("to") == my_node_num
+                    is_pki_dm = is_direct_to_me and packet.get("pkiEncrypted", False)
+                    if (
+                        self.channel_index is not None
+                        and ch != self.channel_index
+                        and not is_pki_dm
+                    ):
                         # Ignore packets from different channels (e.g. public channel 0).
+                        # PKI direct messages omit the channel field and decode as
+                        # channel 0 even when sent from a secondary channel.
                         return
 
                     from_id = packet.get("fromId") or f"!{packet.get('from', 0):08x}"
-                    my_info = getattr(interface, "myInfo", None)
-                    my_node_num = getattr(my_info, "my_node_num", 0)
                     if packet.get("from") == my_node_num:
                         # Prevent self-message loops.
                         return
@@ -237,7 +245,7 @@ def get_adapter_class():
                         return
 
                     to_id = packet.get("toId")
-                    is_dm = to_id != "^all" and packet.get("to") == my_node_num
+                    is_dm = to_id != "^all" and is_direct_to_me
                     chat_id = from_id if is_dm else f"channel_{ch}"
                     chat_type = "dm" if is_dm else "channel"
 

plugin/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "hermes-meshtastic"
-version = "0.1.3"
+version = "0.1.4"
 description = "Meshtastic LoRa mesh gateway adapter plugin for Hermes Agent"
 readme = "README.md"
 requires-python = ">=3.10"

plugin/tests/contract/test_gateway_contract.py

@@ -36,6 +36,7 @@ so the gate can never silently skip.
 
 import inspect
 import os
+from types import SimpleNamespace
 
 import pytest
 
@@ -228,3 +229,40 @@ def test_inbound_event_construction_is_valid(registered_entry):
     )
     assert event.source.platform.value == "meshtastic"
     assert event.text == "hello over LoRa"
+
+
+def test_pki_direct_message_without_channel_reaches_gateway(
+    registered_entry, monkeypatch
+):
+    """PKI DMs omit channel and must bypass the secondary-channel filter."""
+    adapter = _make_adapter(registered_entry)
+    adapter._loop = SimpleNamespace(is_running=lambda: True)
+    captured = {}
+
+    async def handle_message(event):
+        captured["event"] = event
+
+    def submit(coro, loop):
+        captured["loop"] = loop
+        with pytest.raises(StopIteration):
+            coro.send(None)
+
+    adapter.handle_message = handle_message
+    monkeypatch.setattr(plugin_adapter.asyncio, "run_coroutine_threadsafe", submit)
+
+    interface = SimpleNamespace(myInfo=SimpleNamespace(my_node_num=0x1BA1BC60))
+    adapter._on_meshtastic_receive(
+        {
+            "from": 0x1BA1496C,
+            "to": 0x1BA1BC60,
+            "toId": "!1ba1bc60",
+            "pkiEncrypted": True,
+            "decoded": {"text": "hello over PKI"},
+            "id": 42,
+        },
+        interface,
+    )
+
+    assert captured["loop"] is adapter._loop
+    assert captured["event"].text == "hello over PKI"
+    assert captured["event"].source.chat_id == "!1ba1496c"