fix(meshcore): ACK-confirmed DM fast path (~2s), discover only on no-ACK

Stop waiting 25s for a PATH_RESPONSE that never arrives. Send the reply
directly, wait ~6s for the delivery ACK the lib exposes; on ACK we're done
(~1-3s, the common case). Only on no-ACK do we run path discovery + resend —
and discovery's wait drops from 25s to a config default (8s). Both timeouts
are config knobs (meshcore_ack_wait_seconds / meshcore_discovery_wait_seconds)
for live tuning. Fixes the real bug behind PR #57 (checked is_error, should
have checked ACK).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

work/meshai/config.py

@@ -50,6 +50,8 @@ class ConnectionConfig:
     meshcore_baud: int = 115200
     meshcore_ble_address: str = ""             # optional; for ble
     meshcore_auto_add_contacts: bool = True     # firmware auto-adds every node it hears an advert from (so AIDA can DM anyone)
+    meshcore_ack_wait_seconds: float = 6.0       # wait for delivery ACK before falling back to path discovery
+    meshcore_discovery_wait_seconds: float = 8.0  # path-discovery timeout on the no-ACK fallback (was hardcoded 25s)
 
     def __post_init__(self):
         if self.meshcore_conn_type not in {"tcp", "serial", "ble"}:

work/meshai/transport/meshcore_transport.py

@@ -76,6 +76,9 @@ class MeshCoreTransport(MeshTransport):
         # or None). None = pass-through (no filtering). Injected at construction
         # time by the factory; can also be (re)set via set_context_config().
         self._mc_context = meshcore_context
+        # DM reply tuning (config knobs; getattr defaults keep old test configs valid).
+        self._ack_wait = float(getattr(config, "meshcore_ack_wait_seconds", 6.0))
+        self._discovery_wait = float(getattr(config, "meshcore_discovery_wait_seconds", 8.0))
         self._mc = None                          # meshcore.MeshCore instance
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._loop_thread: Optional[threading.Thread] = None
@@ -185,8 +188,8 @@ class MeshCoreTransport(MeshTransport):
         # response, so a subsequent send_msg will use the direct route.
         try:
             path_event = self._run_coro(
-                self._mc.commands.send_path_discovery_sync(contact, timeout=25),
-                timeout=30,
+                self._mc.commands.send_path_discovery_sync(contact, timeout=self._discovery_wait),
+                timeout=self._discovery_wait + 3,
             )
             if path_event is not None and not path_event.is_error():
                 logger.info(
@@ -236,6 +239,90 @@ class MeshCoreTransport(MeshTransport):
         except Exception as exc:
             logger.debug("MeshCore: _establish_direct_path step-2 error for %s: %s", dst, exc)
 
+    @staticmethod
+    def _extract_expected_ack(result):
+        """Pull the ``expected_ack`` value off a MSG_SENT event, or None.
+
+        The lib's reader puts the raw 4-byte ack code at
+        ``payload["expected_ack"]`` (see reader.py MSG_SENT branch). Returns it
+        as-is (bytes on the real lib; may be a hex str in tests) or None if the
+        event has no dict payload / no ack.
+        """
+        payload = getattr(result, "payload", None)
+        if isinstance(payload, dict):
+            return payload.get("expected_ack")
+        return None
+
+    def _wait_for_ack(self, expected_ack, timeout: float) -> bool:
+        """Block until the delivery ACK matching *expected_ack* arrives, or timeout.
+
+        Mirrors ``send_msg_with_retry``'s ACK wait (messaging.py): the firmware
+        emits ``EventType.ACK`` with a hex ``code`` attribute equal to the
+        MSG_SENT ``expected_ack`` rendered as ``.hex()`` (reader.py ACK branch).
+        We run the lib's dispatcher wait on the transport loop via ``_run_coro``.
+
+        Returns True only if a matching ACK is dispatched before *timeout*;
+        False when *expected_ack* is falsy (can't confirm) or on any
+        timeout/exception. Coexists with the standing ``_on_ack_event``
+        subscription — the dispatcher supports concurrent subscribers + waiters.
+        """
+        if not expected_ack:
+            return False
+        from meshcore import EventType  # noqa: PLC0415 (lazy import, matches module)
+        try:
+            code = (
+                expected_ack.hex()
+                if isinstance(expected_ack, (bytes, bytearray))
+                else str(expected_ack)
+            )
+        except Exception:
+            return False
+        try:
+            event = self._run_coro(
+                self._mc.dispatcher.wait_for_event(
+                    EventType.ACK,
+                    attribute_filters={"code": code},
+                    timeout=timeout,
+                ),
+                timeout=timeout + 2,
+            )
+            return event is not None
+        except Exception:
+            logger.debug(
+                "MeshCore: ACK wait failed for code %r", expected_ack, exc_info=True
+            )
+            return False
+
+    def _send_dm_once(self, contact: dict, text: str, destination: str):
+        """Send one DM frame (no discovery, no retry) and log its route.
+
+        Uses plain ``send_msg`` — NOT ``send_msg_with_retry``, which calls
+        ``reset_path`` and forces flood, defeating any direct route we hold.
+        Returns the MSG_SENT event, or None if the radio produced no result.
+        """
+        result = self._run_coro(
+            self._mc.commands.send_msg(contact, text),
+            timeout=15,
+        )
+        if result is None:
+            logger.warning("MeshCoreTransport: DM to %s — no send result", destination)
+            return None
+        # Log whether the radio sent DIRECT or FLOOD from the MSG_SENT type field.
+        try:
+            sent_type = (
+                result.payload.get("type")
+                if hasattr(result, "payload") and isinstance(result.payload, dict)
+                else None
+            )
+            logger.info(
+                "MeshCore: DM to %s sent (route=%s)",
+                contact.get("adv_name") or destination,
+                "flood" if sent_type == 1 else ("direct" if sent_type == 0 else "?"),
+            )
+        except Exception:
+            pass
+        return result
+
     # ------------------------------------------------------------------
     # Channel table enumeration
     # ------------------------------------------------------------------
@@ -640,38 +727,40 @@ class MeshCoreTransport(MeshTransport):
                         destination,
                     )
                     return False
