fix(stream): payload-length cap, quiet-port resync, main-thread error teardown

src/pygui/backend/controllers/session_controller.py

@@ -21,6 +21,7 @@ from pygui.backend.models.sensor_editor import SensorEditor
 from pygui.backend.models.session_model import SessionModel, SessionUpdate
 from pygui.backend.models.threshold_classifier import ThresholdConfig
 from pygui.backend.utils import slot_error_boundary
+from pygui.backend.wafer.family_spec import sensor_count_for
 from pygui.backend.wafer.wafer_layouts import ec_pairs_for_wafer_id
 from pygui.backend.wafer.zwafer_models import Sensor
 from pygui.backend.wafer.zwafer_parser import ZWaferParser
@@ -43,6 +44,7 @@ class SessionController(QObject):
     loadFileError = Signal(str)
     clusterAveragingEnabledChanged = Signal()
     liveStatsChanged = Signal()  # emitted when receivedCount or errorCount change
+    liveError = Signal(str)  # fatal live-stream error message, for Activity Log
     # trend: per-frame avg for live graph
     trendDelta = Signal(str)  # JSON [[elapsed_s, avg]] — one new point per live frame
     trendReset = Signal()  # live trend buffer cleared (new stream started)
@@ -53,7 +55,7 @@ class SessionController(QObject):
     segmentExported = Signal(dict)  # dict with success, path | error
     # private: marshal a worker-thread frame onto the main thread
     _liveFrame = Signal(object)    # Frame
-    _liveError = Signal()          # worker-thread error ping
+    _liveError = Signal(str)       # worker-thread error ping, carries exception text
 
     def __init__(self, parent: QObject | None = None) -> None:
         super().__init__(parent)
@@ -700,9 +702,16 @@ class SessionController(QObject):
         if not self._active_clusters:
             self._active_clusters = group_sensors_by_radius(self._sensors)
 
-        # The new binary protocol sends payload of (sensorCount * 2) bytes
+        # The binary protocol's payload carries a fixed-size channel array
-        # Each sensor is a big-endian 16-bit value (Sign-Magnitude).
+        # (sensorCount * 2 bytes, big-endian Sign-Magnitude each) that's wider
-        expected_sensors = 80 if (family_code or "") == "X" else 244
+        # than any one wafer family's real sensor count — trailing channels
+        # are unpopulated hardware slots, not real readings. Bound decoding to
+        # the just-loaded layout's actual sensor count so those float/noise
+        # channels never reach values (and therefore never reach MIN/MAX/AVG
+        # in compute_stats, which has no sensor-count knowledge of its own).
+        # Fall back to the family YAML lookup only if no layout loaded yet
+        # (e.g. startStream called with an unknown family_code).
+        expected_sensors = len(self._sensors) or sensor_count_for(family_code or "")
 
         def parse_binary_frame(payload: bytes, seq: int) -> Frame:
             values = []
@@ -744,8 +753,11 @@ class SessionController(QObject):
             log.error("Live stream error: %s", exc)
             ref = weak_self()
             if ref is not None:
-                ref._liveError.emit()
+                # Marshal onto the main thread via the queued signal — this
-                ref.stopStream()
+                # callback runs on StreamReader's own worker thread, and
+                # stopStream() joins that same thread, so calling it directly
+                # here would deadlock/raise ("cannot join current thread").
+                ref._liveError.emit(str(exc))
 
         def on_frame(frame: Frame):
             ref = weak_self()
@@ -815,10 +827,15 @@ class SessionController(QObject):
         self.frameUpdated.emit()
         self.stateChanged.emit()
 
-    def _on_live_error(self) -> None:
+    def _on_live_error(self, message: str) -> None:
-        """Main-thread callback for worker-thread stream errors."""
+        """Main-thread callback for worker-thread stream errors. StreamReader
+        only calls its on_error for resync exhaustion or an unhandled parse
+        exception — both unrecoverable for the current stream — so this ends
+        the session and surfaces the reason (Activity Log wiring in QML)."""
         self._error_count += 1
         self.liveStatsChanged.emit()
+        self.liveError.emit(message)
+        self.stopStream()
 
     # ---- recording ----
     @Slot(str, str)

src/pygui/backend/license/license_model.py

@@ -207,3 +207,12 @@ class LicenseModel(QObject):
             return True
         age_days = (time.time() - marker.stat().st_mtime) / 86400.0
         return age_days > TRIAL_DAYS
