tooling(micromate): exercise the event chain, and test the 0x06 count candidate

mm_client_check now drives the step-4 client methods rather than the protocol
layer directly -- list_events(), iter_events(), get_event() and decode_error()
-- so a bench run exercises the code that will actually be used.

Prints each ref with the filename it would be stored under, and reports the PVS
self-check result per event.  The download uses iter_events(), which reproduces
THOR's interleaved order.

Adds the 0x06 test: reads storage range BEFORE the chain walk, prints
content[0:4] as the candidate event count and content[4:8] as the unexplained
companion value, then reports AGREES or DISAGREES against the chain length.
Two samples on one unit said the count is right; a third value from a different
unit either confirms it or kills it, and the tool now answers that in one run.

The 0x06 read is wrapped in the same step() helper as everything else, so a unit
that does not answer it degrades to a FAILED line rather than aborting the run.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ru8Lg9HkkYvX9VWWo65SmL

bridges/mm_client_check.py

@@ -314,25 +314,55 @@ def main() -> int:
         if setups:
             print(f"       first={setups[0]!r}  last={setups[-1]!r}")
 
+        # ── SUB 0x06: is content[0:4] really the event count? ─────────────
+        # Two samples on one unit said yes, and THOR reads it BEFORE the chain
+        # walk then stops without ever reading the sentinel.  A third value
+        # either confirms it or kills it.
+        claimed = None
+        raw06 = step("0x06 storage range", mm.protocol.read_storage_range)
+        if raw06:
+            c = _content(raw06)
+            claimed = int.from_bytes(c[0:4], "big")
+            print(f"       content[0:4] = {claimed}   <- CANDIDATE: event count")
+            print(f"       content[4:8] = {int.from_bytes(c[4:8],'big')}   "
+                  f"<- unexplained (read 9 alongside a 6 on UM12947)")
+
         if a.download:
-            print("\n  event chain (read-only):")
-            proto = mm.protocol
-            proto.arm_event()
-            hdr = _content(proto.read_event_first())
-            key, size = hdr[0:4], int.from_bytes(hdr[4:8], "big")
-            if not size:
-                print("       no events stored")
-            else:
-                rec = _content(proto.read_event_record(key))
-                print(f"       first event key={key.hex()} size={size} B "
-                      f"type={_event_type(rec)}")
+            print("\n  event chain (read-only), via MicromateClient:")
             t1 = time.monotonic()
-                blob = proto.read_event_file(key, size)
-                dt = time.monotonic() - t1
-                print(f"       downloaded {len(blob)} B in {dt:.1f} s "
-                      f"({len(blob)/dt/1024:.1f} KiB/s)")
-                assert len(blob) == size
-                _decode(blob, key, rec)
+            refs = mm.list_events()                 # walks to the sentinel
+            walk = time.monotonic() - t1
+            print(f"       {len(refs)} events in {walk:.1f} s "
+                  f"({walk/max(len(refs),1):.2f} s each, 3 round trips per event)")
+
+            if claimed is not None:
+                verdict = ("✓ AGREES" if claimed == len(refs)
+                           else f"✗ DISAGREES (0x06 said {claimed})")
+                print(f"       0x06 count vs chain length: {verdict}")
+
+            for ref in refs:
+                print(f"       {ref}")
+                print(f"         would be filed as {ref.filename}")
+
+            if refs:
+                # Download via iter_events, which reproduces THOR's interleaved
+                # order -- the one with captures behind it.
+                print("\n  download (first event, THOR's interleaved order):")
+                for ref in mm.iter_events():
+                    t1 = time.monotonic()
+                    try:
+                        result = mm.get_event(ref)       # verify=True
+                    except Exception as e:
+                        print(f"       {ref.key_hex}: {type(e).__name__}: {e}")
+                        break
+                    dt = max(time.monotonic() - t1, 1e-6)
+                    n = sum(len(v) for v in getattr(result, "samples", {}).values())
+                    err = mm.decode_error(ref, result)
+                    check = ("PVS agrees to %+.4f%%" % (100 * err)
+                             if err is not None else "PVS check n/a (histogram)")
+                    print(f"       {ref.key_hex} {ref.record_type:9} {ref.size:6} B "
+                          f"in {dt:.1f} s -> {n} samples, {check}")
+                    break
 
     except ProtocolError as e:
         print(f"\n  ABORTED  {type(e).__name__}: {e}")

micromate/client.py

@@ -32,11 +32,14 @@ from __future__ import annotations
 
 import datetime
 import logging
+import math
+import os
+import struct
 from typing import Optional
 
 from minimateplus.transport import BaseTransport
 
