fix(transport): skip async debug frames — the real cause of garbage LF reads

Debug-heavy firmware emits DEBUG_PRINT_STRING/INTEGERS/BYTES (0x0100-0x0102) frames
asynchronously, independent of any command. pm3py's read_response returned the FIRST
frame it saw, so a debug line landing mid-command was misparsed as the reply — observed
on hardware as garbage LF sample counts (e.g. 1593903105), a config-SET that "timed
out", and desynced BigBuf downloads (reading firmware memory / the 0xDEADBEEF stack
canary). The C client routes debug frames to its logger and keeps reading for the real
reply; pm3py did not.

read_response and _read_any_response now loop past DEBUG_PRINT_* (and malformed) frames
up to a bound, returning the actual command reply — for both NG replies and the OLD-frame
bulk-download stream. This is what makes LF read/download reliable on this device.

Tests: single + multiple interleaved debug frames are skipped to reach the real reply.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

pm3py/core/transport.py

@@ -14,6 +14,14 @@ from .protocol import (
     OLD_FRAME_FMT, OLD_FRAME_HDR_SIZE, OLD_FRAME_SIZE,
 )
 
+# Frames the device emits asynchronously (debug output), independent of any command. They can
+# arrive in the middle of a command's response stream on debug-heavy firmware, so the response
+# readers skip them to return the real reply (mirrors the C client routing debug to its logger).
+_ASYNC_FRAME_CMDS = frozenset({
+    Cmd.DEBUG_PRINT_STRING, Cmd.DEBUG_PRINT_INTEGERS, Cmd.DEBUG_PRINT_BYTES,
+})
+_MAX_ASYNC_SKIP = 64        # bound the skip loop so a chatty/looping device can't hang a read
+
 
 def _crc_to_wire(crc: int) -> int:
     """Byte-swap CRC to match PM3 wire format: (low << 8) | high."""
@@ -188,19 +196,27 @@ class PM3Transport:
         await self._writer.drain()
 
     async def read_response(self, timeout: float = 2.0) -> PM3Response:
-        """Read and decode one response frame from the device."""
+        """Read one command reply, skipping async debug-print frames the device interleaves.
+
+        Debug-heavy firmware emits DEBUG_PRINT_* frames asynchronously, which can land in the
+        middle of a command's response stream. The C client routes those to its logger and keeps
+        reading for the real reply; pm3py must do the same, or it returns a debug line misparsed
+        as the reply — the cause of garbage sample counts and desynced downloads."""
         if not self._reader:
             raise PM3Error("Not connected")
-
+        for _ in range(_MAX_ASYNC_SKIP):
             try:
-            # Read until we find response magic
-            raw = await asyncio.wait_for(
-                self._read_response_frame(), timeout=timeout,
-            )
-            d = decode_response_frame(raw)
-            return PM3Response.from_dict(d)
+                raw = await asyncio.wait_for(self._read_response_frame(), timeout=timeout)
             except asyncio.TimeoutError:
                 raise PM3Error("Response timeout", status=-4)
+            try:
+                resp = PM3Response.from_dict(decode_response_frame(raw))
+            except ValueError:
+                continue                               # malformed frame — resync on the next magic
+            if resp.cmd in _ASYNC_FRAME_CMDS:
+                continue                               # device debug output — not our reply
+            return resp
+        raise PM3Error("Too many async frames without a reply", status=-4)
 
     async def _read_response_frame(self) -> bytes:
         """Read a complete response frame, scanning for magic bytes."""
@@ -337,19 +353,28 @@ class PM3Transport:
             return bytes(buf[:count]) if count else bytes(buf)
 
     async def _read_any_response(self, timeout: float = 5.0) -> PM3Response:
-        """Read one frame that may be NG or OLD and return a PM3Response."""
+        """Read one NG-or-OLD frame, skipping async debug-print frames (as read_response does)."""
         if not self._reader:
             raise PM3Error("Not connected")
+        for _ in range(_MAX_ASYNC_SKIP):
             try:
                 raw = await asyncio.wait_for(self._read_any_frame(), timeout=timeout)
             except asyncio.TimeoutError:
                 raise PM3Error("Response timeout", status=-4)
             magic = struct.unpack_from("<I", raw, 0)[0]
+            try:
                 if magic == RESP_PREAMBLE_MAGIC:
-            return PM3Response.from_dict(decode_response_frame(raw))
+                    resp = PM3Response.from_dict(decode_response_frame(raw))
+                else:
                     d = decode_old_frame(raw)
-        return PM3Response(cmd=d["cmd"], status=0, reason=0, ng=False,
+                    resp = PM3Response(cmd=d["cmd"], status=0, reason=0, ng=False,
                                        data=d["data"], oldarg=d["oldarg"])
+            except ValueError:
+                continue
+            if resp.cmd in _ASYNC_FRAME_CMDS:
+                continue                               # device debug output — not our reply
+            return resp
+        raise PM3Error("Too many async frames without a reply", status=-4)
 
     async def _read_any_frame(self) -> bytes:
         """Read a complete frame, branching NG vs OLD on the leading 4-byte magic.

tests/test_transport.py

@@ -151,3 +151,36 @@ def test_download_bulk_drains_stale_input_first():
     t.send_frame = tracking_send
     _run(t.download_bulk(Cmd.DOWNLOAD_BIGBUF, 0, 100, Cmd.DOWNLOADED_BIGBUF, Cmd.ACK))
     assert calls[0] == "drain" and calls[1] == "send"   # drained BEFORE the request went out
+
+
+def _resp_frame(cmd, data=b"", status=0):
+    # build a no-CRC NG response frame the way the device sends one
+    length_ng = len(data) | 0x8000
+    pre = struct.pack("<IHbbH", RESP_PREAMBLE_MAGIC, length_ng, status, 0, cmd)
+    return pre + data + struct.pack("<H", RESP_POSTAMBLE_NOCRC)
+
+
+def test_read_response_skips_async_debug_frames():
+    # a debug-print frame interleaved before the real reply must be skipped, not misread as data
+    from unittest.mock import AsyncMock as _AM
+    t = PM3Transport("/dev/null")
+    t._reader = object()                               # truthy so read_response proceeds
+    debug = _resp_frame(int(Cmd.DEBUG_PRINT_STRING), b"LF Sampling config q=95")
+    reply = _resp_frame(int(Cmd.LF_ACQ_RAW_ADC), struct.pack("<I", 12288))
+    t._read_response_frame = _AM(side_effect=[debug, reply])
+    resp = _run(t.read_response())
+    assert resp.cmd == int(Cmd.LF_ACQ_RAW_ADC)
+    assert struct.unpack_from("<I", resp.data, 0)[0] == 12288   # the real sample count, not debug bytes
+
+
+def test_read_response_skips_multiple_debug_then_reply():
+    from unittest.mock import AsyncMock as _AM
+    t = PM3Transport("/dev/null")
+    t._reader = object()
+    frames = [_resp_frame(int(Cmd.DEBUG_PRINT_STRING), b"a"),
+              _resp_frame(int(Cmd.DEBUG_PRINT_INTEGERS), struct.pack("<I", 999)),
+              _resp_frame(int(Cmd.DEBUG_PRINT_BYTES), b"\x01\x02"),
+              _resp_frame(int(Cmd.PING), b"pong")]
+    t._read_response_frame = _AM(side_effect=frames)
+    resp = _run(t.read_response())
+    assert resp.cmd == int(Cmd.PING) and resp.data == b"pong"