-                # Establish a real route so the reply goes DIRECT — flood DMs are
-                # silently rejected on this mesh.
+                label = contact.get("adv_name") or destination
+                # FAST PATH: send the reply DIRECTLY (no discovery) and confirm
+                # by the delivery ACK the firmware returns. This is the common
+                # case (~1-3s) — the firmware already holds a usable route after
+                # the inbound DM primed it. We do NOT pay the path-discovery cost
+                # up front (that 25s wait was the ~25s reply-latency bug).
+                result = self._send_dm_once(contact, text, destination)
+                if result is None:
+                    return False
+                exp_ack = self._extract_expected_ack(result)
+                if self._wait_for_ack(exp_ack, self._ack_wait):
+                    logger.info("MeshCore: DM to %s ACKed (direct)", label)
+                    return True
+                # No ACK -> route is stale/unknown. NOW run path discovery to
+                # learn a direct route, resend, and re-confirm by ACK.
+                logger.info(
+                    "MeshCore: no ACK from %s in %.1fs; running path discovery and retrying",
+                    label, self._ack_wait,
+                )
                 self._establish_direct_path(contact, destination)
                 # Re-resolve so we send with the freshly-learned out_path.
                 contact = self._resolve_contact(destination) or contact
-                # Plain send_msg (NOT send_msg_with_retry — that calls reset_path
-                # which forces flood, defeating the path we just established).
-                result = self._run_coro(
-                    self._mc.commands.send_msg(contact, text),
-                    timeout=15,
-                )
+                result = self._send_dm_once(contact, text, destination)
                 if result is None:
-                    logger.warning("MeshCoreTransport: DM to %s — no send result", destination)
                     return False
-                # Log whether the radio sent DIRECT or FLOOD from RESP_CODE_SENT type field.
-                try:
-                    sent_type = (
-                        result.payload.get("type")
-                        if hasattr(result, "payload") and isinstance(result.payload, dict)
-                        else None
-                    )
-                    logger.info(
-                        "MeshCore: DM to %s sent (route=%s)",
-                        contact.get("adv_name") or destination,
-                        "flood" if sent_type == 1 else ("direct" if sent_type == 0 else "?"),
-                    )
-                except Exception:
-                    pass
-                success = not result.is_error()
-                if not success:
-                    logger.warning("MeshCoreTransport: DM send returned error event")
-                return success
+                exp_ack = self._extract_expected_ack(result)
+                acked = self._wait_for_ack(exp_ack, self._ack_wait)
+                logger.info(
+                    "MeshCore: DM to %s %s after discovery",
+                    contact.get("adv_name") or destination,
+                    "ACKed" if acked else "no ACK (sent best-effort)",
+                )
+                # Best-effort success if the frame was at least accepted by the radio.
+                return acked or (not result.is_error())
             else:
                 # Channel broadcast.
                 # Channel-index semantics do NOT cross transports: the passed