+
+    @Slot(result=bool)
+    def reviewAccessBlocked(self) -> bool:
+        """True only once a started trial has actually expired (ADR-0005).
+
+        A "none" state (trial never started) is not a block — nothing to
+        enforce yet. "permanent" (a real license) never blocks.
+        """
+        return self.replayLicenseState() == "temporary" and self.replayTrialExpired()

src/pygui/serialcomm/stream_reader.py

@@ -15,8 +15,30 @@ log = logging.getLogger(__name__)
 ParseFrame = Callable[[bytes, int], Frame]
 
 # Max consecutive resync attempts before giving up on a corrupt stream.
+# Counted only when data WAS read and still no sync marker (pure garbage);
+# an idle port never counts — C#'s findMessageStart waits forever on quiet.
 _MAX_RESYNC_ATTEMPTS = 50
 
+# Largest believable payload_len. Real frames are ≤ 488 bytes (244 words);
+# anything bigger means we latched onto a false 0xAA 0x88 inside data and
+# read garbage as a length — without this cap, _accumulate would swallow up
+# to 64KB of good frames waiting for a packet that doesn't exist.
+# ponytail: loose cap; exact per-family length check if false frames slip CRC.
+_MAX_PAYLOAD_LEN = 4096
+
+
+def _crc16(data: bytes) -> int:
+    """CRC16/MODBUS (poly 0xA001, init 0xFFFF) — mirrors C# CalculateCRC16."""
+    crc = 0xFFFF
+    for byte in data:
+        crc ^= byte
+        for _ in range(8):
+            if crc & 1:
+                crc = (crc >> 1) ^ 0xA001
+            else:
+                crc >>= 1
+    return crc
+
 class StreamReader:
 
     def __init__(self, transport, parse_frame: ParseFrame,
@@ -72,6 +94,8 @@ class StreamReader:
                 peek_buf.extend(b)
                 if b[0] == 10:  # newline
                     break
+                if len(peek_buf) == 2 and peek_buf[0] == 0xAA and peek_buf[1] == 0x88:
+                    break  # binary sync marker — decisive, stop peeking
 
             if self._stop.is_set() or not peek_buf:
                 return
@@ -95,11 +119,17 @@ class StreamReader:
             # Default: ASCII hex dump stream.
             self._run_ascii(raw)
         except Exception as exc:
-            log.exception("StreamReader thread encountered error: %s", exc)
+            if self._stop.is_set():
-            try:
+                # stop() closes the transport to unblock a pending read —
-                self._on_error(exc)
+                # that self-inflicted close raises here too (e.g. EBADF).
-            except Exception:
+                # Not a real stream fault, so don't report it as one.
-                pass
+                log.debug("StreamReader thread exiting after stop(): %s", exc)
+            else:
+                log.exception("StreamReader thread encountered error: %s", exc)
+                try:
+                    self._on_error(exc)
+                except Exception:
+                    pass
         finally:
             try:
                 self._transport.close()
@@ -133,6 +163,12 @@ class StreamReader:
                     return
 
                 payload_len = int.from_bytes(buf[5:7], byteorder='little')
+                if payload_len > _MAX_PAYLOAD_LEN:
+                    # False sync marker inside data — the "length" is garbage.
+                    # Skip past the marker and rescan what we already have.
+                    self.resync_count += 1
+                    buf = buf[2:]
+                    continue
                 total_packet_len = 2 + 5 + payload_len + 2
 
                 # Accumulate the rest of the packet in one buffered read.
@@ -144,6 +180,22 @@ class StreamReader:
                 if self._stop.is_set():
                     return
 
+                # CRC16 over sync+header+payload, trailing 2 bytes LE —
+                # mirrors C# VerifyData. A failed CRC is a corrupt or
+                # misframed packet: drop it, rescan, keep streaming.
+                received_crc = int.from_bytes(
+                    buf[7 + payload_len:9 + payload_len], byteorder='little'
+                )
+                if _crc16(bytes(buf[:7 + payload_len])) != received_crc:
+                    self.error_count += 1
+                    if self.error_count <= 10:
+                        log.warning(
+                            "CRC mismatch on binary packet (seq bytes %s) — dropped",
+                            bytes(buf[3:5]).hex(),
+                        )
+                    buf = buf[2:]
+                    continue
+
                 header = buf[2:7]
                 seq = int.from_bytes(header[1:3], byteorder='little')
                 payload = buf[7:7 + payload_len]
@@ -160,6 +212,17 @@ class StreamReader:
                 buf = buf[total_packet_len:]
             else:
                 # Resync: keep only trailing 0xAA if present, then read more.
+                if len(buf) > 0 and buf[-1] == 0xAA:
+                    buf = bytearray([0xAA])
+                else:
+                    buf.clear()
+
+                chunk = self._read_chunk()
+                if not chunk:
+                    # Quiet port — waiting for data is not a resync failure.
+                    # The device pauses between scans; C#'s findMessageStart
+                    # waits forever. Only actual garbage counts toward the cap.
+                    continue
                 resync_attempts += 1
                 self.resync_count += 1
                 if resync_attempts > _MAX_RESYNC_ATTEMPTS:
@@ -172,12 +235,6 @@ class StreamReader:
                         TimeoutError("Binary stream resync failed: no sync marker")
                     )
                     return
