Merge pull request #96 from agessaman/fix/anon-req-reply-path-encoding

Four fixes to anon requests and the BLE transport

src/meshcore/ble_cx.py

@@ -4,6 +4,7 @@ mccli.py : CLI interface to MeschCore BLE companion app
 
 import asyncio
 import logging
+from typing import Optional
 
 
 # Make bleak optional - only fail if BLE operations are attempted
@@ -27,6 +28,11 @@ UART_RX_CHAR_UUID = "6E400002-B5A3-F393-E0A9-E50E24DCCA9E"
 UART_TX_CHAR_UUID = "6E400003-B5A3-F393-E0A9-E50E24DCCA9E"
 
 class BLEConnection:
+    # Upper bound on a single write (lock acquisition included). Healthy writes
+    # measured 0.06-0.17s against real hardware; observed stalls ran 20s to
+    # minutes, so this preempts them rather than waiting for CoreBluetooth.
+    WRITE_TIMEOUT = 10.0
+
     def __init__(self, address=None, device=None, client=None, pin=None):
         """
         Constructor: specify address or an existing BleakClient.
@@ -52,6 +58,27 @@ class BLEConnection:
         self.rx_char = None
         self._disconnect_callback = None
         self._background_tasks: set[asyncio.Task] = set()
+        self._write_lock_obj: Optional[asyncio.Lock] = None
+
+    @property
+    def _write_lock(self) -> asyncio.Lock:
+        """Serialises write_gatt_char().
+
+        Two overlapping writes to the same characteristic drop the link outright
+        (observed on macOS/CoreBluetooth: "BLE write failed: 19", connection
+        gone). Nothing above this layer guarantees callers are sequential --
+        schedulers, health checks and user commands all issue independently --
+        so the transport has to enforce it.
+
+        Lazily created so it binds to the running loop, mirroring the
+        _mesh_request_lock property in commands/base.py. Read through getattr so
+        an instance built without __init__ still works.
+        """
+        lock = getattr(self, "_write_lock_obj", None)
+        if lock is None:
+            lock = asyncio.Lock()
+            self._write_lock_obj = lock
+        return lock
 
     def _spawn_background(self, coro) -> asyncio.Task:
         """Create a tracked background task (prevents GC of fire-and-forget tasks)."""
@@ -190,6 +217,10 @@ class BLEConnection:
         if self.reader is not None:
             self._spawn_background(self.reader.handle_rx(data))
 
+    async def _write_locked(self, data):
+        async with self._write_lock:
+            await self.client.write_gatt_char(self.rx_char, bytes(data), response=True)
+
     async def send(self, data):
         if not self.client:
             logger.error("Client is not connected")
@@ -199,8 +230,24 @@ class BLEConnection:
         if not self.rx_char:
             logger.error("RX characteristic not found")
             return False
+        # Bound the whole acquire-plus-write. A stalled write has been seen to
+        # hang for minutes, and CommandHandler's own timeout does not cover this
+        # -- it starts only after _sender_func returns -- so without a bound the
+        # serialising lock would queue every other command behind the stall
+        # indefinitely, with nothing logged and no disconnect raised. Turning one
+        # hung command into a silent whole-client stall would be worse than the
+        # overlap the lock exists to prevent.
         try:
-            await self.client.write_gatt_char(self.rx_char, bytes(data), response=True)
+            await asyncio.wait_for(self._write_locked(data), timeout=self.WRITE_TIMEOUT)
+        except asyncio.TimeoutError:
+            # Do not simply release and carry on: the underlying write may still
+            # be in flight, and a second write racing it re-creates the exact
+            # overlap that kills the link. Tear the connection down so the
+            # reconnect path takes over -- bounded and self-healing.
+            logger.warning(f"BLE write timed out after {self.WRITE_TIMEOUT}s")
+            if self._disconnect_callback:
+                await self._disconnect_callback("ble_write_timeout")
+            return False
         except Exception as exc:
             logger.warning(f"BLE write failed: {exc}")
             if self._disconnect_callback:

src/meshcore/commands/base.py

@@ -57,6 +57,64 @@ def _validate_destination(dst: DestinationType, prefix_length: int = 6) -> bytes
         )
 
 
+# Size of the server-side reply_path buffer (uint8_t reply_path[64] in
+# simple_repeater/MyMesh.h). It is memcpy'd into without a length check.
+MAX_REPLY_PATH_BYTES = 64
+# reply_path_len is the low 6 bits of the header byte.
+MAX_REPLY_PATH_HOPS = 63
+
+
+def encode_reply_path(out_path_len: int, out_path_hex: str, out_path_hash_mode: int) -> bytes:
+    """Encode the reply path a server should use when answering us.
+
+    The leading byte packs two fields, which the server unpacks as:
+
+        reply_path_len       = byte & 63
+        reply_path_hash_size = (byte >> 6) + 1
+
+    so the hash mode has to travel in the top two bits. Omitting it makes the
+    server read a hash size of 1 regardless of the real mode, take the wrong
+    number of bytes per hop, and route its reply to hops that do not exist.
+
+    The path itself is reversed by *hop*, not by byte: a return path visits the
+    same hops in the opposite order, and each hop's multi-byte hash must stay
+    intact. (For single-byte hops the two are indistinguishable, which is most
+    of why this went unnoticed - mode 0 is the default.)
+    """
+    hash_mode = max(out_path_hash_mode, 0)      # -1 means "flood", i.e. no path
+    if hash_mode > 2:
+        # The server computes hash_size = mode + 1, and Packet::isValidPathLen
+        # rejects 4-byte hops outright, so such a path is unusable on the wire.
+        logger.warning(
+            f"Unsupported out_path_hash_mode {out_path_hash_mode}; "
+            "requesting a zero-hop reply path instead"
+        )
+        return b"\x00"
+    hash_size = hash_mode + 1
+    # Saturate rather than mask: `& 63` would silently wrap a 64-hop path to
+    # zero hops, i.e. a zero-hop reply for a distant node.
+    hops = min(max(out_path_len, 0), MAX_REPLY_PATH_HOPS)
+
+    raw = bytes.fromhex(out_path_hex or "")
+    # Never read past what the contact actually carries; a truncated or padded
+    # field would otherwise yield short trailing hops.
+    hops = min(hops, len(raw) // hash_size)
+    # The server memcpys into a fixed 64-byte reply_path with no bounds check
+    # (simple_repeater/MyMesh.cpp), so never describe more than fits.
+    max_hops = min(MAX_REPLY_PATH_HOPS, MAX_REPLY_PATH_BYTES // hash_size)
+    if hops > max_hops:
+        logger.warning(
+            f"Reply path of {hops} hops x {hash_size}B exceeds the "
+            f"{MAX_REPLY_PATH_BYTES}B the server can hold; truncating to {max_hops}"
+        )
+        hops = max_hops
+
+    path = b"".join(
+        raw[i * hash_size:(i + 1) * hash_size] for i in range(hops - 1, -1, -1)
+    )
+    return bytes([hops | (hash_mode << 6)]) + path
+
+
 class CommandHandlerBase:
     """Base class for command handlers.
 
