pureboot: the info block is what proves a knock landed

A prompt byte alone does not: one left over from a previous session can
still be in the pipeline while the port opening resets the device into a
fresh window, where the bare command that follows is discarded. Each
attempt is now the whole handshake, retried until the block comes back.

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

pureboot/pureboot.py

@@ -364,26 +364,37 @@ class Loader:
         self.info = None
 
     def connect(self, wait):
-        """Knock until the window answers, then read the info block. Also
-        converges into a live session: the knock bytes are ignored there and
-        the drain absorbs whatever they produced."""
-        self.port.flush_input()
+        """Knock until the info block comes back. The block is what proves the
+        loader is listening — a prompt byte alone does not, since one left over
+        from a previous session can still be in the pipeline while the port
+        opening resets the device into a fresh activation window, where a
+        command without its knock is discarded. Each attempt is therefore the
+        whole handshake, retried until it produces the block or the window
+        closes. Also converges into a live session: the knock bytes are ignored
+        there and the drain absorbs whatever they produced."""
         deadline = time.monotonic() + wait
         knocks = 0
         while True:
+            self.port.flush_input()
             self.port.write(b"pb")
             knocks += 1
             if PROMPT in self.port.read_available(0.4):
-                break
-            if time.monotonic() > deadline:
-                raise Error("no answer — reset the device within its activation window")
                 while self.port.read_available(0.3):
                     pass
                 self.port.write(b"b")
-        self.info = Info(self.port.read_exact(12, 2.0))
+                try:
+                    block = self.port.read_exact(12, 2.0)
+                except Error:
+                    block = b""
+                # A version the tool cannot speak is the loader's own answer,
+                # not a failed knock: Info reports it rather than retrying.
+                if block[0:2] == b"PB":
+                    self.info = Info(block)
                     self._expect_prompt()
                     verbose(f"loader answered knock {knocks}; info block read")
                     return self.info
+            if time.monotonic() > deadline:
+                raise Error("no answer — reset the device within its activation window")
 
     def _expect_prompt(self, timeout=2.0):
         byte = self.port.read_exact(1, timeout)

test/test_planner.py

@@ -280,6 +280,68 @@ def main():
     if device.writes != pb.RETRIES + 1:
         fail(f"unrepairable page took {device.writes} writes, expected {pb.RETRIES + 1}")
 
+    # The knock handshake against a device that is not listening yet — the
+    # state a port open leaves behind: it resets the chip into a fresh
+    # activation window while the previous session's prompt is still in
+    # flight, so the first knock is lost and a prompt arrives anyway.
+    class FakePort:
+        """A loader in its activation window, plus `lost` leading writes the
+        reset swallows and one stale prompt still on the wire."""
+
+        def __init__(self, info_raw, lost=0, stale=b"", active=False):
+            self.info_raw = info_raw
+            self.lost = lost
+            self.inflight = bytearray(stale)
+            self.rx = bytearray()
+            self.active = active
+            self.last = None
+
+        def flush_input(self):
+            self.rx.clear()
+
+        def write(self, data):
+            if self.lost:
+                self.lost -= 1
+                return
+            for byte in bytes(data):
+                if not self.active:
+                    self.active = self.last == ord("p") and byte == ord("b")
+                    self.last = byte
+                    if self.active:
+                        self.rx += pb.PROMPT
+                elif byte == ord("b"):
+                    self.rx += self.info_raw + pb.PROMPT
+                else:
+                    self.rx += pb.PROMPT
+
+        def read_available(self, wait):
+            self.rx = self.inflight + self.rx  # the stale prompt lands late
+            self.inflight.clear()
+            out, self.rx = bytes(self.rx), bytearray()
+            return out
+
+        def read_exact(self, count, timeout):
+            if len(self.rx) < count:
+                raise pb.Error(f"timeout: got {len(self.rx)} of {count} bytes")
+            out, self.rx = bytes(self.rx[:count]), self.rx[count:]
+            return out
+
+    raw = info_of(pb, 0x7E00, 128, False, 0x8000).raw
+    for what, port in (
+        ("clean window", FakePort(raw)),
+        ("stale prompt over a lost knock", FakePort(raw, lost=1, stale=pb.PROMPT)),
+        ("live session", FakePort(raw, active=True)),
+    ):
+        info = pb.Loader(port).connect(5)
+        if info.raw != raw:
+            fail(f"connect ({what}) returned {info.raw.hex()}")
+
+    # A device that never answers still says so, and a version the tool cannot
+    # speak is reported as such rather than retried into a timeout.
+    expect_error("dead device", lambda: pb.Loader(FakePort(raw, lost=99)).connect(0), "no answer")
+    old = bytes(raw[:2]) + bytes((pb.NEWEST_LOADER + 1,)) + bytes(raw[3:])
+    expect_error("unspeakable version", lambda: pb.Loader(FakePort(old)).connect(5), "needs a newer tool")
+
     print("test_planner: all planner and policy checks pass")