fix(meshcore): deliver DM replies (flood/ACK), save meshcore_context, test-llm, inbound logging (#21)

Four fixes surfaced by live testing (a MeshCore DM got no reply):

- DM REPLY DELIVERY (root cause): reply used the meshcore lib's fire-and-forget
  send_msg (MSG_SENT != delivered, no flood, no ACK) so replies to nodes without
  an established direct path silently vanished. Switch to send_msg_with_retry
  (contact resolve + flood fallback + ACK wait); a None return (no ACK) is now a
  real failure, not silent success. _run_coro timeout raised to 40s for the ACK cycle.
- 422 on save: register meshcore_context in config_loader SECTION_TO_FILE
  (config.yaml) — it was in VALID_SECTIONS but not the save-routing table.
- test-llm endpoint: called backend.generate() with (str, []) instead of
  (messages:list, system_prompt:str) → "string indices" error; fixed the call.
- Inbound observability + robustness: subscribe to CONTACT_MSG_RECV BEFORE
  start_auto_message_fetching (+ ensure_contacts) so a DM queued at connect isn't
  drained before the handler registers; add INFO/DEBUG logging across the inbound
  DM + dispatch + send path (was entirely unlogged).

Tests: +test_meshcore_dm_delivery, +test_fix_meshcore_save_and_llm_test; 0 new failures.

Co-authored-by: Matt Johnson <mj@k7zvx.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/config_loader.py

@@ -41,6 +41,7 @@ SECTION_TO_FILE: dict[str, str] = {
     "history": "config.yaml",
     "memory": "config.yaml",
     "context": "config.yaml",
+    "meshcore_context": "config.yaml",
     "weather": "config.yaml",
     "meshmonitor": "config.yaml",
     "knowledge": "config.yaml",

work/meshai/dashboard/api/config_routes.py

@@ -218,8 +218,11 @@ async def test_llm_connection(request: Request):
         else:
             return {"success": False, "error": f"Unknown backend: {backend_name}"}
 
-        # Send test prompt
+        # Send test prompt — generate(messages: list[dict], system_prompt: str)
-        response = await backend.generate("Reply with 'OK' if you can read this.", [])
+        response = await backend.generate(
+            [{"role": "user", "content": "Reply with 'OK' if you can read this."}],
+            "",
+        )
         await backend.close()
 
         return {"success": True, "response": response}

work/meshai/transport/meshcore_transport.py

@@ -312,13 +312,22 @@ class MeshCoreTransport(MeshTransport):
         return mc
 
     async def _setup_subscriptions(self) -> None:
-        """Start auto message fetching and subscribe to inbound events."""
+        """Subscribe to inbound events, then start auto message fetching."""
         from meshcore import EventType  # noqa: PLC0415
-        await self._mc.start_auto_message_fetching()
+        # Subscribe BEFORE starting auto-fetch: start_auto_message_fetching() drains
+        # the companion queue immediately, which would dispatch CONTACT_MSG_RECV before
+        # our handler is registered and silently lose a DM queued at connect time.
+        # (Canonical lib order: subscribe -> ensure_contacts -> start fetching.)
         self._mc.subscribe(EventType.CONTACT_MSG_RECV, self._on_dm_event)
         self._mc.subscribe(EventType.CHANNEL_MSG_RECV, self._on_channel_event)
         self._mc.subscribe(EventType.DISCONNECTED, self._on_disconnect_event)
         self._mc.subscribe(EventType.CONNECTED, self._on_connect_event)
+        try:
+            await self._mc.ensure_contacts()
+        except Exception:
+            logger.debug("MeshCore: ensure_contacts failed (non-fatal)", exc_info=True)
+        await self._mc.start_auto_message_fetching()
+        logger.info("MeshCore: subscriptions registered; auto message-fetch started")
 
     async def _do_disconnect(self) -> None:
         """Stop fetching and close the meshcore connection."""
@@ -459,10 +468,26 @@ class MeshCoreTransport(MeshTransport):
 
         try:
             if destination:
-                # DM: meshcore_channel is irrelevant; route by pubkey.
+                # DM: use the retry/flood-capable send so replies actually deliver.
+                # Plain send_msg is fire-and-forget (MSG_SENT != delivered) with no flood
+                # fallback, so DMs to nodes without an established direct path silently drop.
+                # send_msg_with_retry resolves the contact, floods when needed, waits for ACK,
+                # and returns None if no ACK (delivery not confirmed).
+                logger.debug("MeshCore: sending DM to %s", destination[:40])
                 result = self._run_coro(
-                    self._mc.commands.send_msg(destination, text)
+                    self._mc.commands.send_msg_with_retry(destination, text),
+                    timeout=40,
                 )
+                if result is None:
+                    logger.warning(
+                        "MeshCoreTransport: DM to %s not ACKed (delivery not confirmed)",
+                        destination,
+                    )
+                    return False
+                success = not result.is_error()
+                if not success:
+                    logger.warning("MeshCoreTransport: DM send returned error event")
+                return success
             else:
                 # Channel broadcast.
                 # Channel-index semantics do NOT cross transports: the passed
@@ -587,10 +612,22 @@ class MeshCoreTransport(MeshTransport):
 
     def _on_dm_event(self, event) -> None:
         """Handle CONTACT_MSG_RECV: normalize, filter, and dispatch to meshai."""
+        try:
+            _p = event.payload or {}
+            _sender = _p.get("pubkey_prefix", "?")
+            _preview = str(_p.get("text", ""))[:40]
+        except Exception:
+            _sender = repr(event)[:40]
+            _preview = ""
+        logger.info("MeshCore: inbound DM from %s: %r", _sender, _preview)
         msg = self._normalize_dm_event(event)
-        if msg is None or not mc_context_allows(
+        if msg is None:
+            logger.debug("MeshCore: DM from %s dropped (normalize returned None)", _sender)
+            return
+        if not mc_context_allows(
             self._mc_context, msg, {v: k for k, v in self._chan_name_to_idx.items()}
         ):
+            logger.debug("MeshCore: DM from %s dropped by context gate", _sender)
             return
         self._dispatch_message(msg)
 
@@ -610,11 +647,17 @@ class MeshCoreTransport(MeshTransport):
             loop.call_soon_threadsafe(lambda m=msg: asyncio.create_task(cb(m)))
         """
         if msg is None or self._message_callback is None or self._callback_loop is None:
+            logger.debug(
+                "MeshCoreTransport: _dispatch_message dropped (msg=%s callback=%s loop=%s)",
+                msg is not None, self._message_callback is not None,
+                self._callback_loop is not None,
+            )
             return
         try:
             self._callback_loop.call_soon_threadsafe(
                 lambda m=msg: asyncio.create_task(self._message_callback(m))
             )
+            logger.debug("MeshCoreTransport: dispatched message to meshai")
         except Exception as exc:
             logger.error("MeshCoreTransport: error dispatching message: %s", exc)
 

work/tests/test_fix_meshcore_save_and_llm_test.py

@@ -0,0 +1,145 @@
+"""Regression tests for two bug fixes:
+
+Bug 1 — save_section raises ValueError for 'meshcore_context'
+  Covered by test_meshcore_context_save_section_no_error (uses tmp config dir).
+
+Bug 2 — POST /api/config/test-llm "string indices must be integers, not 'str'"
+  The handler was calling backend.generate(str, list) instead of
+  generate(list[dict], str).  The google backend then iterated over the string
+  char-by-char and blew up on `msg["role"]`.  After the fix the handler calls
+  generate([{"role": "user", "content": "..."}], ""), which returns a plain
+  string — the correct return type.
+"""
+from __future__ import annotations
+
+import sys
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+# -----------------------------------------------------------------------
+# Stub heavy optional deps so config_routes can be imported without them.
+# -----------------------------------------------------------------------
+for _mod in ("openai", "aiosqlite", "anthropic", "google", "google.genai"):
+    sys.modules.setdefault(_mod, MagicMock())
+
+
+# ==========================================================================
+# Bug 1 — meshcore_context in SECTION_TO_FILE
+# ==========================================================================
+
+
+def test_meshcore_context_in_section_to_file():
+    """'meshcore_context' must be present in SECTION_TO_FILE mapping to
+    config.yaml, otherwise save_section raises ValueError (HTTP 422)."""
+    from meshai.config_loader import SECTION_TO_FILE
+
+    assert "meshcore_context" in SECTION_TO_FILE, (
+        "meshcore_context missing from SECTION_TO_FILE"
+    )
+    assert SECTION_TO_FILE["meshcore_context"] == "config.yaml"
+
+
+def test_meshcore_context_save_section_no_error(tmp_path):
+    """save_section('meshcore_context', ...) must not raise ValueError.
+
+    Uses a minimal on-disk config.yaml so the saver has something to
+    round-trip without needing !include or live config infrastructure.
+    """
+    import yaml
+    from meshai.config_loader import save_section
+
+    cfg_dir = tmp_path / "config"
+    cfg_dir.mkdir()
+    # Minimal seed so _load_yaml_preserve finds the file.
+    (cfg_dir / "config.yaml").write_text(
+        yaml.safe_dump({"timezone": "UTC"})
+    )
+
+    # Should not raise
+    result = save_section(
+        "meshcore_context",
+        {
+            "enable_passive_context": True,
+            "observe_channels": ["#general"],
+            "ignore_contacts": [],
+            "respond_to_dms": False,
+        },
+        cfg_dir,
+    )
+    assert result["saved"] is True
+    assert any("config.yaml" in f for f in result["files_written"])
+
+    # Verify the value was actually written to config.yaml
+    on_disk = yaml.safe_load((cfg_dir / "config.yaml").read_text())
+    assert "meshcore_context" in on_disk
+    assert on_disk["meshcore_context"]["observe_channels"] == ["#general"]
+
+
+# ==========================================================================
+# Bug 2 — POST /api/config/test-llm
+# ==========================================================================
+
+
+def _build_test_app():
+    """Build a minimal FastAPI app wired to config_routes with a google config."""
+    from fastapi import FastAPI
+    from meshai.dashboard.api.config_routes import router
+    from meshai.config import Config, LLMConfig
+
+    app = FastAPI()
+    app.include_router(router, prefix="/api")
+
+    cfg = Config()
+    cfg.llm = LLMConfig(backend="google", model="gemini-2.5-flash", api_key="fake-key")
+    app.state.config = cfg
+    app.state.config_path = None  # not needed for this route
+    return app
+
+
+def test_test_llm_google_backend_returns_success():
+    """POST /api/config/test-llm with google backend must return success:true.
+
+    The GoogleBackend.generate() returns a plain string.  The old handler
+    passed (str, list) to generate() and crashed; the fixed handler passes
+    ([{"role": "user", "content": "..."}], ""), which the mocked backend
+    receives as a proper list and returns the string reply.
+    """
+    from fastapi.testclient import TestClient
+
+    app = _build_test_app()
+
+    # resolve_api_key() must return a non-empty string so the handler proceeds.
+    # GoogleBackend is mocked so no real HTTP call is made.
+    with (
+        patch.object(
+            app.state.config.__class__,
+            "resolve_api_key",
+            return_value="fake-key",
+        ),
+        patch(
+            "meshai.backends.GoogleBackend",
+        ) as MockGoogleBackend,
+    ):
+        mock_instance = MagicMock()
+        # generate() returns a plain string — that is the real return type.
+        mock_instance.generate = AsyncMock(return_value="OK")
+        mock_instance.close = AsyncMock()
+        MockGoogleBackend.return_value = mock_instance
+
+        client = TestClient(app)
+        r = client.post("/api/config/test-llm")
+
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert body["success"] is True
+    assert body["response"] == "OK"
+
+    # Confirm generate was called with a list as first arg (not a string)
+    call_args = mock_instance.generate.call_args
+    messages_arg = call_args[0][0] if call_args[0] else call_args[1].get("messages")
+    assert isinstance(messages_arg, list), (
+        f"generate() first arg must be list[dict], got {type(messages_arg)}"
+    )
+    assert messages_arg[0]["role"] == "user"

work/tests/test_meshcore_dm_delivery.py

@@ -0,0 +1,181 @@
+"""Focused tests for the MeshCore DM delivery fix.
+
+Verifies that send_message(..., destination=...) calls send_msg_with_retry
+(not the old fire-and-forget send_msg) and correctly maps its return value
+to True/False:
+  - non-error Event returned  → True  (ACKed, delivered)
+  - None returned             → False (no ACK, delivery not confirmed)
+
+The meshcore lib is mocked via sys.modules (same pattern as the existing
+transport test module).  _run_coro is patched to execute the coroutine
+synchronously so no background event-loop thread is needed.
+"""
+
+import asyncio
+import sys
+import types
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Minimal fake meshcore module (guards against import errors if the real lib
+# is absent, and avoids side-effects from the sys.modules entry in the other
+# transport test module racing this one).
+# ---------------------------------------------------------------------------
+
+def _ensure_fake_meshcore():
+    if "meshcore" in sys.modules:
+        return
+    mod = types.ModuleType("meshcore")
+
+    class EventType:
+        CONTACT_MSG_RECV = "CONTACT_MSG_RECV"
+        CHANNEL_MSG_RECV = "CHANNEL_MSG_RECV"
+        DISCONNECTED = "DISCONNECTED"
+        CONNECTED = "CONNECTED"
+
+    mod.EventType = EventType
+
+    class _FakeMeshCore:
+        self_info = {"public_key": "aabbccdd1122", "name": "FakeNode"}
+        contacts = {}
+
+        async def start_auto_message_fetching(self):
+            pass
+
+        async def stop_auto_message_fetching(self):
+            pass
+
+        async def disconnect(self):
+            pass
+
+        def subscribe(self, event_type, callback):
+            pass
+
+        def get_contact_by_key_prefix(self, prefix):
+            return None
+
+        @classmethod
+        async def create_tcp(cls, host, port,
+                             auto_reconnect=True, max_reconnect_attempts=5):
+            return cls()
+
+        class commands:
+            @staticmethod
+            async def send_chan_msg(chan_idx, text):
+                result = MagicMock()
+                result.is_error.return_value = False
+                return result
+
+            @staticmethod
+            async def send_msg(dst, text):
+                result = MagicMock()
+                result.is_error.return_value = False
+                return result
+
+            @staticmethod
+            async def send_advert(flood=False):
+                pass
+
+    mod.MeshCore = _FakeMeshCore
+    sys.modules["meshcore"] = mod
+
+
+_ensure_fake_meshcore()
+
+from meshai.config import ConnectionConfig                            # noqa: E402
+from meshai.transport.meshcore_transport import MeshCoreTransport    # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _mc_config():
+    return ConnectionConfig(meshcore_host="127.0.0.1", meshcore_port=5050)
+
+
+def _transport_with_mc_mock():
+    """Return a MeshCoreTransport with _mc as a MagicMock (no loop thread).
+
+    _run_coro is patched on the instance to run the coroutine synchronously via
+    asyncio.get_event_loop().run_until_complete(), bypassing the thread bridge.
+    This keeps tests fast and deterministic.
+    """
+    cfg = _mc_config()
+    t = MeshCoreTransport(cfg)
+    mc = MagicMock()
+    t._mc = mc
+    t._connected = True
+
+    def _sync_run_coro(coro, timeout=None):
+        loop = asyncio.new_event_loop()
+        try:
+            return loop.run_until_complete(coro)
+        finally:
+            loop.close()
+
+    t._run_coro = _sync_run_coro
+    return t, mc
+
+
+# ---------------------------------------------------------------------------
+# DM delivery tests
+# ---------------------------------------------------------------------------
+
+class TestMeshCoreDMDelivery:
+    """send_message with destination= must use send_msg_with_retry, not send_msg."""
+
+    def test_acked_dm_returns_true_and_uses_send_msg_with_retry(self):
+        """ACK received (non-error Event) → returns True; send_msg_with_retry called."""
+        t, mc = _transport_with_mc_mock()
+
+        ok_event = MagicMock()
+        ok_event.is_error.return_value = False
+        mc.commands.send_msg_with_retry = AsyncMock(return_value=ok_event)
+        mc.commands.send_msg = AsyncMock()  # must NOT be called
+
+        result = t.send_message("reply text", destination="aabbccdd1122")
+
+        assert result is True
+        mc.commands.send_msg_with_retry.assert_awaited_once_with("aabbccdd1122", "reply text")
+        mc.commands.send_msg.assert_not_awaited()
+
+    def test_no_ack_returns_false(self):
+        """No ACK (send_msg_with_retry returns None) → returns False."""
+        t, mc = _transport_with_mc_mock()
+
+        mc.commands.send_msg_with_retry = AsyncMock(return_value=None)
+
+        result = t.send_message("reply text", destination="aabbccdd1122")
+
+        assert result is False
+        mc.commands.send_msg_with_retry.assert_awaited_once_with("aabbccdd1122", "reply text")
+
+    def test_error_event_returns_false(self):
+        """Error event returned by send_msg_with_retry → returns False."""
+        t, mc = _transport_with_mc_mock()
+
+        err_event = MagicMock()
+        err_event.is_error.return_value = True
+        mc.commands.send_msg_with_retry = AsyncMock(return_value=err_event)
+
+        result = t.send_message("fail text", destination="deadbeef0011")
+
+        assert result is False
+
+    def test_no_ack_logs_warning(self, caplog):
+        """No ACK → a warning mentioning the destination is logged."""
+        import logging
+        t, mc = _transport_with_mc_mock()
+        mc.commands.send_msg_with_retry = AsyncMock(return_value=None)
+
+        with caplog.at_level(logging.WARNING):
+            t.send_message("msg", destination="deadbeef0011")
+
+        assert any(
+            "not ACKed" in r.getMessage() or "delivery not confirmed" in r.getMessage()
+            for r in caplog.records
+        )

work/tests/test_meshcore_transport.py

@@ -326,24 +326,24 @@ class TestSendMessageChannel:
 # ---------------------------------------------------------------------------
 
 class TestSendMessageDM:
-    def test_dispatches_send_msg(self):
+    def test_dispatches_send_msg_with_retry(self):
         t, mc, _ = _transport_with_mock_mc()
         try:
             ok = MagicMock()
             ok.is_error.return_value = False
-            mc.commands.send_msg = AsyncMock(return_value=ok)
+            mc.commands.send_msg_with_retry = AsyncMock(return_value=ok)
             result = t.send_message("hi DM", destination="aabbcc")
             assert result is True
-            mc.commands.send_msg.assert_awaited_once_with("aabbcc", "hi DM")
+            mc.commands.send_msg_with_retry.assert_awaited_once_with("aabbcc", "hi DM")
         finally:
             _cleanup(t)
 
-    def test_send_msg_error_returns_false(self):
+    def test_send_msg_with_retry_error_returns_false(self):
         t, mc, _ = _transport_with_mock_mc()
         try:
             err = MagicMock()
             err.is_error.return_value = True
-            mc.commands.send_msg = AsyncMock(return_value=err)
+            mc.commands.send_msg_with_retry = AsyncMock(return_value=err)
             assert t.send_message("hi", destination="deadbeef") is False
         finally:
             _cleanup(t)