-                if len(buf) > 0 and buf[-1] == 0xAA:
-                    buf = bytearray([0xAA])
-                else:
-                    buf.clear()
-
-                chunk = self._read_chunk()
                 buf.extend(chunk)
 
     def _run_ascii(self, initial_bytes: bytes) -> None:
@@ -276,6 +333,22 @@ class StreamReader:
 
     def stop(self) -> None:
         self._stop.set()
+        # Close the port out from under any blocked transport.read() call —
+        # relying on _stop alone means a thread stuck in a longer-than-
+        # expected read (or one the OS driver is slow to unblock) keeps the
+        # port open, and a subsequent startStream() then opens a *second*
+        # handle to the same port, racing the still-live old reader for
+        # bytes (garbled/duplicated frames on the next Live session).
+        try:
+            self._transport.close()
+        except Exception:
+            pass
         if self._thread is not None and self._thread.is_alive():
             self._thread.join(timeout = 2.0)
+            if self._thread.is_alive():
+                log.error(
+                    "StreamReader: worker thread still alive %.1fs after "
+                    "stop() — port may not be released for the next session",
+                    2.0,
+                )
             self._thread = None

tests/test_session_controller.py

@@ -96,6 +96,45 @@ def test_leaving_live_mode_stops_stream(controller):
     assert not controller._repaint_timer.isActive()
 
 
+def test_binary_frame_decode_bounded_to_loaded_layout(controller, monkeypatch):
+    """Regression for phantom sensor #239 polluting MIN/MAX/AVG: the wire
+    payload carries a fixed-width channel array (up to 244 slots) wider than
+    any single wafer family's real sensor count. Family "A" (aepwafer.yaml)
+    has 48 real sensors — everything past that in the payload is an
+    unpopulated hardware slot and must never reach Frame.values."""
+    fake_transport = MagicMock()
+    fake_transport.read.return_value = b""
+    monkeypatch.setattr(
+        "pygui.serialcomm.serial_port.SerialPort.open_port",
+        lambda *a, **k: fake_transport,
+    )
+
+    controller.startStream("COM_FAKE", "A")
+    try:
+        assert len(controller._sensors) == 48  # sanity: aepwafer.yaml loaded
+        parse = controller._reader._parse
+        payload = bytes(244 * 2)  # full-width wire payload, well beyond the real 48
+        frame = parse(payload, seq=1)
+        assert len(frame.values) == 48
+    finally:
+        controller.stopStream()
+
+
+def test_live_error_surfaces_message_and_stops_stream(controller):
+    """A fatal StreamReader error must reach QML with its message (for the
+    Activity Log) and must tear the stream down cleanly."""
+    controller.setMode("live")
+    controller._reader = MagicMock()
+
+    error_seen = MagicMock()
+    controller.liveError.connect(error_seen)
+
+    controller._on_live_error("Binary stream resync failed: no sync marker")
+
+    error_seen.assert_called_once_with("Binary stream resync failed: no sync marker")
+    assert controller._reader is None
+
+
 def test_compare_files_integration(controller):
     # Mock the comparisonResult signal
     mock_slot = MagicMock()

tests/test_stream_reader.py

@@ -1,7 +1,7 @@
 import time
 
 from pygui.backend.models.frame import Frame
-from pygui.serialcomm.stream_reader import StreamReader
+from pygui.serialcomm.stream_reader import StreamReader, _crc16
 
 
 class FakeTransport:
@@ -47,3 +47,121 @@ def test_reads_frames_and_counts_errors():
     assert [f.values for f in got] == [[149.0, 148.0], [150.0, 149.0]]
     assert r.error_count == 1
     assert transport.closed