@@ -299,23 +357,61 @@ class CommandHandlerBase:
         return result
 
     async def send_anon_req(self, dst: DestinationType, request_type: AnonReqType, data: Optional[bytes] = None, context={}, timeout=None, min_timeout=0) -> Event:
+        """Send an anonymous request to *dst*.
+
+        *dst* need not be a known contact. When it is, that contact's out path is
+        used as the reply path; otherwise a zero-hop direct reply path is
+        requested (see the comment below).
+
+        Note: *data* is currently ignored -- the request body is the reply path,
+        which is derived here rather than supplied by the caller.
+        """
         dst_bytes = _validate_destination(dst, prefix_length=32)
         pubkey_prefix = _validate_destination(dst, prefix_length=6)
         logger.debug(f"Anon Binary request to {dst_bytes.hex()}")
 
-        contact = self._get_contact_by_prefix(dst_bytes.hex()) # need a contact for return path
-        if contact is None:
-            logger.error("No contact found")
-            return Event(EventType.ERROR, {"reason": "contact_not_found"})
+        # The contact is consulted only to build the reply path appended to the
+        # request; it is not required to send one. Companion firmware from
+        # FIRMWARE_VER_CODE 13 synthesises a transient anon contact for an unknown
+        # pubkey (out_path_len = 0, zero-hop direct), so an unknown destination is
+        # reachable as long as we ask it to reply zero-hop. Refusing here would
+        # block probing any node the client has not already added - for instance
+        # asking a freshly discovered neighbour for its regions.
+        contact = self._get_contact_by_prefix(dst_bytes.hex())
 
         zero_hop = False