-from .models import MicromateDeviceInfo, MicromateState
+from .models import MicromateDeviceInfo, MicromateEventRef, MicromateState
 from .protocol import MicromateProtocol, ProtocolError
 
 log = logging.getLogger(__name__)
@@ -57,6 +60,27 @@ _MS_BATTERY = slice(34, 36)     # uint16 BE, volts × 100
 _MS_MEM_TOTAL = slice(36, 40)   # uint32 BE
 _MS_MEM_FREE = slice(40, 44)    # uint32 BE
 
+# `SUB 0x0C` record, relative to content.  Established against 7 events across
+# both firmware lines.
+_REC_DAY, _REC_MONTH, _REC_YEAR = 0, 1, slice(2, 4)
+_REC_UNKNOWN_4 = 4              # ⚠ same shape as 0x1C's content[6]; undecoded
+_REC_HOUR, _REC_MIN, _REC_SEC = 5, 6, 7
+_REC_TYPE = 11                  # 0x07 waveform, 0x08 histogram
+_REC_LOCATION = 12
+_REC_SETUP = 34
+_REC_SERIAL = 76
+_RECORD_TYPES = {0x07: "waveform", 0x08: "histogram"}
+
+# The channel labels the record carries, in the order they appear.  The peak
+# float sits `label + 6`; the peak vector sum sits 12 bytes BEFORE "Tran".
+_REC_CHANNELS = (b"Tran", b"Vert", b"Long", b"Mic")
+_REC_PVS_BACK = 12
+
+# A chain walk terminates on an all-zero key.  The ceilings below are guards
+# against a device cursor that never advances, not fleet limits.
+_NULL_KEY = bytes(4)
+_MAX_EVENTS = 4096
+
 # A setup-list walk that does not terminate is a bug, not a big fleet.  The
 # bench unit holds 22 setups; this is a generous ceiling, not a limit.
 _MAX_SETUPS = 512
@@ -71,6 +95,10 @@ def _cstring(buf: bytes, offset: int = 0) -> str:
     return buf[offset:].split(b"\x00")[0].decode("ascii", "replace").strip()
 
 