+
+
+class WedgedTransport:
+    """Simulates a worker thread stuck in a blocking read() that ignores
+    the stop event — read() never returns on its own. stop() must close
+    the port to unblock it rather than relying on the read loop to notice."""
+    def __init__(self):
+        self.closed = False
+
+    def read(self, n=1):
+        while not self.closed:
+            time.sleep(0.001)
+        raise OSError("port closed")
+
+    def close(self):
+        self.closed = True
+
+
+def test_stop_closes_transport_even_if_worker_thread_is_wedged():
+    r = StreamReader(WedgedTransport(), parse_line,
+                    on_frame=lambda f: None, on_error=lambda e: None)
+    r.start()
+    time.sleep(0.05)  # let the thread enter the blocking read
+    r.stop()
+    assert r._transport.closed, (
+        "stop() must close the transport itself — otherwise a wedged "
+        "worker thread keeps the port open and a subsequent startStream() "
+        "opens a second competing handle to the same port"
+    )
+
+
+def make_packet(payload: bytes, seq: int = 1) -> bytes:
+    """Build a wire packet exactly as the device does: AA 88, msg type,
+    seq (LE), payload len (LE), payload, CRC16/MODBUS (LE) over the rest."""
+    body = (b"\xaa\x88\x01" + seq.to_bytes(2, "little")
+            + len(payload).to_bytes(2, "little") + payload)
+    return body + _crc16(body).to_bytes(2, "little")
+
+
+def parse_payload(payload: bytes, seq: int) -> Frame:
+    return Frame(seq=seq, time=0.0, values=[float(b) for b in payload])
+
+
+def test_crc16_known_vector():
+    # CRC16/MODBUS check value for "123456789" is 0x4B37.
+    assert _crc16(b"123456789") == 0x4B37
+
+
+def test_binary_stream_parses_crc_valid_packets():
+    got = []
+    data = make_packet(b"\x01\x02", seq=1) + make_packet(b"\x03\x04", seq=2)
+    r = StreamReader(FakeTransport([data]), parse_payload,
+                    on_frame=got.append, on_error=lambda e: None)
+    r.start()
+    time.sleep(0.2)
+    r.stop()
+    assert [f.seq for f in got] == [1, 2]
+    assert r.error_count == 0
+
+
+def test_corrupt_crc_packet_dropped_and_stream_continues():
+    got = []
+    bad = bytearray(make_packet(b"\x09\x09", seq=1))
+    bad[-1] ^= 0xFF  # break the CRC
+    data = bytes(bad) + make_packet(b"\x01\x02", seq=2)
+    r = StreamReader(FakeTransport([data]), parse_payload,
+                    on_frame=got.append, on_error=lambda e: None)
+    r.start()
+    time.sleep(0.2)
+    r.stop()
+    assert [f.seq for f in got] == [2]
+    assert r.error_count == 1
+
+
+def test_bogus_giant_payload_len_does_not_stall_the_stream():
+    """A false 0xAA 0x88 inside data yields a garbage payload_len (up to
+    64KB). Without a sanity cap the reader waits forever accumulating a
+    packet that doesn't exist — the 'frames stop, no error' symptom."""
+    got = []
+    # False sync + header claiming a 0xFFFF-byte payload, then a real packet.
+    bogus = b"\xaa\x88\x01\x00\x00\xff\xff"
+    data = bogus + make_packet(b"\x01\x02", seq=7)
+    r = StreamReader(FakeTransport([data]), parse_payload,
+                    on_frame=got.append, on_error=lambda e: None)
+    r.start()
+    time.sleep(0.2)
+    r.stop()
+    assert [f.seq for f in got] == [7]
+
+
+def test_quiet_port_between_scans_is_not_a_resync_failure():
+    """Device pauses between measurement scans. Empty reads must not count
+    toward the resync give-up cap (C#'s findMessageStart waits forever) —
+    the old code died with TimeoutError after ~50 empty reads."""
+    errors = []
+    got = []
+    r = StreamReader(FakeTransport([make_packet(b"\x01\x02", seq=1)]),
+                    parse_payload, on_frame=got.append, on_error=errors.append)
+    r.start()
+    time.sleep(0.5)  # ~100 empty reads after the single packet drains
+    r.stop()
+    assert [f.seq for f in got] == [1]
+    assert errors == []
+
+
+def test_stop_does_not_report_the_close_it_caused_as_an_error():
+    """stop()'s own transport.close() unblocks the worker thread's read()
+    by making it raise — that's expected teardown, not a stream fault, and
+    must not reach on_error (Activity Log would otherwise show a fake
+    "Live stream error" on every ordinary Stop)."""
+    errors = []
+    r = StreamReader(WedgedTransport(), parse_line,
+                    on_frame=lambda f: None, on_error=errors.append)
+    r.start()
+    time.sleep(0.05)
+    r.stop()
+    time.sleep(0.05)  # let the worker thread's except-handler run
+    assert errors == []