-        if contact["out_path_len"] == -1: 
-            logger.info("No path set trying zero hop")
-            zero_hop = True
-            await self.change_contact_path(contact, "")
+        if contact is None:
+            logger.debug("No contact found, requesting a zero-hop direct reply path")
+            out_path_len = 0
+            reply_path = encode_reply_path(0, "", 0)
+        else:
+            if contact["out_path_len"] == -1:
+                logger.info("No path set trying zero hop")
+                zero_hop = True
+                path_res = await self.change_contact_path(contact, "")
+                if path_res is not None and path_res.type == EventType.ERROR:
+                    # The device still has this contact as flood, so sendAnonReq
+                    # will flood the request -- and the server gates REGIONS,
+                    # OWNER and BASIC behind isRouteDirect(), silently dropping
+                    # it. Better to fail here than to wait out a full timeout
+                    # for a reply that cannot come.
+                    logger.error("Could not set zero-hop path, aborting anon request")
+                    return Event(EventType.ERROR, {"reason": "path_reset_failed"})
+            # update_contact() normally reflects the change back onto the dict, so
+            # out_path_len reads 0 here. Clamp anyway: if that call failed (e.g. the
+            # device query inside it errored) the dict is still -1, and the unsigned
+            # to_bytes below would raise OverflowError -- which would skip the
+            # reset_path at the end of this method and leave the contact pinned to
+            # zero-hop on the device. Zero is the right value to send regardless,
+            # since zero-hop is exactly what we just asked for.
+            out_path_len = max(contact["out_path_len"], 0)
+            reply_path = encode_reply_path(
+                out_path_len,
+                contact["out_path"],
+                contact.get("out_path_hash_mode", 0),
+            )
 
-        data = contact["out_path_len"].to_bytes(1, "little") + bytes.fromhex(contact["out_path"])[::-1]
-        data = b"\x39" + dst_bytes + request_type.value.to_bytes(1, "little", signed=False) + (data if data else b"")
+        data = b"\x39" + dst_bytes + request_type.value.to_bytes(1, "little", signed=False) + reply_path
 
         result = await self.send(data, [EventType.MSG_SENT, EventType.ERROR])
 
@@ -326,7 +422,8 @@ class CommandHandlerBase:
 
             exp_tag = result.payload["expected_ack"].hex()
             # Use provided timeout or fallback to suggested timeout (with 5s default)
-            result.payload["suggested_timeout"] = result.payload.get("suggested_timeout", 4000) * (contact["out_path_len"] + 1) # update timeout from path_len
+            emitted_hops = reply_path[0] & 63   # what actually went on the wire
+            result.payload["suggested_timeout"] = result.payload.get("suggested_timeout", 4000) * (emitted_hops + 1) # update timeout from path_len
             actual_timeout = timeout if timeout is not None and timeout > 0 else result.payload.get("suggested_timeout", 4000) / 800.0
             actual_timeout = min_timeout if actual_timeout < min_timeout else actual_timeout
             self._reader.register_binary_request(pubkey_prefix.hex(), exp_tag, request_type, actual_timeout, context=context, is_anon=True)

src/meshcore/reader.py

@@ -112,7 +112,17 @@ class MessageReader:
                 else:
                     c["out_path_hash_mode"] = plen >> 6 
                     c["out_path_len"] = plen & 0x3F # 6 LSB
