fix(transport): auto-resync after a read timeout (the "04 BB stale UID" desync)

On the flaky NG link a command's response can arrive *after* its read times out, and the
next command's read then picks up that stale frame — observed in rawcli as every raw 14a
command returning "04 BB" (the truncated UID from _ensure_selected's timed-out scan) instead
of its own response.

read_response/_read_any_response now set a _resync_needed flag on timeout, and the next
send_frame drains any late/stale input before writing. Cheap — only fires after a timeout
(rare on a healthy link) and no-ops when the stream is clean. Hardware-verified: identify ->
NTAG216, READ(4) 6/6 correct, GET_VERSION correct, with the hardening in place.

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

pm3py/core/transport.py

@@ -163,6 +163,10 @@ class PM3Transport:
         self._reader: asyncio.StreamReader | None = None
         self._writer: asyncio.StreamWriter | None = None
         self._lock = asyncio.Lock()
+        # set when a read times out: the command's response may still be in flight and would be
+        # misread as the *next* command's reply (the "04 BB = stale UID" desync). The next send
+        # drains it first. Cheap: only fires after a timeout, which is rare on a healthy link.
+        self._resync_needed = False
 
     async def connect(self) -> None:
         import serial_asyncio
@@ -192,6 +196,9 @@ class PM3Transport:
     async def send_frame(self, frame: bytes) -> None:
         if not self._writer:
             raise PM3Error("Not connected")
+        if self._resync_needed:            # a prior command timed out; clear its late response first
+            await self._drain_input()
+            self._resync_needed = False
         self._writer.write(frame)
         await self._writer.drain()
 
@@ -208,6 +215,7 @@ class PM3Transport:
             try:
                 raw = await asyncio.wait_for(self._read_response_frame(), timeout=timeout)
             except asyncio.TimeoutError:
+                self._resync_needed = True     # late response would desync the next read — drain it
                 raise PM3Error("Response timeout", status=-4)
             try:
                 resp = PM3Response.from_dict(decode_response_frame(raw))
@@ -360,6 +368,7 @@ class PM3Transport:
             try:
                 raw = await asyncio.wait_for(self._read_any_frame(), timeout=timeout)
             except asyncio.TimeoutError:
+                self._resync_needed = True     # late response would desync the next read — drain it
                 raise PM3Error("Response timeout", status=-4)
             magic = struct.unpack_from("<I", raw, 0)[0]
             try:

tests/test_transport.py

@@ -184,3 +184,30 @@ def test_read_response_skips_multiple_debug_then_reply():
     t._read_response_frame = _AM(side_effect=frames)
     resp = _run(t.read_response())
     assert resp.cmd == int(Cmd.PING) and resp.data == b"pong"
+
+
+def test_read_timeout_sets_resync_next_send_drains():
+    # a read timeout flags a resync; the next send drains the late response before writing
+    import asyncio as _a
+    from unittest.mock import AsyncMock as _AM
+    t = PM3Transport("/dev/null")
+    t._reader = object()
+    async def _timeout():
+        raise _a.TimeoutError
+    t._read_response_frame = _timeout
+    try:
+        _run(t.read_response(timeout=0.01))
+    except Exception:
+        pass
+    assert t._resync_needed is True                     # timeout armed a resync
+
+    t._writer = _AM()
+    t._writer.write = lambda f: None
+    t._writer.drain = _AM()
+    drained = []
+    async def _fake_drain(quiet=0.08):
+        drained.append(True)
+        return 0
+    t._drain_input = _fake_drain
+    _run(t.send_frame(b"\x00"))
+    assert drained == [True] and t._resync_needed is False   # drained once, flag cleared