From ddd47afa8713ce55b3470af3ecf2382b16c536f0 Mon Sep 17 00:00:00 2001 From: malice Date: Fri, 3 Jul 2026 19:54:16 -0600 Subject: [PATCH] fix(meshcore): deliver DM replies (flood/ACK), save meshcore_context, test-llm, inbound logging (#21) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Co-authored-by: Claude Opus 4.8 (1M context) --- work/meshai/config_loader.py | 1 + work/meshai/dashboard/api/config_routes.py | 7 +- work/meshai/transport/meshcore_transport.py | 53 ++++- .../test_fix_meshcore_save_and_llm_test.py | 145 ++++++++++++++ work/tests/test_meshcore_dm_delivery.py | 181 ++++++++++++++++++ work/tests/test_meshcore_transport.py | 10 +- 6 files changed, 385 insertions(+), 12 deletions(-) create mode 100644 work/tests/test_fix_meshcore_save_and_llm_test.py create mode 100644 work/tests/test_meshcore_dm_delivery.py diff --git a/work/meshai/config_loader.py b/work/meshai/config_loader.py index 220c3e8..69f5ac9 100644 --- a/work/meshai/config_loader.py +++ b/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", diff --git a/work/meshai/dashboard/api/config_routes.py b/work/meshai/dashboard/api/config_routes.py index 660983b..6207994 100644 --- a/work/meshai/dashboard/api/config_routes.py +++ b/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 - response = await backend.generate("Reply with 'OK' if you can read this.", []) + # Send test prompt — generate(messages: list[dict], system_prompt: str) + response = await backend.generate( + [{"role": "user", "content": "Reply with 'OK' if you can read this."}], + "", + ) await backend.close() return {"success": True, "response": response} diff --git a/work/meshai/transport/meshcore_transport.py b/work/meshai/transport/meshcore_transport.py index 666c129..bc66346 100644 --- a/work/meshai/transport/meshcore_transport.py +++ b/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) diff --git a/work/tests/test_fix_meshcore_save_and_llm_test.py b/work/tests/test_fix_meshcore_save_and_llm_test.py new file mode 100644 index 0000000..c2c40a8 --- /dev/null +++ b/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" diff --git a/work/tests/test_meshcore_dm_delivery.py b/work/tests/test_meshcore_dm_delivery.py new file mode 100644 index 0000000..5a200c0 --- /dev/null +++ b/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 + ) diff --git a/work/tests/test_meshcore_transport.py b/work/tests/test_meshcore_transport.py index 36b2d2e..e6c4887 100644 --- a/work/tests/test_meshcore_transport.py +++ b/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)