-                c["out_path"] = dbuf.read(64).replace(b"\0", b"").hex()
+                # The field is a fixed 64 bytes, NUL-padded past the real path.
+                # Take exactly the bytes the path occupies rather than stripping
+                # NULs: a hop hash may legitimately contain 0x00, and dropping
+                # those shortens the path and shifts every hop after it.
+                # (PATH_DISCOVERY_RESPONSE below already reads opl*opl_hlen.)
+                path_bytes = dbuf.read(64)
+                if c["out_path_len"] > 0:
+                    used = c["out_path_len"] * (c["out_path_hash_mode"] + 1)
+                    c["out_path"] = path_bytes[:used].hex()
+                else:
+                    c["out_path"] = ""
                 c["adv_name"] = dbuf.read(32).decode("utf-8", "ignore").replace("\0", "")
                 c["last_advert"] = int.from_bytes(dbuf.read(4), byteorder="little")
                 c["adv_lat"] = (

tests/unit/test_error_handling.py

@@ -203,21 +203,195 @@ async def test_set_multi_acks_error(
     assert result.is_error()
 
 
-# ── send_anon_req returns ERROR on contact not found ─────────
+# ── send_anon_req falls back to zero-hop when no contact exists ─────────
 
-async def test_send_anon_req_contact_not_found(
-    command_handler, mock_dispatcher
+async def test_send_anon_req_without_contact_sends_zero_hop(
+    command_handler, mock_connection, mock_dispatcher
 ):
-    """send_anon_req returns ERROR event when contact prefix not found,
-    instead of raising TypeError on NoneType subscript."""
+    """An unknown destination is sent with a zero-hop reply path.
+
+    The contact is only used to build the reply path. Companion firmware from
+    FIRMWARE_VER_CODE 13 synthesises a transient anon contact for an unknown
+    pubkey, so refusing to send here would needlessly block probing any node
+    the client has not already added. Must still not raise TypeError on the
+    NoneType subscript that this test originally guarded.
+    """
     command_handler._get_contact_by_prefix = MagicMock(return_value=None)
+    command_handler.change_contact_path = AsyncMock()
+    command_handler.reset_path = AsyncMock()
+    setup_event_response(
+        mock_dispatcher, EventType.MSG_SENT,
+        {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000},
+    )
 
     result = await command_handler.send_anon_req(
         VALID_PUBKEY_HEX, MagicMock(value=1)
     )
 
+    assert not result.is_error()
+    sent = mock_connection.send.await_args.args[0]
+    # \x39 | 32-byte pubkey | request type | reply-path-len 0 (no path bytes)
+    assert sent == b"\x39" + bytes.fromhex(VALID_PUBKEY_HEX) + b"\x01" + b"\x00"
+
+    # No contact to mutate, so no device round-trips for path changes.
+    command_handler.change_contact_path.assert_not_awaited()
+    command_handler.reset_path.assert_not_awaited()
+
+
+async def test_send_anon_req_without_contact_does_not_scale_timeout(
+    command_handler, mock_dispatcher
+):
+    """suggested_timeout must not be multiplied by a path length we don't have."""
+    command_handler._get_contact_by_prefix = MagicMock(return_value=None)
+    reader = MagicMock()
+    reader.register_binary_request = MagicMock()
+    command_handler._reader = reader
+    setup_event_response(
+        mock_dispatcher, EventType.MSG_SENT,
+        {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000},
+    )
+
+    result = await command_handler.send_anon_req(
+        VALID_PUBKEY_HEX, MagicMock(value=1)
+    )
+
+    # out_path_len 0 -> (0 + 1) -> unchanged from the device's own estimate.
+    assert result.payload["suggested_timeout"] == 4000
+    reader.register_binary_request.assert_called_once()
+
+
+async def test_send_anon_req_with_contact_still_uses_its_path(
+    command_handler, mock_connection, mock_dispatcher
+):
+    """A known contact's reply path and timeout scaling are both unchanged.
+
+    _reader must be set: the out_path_len-based timeout scaling is gated behind
+    it, and that scaling is the only behaviour the no-contact refactor touches on
+    this path -- without a reader it is never executed.
+    """
+    command_handler._get_contact_by_prefix = MagicMock(return_value={
+        "public_key": VALID_PUBKEY_HEX,
+        "out_path_len": 3,
+        "out_path": "aabbcc",
+    })
+    command_handler.change_contact_path = AsyncMock()
+    command_handler.reset_path = AsyncMock()
+    reader = MagicMock()
+    command_handler._reader = reader
+    setup_event_response(
+        mock_dispatcher, EventType.MSG_SENT,
+        {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000},
+    )
+
+    result = await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1))
+
+    sent = mock_connection.send.await_args.args[0]
+    # \x39 | pubkey | request type | reply-path-len 3 | path reversed
+    assert sent == (
+        b"\x39" + bytes.fromhex(VALID_PUBKEY_HEX) + b"\x01"
+        + b"\x03" + bytes.fromhex("ccbbaa")
+    )
+    # Scaled by out_path_len + 1 = 4.
+    assert result.payload["suggested_timeout"] == 16000
+    reader.register_binary_request.assert_called_once()
+    assert reader.register_binary_request.call_args.args[3] == 16000 / 800.0
+    command_handler.change_contact_path.assert_not_awaited()
+    command_handler.reset_path.assert_not_awaited()
+
+
+async def test_send_anon_req_timeout_uses_path_len_as_sent(
+    command_handler, mock_connection, mock_dispatcher
+):
+    """The timeout multiplier must match the path length actually transmitted.
+
+    The contact dict is a live reference that other commands mutate in place, so
+    re-reading it after the await could scale the timeout by a path length that
+    was never sent.
+    """
+    contact = {"public_key": VALID_PUBKEY_HEX, "out_path_len": 3, "out_path": "aabbcc"}
+    command_handler._get_contact_by_prefix = MagicMock(return_value=contact)
+    command_handler._reader = MagicMock()
+
+    def fake_subscribe(evt_type, handler, attr_filters=None):
+        sub = MagicMock(spec=Subscription)
+        sub.unsubscribe = MagicMock()
+        if evt_type == EventType.MSG_SENT:
+            # Simulate another command flipping the contact to flood mid-send.
+            contact["out_path_len"] = -1
+            contact["out_path"] = ""
+            asyncio.get_event_loop().call_soon(
+                handler,
+                Event(EventType.MSG_SENT,
+                      {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000}),
+            )
+        return sub
+
+    mock_dispatcher.subscribe = MagicMock(side_effect=fake_subscribe)
+
+    result = await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1))
+
+    sent = mock_connection.send.await_args.args[0]
+    assert sent.endswith(b"\x03" + bytes.fromhex("ccbbaa"))
+    # 4 x 4000, matching the 3-hop path in the frame -- not 0 from the mutated -1.
+    assert result.payload["suggested_timeout"] == 16000
+
+
+async def test_send_anon_req_flood_contact_still_forces_and_restores_zero_hop(
+    command_handler, mock_connection, mock_dispatcher
+):
+    """out_path_len == -1 keeps its existing force-zero-hop-then-restore path."""
+    contact = {
+        "public_key": VALID_PUBKEY_HEX,
+        "out_path_len": -1,
+        "out_path": "",
+    }
+
+    async def fake_change_path(c, path):
+        # update_contact() reflects the change onto the dict; -1 would otherwise
+        # raise OverflowError on the unsigned to_bytes below.
+        c["out_path_len"] = 0
+        c["out_path"] = ""
+
+    command_handler._get_contact_by_prefix = MagicMock(return_value=contact)
+    command_handler.change_contact_path = AsyncMock(side_effect=fake_change_path)
+    command_handler.reset_path = AsyncMock()
+    setup_event_response(
+        mock_dispatcher, EventType.MSG_SENT,
+        {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000},
+    )
+
+    await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1))
+
+    command_handler.change_contact_path.assert_awaited_once()
+    command_handler.reset_path.assert_awaited_once()
+    sent = mock_connection.send.await_args.args[0]
+    assert sent == b"\x39" + bytes.fromhex(VALID_PUBKEY_HEX) + b"\x01" + b"\x00"
+
+
+async def test_send_anon_req_aborts_when_change_contact_path_fails(
+    command_handler, mock_connection, mock_dispatcher
+):
+    """A failed zero-hop switch must abort, not send an unanswerable request.
+
+    If change_contact_path() fails the device still has the contact as flood, so
+    sendAnonReq() floods the request -- and the server gates REGIONS/OWNER/BASIC
+    behind isRouteDirect(), dropping it silently. Sending anyway would burn a
+    full path-scaled timeout waiting for a reply that cannot arrive.
+    """
+    contact = {"public_key": VALID_PUBKEY_HEX, "out_path_len": -1, "out_path": ""}
+    command_handler._get_contact_by_prefix = MagicMock(return_value=contact)
+    command_handler.change_contact_path = AsyncMock(
+        return_value=Event(EventType.ERROR, {"reason": "device_query_failed"})
+    )
+    command_handler.reset_path = AsyncMock()
+
+    result = await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1))
+
     assert result.is_error()