+class DecodeMismatch(ProtocolError):
+    """Our decoded peak disagrees with the one the device computed itself."""
+
+
 class MicromateClient:
     """High-level read-only client for one Micromate.
 
@@ -88,6 +116,7 @@ class MicromateClient:
             transport, recv_timeout=recv_timeout, strict_checksums=strict_checksums
         )
         self._firmware_line: Optional[str] = None
+        self._serial: Optional[str] = None
 
     # ── Lifecycle ─────────────────────────────────────────────────────────────
 
@@ -140,6 +169,7 @@ class MicromateClient:
 
         manufacturer, model = self._parse_poll(poll.data)
         serial = _cstring(_content(self._proto.read_serial()))
+        self._serial = serial
         monitoring = self._parse_state(self._proto.read_state())
 
         info = MicromateDeviceInfo(
@@ -293,3 +323,246 @@ class MicromateClient:
             f"setup list did not terminate after {_MAX_SETUPS} entries — the "
             f"device cursor is not advancing"
         )
+
+    # ── Events ────────────────────────────────────────────────────────────────
+
+    def serial(self) -> str:
+        """The unit's serial, cached from `connect()` or read on demand.
+
+        Needed by anything that handles an event, because an event key is
+        ambiguous without it — see `MicromateEventRef`.
+        """
+        if self._serial is None:
+            self._serial = _cstring(_content(self._proto.read_serial()))
+        return self._serial
+
+    def list_events(self, *, with_records: bool = True) -> list[MicromateEventRef]:
+        """Walk the event chain.  `0x93 → 1E`, then `0x93 → 1F` until the sentinel.
+
+        THOR sends `0x93` before **every** chain read, and this mirrors that.
+        The chain ends on an all-zero key.
+
+        ⚠ `with_records=True` costs **one extra round trip per event** for the
+        `0x0C` read, and over cellular a round trip is ~0.65 s regardless of
+        size.  On a unit with 40 events that is the difference between ~52 s and
+        ~78 s.  Pass False when you only need "what is here and how big" — but
+        note the **record is where the type and timestamp live**, so without it
+        `ref.filename` is None and `get_event()` cannot pick a suffix.
+
+        ⚠ **To download, prefer `iter_events()`.**  This walks the whole chain
+        first; THOR interleaves, downloading each event before advancing.
+        `0x5A` addresses an event by key, so downloading afterwards *should*
+        work — but "should" is doing real work in that sentence, and the
+        interleaved order is the one with captures behind it.  See
+        `iter_events()`.
+        """
+        serial = self.serial()
+        refs: list[MicromateEventRef] = []
+
+        for i in range(_MAX_EVENTS):
+            self._proto.arm_event()
+            raw = (self._proto.read_event_first() if i == 0
+                   else self._proto.read_event_next())
+            c = _content(raw)
+            if len(c) < 8:
+                raise ProtocolError(
+                    f"chain entry {i}: {len(c)} B of content, need 8 (key + size)"
+                )
+            key, size = c[0:4], int.from_bytes(c[4:8], "big")
+            if key == _NULL_KEY:
+                return refs                       # the sentinel, not an error
+
+            ref = MicromateEventRef(index=i, key=key, size=size, serial=serial)
+            if with_records:
+                self._read_record_into(ref)
+            refs.append(ref)
+
+        raise ProtocolError(
+            f"event chain did not terminate after {_MAX_EVENTS} entries — the "
+            f"device cursor is not advancing"
+        )
+
+    def iter_events(self, *, with_records: bool = True):
+        """Walk the chain, yielding each event **at the cursor position THOR uses.**
+
+            for ref in mm.iter_events():
+                if ref.is_histogram:
+                    continue
+                data = mm.download_event(ref)      # ← safe here
+
+        Why this exists alongside `list_events()`: THOR's captured order is
+
+            0x93 → 1E → 0C → 5A×n → 0x93 → 1F → 0C → 5A×n → …
+
+        — it downloads each event *before* advancing the chain.  `list_events()`
+        walks to the end first, which is fine for browsing (the browse walk is
+        separately attested) but means a later download happens with the device
+        cursor parked past the event.  `0x5A` is key-addressed, so it very
+        probably does not care; nothing observed says it does, and nothing
+        observed says it does not.
+
+        Downloading inside this loop reproduces THOR's sequence exactly, so it
+        is the path to use when it matters.  ⚠ Do not advance the generator
+        before finishing with the event it yielded.
+        """
+        serial = self.serial()
+        for i in range(_MAX_EVENTS):
+            self._proto.arm_event()
+            raw = (self._proto.read_event_first() if i == 0
+                   else self._proto.read_event_next())
+            c = _content(raw)
+            if len(c) < 8:
+                raise ProtocolError(
+                    f"chain entry {i}: {len(c)} B of content, need 8 (key + size)"
+                )
+            key, size = c[0:4], int.from_bytes(c[4:8], "big")
+            if key == _NULL_KEY:
+                return
+            ref = MicromateEventRef(index=i, key=key, size=size, serial=serial)
+            if with_records:
+                self._read_record_into(ref)
+            yield ref
+
+        raise ProtocolError(
+            f"event chain did not terminate after {_MAX_EVENTS} entries — the "
+            f"device cursor is not advancing"
+        )
+
+    def _read_record_into(self, ref: MicromateEventRef) -> None:
+        """`SUB 0x0C` — 210 B of content: timestamp, type, names, peaks."""
+        raw = self._proto.read_event_record(ref.key)
+        c = _content(raw)
+        if len(c) <= _REC_SERIAL:
+            raise ProtocolError(f"event record is {len(c)} B, too short to decode")
+        ref.raw_record = raw
+
+        ref.record_type = _RECORD_TYPES.get(c[_REC_TYPE])
+        if ref.record_type is None:
+            # Worth saying out loud rather than filing the event as a waveform:
+            # the suffix decides which codec runs.
+            log.warning("event %s: unknown record type 0x%02x at content[%d]",
+                        ref.key_hex, c[_REC_TYPE], _REC_TYPE)
+
+        try:
+            ref.timestamp = datetime.datetime(
+                year=int.from_bytes(c[_REC_YEAR], "big"),
+                month=c[_REC_MONTH], day=c[_REC_DAY],
+                hour=c[_REC_HOUR], minute=c[_REC_MIN], second=c[_REC_SEC],
+            )
+        except ValueError as e:
+            log.warning("event %s: bad timestamp (%s): %s",
+                        ref.key_hex, e, c[:8].hex(" "))
+
+        ref.sensor_location = _cstring(c, _REC_LOCATION) or None
+        ref.setup = _cstring(c, _REC_SETUP) or None
+        # Prefer the record's own serial over the cached one — they have always
+        # agreed, but the record is the event's own account of where it came from.
+        if rec_serial := _cstring(c, _REC_SERIAL):
+            ref.serial = rec_serial
+
+        peaks: dict[str, float] = {}
+        for label in _REC_CHANNELS:
+            i = c.find(label)
+            if i < 0 or i + len(label) + 10 > len(c):
+                continue
+            peaks[label.decode()] = struct.unpack(
+                ">f", c[i + len(label) + 2: i + len(label) + 6])[0]
+        ref.peaks_ips = peaks or None
+
+        tran = c.find(b"Tran")
+        if tran >= _REC_PVS_BACK:
+            ref.peak_vector_sum_ips = struct.unpack(
+                ">f", c[tran - _REC_PVS_BACK: tran - _REC_PVS_BACK + 4])[0]
+
+    def download_event(self, ref: MicromateEventRef) -> bytes:
+        """The raw `.IDFW`/`.IDFH` bytes, exactly as THOR would have stored them.
+
+        Feeds `micromate.idf_file.read_idf_file()` and `/db/import/idf_file`
+        unchanged — no new codec work is needed for a directly downloaded event.
+        """
+        return self._proto.read_event_file(ref.key, ref.size)
+
+    def get_event(self, ref: MicromateEventRef, *, verify: bool = True,
+                  tolerance: float = 0.01):
+        """Download and decode one event.
+
+        Returns the codec's `IdfReadResult`.  Needs `ref.record_type`, since
+        `read_idf_file()` dispatches on the filename suffix and there is no
+        filename on the wire — so call `list_events(with_records=True)` first.
+
+        ⚠ `verify=True` re-computes the **peak vector sum** from the decoded
+        samples and compares it against the float the *device* put in the `0x0C`
+        record.  Those are two independent computations over the same samples —
+        the device's from its own firmware, ours from our codec — so a
+        disagreement means our decode is wrong.  Measured agreement on the bench
+        events is **0.000%**.
+
+        This is cheap insurance in a codebase whose decode failures have
+        historically been *silent*: unhandled block tags shorten a channel and
+        nothing raises.  ⚠ It is a decode-correctness check, **not** a
+        truncation detector — a channel cut after its peak still yields the
+        right PVS.
+
+        Histograms are not verified: `samples` is empty for them.
+        """
+        if not ref.record_type:
+            raise ValueError(
+                f"event {ref.key_hex}: record_type is unknown, so the codec "
+                f"cannot be dispatched.  Use list_events(with_records=True)."
+            )
+        blob = self.download_event(ref)
+
+        import tempfile
+        from .idf_file import read_idf_file
+
+        # read_idf_file dispatches on the suffix, so the bytes need a name.
+        with tempfile.NamedTemporaryFile(suffix=ref.suffix, delete=False) as f:
+            f.write(blob)
+            tmp = f.name
+        try:
+            result = read_idf_file(tmp)
+        finally:
+            os.unlink(tmp)
+
+        if verify and not ref.is_histogram:
+            err = self.decode_error(ref, result)
+            if err is not None and abs(err) > tolerance:
+                raise DecodeMismatch(
+                    f"event {ref.uid}: decoded peak vector sum differs from the "
+                    f"device's own by {100 * err:+.3f}% (tolerance "
+                    f"{100 * tolerance:.1f}%) — the decode is suspect, not the "
+                    f"device.  Stored {ref.peak_vector_sum_ips:.5f} in/s."
+                )
+            if err is not None:
+                log.debug("event %s: PVS agrees to %+.4f%%", ref.uid, 100 * err)
+        return result
+
+    @staticmethod
+    def decode_error(ref: MicromateEventRef, result) -> Optional[float]:
+        """Relative error between our decoded PVS and the device's stored one.
+
+        None when either side is unavailable.  Positive means the device's
+        figure is higher than ours.
+        """
+        from .idf_file import geo_count_to_ips
+
+        stored = ref.peak_vector_sum_ips
+        if not stored or not getattr(result, "samples", None):
+            return None
+
+        ch = {k.lower(): v for k, v in result.samples.items()}
+        try:
+            t, v, l = ch["tran"], ch["vert"], ch["long"]
+        except KeyError:
+            return None
+        n = min(len(t), len(v), len(l))
+        if not n:
+            return None
+
+        pvs = max(
+            math.sqrt(geo_count_to_ips(t[i]) ** 2
+                      + geo_count_to_ips(v[i]) ** 2
+                      + geo_count_to_ips(l[i]) ** 2)
+            for i in range(n)
+        )
+        return (stored - pvs) / pvs if pvs else None

micromate/models.py

@@ -478,3 +478,76 @@ class MicromateState:
         if frac is not None:
             bits.append(f"memory {frac * 100:.1f}% used")
         return "  ".join(bits)
+
+
+@dataclass
+class MicromateEventRef:
+    """One entry in a unit's event chain, from `1E`/`1F` and optionally `0x0C`.
+
+    ⚠ **`key` is NOT unique across units.**  The event counter starts from the
+    same value on every Micromate — UM12947 and UM20147 both have an event
+    `055d4a81`, with different sizes and different contents.  Use `uid`, or key
+    on `(serial, key_hex)`, for anything that stores or deduplicates.  A store
+    keyed on the event key alone silently treats one unit's event as a duplicate
+    of another's, and nothing raises.
+    """
+
+    index: int
+    key: bytes                     # 4-byte event key from the chain walk
+    size: int                      # bytes the device will send for this event
+    serial: Optional[str] = None   # the unit, because `key` alone is ambiguous
+
+    # From `SUB 0x0C` — one extra round trip per event, so optional.
+    record_type: Optional[str] = None            # "waveform" | "histogram"
+    timestamp: Optional[datetime.datetime] = None
+    setup: Optional[str] = None                  # setup file name, no extension
+    sensor_location: Optional[str] = None
+    peak_vector_sum_ips: Optional[float] = None  # per-sample PVS, device-computed
+    peaks_ips: Optional[Dict[str, float]] = None # {"Tran": …, "Vert": …, …}
+    raw_record: Optional[bytes] = field(default=None, repr=False)
+
+    @property
+    def key_hex(self) -> str:
+        return self.key.hex()
+
+    @property
+    def uid(self) -> str:
+        """`SERIAL:key` — safe to use as a primary key.  See the class note."""
+        return f"{self.serial or '?'}:{self.key_hex}"
+
+    @property
+    def is_histogram(self) -> Optional[bool]:
+        if self.record_type is None:
+            return None
+        return self.record_type == "histogram"
+
+    @property
+    def suffix(self) -> Optional[str]:
+        return {"waveform": ".IDFW", "histogram": ".IDFH"}.get(self.record_type or "")
+
+    @property
+    def filename(self) -> Optional[str]:
+        """The name THOR would have given this event.
+
+        `<serial>_<YYYYMMDDHHMMSS>.IDF{W,H}` — e.g.
+        `UM12947_20260923163319.IDFW`.  Verified against the production store
+        for all five bench events.
+
+        ⚠ The type comes from the **protocol**, not the payload, so it has to be
+        carried here from the `0x0C` read.  Returns None without it: guessing
+        the suffix would file a histogram as a waveform, and `read_idf_file()`
+        dispatches on exactly that.
+        """
+        if not (self.serial and self.timestamp and self.suffix):
+            return None
+        return f"{self.serial}_{self.timestamp:%Y%m%d%H%M%S}{self.suffix}"
+
+    def __str__(self) -> str:
+        bits = [self.uid, f"{self.size} B"]
+        if self.record_type:
+            bits.append(self.record_type)
+        if self.timestamp:
+            bits.append(self.timestamp.strftime("%Y-%m-%d %H:%M:%S"))
+        if self.peak_vector_sum_ips is not None:
+            bits.append(f"PVS {self.peak_vector_sum_ips:.4f} in/s")
+        return "  ".join(bits)

tests/test_micromate_events.py

@@ -0,0 +1,343 @@
+"""Event-chain tests for the Micromate (series-4) client.
+
+The load-bearing test here replays THOR's captured six-event download session
+through `MicromateClient.iter_events()` + `get_event()` and asserts **every byte
+we put on the wire matches what THOR put on the wire** — 99 frames — while also
+decoding all six events and cross-checking each waveform's peak vector sum
+against the one the device computed itself.
+
+That capture is gitignored, so those tests skip on a fresh clone; the offline
+tests below use embedded real response bytes and cover the same logic.
+"""
+from __future__ import annotations
+
+import datetime
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from micromate.client import CONTENT, DecodeMismatch, MicromateClient
+from micromate.framing import ACK, DLE, ETX, STX, checksum, stuff
+from micromate.models import MicromateEventRef
+from micromate.protocol import ProtocolError
+
+from test_micromate_client import ScriptedTransport, SERIAL, frame
+
+# The real 0x0C record from UM20147 (11.0BD), captured over USB 2026-09-30.
+# A histogram: content[11] = 0x08.
+RECORD_BD = bytes.fromhex(
+    "d200000000055d4a8100001e0907eab30d1b21000000084c6f636174696f6e00"
+    "0000000000000000000000000074657374320000000000000000000000000000"
+    "0000000000000000000000000000000000000000000000554d32303134370000"
+    "003fcb3bde00000000053f000f5472616e00003e698cdb000300005665727400"
+    "003fc76e65000300004c6f6e6700003e567c21000300004d69630000003956b9"
+    "7c00050000000000000000000000000000000000000000000000000000000000"
+    "0000000000000000000000000000000000000000000000000000000000"
+)
+
+
+def chain_entry(key: bytes, size: int) -> bytes:
+    """A 1E/1F response data section: 11-byte prefix then key + size."""
+    return (bytes([0x08]) + bytes(7) + bytes([0xFE]) + bytes(2)
+            + key + size.to_bytes(4, "big"))
+
+
+def client(responses):
+    t = ScriptedTransport(responses)
+    return MicromateClient(t, recv_timeout=0.5), t
+
+
+# ── The chain walk ────────────────────────────────────────────────────────────
+
+def test_chain_walk_arms_before_every_entry():
+    """⚠ THOR sends 0x93 before EVERY 1E/1F, and this mirrors that."""
+    responses = [
+        frame(0xEA, SERIAL),                                    # serial()
+        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4076)),
+        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes.fromhex("055d4a82"), 11032)),
+        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),   # sentinel
+    ]
+    mm, t = client(responses)
+    refs = mm.list_events(with_records=False)
+
+    subs = [w[5] for w in t.written]
+    assert subs == [0x15, 0x93, 0x1E, 0x93, 0x1F, 0x93, 0x1F]
+    assert [r.key_hex for r in refs] == ["055d4a81", "055d4a82"]
+    assert [r.size for r in refs] == [4076, 11032]
+
+
+def test_an_all_zero_key_ends_the_chain_and_is_not_an_error():
+    mm, _ = client([
+        frame(0xEA, SERIAL),
+        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes(4), 0)),
+    ])
+    assert mm.list_events(with_records=False) == []
+
+
+def test_refs_carry_the_serial_because_a_key_alone_is_ambiguous():
+    """⚠ UM12947 and UM20147 BOTH have an event 055d4a81.
+
+    A store keyed on the event key alone treats one unit's event as a duplicate
+    of the other's, and nothing raises.  `uid` is the safe identifier.
+    """
+    mm, _ = client([
+        frame(0xEA, SERIAL),
+        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4076)),
+        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
+    ])
+    (ref,) = mm.list_events(with_records=False)
+    assert ref.serial == "UM12947"
+    assert ref.uid == "UM12947:055d4a81"
+
+
+def test_a_cursor_that_never_advances_raises_rather_than_hanging():
+    from micromate import client as C
+    responses = [frame(0xEA, SERIAL)]
+    entry = chain_entry(bytes.fromhex("055d4a81"), 4076)
+    responses += [frame(0x6C, bytes(11)), frame(0xE1, entry)]      # the 1E read
+    for _ in range(C._MAX_EVENTS + 2):                             # then 1F forever
+        responses += [frame(0x6C, bytes(11)), frame(0xE0, entry)]
+    mm, _ = client(responses)
+    with pytest.raises(ProtocolError, match="not advancing"):
+        mm.list_events(with_records=False)
+
+
+def test_a_truncated_chain_entry_raises():
+    mm, _ = client([
+        frame(0xEA, SERIAL), frame(0x6C, bytes(11)), frame(0xE1, bytes(14)),
+    ])
+    with pytest.raises(ProtocolError, match="need 8"):
+        mm.list_events(with_records=False)
+
+
+def test_iter_events_does_not_read_ahead():
+    """It must yield at the cursor position, one arm/advance per event."""
+    mm, t = client([
+        frame(0xEA, SERIAL),
+        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4076)),
+        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
+    ])
+    it = mm.iter_events(with_records=False)
+    first = next(it)
+    assert [w[5] for w in t.written] == [0x15, 0x93, 0x1E], "no read-ahead"
+    assert first.key_hex == "055d4a81"
+    assert list(it) == []
+
+
+# ── The 0x0C record ───────────────────────────────────────────────────────────
+
+def test_record_decode_on_real_bd_bytes():
+    mm, _ = client([
+        frame(0xEA, SERIAL),
+        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4796)),
+        frame(0xF3, RECORD_BD),
+        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
+    ])
+    (ref,) = mm.list_events()
+
+    assert ref.record_type == "histogram"      # content[11] = 0x08
+    assert ref.is_histogram is True
+    assert ref.suffix == ".IDFH"
+    assert ref.timestamp == datetime.datetime(2026, 9, 30, 13, 27, 33)
+    assert ref.sensor_location == "Location"
+    assert ref.setup == "test2"
+    assert ref.serial == "UM20147", "the record's own serial wins over the cache"
+    assert ref.peak_vector_sum_ips == pytest.approx(1.587765, abs=1e-5)
+    assert ref.peaks_ips["Tran"] == pytest.approx(0.228076, abs=1e-5)
+    assert ref.peaks_ips["Vert"] == pytest.approx(1.558056, abs=1e-5)
+    assert ref.peaks_ips["Long"] == pytest.approx(0.209458, abs=1e-5)
+
+
+def test_the_pvs_is_not_reconstructible_from_the_reported_peaks():
+    """⚠ Why the PVS field matters: it cannot be recomputed from the peaks.
+
+    It is the PER-SAMPLE peak vector sum.  `sqrt(Σpeak²)` is only an upper
+    bound, because the channel maxima do not occur at the same instant, and
+    `max(T,V,L)` is a lower bound.  On THIS event the two happen to be within
+    0.05%, which is exactly the coincidence that made the field look like
+    `sqrt(Σpeak²)` on first inspection.  Six other events separated them.
+    """
+    import math
+    mm, _ = client([
+        frame(0xEA, SERIAL),
+        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 4796)),
+        frame(0xF3, RECORD_BD),
+        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
+    ])
+    (ref,) = mm.list_events()
+    p = ref.peaks_ips
+    upper = math.sqrt(p["Tran"] ** 2 + p["Vert"] ** 2 + p["Long"] ** 2)
+    lower = max(p["Tran"], p["Vert"], p["Long"])
+    assert lower < ref.peak_vector_sum_ips < upper
+
+
+def test_filename_matches_thors_convention():
+    ref = MicromateEventRef(index=0, key=bytes.fromhex("055d4a82"), size=11032,
+                            serial="UM12947", record_type="waveform",
+                            timestamp=datetime.datetime(2026, 9, 23, 16, 33, 19))
+    assert ref.filename == "UM12947_20260923163319.IDFW"
+    ref.record_type = "histogram"
+    assert ref.filename == "UM12947_20260923163319.IDFH"
+
+
+def test_filename_is_none_without_a_type_rather_than_guessing():
+    """⚠ Guessing would file a histogram as a waveform, and read_idf_file()
+    dispatches on exactly that suffix."""
+    ref = MicromateEventRef(index=0, key=bytes(4), size=1, serial="UM12947",
+                            timestamp=datetime.datetime(2026, 1, 1))
+    assert ref.record_type is None
+    assert ref.suffix is None
+    assert ref.filename is None
+
+
+def test_an_unknown_record_type_warns_and_leaves_it_none(caplog):
+    bad = bytearray(RECORD_BD)
+    bad[CONTENT + 11] = 0x99
+    mm, _ = client([
+        frame(0xEA, SERIAL),
+        frame(0x6C, bytes(11)), frame(0xE1, chain_entry(bytes.fromhex("055d4a81"), 10)),
+        frame(0xF3, bytes(bad)),
+        frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0)),
+    ])
+    with caplog.at_level("WARNING"):
+        (ref,) = mm.list_events()
+    assert ref.record_type is None
+    assert "unknown record type 0x99" in caplog.text
+
+
+def test_get_event_refuses_without_a_record_type():
+    mm, _ = client([])
+    ref = MicromateEventRef(index=0, key=bytes.fromhex("055d4a81"), size=4076)
+    with pytest.raises(ValueError, match="record_type is unknown"):
+        mm.get_event(ref)
+
+
+# ── Against the real captured session ─────────────────────────────────────────
+
+_CAPTURES = (
+    Path(__file__).resolve().parents[1]
+    / "bridges" / "captures" / "9-24-26 - micromate2"
+)
+_DOWNLOAD = "raw_bw_20260925_011403_Download_events_then_delete_1_event.bin"
+
+
+def _destuffed(blob: bytes, is_req: bool):
+    i, n = 0, len(blob)
+    while i < n:
+        if is_req:
+            if not (blob[i] == ACK and i + 1 < n and blob[i + 1] == STX):
+                i += 1
+                continue
+            j = i + 2
+        else:
+            if blob[i] != STX:
+                i += 1
+                continue
+            j = i + 1
+        out = bytearray()
+        while j < n:
+            if blob[j] == DLE and j + 1 < n:
+                out.append(blob[j + 1])
+                j += 2
+                continue
+            if blob[j] == ETX:
+                break
+            out.append(blob[j])
+            j += 1
+        if len(out) >= 6:
+            yield blob[i:j + 1], bytes(out[:-1])
+        i = j + 1
+
+
+@pytest.mark.skipif(
+    not (_CAPTURES / _DOWNLOAD).is_file(),
+    reason="capture is gitignored; present only on a dev box",
+)
+def test_replaying_thors_session_reproduces_every_byte_and_decodes_every_event():
+    """The whole point of step 4.
+
+    Feed THOR's own responses to `iter_events()` + `get_event()`, and assert:
+      * every request byte we emit matches THOR's, in order
+      * all six events decode
+      * each waveform's decoded PVS matches the device's stored float
+    """
+    reqs = list(_destuffed((_CAPTURES / _DOWNLOAD).read_bytes(), True))
+    rsps = list(_destuffed(
+        (_CAPTURES / _DOWNLOAD.replace("raw_bw", "raw_s3")).read_bytes(), False))
+
+    # THOR's session opens with commands our client does not send (POLL, 0x1C,
+    # 0x06 …) and ends with a delete.  Take the contiguous run from the first
+    # 0x93 to the last 0x5A -- that is the event walk, and it is what we mirror.
+    subs = [p[2] for _, p in reqs]
+    lo = subs.index(0x93)
+    hi = len(subs) - 1 - subs[::-1].index(0x5A)
+    want_wire = [w for w, _ in reqs[lo:hi + 1]]
+    replay = [bytes([STX]) + stuff(p + bytes([checksum(p)])) + bytes([ETX])
+              for _, p in rsps[lo:hi + 1]]
+
+    # ⚠ THOR never reads the chain sentinel in this capture -- it downloaded
+    # exactly six events and stopped, so it knew the count in advance.  It read
+    # `SUB 0x06` (storage range) at frame 7, BEFORE the walk, and that response
+    # begins `00 00 00 06` -- the event count.  Our generator walks until the
+    # sentinel instead, so append one and compare only the overlapping frames.
+    replay += [frame(0x6C, bytes(11)), frame(0xE0, chain_entry(bytes(4), 0))]
+
+    # serial() would emit a 0x15 that is not in this slice, so prime the cache.
+    t = ScriptedTransport(replay)
+    mm = MicromateClient(t, recv_timeout=1.0)
+    mm._serial = "UM12947"
+
+    decoded, checked = 0, 0
+    for ref in mm.iter_events():
+        result = mm.get_event(ref)          # verify=True by default
+        decoded += 1
+        assert ref.serial == "UM12947"
+        assert ref.filename and ref.filename.endswith(ref.suffix)
+        if not ref.is_histogram:
+            err = mm.decode_error(ref, result)
+            assert err is not None
+            assert abs(err) < 1e-4, f"{ref.uid}: PVS off by {100 * err:+.4f}%"
+            checked += 1
+
+    assert decoded == 6, "four waveforms and two histograms"
+    assert checked == 4
+    # Our trailing sentinel read is two frames THOR did not send; everything
+    # up to it must match byte for byte, in order.
+    assert t.written[:len(want_wire)] == want_wire, (
+        f"we emitted {len(t.written)} frames, THOR emitted {len(want_wire)}"
+    )
+    assert len(t.written) == len(want_wire) + 2, "only the sentinel read is extra"
+
+
+@pytest.mark.skipif(
+    not (_CAPTURES / _DOWNLOAD).is_file(),
+    reason="capture is gitignored; present only on a dev box",
+)
+def test_a_corrupted_stored_peak_is_caught_by_the_self_check():
+    """Prove the verify path actually fires -- otherwise it is decoration."""
+    reqs = list(_destuffed((_CAPTURES / _DOWNLOAD).read_bytes(), True))
+    rsps = list(_destuffed(
+        (_CAPTURES / _DOWNLOAD.replace("raw_bw", "raw_s3")).read_bytes(), False))
+    subs = [p[2] for _, p in reqs]
+    lo = subs.index(0x93)
+    hi = len(subs) - 1 - subs[::-1].index(0x5A)
+    replay = [bytes([STX]) + stuff(p + bytes([checksum(p)])) + bytes([ETX])
+              for _, p in rsps[lo:hi + 1]]
+
+    t = ScriptedTransport(replay)
+    mm = MicromateClient(t, recv_timeout=1.0)
+    mm._serial = "UM12947"
+
+    for ref in mm.iter_events():
+        if ref.is_histogram:
+            mm.download_event(ref)          # keep the replay in step
+            continue
+        ref.peak_vector_sum_ips *= 1.5      # as a bad decode would look
+        with pytest.raises(DecodeMismatch, match="decode is suspect"):
+            mm.get_event(ref)
+        return
+    pytest.fail("no waveform event found in the capture")