work/tests/test_meshcore_dm_delivery.py

@@ -1,14 +1,18 @@
-"""Focused tests for the MeshCore DM direct-route delivery fix.
+"""Focused tests for the MeshCore DM ACK-confirmed fast-path delivery.
 
 Verifies that send_message(..., destination=...) resolves the destination to
-the full contact object, calls _establish_direct_path (path discovery via
-send_path_discovery_sync) BEFORE send_msg, uses plain send_msg (NOT
+the full contact object, sends the reply DIRECTLY first (fast path) and only
+runs _establish_direct_path (path discovery via send_path_discovery_sync) on
+the no-ACK fallback — never before the first send.  Uses plain send_msg (NOT
 send_msg_with_retry), and correctly maps the return value to True/False:
   - non-error Event returned  → True  (sent, route logged)
   - None returned             → False (no send result)
   - error Event returned      → False
   - contact not in roster     → False (logged warning, send never called)
 
+Note: these mocks provide no dispatcher ACK, so _wait_for_ack returns False and
+every send exercises the no-ACK fallback (discovery + resend) leg.
+
 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.
@@ -195,7 +199,8 @@ def _transport_with_mc_mock(contact=_CONTACT_DICT):
 # ---------------------------------------------------------------------------
 
 class TestMeshCoreDMDelivery:
-    """send_message with destination= must call path discovery then plain send_msg."""
+    """send_message with destination= sends directly first (ACK fast path); path
+    discovery is a no-ACK fallback that runs after the first send, not before."""
 
     def test_successful_dm_returns_true(self):
         """Non-error send_msg result → True."""
@@ -205,8 +210,14 @@ class TestMeshCoreDMDelivery:
 
         assert result is True
 
-    def test_path_discovery_called_before_send_msg(self):
-        """send_path_discovery_sync must be called before send_msg."""
+    def test_fast_path_sends_before_discovery(self):
+        """ACK-confirmed fast path: send_msg goes out FIRST; path discovery only
+        runs on the no-ACK fallback (i.e. AFTER the first send), never before it.
+
+        (This mock has no dispatcher ACK, so _wait_for_ack returns False and the
+        no-ACK fallback always fires — which is exactly what exercises the
+        discovery-then-resend leg here.)
+        """
         t, mc = _transport_with_mc_mock()
         call_order = []
 
@@ -223,10 +234,11 @@ class TestMeshCoreDMDelivery:
 
         t.send_message("hello", destination="aabbccdd1122")
 
-        assert "discovery" in call_order, "send_path_discovery_sync was never called"
         assert "send_msg" in call_order, "send_msg was never called"
-        assert call_order.index("discovery") < call_order.index("send_msg"), (
-            "send_path_discovery_sync must be called before send_msg"
+        assert "discovery" in call_order, "send_path_discovery_sync was never called (no-ACK fallback)"
+        # The reply is sent DIRECTLY first; discovery is a fallback that comes after.
+        assert call_order.index("send_msg") < call_order.index("discovery"), (
+            "fast path must send the reply before running path discovery"
         )
 
     def test_send_msg_used_not_send_msg_with_retry(self):

work/tests/test_meshcore_transport.py

@@ -339,43 +339,82 @@ class TestSendMessageChannel:
 _DM_CONTACT = {"public_key": "a" * 64, "adv_name": "TestContact", "out_path_len": -1}
 
 
+def _msg_sent(expected_ack=b"\x01\x02\x03\x04", type_=0, is_error=False):
+    """Build a fake MSG_SENT event carrying an expected_ack (like reader.py)."""
+    ev = MagicMock()
+    ev.is_error.return_value = is_error
+    ev.payload = {"type": type_, "expected_ack": expected_ack}
+    return ev
+
+
 class TestSendMessageDM:
-    def test_dispatches_send_msg_with_retry(self):
-        """DM send: path discovery then plain send_msg (not send_msg_with_retry)."""
+    def test_fast_path_acked_no_discovery(self):
+        """ACK to the direct send → success WITHOUT path discovery, send_msg once."""
         t, mc, _ = _transport_with_mock_mc()
         try:
-            # Supply a resolved contact so _resolve_contact succeeds.
             mc.get_contact_by_key_prefix.return_value = _DM_CONTACT
             mc.ensure_contacts = AsyncMock(return_value=True)
-            # Path discovery must be awaitable.
-            path_ev = MagicMock()
-            path_ev.is_error.return_value = False
-            mc.commands.send_path_discovery_sync = AsyncMock(return_value=path_ev)
-            ok = MagicMock()
-            ok.is_error.return_value = False
-            ok.payload = {"type": 0, "expected_ack": "00000000"}
-            mc.commands.send_msg = AsyncMock(return_value=ok)
+            mc.commands.send_msg = AsyncMock(return_value=_msg_sent())
+            # Delivery ACK arrives → wait_for_event returns a matching ACK event.
+            ack_ev = MagicMock()
+            mc.dispatcher.wait_for_event = AsyncMock(return_value=ack_ev)
+            # Spy: discovery must NOT run on the fast path.
+            t._establish_direct_path = MagicMock()
+
             result = t.send_message("hi DM", destination="aabbcc")
+
             assert result is True
-            # Must use send_msg (not send_msg_with_retry) with the CONTACT OBJECT.
+            t._establish_direct_path.assert_not_called()
             mc.commands.send_msg.assert_awaited_once_with(_DM_CONTACT, "hi DM")
+            # ACK was matched on the hex of expected_ack (b"\x01\x02\x03\x04").
+            _, kwargs = mc.dispatcher.wait_for_event.await_args
+            assert kwargs["attribute_filters"] == {"code": "01020304"}
         finally:
             _cleanup(t)
 
-    def test_send_msg_with_retry_error_returns_false(self):
-        """Error event from send_msg → False."""
+    def test_no_ack_falls_back_to_discovery_then_acks(self):
+        """No ACK on the direct send → discovery + resend; 2nd ACK → success."""
         t, mc, _ = _transport_with_mock_mc()
         try:
             mc.get_contact_by_key_prefix.return_value = _DM_CONTACT
             mc.ensure_contacts = AsyncMock(return_value=True)
-            path_ev = MagicMock()
-            path_ev.is_error.return_value = False
-            mc.commands.send_path_discovery_sync = AsyncMock(return_value=path_ev)
-            err = MagicMock()
-            err.is_error.return_value = True
-            err.payload = {"reason": "test"}
-            mc.commands.send_msg = AsyncMock(return_value=err)
+            mc.commands.send_msg = AsyncMock(return_value=_msg_sent())
+            ack_ev = MagicMock()
+            # 1st wait times out (None); 2nd (post-discovery) returns an ACK.
+            mc.dispatcher.wait_for_event = AsyncMock(side_effect=[None, ack_ev])
+            t._establish_direct_path = MagicMock()
+
+            result = t.send_message("hi DM", destination="aabbcc")
+
+            assert result is True
+            t._establish_direct_path.assert_called_once()
+            assert mc.commands.send_msg.await_count == 2
+        finally:
+            _cleanup(t)
+
+    def test_wait_for_ack_false_on_none_and_on_exception(self):
+        """_wait_for_ack: falsy exp_ack → False; wait_for_event raising → False."""
+        t, mc, _ = _transport_with_mock_mc()
+        try:
+            # Falsy expected_ack short-circuits without touching the dispatcher.
+            assert t._wait_for_ack(None, 1.0) is False
+            assert t._wait_for_ack(b"", 1.0) is False
+            # A raising wait_for_event is swallowed → False.
+            mc.dispatcher.wait_for_event = AsyncMock(side_effect=RuntimeError("boom"))
+            assert t._wait_for_ack(b"\x01\x02\x03\x04", 1.0) is False
+        finally:
+            _cleanup(t)
+
+    def test_direct_send_no_result_returns_false(self):
+        """send_msg returning None (no radio result) → False, no discovery."""
+        t, mc, _ = _transport_with_mock_mc()
+        try:
+            mc.get_contact_by_key_prefix.return_value = _DM_CONTACT
+            mc.ensure_contacts = AsyncMock(return_value=True)
+            mc.commands.send_msg = AsyncMock(return_value=None)
+            t._establish_direct_path = MagicMock()
             assert t.send_message("hi", destination="deadbeef") is False
+            t._establish_direct_path.assert_not_called()
         finally:
             _cleanup(t)