-    assert result.payload["reason"] == "contact_not_found"
+    assert result.payload["reason"] == "path_reset_failed"
+    mock_connection.send.assert_not_awaited()
+    # Nothing was changed on the device, so there is nothing to restore.
+    command_handler.reset_path.assert_not_awaited()
 
 
 # ── send_trace handles unknown path_hash_len without NameError ──
@@ -234,3 +408,309 @@ async def test_send_trace_unknown_path_hash_len(
 
     assert result.is_error()
     assert result.payload["reason"] == "invalid_path_format"
+
+
+# ── BLE transport serialises writes ──────────────────────────
+
+async def test_ble_send_serialises_concurrent_writes():
+    """Overlapping write_gatt_char() calls must not interleave.
+
+    Two concurrent writes to the same characteristic drop the link outright
+    (observed on macOS/CoreBluetooth: "BLE write failed: 19"). Nothing above the
+    transport guarantees callers are sequential, so the transport must.
+    """
+    from meshcore.ble_cx import BLEConnection
+
+    conn = BLEConnection.__new__(BLEConnection)   # bypass bleak availability check
+    conn._disconnect_callback = None
+    conn.rx_char = object()
+
+    overlap = {"current": 0, "max": 0}
+
+    class FakeClient:
+        async def write_gatt_char(self, char, data, response=True):
+            overlap["current"] += 1
+            overlap["max"] = max(overlap["max"], overlap["current"])
+            await asyncio.sleep(0.01)     # a real write is not instantaneous
+            overlap["current"] -= 1
+
+    conn.client = FakeClient()
+
+    await asyncio.gather(*(conn.send(b"\x01\x02") for _ in range(8)))
+
+    assert overlap["max"] == 1, f"{overlap['max']} writes overlapped"
+
+
+async def test_ble_send_releases_lock_on_write_failure():
+    """A failed write must not wedge the lock for every later command."""
+    from meshcore.ble_cx import BLEConnection
+
+    conn = BLEConnection.__new__(BLEConnection)
+    conn.rx_char = object()
+    reasons = []
+
+    async def capture(reason):
+        reasons.append(reason)
+
+    conn._disconnect_callback = capture
+
+    class BoomThenOK:
+        def __init__(self):
+            self.calls = 0
+
+        async def write_gatt_char(self, char, data, response=True):
+            self.calls += 1
+            if self.calls == 1:
+                raise RuntimeError("19")
+
+    conn.client = BoomThenOK()
+
+    await conn.send(b"\x01")
+    assert reasons == ["ble_write_failed: 19"]
+    # Second write must still be able to acquire the lock.
+    await asyncio.wait_for(conn.send(b"\x02"), timeout=1.0)
+    assert conn.client.calls == 2
+
+
+# ── reply-path encoding (hash mode + hop-wise reversal) ──────
+
+from meshcore.commands.base import encode_reply_path  # noqa: E402
+
+
+async def test_encode_reply_path_zero_hop():
+    # len 0, mode 0 -> a single 0x00 byte, the zero-hop direct request.
+    assert encode_reply_path(0, "", 0) == b"\x00"
+
+
+async def test_encode_reply_path_mode0_reverses_hops():
+    # hops aa, bb -> reply visits bb then aa. Mode 0 leaves the top bits clear.
+    assert encode_reply_path(2, "aabb", 0) == b"\x02" + bytes.fromhex("bbaa")
+
+
+async def test_encode_reply_path_carries_hash_mode_in_top_bits():
+    """The server reads hash size from bits 6-7; omitting it truncates each hop.
+
+    Mode 2 = 3 bytes per hop. Without the mode the server would read hash_size 1
+    and reply to two 1-byte hops that do not exist.
+    """
+    out = encode_reply_path(2, "aabbccddeeff", 2)
+    assert out[0] == 0x82                      # 2 hops | (mode 2 << 6)
+    assert out[0] & 63 == 2                    # server: reply_path_len
+    assert (out[0] >> 6) + 1 == 3              # server: reply_path_hash_size
+    # Hop order reversed, each 3-byte hash intact.
+    assert out[1:] == bytes.fromhex("ddeeff") + bytes.fromhex("aabbcc")
+
+
+async def test_encode_reply_path_mode1_two_byte_hops():
+    out = encode_reply_path(3, "aabbccddeeff", 1)
+    assert out[0] == (3 | (1 << 6))
+    assert out[1:] == bytes.fromhex("eeff") + bytes.fromhex("ccdd") + bytes.fromhex("aabb")
+
+
+async def test_encode_reply_path_flood_mode_is_clamped():
+    # out_path_hash_mode is -1 for a flood contact; must not produce a negative shift.
+    assert encode_reply_path(0, "", -1) == b"\x00"
+
+
+async def test_encode_reply_path_ignores_padding_beyond_the_path():
+    # Real device fields are NUL-padded to 64 bytes; only the used hops count.
+    padded = "aabb" + "00" * 60
+    assert encode_reply_path(2, padded, 0) == b"\x02" + bytes.fromhex("bbaa")
+
+
+async def test_encode_reply_path_truncated_field_does_not_emit_short_hops():
+    # Claims 4 hops of 3 bytes but only 6 bytes present -> emit the 2 it has.
+    out = encode_reply_path(4, "aabbccddeeff", 2)
+    assert out[0] & 63 == 2
+    assert out[1:] == bytes.fromhex("ddeeff") + bytes.fromhex("aabbcc")
+
+
+async def test_send_anon_req_reply_path_uses_hash_mode(
+    command_handler, mock_connection, mock_dispatcher
+):
+    """End-to-end: a mode-2 multi-hop contact must get a correct reply path."""
+    command_handler._get_contact_by_prefix = MagicMock(return_value={
+        "public_key": VALID_PUBKEY_HEX,
+        "out_path_len": 2,
+        "out_path": "aabbccddeeff",
+        "out_path_hash_mode": 2,
+    })
+    command_handler.change_contact_path = AsyncMock()
+    command_handler.reset_path = AsyncMock()
+    setup_event_response(
+        mock_dispatcher, EventType.MSG_SENT,
+        {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 4000},
+    )
+
+    await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1))
+
+    sent = mock_connection.send.await_args.args[0]
+    assert sent == (
+        b"\x39" + bytes.fromhex(VALID_PUBKEY_HEX) + b"\x01"
+        + b"\x82" + bytes.fromhex("ddeeff") + bytes.fromhex("aabbcc")
+    )
+
+
+async def test_reader_contact_out_path_keeps_zero_bytes():
+    """A hop hash containing 0x00 must survive parsing by the real reader.
+
+    The 64-byte field is NUL-padded, but stripping every NUL also eats
+    legitimate hash bytes, shortening the path and shifting every hop after it.
+    """
+    from meshcore.reader import MessageReader
+    from meshcore.packets import PacketType
+
+    hops = bytes.fromhex("aa00bb")           # middle byte is a legitimate 0x00
+    frame = (
+        bytes([PacketType.CONTACT.value])
+        + bytes(32)                          # public_key
+        + b"\x02"                            # type
+        + b"\x00"                            # flags
+        + bytes([1 | (2 << 6)])              # 1 hop, hash mode 2 -> 3 bytes/hop
+        + hops + bytes(64 - len(hops))       # out_path, NUL-padded to 64
+        + b"name".ljust(32, b"\0")           # adv_name
+        + (0).to_bytes(4, "little")          # last_advert
+        + (0).to_bytes(4, "little", signed=True)   # adv_lat
+        + (0).to_bytes(4, "little", signed=True)   # adv_lon
+        + (0).to_bytes(4, "little")          # lastmod
+    )
+
+    seen = []
+
+    class Dispatcher:
+        async def dispatch(self, event):
+            seen.append(event)
+
+    reader = MessageReader(Dispatcher())
+    reader.contacts = {}
+    await reader.handle_rx(frame)
+
+    contact = next(e.payload for e in seen if e.type == EventType.NEXT_CONTACT)
+    assert contact["out_path_len"] == 1
+    assert contact["out_path_hash_mode"] == 2
+    assert contact["out_path"] == "aa00bb", "the 0x00 inside the hop hash was lost"
+
+
+# ── BLE write bound (a stalled write must not wedge the client) ──
+
+async def test_ble_write_timeout_tears_down_the_link():
+    """A stalled write must be bounded and must drop the link, not just release.
+
+    CommandHandler's timeout starts only after the write returns, so nothing
+    else bounds this. Releasing the lock alone would be wrong too: the
+    underlying write may still be in flight, and a second write racing it
+    re-creates the overlap the lock exists to prevent.
+    """
+    from meshcore.ble_cx import BLEConnection
+
+    conn = BLEConnection.__new__(BLEConnection)
+    conn.rx_char = object()
+    reasons = []
+
+    async def capture(reason):
+        reasons.append(reason)
+
+    conn._disconnect_callback = capture
+    conn.WRITE_TIMEOUT = 0.05
+
+    class Stalling:
+        async def write_gatt_char(self, char, data, response=True):
+            await asyncio.Event().wait()      # never completes
+
+    conn.client = Stalling()
+
+    result = await asyncio.wait_for(conn.send(b"\x01"), timeout=2.0)
+    assert result is False
+    assert reasons == ["ble_write_timeout"]
+
+
+async def test_ble_stalled_write_does_not_block_later_commands_forever():
+    """Head-of-line: queued writes must not inherit an unbounded stall."""
+    from meshcore.ble_cx import BLEConnection
+
+    conn = BLEConnection.__new__(BLEConnection)
+    conn.rx_char = object()
+    conn._disconnect_callback = None
+    conn.WRITE_TIMEOUT = 0.05
+
+    class Stalling:
+        async def write_gatt_char(self, char, data, response=True):
+            await asyncio.Event().wait()
+
+    conn.client = Stalling()
+
+    # Five writers behind one stalled holder must all return, not hang.
+    done = await asyncio.wait_for(
+        asyncio.gather(*(conn.send(bytes([i])) for i in range(5))), timeout=3.0
+    )
+    assert done == [False] * 5
+
+
+async def test_ble_connection_init_provides_the_lock():
+    """__init__ must set up the lock slot; the tests must not paper over it."""
+    from meshcore.ble_cx import BLEConnection
+
+    conn = BLEConnection.__new__(BLEConnection)
+    BLEConnection.__init__(conn, address="AA:BB:CC:DD:EE:FF")
+    assert conn._write_lock_obj is None          # lazily created
+    assert isinstance(conn._write_lock, asyncio.Lock)
+    assert conn._write_lock is conn._write_lock  # stable across reads
+
+
+# ── reply-path bounds ────────────────────────────────────────
+
+async def test_encode_reply_path_rejects_unsupported_hash_mode():
+    # mode 3 -> 4-byte hops, which Packet::isValidPathLen refuses outright.
+    assert encode_reply_path(2, "aabbccddeeffaabb", 3) == b"\x00"
+
+
+async def test_encode_reply_path_never_overflows_the_server_buffer():
+    """The server memcpys into a fixed 64-byte reply_path with no length check."""
+    for mode in (0, 1, 2):
+        hash_size = mode + 1
+        out = encode_reply_path(63, "ab" * 200, mode)
+        body = out[1:]
+        assert len(body) <= 64, f"mode {mode} emitted {len(body)} bytes"
+        assert (out[0] & 63) * hash_size == len(body), "header disagrees with body"
+
+
+async def test_send_anon_req_timeout_tracks_emitted_hops_not_claimed_length():
+    """Scaling must follow the hops actually sent, not a length that got clamped."""
+    command_handler = CommandHandler()
+    command_handler.dispatcher = MagicMock()
+    command_handler._reader = MagicMock()
+    conn = MagicMock()
+    conn.send = AsyncMock()
+
+    async def sender(data):
+        await conn.send(data)
+
+    command_handler._sender_func = sender
+    # Claims 8 hops but carries only 3 bytes -> 3 hops actually emitted.
+    command_handler._get_contact_by_prefix = MagicMock(return_value={
+        "public_key": VALID_PUBKEY_HEX,
+        "out_path_len": 8,
+        "out_path": "aabbcc",
+        "out_path_hash_mode": 0,
+    })
+    setup_event_response(
+        command_handler.dispatcher, EventType.MSG_SENT,
+        {"expected_ack": b"\x01\x02\x03\x04", "suggested_timeout": 1000},
+    )
+
+    result = await command_handler.send_anon_req(VALID_PUBKEY_HEX, MagicMock(value=1))
+
+    sent = conn.send.await_args.args[0]
+    assert sent[-4] & 63 == 3                     # 3 hops on the wire
+    assert result.payload["suggested_timeout"] == 4000   # (3 + 1) x 1000, not 9x
+
+
+async def test_encode_reply_path_saturates_rather_than_wrapping():
+    """hops is a 6-bit field; masking would wrap 64 to 0.
+
+    A 64-hop path would then be described as zero hops -- a zero-hop reply
+    request for a distant node -- instead of being clamped and logged.
+    """
+    out = encode_reply_path(64, "ab" * 64, 0)
+    assert out[0] & 63 == 63
+    assert len(out[1:]) == 63