Merge branch 'master' into merge_osmerge_os

1 files changed, 1080 insertions, 514 deletions

diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py
index 5d9407883..174251343 100755
--- a/test/hil/hil_test.py
+++ b/test/hil/hil_test.py
@@ -45,7 +45,10 @@ import os
 import random
 import re
 import select
+import signal
+import shlex
 import sys
+import tempfile
 import time
 from contextlib import redirect_stdout
 from pathlib import Path
@@ -53,30 +56,29 @@ from typing import TypedDict, NotRequired, cast
 
 import serial
 import subprocess
+import traceback
 import json
 import glob
 import multiprocessing
 from multiprocessing import TimeoutError as MpTimeoutError
 
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))  # PYTHONSAFEPATH drops it
 import hil_flash
-import hil_lock
-from hil_examples import device_tests, dual_tests, host_test
+from helper import hil_health, hil_lock, hil_util
+from helper.hil_util import device_tests, dual_tests, host_test
 
-# Raw Lock/Semaphore objects passed via Pool initargs are inheritable only under the fork
-# start method (spawn/forkserver pickle them and fail at Pool creation) — pin it so a
-# future interpreter default change cannot break the run at startup.
-_mp = multiprocessing.get_context('fork')
+# Raw Lock/Semaphore objects in Pool initargs are inheritable only under fork
+# (spawn/forkserver pickle them and fail at Pool creation), so pin it against an
+# interpreter default change. Windows has no fork: fall back so it still IMPORTS there.
+
+_mp = multiprocessing.get_context('fork') if os.name != 'nt' else multiprocessing.get_context()
 Pool, Lock, Semaphore, Manager = _mp.Pool, _mp.Lock, _mp.Semaphore, _mp.Manager
-import hashlib
-import ctypes
-from pymtp import LIBMTP_DeviceEntry, LIBMTP_RawDevice, MTP
 import string
 
-# Enumeration wait budget. The first attempt gets ENUM_TIMEOUT; retry attempts get the
-# shorter ENUM_TIMEOUT_RETRY - the board was just re-flashed again, and a device that is
-# going to enumerate shows up within a few seconds, so a failing test costs ~3-5x a
-# passing one instead of 10-30x. Per-attempt value is set by test_example(); each pool
-# worker is its own process, so a module global is safe.
+# Enumeration wait budget: first attempt ENUM_TIMEOUT, retries the shorter
+# ENUM_TIMEOUT_RETRY -- a device that will enumerate shows up within seconds, so a failing
+# test costs ~3-5x a passing one instead of 10-30x. Set per attempt by test_example(); a
+# module global is safe because each pool worker is its own process.
 ENUM_TIMEOUT = 8
 ENUM_TIMEOUT_RETRY = 4
 _enum_timeout = ENUM_TIMEOUT
@@ -104,26 +106,36 @@ STATUS_OK = "\033[32mOK\033[0m"
 STATUS_FAILED = "\033[31mFailed\033[0m"
 STATUS_SKIPPED = "\033[33mSkipped\033[0m"
 
-# Plain (non-ANSI) cell symbols for the markdown matrix report (hil_report.md).
-# A missing binary is reported as skipped too.
+# Plain (non-ANSI) cell symbols for hil_report.md; a missing binary counts as skipped.
 REPORT_CELL = {'pass': '✅', 'fail': '❌', 'skip': '⚪'}
 
 
 class TestFail(AssertionError):
     """Fail a test but still surface a metric string in its report cell (e.g. usbtest's '❌ 29/30'
     instead of a bare ❌). The cell metric is icon-prefixed so render/tally treat it as a failure."""
-    def __init__(self, msg: str, metric: str | None = None):
+    def __init__(self, msg: str, metric: str | None = None, parsed: bool = False):
         super().__init__(msg)
         self.metric = metric
+        # parsed=True: a real per-case verdict, so a retry would only re-observe it
+        # (test_example skips the rest). A failure to RUN the tool stays retryable.
+        self.parsed = parsed
 
 
 verbose = False
+# Set when a HUNG usbtest case could not be recovered: the DUT's usbfs node still has a
+# D-state holder, so every later flash on that board enumerates into it, blocks, survives
+# SIGKILL and becomes another stray. maxtasksperchild=1 gives each board its own worker,
+# so this global is board-scoped; test_board resets it anyway.
+board_wedged = ''
+max_retry = 1   # mirrors argparse's -r default (see main); defined HERE too so
+                # test_example is callable (and testable) without going through main()
 PROFILE = os.environ.get('HIL_PROFILE') == '1'  # timestamped logs + permit/flash timing + ctrl-map dump
 test_only = []
 board_test = {}
 skip_flash = False
 print_lock = None
 shuffle_seed = None  # per-run seed for the per-board test-order shuffle (HIL_SHUFFLE_SEED to replay)
+_current_fw = None  # firmware test_example resolved for the RUNNING test (set before each test fn)
 
 
 def init_worker(lock, seed, b_mutexes, f_sems, cmap, cmeta, hints_by_uid):
@@ -147,13 +159,21 @@ def log_line(msg: str) -> None:
 def compact_output(raw: str) -> str:
     if not raw:
         return ''
-    lines = [ln.strip() for ln in raw.replace('\r', '\n').split('\n') if ln.strip()]
+    # Defense in depth (the emitter already suppresses them, see _ci_log_groups): markers
+    # piped into this capture land mid-row, where GitHub renders them literally.
+    lines = []
+    for ln in raw.replace('\r', '\n').split('\n'):
+        ln = hil_util.strip_workflow_markers(ln.strip()).strip()
+        if ln:
+            lines.append(ln)
     return ' | '.join(lines)
 
 class FlasherCfg(TypedDict):
     name: str
     uid: str
-    args: str
+    args: NotRequired[str]     # stlink entries carry no args
+    vid_pid: NotRequired[str]  # openocd probe pin, verbatim (e.g. "0x2e8a 0x000c")
+    verify: NotRequired[bool]  # openocd read-back verify opt-out (WCH)
 
 
 class AttachedDevCfg(TypedDict, total=False):
@@ -197,9 +217,34 @@ class Board(TypedDict):
 class HilConfig(TypedDict):
     boards: list[Board]
 
-POOL_TIMEOUT = int(os.getenv('HIL_POOL_TIMEOUT', '4200'))  # usbtest batteries are serialized fleet-wide, lengthening the tail
-SERIAL_READ_TIMEOUT = float(os.getenv('HIL_SERIAL_READ_TIMEOUT', '5'))
-SERIAL_WRITE_TIMEOUT = float(os.getenv('HIL_SERIAL_WRITE_TIMEOUT', '10'))
+# Below the CI job ceilings so THIS guard fires first and still writes a report, well
+# above a healthy fleet run (~14 min measured), and deliberately generous: firing early
+# abandons boards that were still in flight (30 min fired on 5 of the last 8 HIL jobs),
+# while firing late costs minutes on an already-wedged run. The drain keeps whatever had
+# already finished either way.
+POOL_TIMEOUT = hil_util.pos_int_env('HIL_POOL_TIMEOUT', 3600)
+
+
+# Headroom on top of a battery's own budget so ONE HUNG recovery (case timeout, SIGKILL
+# wait, bounded reflash, settle) can finish. Only spent when cases actually time out.
+USBTEST_RECOVERY_BUDGET = hil_util.pos_int_env('HIL_USBTEST_RECOVERY_BUDGET', 250)
+# How long usbtest.py may keep starting new cases (--budget). The outer run_cmd timeout is
+# always this PLUS the recovery headroom, never a separate literal, or lowering one eats
+# the reserve the recovery needs. 0 is refused (usbtest.py reads it as "no limit"); the
+# margin over a healthy battery (~200s) keeps contention from becoming BUDGET entries.
+USBTEST_BATTERY_BUDGET = hil_util.pos_int_env('HIL_USBTEST_BATTERY_BUDGET', 260)
+
+# The battery checks its budget BEFORE dispatching a case, so it can overshoot by one
+# already-started case. Our outer kill must sit ABOVE that or we SIGKILL the battery just
+# as it goes to print its JSON, turning ~29 real per-case verdicts into "usbtest did not
+# run" and re-paying the whole battery on retry.
+# Worst case, from usbtest.py: --timeout 60 (the case) + 5s post-SIGKILL reap +
+# dmesg_tail(), which is bounded by HELPER_TIMEOUT=30 and runs on BOTH the FAIL and HUNG
+# timeout paths = 95s. 120 leaves a margin; 75 (my first estimate, taken before checking
+# dmesg_tail) was 20s SHORT and would have killed the battery mid-print.
+USBTEST_OVERSHOOT = 120
+SERIAL_READ_TIMEOUT = hil_util.pos_float_env('HIL_SERIAL_READ_TIMEOUT', 5)
+SERIAL_WRITE_TIMEOUT = hil_util.pos_float_env('HIL_SERIAL_WRITE_TIMEOUT', 10)
 
 
 MSC_README_TXT = \
@@ -207,7 +252,6 @@ b"This is tinyusb's MassStorage Class demo.\r\n\r\n\
 If you find any bugs or get any questions, feel free to file an\r\n\
 issue at github.com/hathach/tinyusb"
 
-# get usb disk by id
 def get_disk_dev(id, vendor_str, lun):
     return f'/dev/disk/by-id/usb-{vendor_str}_Mass_Storage_{id}-0:{lun}'
 
@@ -235,8 +279,7 @@ def open_serial_dev(port: str):
     while timeout > 0:
         if os.path.exists(port):
             try:
-                # write_timeout: a wedged device otherwise blocks ser.write() forever,
-                # hanging the worker until the pool/job timeout kills the whole run
+                # write_timeout: see serial_write_all
                 ser = serial.Serial(port, baudrate=115200, timeout=SERIAL_READ_TIMEOUT,
                                     write_timeout=SERIAL_WRITE_TIMEOUT)
                 break
@@ -252,18 +295,43 @@ def open_serial_dev(port: str):
 
 
 def serial_write_all(ser: serial.Serial, data: bytes):
-    # write_timeout is a total deadline for the whole call (pyserial keeps partial progress
-    # internally). A timeout means the device stopped draining — treat it as fatal: pyserial
-    # loses the partial-write count on raise, so retrying would duplicate bytes on the wire.
+    # write_timeout is a deadline for the whole call. A timeout means the device stopped
+    # draining, and it is fatal: pyserial loses the partial-write count on raise, so
+    # retrying would duplicate bytes on the wire.
     try:
         ser.write(data)
     except serial.SerialTimeoutException:
         raise AssertionError(f'Serial write timeout after {SERIAL_WRITE_TIMEOUT:.1f}s')
 
 
+LP_OPEN_TIMEOUT = 5   # bound on opening the printer lp node; see test_device_printer_to_cdc
+# Runs under hil_util.run_alongside as `python3 -c`. Inline rather than a file so hil_ci.sh's
+# staging list does not need another entry to keep the rig working.
+LP_READER = (
+    'import os, sys\n'
+    'fd = os.open(sys.argv[1], os.O_RDONLY)\n'
+    # readiness marker: the parent must not send a byte before the node is open, or the
+    # bytes are lost. A blind sleep raced CPython start-up on a loaded rig.
+    'open(sys.argv[3], "w").close()\n'
+    'want = int(sys.argv[2])\n'
+    'buf = b""\n'
+    'while len(buf) < want:\n'
+    '    chunk = os.read(fd, min(64, want - len(buf)))\n'
+    '    if not chunk:\n'
+    '        break\n'
+    '    buf += chunk\n'
+    'sys.stdout.buffer.write(buf)\n'
+)
+MTYPE_TIMEOUT = 30  # a README-sized read is <1 s; bounds a D-state hang on a wedged device
+
+
 def read_disk_file(uid: str, lun: int, fname: str) -> bytes:
-    # Reads a file from a FAT volume on a block device without mounting it.
-    # Requires mtools: `apt install mtools` (no pip dependency).
+    # Reads a file from an unmounted FAT volume; needs mtools. run_cmd everywhere in this
+    # file rather than subprocess.run/check_output: its post-timeout reap is an unbounded
+    # communicate() with no killpg (CPython 3.13.5 subprocess.py:558-565 -- kill(), then
+    # communicate() with NO timeout), which never returns on a device wedged in D state,
+    # where the kill is queued and never delivered. binary
+    # keeps the bytes exact, split_stderr keeps mtype warnings out of them.
     dev = get_disk_dev(uid, 'TinyUSB', lun)
     last_err = None
 
@@ -271,101 +339,27 @@ def read_disk_file(uid: str, lun: int, fname: str) -> bytes:
         nonlocal last_err
         if not os.path.exists(dev):
             return None
-        try:
-            data = subprocess.check_output(
-                ['mtype', '-i', dev, f'::/{fname}'], stderr=subprocess.PIPE)
-            assert data, f'Cannot read file {fname} from {dev}'
-            return data
-        except subprocess.CalledProcessError as e:
-            last_err = e.stderr.decode(errors='replace').strip()
-            return None
+        r = hil_util.run_cmd(f"mtype -i {shlex.quote(dev)} ::/{shlex.quote(fname)}",
+                             timeout=MTYPE_TIMEOUT, binary=True, split_stderr=True, quiet=True)
+        if r.returncode == 0:
+            if r.stdout:
+                return r.stdout
+            # rc 0 with no data is an answer (empty file, zeroed sectors), not "not
+            # ready" — fail now instead of spinning the budget
+            raise AssertionError(f'Cannot read file {fname} from {dev}: mtype returned no data')
+        last_err = (r.stderr or b'').decode(errors='replace').strip() or f'mtype rc {r.returncode}'
+        return None
 
     data = wait_until(try_read)
     if data is None:
-        raise AssertionError(f'mtype failed on {dev}: {last_err}' if last_err else f'Storage {dev} not existed')
+        raise AssertionError(f'Cannot read file {fname} from {dev}: {last_err}' if last_err
+                             else f'Storage {dev} not existed')
     return data
 
 
-def open_mtp_dev(uid: str):
-    mtp = MTP()
-    last_detail = None
-    deadline = time.monotonic() + 2 * enum_timeout()
-
-    def find_ready_mtp():
-        nonlocal last_detail
-        for marker_name in glob.glob('/dev/libmtp-*'):
-            marker = Path(marker_name)
-            serial = ''
-            try:
-                # libmtp-runtime publishes libmtp-%k only after its synchronous
-                # mtp-probe has accepted the device. Starting from that small, ready-only
-                # set avoids a broad sysfs scan racing unrelated parallel re-enumerations.
-                sysname = marker.name[len('libmtp-'):]
-                dev_path = Path('/sys/bus/usb/devices') / sysname
-                serial = (dev_path / 'serial').read_text().strip()
-                if (serial.lower() != uid.lower()
-                        or (dev_path / 'idVendor').read_text().strip() != 'cafe'
-                        or (dev_path / 'idProduct').read_text().strip() != '4017'):
-                    continue
-
-                busnum = int((dev_path / 'busnum').read_text())
-                devnum = int((dev_path / 'devnum').read_text())
-                usb_node = Path('/dev/bus/usb') / f'{busnum:03d}' / f'{devnum:03d}'
-                if marker.resolve(strict=True) != usb_node or not os.access(
-                        usb_node, os.R_OK | os.W_OK):
-                    last_detail = f'{marker} did not resolve to an accessible {usb_node}'
-                    continue
-                return busnum, devnum
-            except (OSError, ValueError) as e:
-                # A marker can disappear while another board flashes. Only retain
-                # diagnostics for this board's marker, not unrelated MTP devices.
-                if serial.lower() == uid.lower():
-                    last_detail = f'{marker}: {e}'
-        return None
-
-    def remaining() -> float:
-        return max(0.0, deadline - time.monotonic())
-
-    target = wait_until(find_ready_mtp, step=0.05, timeout=remaining())
-    if target is None:
-        detail = f': {last_detail}' if last_detail else '; install libmtp-runtime'
-        raise AssertionError(f'MTP udev device not ready for {uid}{detail}')
-
-    # A desktop GVFS session may claim MTP after udev probing. This is a no-op on
-    # headless runners, but preserves support for rigs where the mount exists.
-    try:
-        subprocess.run(['gio', 'mount', '-u', f'mtp://TinyUsb_TinyUsb_Device_{uid}/'],
-                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=2)
-    except (FileNotFoundError, subprocess.TimeoutExpired):
-        pass
-
-    # GIO can race a disconnect/re-enumeration. Resolve the completed marker again
-    # rather than opening a stale bus/device tuple.
-    target = wait_until(find_ready_mtp, step=0.05, timeout=remaining())
-    if target is None:
-        raise AssertionError(f'MTP udev device disappeared for {uid}')
-    busnum, devnum = target
-
-    # TinyUSB needs no libmtp device quirks. Construct its raw entry directly so
-    # this test never probes another MTP board that is still being initialized.
-    entry = LIBMTP_DeviceEntry(None, 0xcafe, None, 0x4017, 0)
-    raw = LIBMTP_RawDevice(entry, busnum, devnum)
-    mtp.device = mtp.mtp.LIBMTP_Open_Raw_Device(ctypes.byref(raw))
-    if not mtp.device:
-        raise AssertionError(f'libmtp could not open MTP {uid} at {busnum:03d}/{devnum:03d}')
-
-    try:
-        serial_raw = mtp.get_serialnumber()
-        serial = serial_raw.decode('utf-8') if serial_raw else ''
-        if serial.lower() != uid.lower():
-            raise AssertionError(f'MTP serial mismatch at {busnum:03d}/{devnum:03d}: {serial}')
-    except Exception:
-        try:
-            mtp.disconnect()
-        except Exception:
-            pass
-        raise
-    return mtp
+# ~5 KB of transfers plus libmtp setup takes seconds, not minutes; a larger value makes a
+# wedged MTP board cost that much on every retry, all charged to the pool guard.
+MTP_SESSION_MARGIN = 30  # transfer budget after enumeration; past it the session is killed
 
 
 def get_printer_dev(id: str, vendor_str, product_str, ifnum: int):
@@ -374,10 +368,16 @@ def get_printer_dev(id: str, vendor_str, product_str, ifnum: int):
     product_str = product_str.replace(' ', '_') if product_str else ''
     for lp in glob.glob('/sys/class/usbmisc/lp*'):
         try:
-            sn = open(f'{lp}/device/../serial').read().strip()
+            # bounded: same device_lock() exposure as the sibling reads (see read_sysfs)
+            sn = hil_util.read_sysfs(f'{lp}/device/../serial')
+            # UNKNOWN is not None: the sentinel has no __eq__, so an unanswered read
+            # would fall through both tests and read as 'not this board' -- the exact
+            # absence/unknown conflation read_sysfs exists to prevent.
+            if sn is None or sn is hil_util.SYSFS_UNKNOWN:
+                continue
             if sn == id:
                 return f'/dev/usb/{os.path.basename(lp)}'
-        except (FileNotFoundError, PermissionError, ValueError):
+        except OSError:   # read_sysfs swallows its own OSError/ValueError; glob can race
             pass
     return None
 
@@ -389,7 +389,8 @@ def open_printer_dev(id: str, vendor_str, product_str, ifnum: int) -> str:
         return lp_dev if lp_dev and os.path.exists(lp_dev) else None
 
     lp_dev = wait_until(try_find)
-    assert lp_dev, f'Printer device not found for {id} if{ifnum:02d}'
+    assert lp_dev, (f'Printer device not found for {id} if{ifnum:02d}'
+                    + hil_util.sysfs_blind_note())
     return lp_dev
 
 
@@ -399,18 +400,16 @@ def open_printer_dev(id: str, vendor_str, product_str, ifnum: int) -> str:
 def test_dual_host_info_to_device_cdc(board):
     uid = board['uid']
     declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
-    port = hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
+    port = hil_util.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
     ser = open_serial_dev(port)
     ser.timeout = 0.1
 
-    # read until all expected devices are enumerated
     data = b''
     timeout = enum_timeout()
     while timeout > 0:
         new_data = ser.read(ser.in_waiting or 1)
         if new_data:
             data += new_data
-        # check if all devices found
         enum_dev_sn = []
         for l in data.decode('utf-8', errors='ignore').splitlines():
             vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
@@ -447,7 +446,7 @@ def test_host_device_info(board):
     flasher = board['flasher']
     declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
 
-    port = hil_flash.get_serial_dev(flasher["uid"], None, None, 0)
+    port = hil_util.get_serial_dev(flasher["uid"], None, None, 0)
     ser = open_serial_dev(port)
     ser.timeout = 0.1
 
@@ -455,14 +454,12 @@ def test_host_device_info(board):
     ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
     assert ret.returncode == 0, 'Failed to reset device'
 
-    # read until all expected devices are enumerated
     data = b''
     timeout = enum_timeout()
     while timeout > 0:
         new_data = ser.read(ser.in_waiting or 1)
         if new_data:
             data += new_data
-        # check if all devices found
         enum_dev_sn = []
         for l in data.decode('utf-8', errors='ignore').splitlines():
             vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
@@ -526,7 +523,7 @@ def test_host_cdc_msc_hid(board):
     if not cdc_devs and not msc_devs:
         return 'skipped'
 
-    port = hil_flash.get_serial_dev(flasher["uid"], None, None, 0)
+    port = hil_util.get_serial_dev(flasher["uid"], None, None, 0)
     ser = open_serial_dev(port)
     ser.timeout = 0.1
 
@@ -534,7 +531,6 @@ def test_host_cdc_msc_hid(board):
     ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
     assert ret.returncode == 0, 'Failed to reset device'
 
-    # Wait for all expected mount messages
     data = b''
     timeout = enum_timeout()
     wait_cdc = len(cdc_devs) > 0
@@ -550,7 +546,6 @@ def test_host_cdc_msc_hid(board):
         time.sleep(0.1)
         timeout -= 0.1
 
-    # Lookup serial chip name from vid_pid
     vid_pid_name = {
         '0403_6001': 'FTDI', '0403_6010': 'FTDI', '0403_6011': 'FTDI', '0403_6014': 'FTDI',
         '10c4_ea60': 'CP210x', '10c4_ea70': 'CP210x',
@@ -561,7 +556,6 @@ def test_host_cdc_msc_hid(board):
 
     lines = data.decode('utf-8', errors='ignore').splitlines()
 
-    # Verify and print CDC mount
     if cdc_devs:
         assert b'CDC Interface is mounted' in data, 'CDC device not mounted on host'
         dev = cdc_devs[0]
@@ -570,7 +564,6 @@ def test_host_cdc_msc_hid(board):
             if 'CDC Interface is mounted' in l:
                 print(f'\r\n  {chip_name}: {l} ', end='')
 
-    # Verify and print MSC mount (inquiry + disk size)
     if msc_devs:
         assert b'MassStorage device is mounted' in data, 'MSC device not mounted on host'
         assert b'Disk Size' in data, 'MSC Disk Size not reported'
@@ -590,7 +583,6 @@ def test_host_cdc_msc_hid(board):
 
     packet_size = 64
 
-    # Echo test: write random 1-packet_size chunks, wait for echo before sending next
     echo_len = 1024
     echo_data = rand_ascii(echo_len)
     ser.reset_input_buffer()
@@ -598,7 +590,6 @@ def test_host_cdc_msc_hid(board):
     while offset < echo_len:
         chunk_size = min(random.randint(1, packet_size), echo_len - offset)
         serial_write_all(ser, echo_data[offset:offset + chunk_size])
-        # wait until this chunk is echoed back
         echo = b''
         t_end = time.monotonic() + 1.0
         while time.monotonic() < t_end and len(echo) < chunk_size:
@@ -619,7 +610,7 @@ def test_host_msc_file_explorer(board):
     if not msc_devs:
         return 'skipped'
 
-    port = hil_flash.get_serial_dev(flasher["uid"], None, None, 0)
+    port = hil_util.get_serial_dev(flasher["uid"], None, None, 0)
     ser = open_serial_dev(port)
     ser.timeout = 0.1
 
@@ -627,7 +618,6 @@ def test_host_msc_file_explorer(board):
     ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
     assert ret.returncode == 0, 'Failed to reset device'
 
-    # Wait for MSC mount (Disk Size message)
     data = b''
     timeout = enum_timeout()
     while timeout > 0:
@@ -664,14 +654,12 @@ def test_host_msc_file_explorer(board):
     if MSC_README_TXT.decode() in resp_text:
         print('README.TXT matched ', end='')
 
-    # MSC throughput test: send dd command to read sectors
     time.sleep(0.5)
     ser.reset_input_buffer()
     for ch in 'dd 1024\r':
         serial_write_all(ser, ch.encode())
         time.sleep(0.002)
 
-    # Read dd output until prompt
     resp = b''
     t = 30.0
     while t > 0:
@@ -706,15 +694,14 @@ def test_host_msc_file_explorer_freertos(board):
 # Tests: device
 # -------------------------------------------------------------
 def test_device_board_test(board):
-    # Dummy test
     pass
 
 
 def test_device_cdc_dual_ports(board):
     uid = board['uid']
     port = [
-        hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0),
-        hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 2)
+        hil_util.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0),
+        hil_util.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 2)
     ]
     ser = [open_serial_dev(p) for p in port]
 
@@ -753,7 +740,7 @@ def test_device_cdc_dual_ports(board):
 def test_device_cdc_msc(board):
     uid = board['uid']
     # CDC Echo test
-    port = hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
+    port = hil_util.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
     ser = open_serial_dev(port)
 
     def rand_ascii(length):
@@ -782,6 +769,20 @@ def test_device_cdc_msc_freertos(board):
     test_device_cdc_msc(board)
 
 
+def link_is_fs(speed) -> bool:
+    """Payload scaling from a `speed` attribute. Anything not positively read as high speed
+    counts as FS -- including None and SYSFS_UNKNOWN: the FS payload merely tests an HS
+    board less, while the HS payload hard-fails a healthy FS board."""
+    return speed not in ('480', '5000', '10000')
+
+
+def dd_timeout(mib: float) -> int:
+    """Bound one dd by what was ASKED for: 2.5 s/MiB is the slowest rate this test has
+    measured (FS CDC, ~420 kB/s), over a 30 s floor. A flat bound fails a healthy board as
+    soon as the payload grows or the leaf-hub uplink is shared."""
+    return int(30 + 2.5 * mib)
+
+
 def test_device_cdc_msc_throughput(board):
     uid = board['uid']
 
@@ -792,7 +793,6 @@ def test_device_cdc_msc_throughput(board):
                 return f'{float(m.group(1)):.1f} {m.group(2)}ps'
         return '?'
 
-    # Wait for MSC disk enumeration
     dev = get_disk_dev(uid, 'TinyUSB', 0)
     timeout = enum_timeout()
     while timeout > 0:
@@ -801,8 +801,7 @@ def test_device_cdc_msc_throughput(board):
         time.sleep(0.1); timeout -= 0.1
     assert timeout > 0, f'Disk {dev} not found'
 
-    # Wait for CDC tty enumeration
-    tty = hil_flash.get_serial_dev(uid, 'TinyUSB', 'Throughput', 0)
+    tty = hil_util.get_serial_dev(uid, 'TinyUSB', 'Throughput', 0)
     timeout = enum_timeout()
     while timeout > 0:
         if os.path.exists(tty):
@@ -810,41 +809,48 @@ def test_device_cdc_msc_throughput(board):
         time.sleep(0.1); timeout -= 0.1
     assert timeout > 0, f'CDC tty {tty} not found'
 
-    # Detect speed (12 Mbps FS / 480 Mbps HS) for payload scaling
-    is_fs = False
-    for f in glob.glob('/sys/bus/usb/devices/*/serial'):
-        try:
-            if open(f).read().strip().lower() == uid.lower():
-                is_fs = (open(os.path.join(os.path.dirname(f), 'speed')).read().strip() == '12')
-                break
-        except (OSError, ValueError):
-            pass
+    # Detect speed (12 Mbps FS / 480 Mbps HS) for payload scaling; a device we never find
+    # keeps the FS payload (see link_is_fs)
+    # usb_scan, not a private glob: it skips root hubs and remembers paths that already
+    # stranded, so one wedged peer cannot spend this worker's blindness budget four reads
+    # at a time.
+    is_fs = True
+    speed_known = False
+    devs, _ = hil_util.usb_scan(vid='cafe', serial=uid)
+    if devs:
+        speed = hil_util.read_sysfs(os.path.join(devs[0]['dir'], 'speed'))
+        is_fs = link_is_fs(speed)
+        speed_known = speed not in (None, hil_util.SYSFS_UNKNOWN)
 
     # Put tty in raw mode so dd sees pure binary throughput.
-    rs = hil_flash.run_cmd(f'timeout 30 stty -F {tty} raw -echo')
-    assert rs.returncode == 0, f'stty failed: {hil_flash.cmd_stdout_text(rs.stdout)}'
+    rs = hil_util.run_cmd(f'timeout 30 stty -F {tty} raw -echo')
+    assert rs.returncode == 0, f'stty failed: {hil_util.cmd_stdout_text(rs.stdout)}'
 
     # Payload aim: ~5 s per direction at FS (~830 kB/s), much less at HS.
     msc_count = 2 if is_fs else 16    # bs=1M
     cdc_count = 16 if is_fs else 128  # bs=64K
 
     tmp_file = f'/tmp/cdc_msc_tp_{uid}.bin'
+    t_cdc, t_msc = dd_timeout(cdc_count / 16), dd_timeout(msc_count)
 
-    rw = hil_flash.run_cmd(f'timeout 30 dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1')
-    assert rw.returncode == 0, f'CDC dd write failed: {hil_flash.cmd_stdout_text(rw.stdout)}'
-    cdc_w = parse_speed(hil_flash.cmd_stdout_text(rw.stdout))
+    rw = hil_util.run_cmd(f'timeout {t_cdc} dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1')
+    assert rw.returncode == 0, f'CDC dd write failed: {hil_util.cmd_stdout_text(rw.stdout)}'
+    cdc_w = parse_speed(hil_util.cmd_stdout_text(rw.stdout))
 
-    rr = hil_flash.run_cmd(f'timeout 30 dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1')
-    assert rr.returncode == 0, f'CDC dd read failed: {hil_flash.cmd_stdout_text(rr.stdout)}'
-    cdc_r = parse_speed(hil_flash.cmd_stdout_text(rr.stdout))
+    rr = hil_util.run_cmd(f'timeout {t_cdc} dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1')
+    assert rr.returncode == 0, f'CDC dd read failed: {hil_util.cmd_stdout_text(rr.stdout)}'
+    cdc_r = parse_speed(hil_util.cmd_stdout_text(rr.stdout))
 
-    rmr = hil_flash.run_cmd(f'dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1')
-    assert rmr.returncode == 0, f'MSC dd read failed: {hil_flash.cmd_stdout_text(rmr.stdout)}'
-    msc_r = parse_speed(hil_flash.cmd_stdout_text(rmr.stdout))
+    # inner bound, like the CDC pair above: run_cmd's SIGKILL is merely QUEUED against a
+    # dd blocked in the block layer on a half-dead device, so without one the call rides
+    # CMD_TIMEOUT and is abandoned holding the disk and usbfs nodes.
+    rmr = hil_util.run_cmd(f'timeout {t_msc} dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1')
+    assert rmr.returncode == 0, f'MSC dd read failed: {hil_util.cmd_stdout_text(rmr.stdout)}'
+    msc_r = parse_speed(hil_util.cmd_stdout_text(rmr.stdout))
 
-    rmw = hil_flash.run_cmd(f'dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1')
-    assert rmw.returncode == 0, f'MSC dd write failed: {hil_flash.cmd_stdout_text(rmw.stdout)}'
-    msc_w = parse_speed(hil_flash.cmd_stdout_text(rmw.stdout))
+    rmw = hil_util.run_cmd(f'timeout {t_msc} dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1')
+    assert rmw.returncode == 0, f'MSC dd write failed: {hil_util.cmd_stdout_text(rmw.stdout)}'
+    msc_w = parse_speed(hil_util.cmd_stdout_text(rmw.stdout))
 
     try:
         os.remove(tmp_file)
@@ -853,8 +859,7 @@ def test_device_cdc_msc_throughput(board):
 
     print(f'  CDC read {cdc_r} write {cdc_w}, MSC read {msc_r} write {msc_w}  ', end='')
 
-    # compact read/write speeds for the report cell, e.g. "✅ C 652/422k M 1.1M/783k"
-    # (C=CDC, M=MSC; the unit is shown once when both sides share it)
+    # report cell, e.g. "✅ C 652/422k M 1.1M/783k" (C=CDC, M=MSC; shared unit shown once)
     def short(s):
         return (s.split()[0].rstrip('0').rstrip('.') + s.split()[-1][0]) if ' ' in s else s
 
@@ -864,20 +869,29 @@ def test_device_cdc_msc_throughput(board):
             r = r[:-1]
         return f'{r}/{w}'
 
-    return f'{REPORT_CELL["pass"]} C {pair(cdc_r, cdc_w)} M {pair(msc_r, msc_w)}'
+    # 'FS?' when the speed could not be read: the numbers below were produced against the FS
+    # payload, so an HS board reads as suspiciously slow. Say so rather than publish a green
+    # cell whose scale is a guess.
+    scale = '' if speed_known else ' FS?'
+    return f'{REPORT_CELL["pass"]} C {pair(cdc_r, cdc_w)} M {pair(msc_r, msc_w)}{scale}'
 
 
 def test_device_dfu(board):
     uid = board['uid']
+    vid_pid = 'cafe:400b'
 
-    # Wait device enum. Deadline-based: dfu-util -l itself takes ~1 s per call, which a
-    # per-iteration countdown would not charge against the budget.
+    # Deadline-based: dfu-util takes ~1 s per call, which a countdown would not charge
+    # against the budget. -d pins enumeration to THIS example's ids: a bare `-l` opens every
+    # DFU-capable node, and one wedged node blocks that open in D state. The pair is doubled
+    # because dfu-util matches run-time and DFU-mode devices against SEPARATE id pairs
+    # (parse_vendprod: an omitted DFU-mode pair matches ANY DFU-mode device). The deadline
+    # is only tested BETWEEN calls, so the per-call bound is what caps a blocked open.
     deadline = time.monotonic() + enum_timeout()
     found = False
     while time.monotonic() < deadline:
-        ret = hil_flash.run_cmd(f'dfu-util -l')
-        stdout = hil_flash.cmd_stdout_text(ret.stdout)
-        if f'serial="{uid}"' in stdout and 'Found DFU: [cafe:400b]' in stdout:
+        ret = hil_util.run_cmd(f'dfu-util -d {vid_pid},{vid_pid} -l', timeout=15)
+        stdout = hil_util.cmd_stdout_text(ret.stdout)
+        if f'serial="{uid}"' in stdout and f'Found DFU: [{vid_pid}]' in stdout:
             found = True
             break
         time.sleep(1)
@@ -887,17 +901,23 @@ def test_device_dfu(board):
     f_dfu0 = f'dfu0_{uid}'
     f_dfu1 = f'dfu1_{uid}'
 
-    # Test upload
     try:
         os.remove(f_dfu0)
         os.remove(f_dfu1)
     except OSError:
         pass
 
-    ret = hil_flash.run_cmd(f'dfu-util -S {uid} -a 0 -U {f_dfu0}')
+    # -d as well as -S: dfu-util matches the SERIAL only after libusb_open() (dfu_util.c
+    # probes the descriptor for iSerialNumber), so -S alone still opens every DFU-capable
+    # node. The id filter runs BEFORE the open; -S then picks our board (see the poll).
+    # Each partition is one short string, so a healthy upload is ~1 s; the bound is there
+    # for a node that stops answering mid-transfer.
+    ret = hil_util.run_cmd(f'dfu-util -d {vid_pid},{vid_pid} -S {uid} -a 0 -U {f_dfu0}',
+                           timeout=30)
     assert ret.returncode == 0, 'Upload failed'
 
-    ret = hil_flash.run_cmd(f'dfu-util -S {uid} -a 1 -U {f_dfu1}')
+    ret = hil_util.run_cmd(f'dfu-util -d {vid_pid},{vid_pid} -S {uid} -a 1 -U {f_dfu1}',
+                           timeout=30)
     assert ret.returncode == 0, 'Upload failed'
 
     with open(f_dfu0) as f:
@@ -912,13 +932,14 @@ def test_device_dfu(board):
 
 def test_device_dfu_runtime(board):
     uid = board['uid']
-    # Wait device enum (deadline-based, see test_device_dfu)
+    vid_pid = 'cafe:400c'
+    # enumeration pinned to this example's ids, same per-call bound (see test_device_dfu)
     deadline = time.monotonic() + enum_timeout()
     found = False
     while time.monotonic() < deadline:
-        ret = hil_flash.run_cmd(f'dfu-util -l')
-        stdout = hil_flash.cmd_stdout_text(ret.stdout)
-        if f'serial="{uid}"' in stdout and 'Found Runtime: [cafe:400c]' in stdout:
+        ret = hil_util.run_cmd(f'dfu-util -d {vid_pid},{vid_pid} -l', timeout=15)
+        stdout = hil_util.cmd_stdout_text(ret.stdout)
+        if f'serial="{uid}"' in stdout and f'Found Runtime: [{vid_pid}]' in stdout:
             found = True
             break
         time.sleep(1)
@@ -931,7 +952,6 @@ def test_device_hid_boot_interface(board):
     kbd = get_hid_dev(uid, 'TinyUSB', 'TinyUSB_Device', 'event-kbd')
     mouse1 = get_hid_dev(uid, 'TinyUSB', 'TinyUSB_Device', 'if01-event-mouse')
     mouse2 = get_hid_dev(uid, 'TinyUSB', 'TinyUSB_Device', 'if01-mouse')
-    # Wait device enum
     timeout = enum_timeout()
     while timeout > 0:
         if os.path.exists(kbd) and os.path.exists(mouse1) and os.path.exists(mouse2):
@@ -948,12 +968,9 @@ def test_device_hid_composite_freertos(id):
 
 
 def test_device_printer_to_cdc(board):
-    import threading
-
     uid = board['uid']
 
-    # Wait for CDC port and printer device
-    cdc_port = hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
+    cdc_port = hil_util.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
     ser = open_serial_dev(cdc_port)
     lp_dev = open_printer_dev(uid, 'TinyUSB', 'TinyUSB_Device', 2)
 
@@ -973,7 +990,6 @@ def test_device_printer_to_cdc(board):
 
     sizes = [32, 64, 128, 256, 512, random.randint(2000, 5000)]
 
-    # flush any stale data
     ser.reset_input_buffer()
 
     # Test 1: Printer -> CDC with multiple sizes, write in random 1-64 byte chunks
@@ -983,7 +999,17 @@ def test_device_printer_to_cdc(board):
         ser.reset_input_buffer()
         rd = b''
         offset = 0
-        lp_fd = os.open(lp_dev, os.O_WRONLY | os.O_NONBLOCK)
+        # bounded: O_NONBLOCK does NOT save us -- usblp_open() takes the device mutex
+        # first -- and this open runs on the worker itself, with no thread to abandon
+        lp_fd = hil_util.bounded_open(lp_dev, os.O_WRONLY | os.O_NONBLOCK, 5)
+        # Three-valued on purpose: an OSError here is a FACT about the node (EBUSY from
+        # usblp's single-opener rule, ENOENT from a re-enumeration race, EACCES from a
+        # udev gap) and must not be reported as a wedge -- that sends the operator to
+        # usb-kernel-recover for hardware that is fine.
+        assert lp_fd is not hil_util.SYSFS_UNKNOWN, (
+            f'printer: opening {lp_dev} for write blocked (device wedged)'
+            f'{hil_util.sysfs_blind_note()}')
+        assert lp_fd is not None, f'printer: {lp_dev} could not be opened for write'
         try:
             while offset < size:
                 chunk_size = min(random.randint(1, 64), size - offset)
@@ -1007,128 +1033,88 @@ def test_device_printer_to_cdc(board):
         assert rd == test_data, (f'Printer->CDC wrong data ({size} bytes):\n'
                                  f'  expected: {test_data[:64]}\n  received: {rd[:64]}')
 
-    # Test 2: CDC -> Printer with multiple sizes, write in random 1-64 byte chunks
-    # Use a thread to read from printer since /dev/usb/lp read blocks
+    # Test 2: CDC -> Printer with multiple sizes, write in random 1-64 byte chunks.
+    # The lp read runs in a PROCESS, not a thread: /dev/usb/lp* blocks on read, usblp
+    # allows a SINGLE opener, and a blocked thread cannot be abandoned without keeping
+    # that fd -- which poisoned the node for every later test this worker ran. A killed
+    # process takes its fd with it.
     ser.reset_input_buffer()
     time.sleep(0.5)
     for size in sizes:
         test_data = rand_ascii(size)
-        rd_result = [b'', None]  # [data, error]
-        reader_ready = threading.Event()
-
-        def lp_reader():
-            try:
-                rd = b''
-                fd = os.open(lp_dev, os.O_RDONLY)
-                reader_ready.set()
-                try:
-                    while len(rd) < size:
-                        chunk = os.read(fd, min(64, size - len(rd)))
-                        if not chunk:
-                            break
-                        rd += chunk
-                finally:
-                    os.close(fd)
-                rd_result[0] = rd
-            except Exception as e:
-                rd_result[1] = e
-                reader_ready.set()
 
-        reader = threading.Thread(target=lp_reader, daemon=True)
-        reader.start()
-        # wait for reader to open lp device before writing
-        reader_ready.wait(timeout=5)
-        time.sleep(0.1)
+        ready = Path(tempfile.gettempdir()) / f'hil-lp-ready-{os.getpid()}-{size}'
+        ready.unlink(missing_ok=True)
 
-        # Write to CDC in small chunks with flush to avoid overflowing device FIFO
-        offset = 0
-        while offset < size:
-            chunk_size = min(random.randint(1, 64), size - offset)
-            serial_write_all(ser, test_data[offset:offset + chunk_size])
-            time.sleep(0.01)
-            offset += chunk_size
+        def write_cdc():
+            # WAIT for the reader to have the node open. The child has to fork, exec and
+            # boot a CPython interpreter; on a loaded rig that routinely exceeds the 0.3s
+            # this used to sleep, and every byte sent early is lost -- surfacing as a
+            # spurious data mismatch rather than a timeout.
+            deadline = time.monotonic() + LP_OPEN_TIMEOUT + 5
+            while not ready.exists():
+                if time.monotonic() > deadline:
+                    return       # reader never opened; the rc/compare below reports it
+                time.sleep(0.02)
+            offset = 0
+            while offset < size:
+                chunk_size = min(random.randint(1, 64), size - offset)
+                serial_write_all(ser, test_data[offset:offset + chunk_size])
+                time.sleep(0.01)
+                offset += chunk_size
 
-        reader.join(timeout=10)
-        assert not reader.is_alive(), f'CDC->Printer timeout ({size} bytes)'
-        assert rd_result[1] is None, f'CDC->Printer read error: {rd_result[1]}'
-        assert rd_result[0] == test_data, (f'CDC->Printer wrong data ({size} bytes):\n'
-                                           f'  expected: {test_data[:64]}\n  received: {rd_result[0][:64]}')
+        try:
+            r = hil_util.run_alongside(
+                [sys.executable, '-c', LP_READER, lp_dev, str(size), str(ready)],
+                write_cdc, LP_OPEN_TIMEOUT + 12)
+        finally:
+            ready.unlink(missing_ok=True)
+        # stderr, not stdout: run_alongside keeps the payload stream clean, so a traceback
+        # from the reader now arrives on its own pipe
+        assert r.returncode == 0, (f'CDC->Printer reader failed ({size} bytes, rc '
+                                   f'{r.returncode}): {hil_util.cmd_stdout_text(r.stderr)[:200]}')
+        assert r.stdout == test_data, (f'CDC->Printer wrong data ({size} bytes):\n'
+                                       f'  expected: {test_data[:64]}\n  received: {r.stdout[:64]}')
         time.sleep(0.2)
 
     ser.close()
 
 
 def test_device_mtp(board):
+    # The whole session lives in mtp_test.py under run_cmd: libmtp calls are synchronous
+    # ctypes that block unkillably (D state) on a wedged device, so a disposable process is
+    # the only thing the harness can walk away from.
     uid = board['uid']
-
-    # --- BEFORE: mute C-level stderr for libmtp vid/pid warnings ---
-    fd = sys.stderr.fileno()
-    _saved = os.dup(fd)
-    _null = os.open(os.devnull, os.O_WRONLY)
-    os.dup2(_null, fd)
-
-    try:
-        mtp = open_mtp_dev(uid)
-    finally:
-        # --- AFTER: restore stderr ---
-        os.dup2(_saved, fd)
-        os.close(_null)
-        os.close(_saved)
-
-    try:
-        assert b"TinyUSB" == mtp.get_manufacturer(), 'MTP wrong manufacturer'
-        assert b"MTP Example" == mtp.get_modelname(), 'MTP wrong model'
-        assert b'1.0' == mtp.get_deviceversion(), 'MTP wrong version'
-        assert b'TinyUSB MTP' == mtp.get_devicename(), 'MTP wrong device name'
-
-        # read and compare readme.txt and logo.png
-        f1_expect = b'TinyUSB MTP Filesystem example'
-        f2_md5_expect = '40ef23fc2891018d41a05d4a0d5f822f' # md5sum of logo.png
-        f1 = uid.encode("utf-8") + b'_file1'
-        f2 = uid.encode("utf-8") + b'_file2'
-        f3 = uid.encode("utf-8") + b'_file3'
-        mtp.get_file_to_file(1, f1)
-        with open(f1, 'rb') as file:
-            f1_data = file.read()
-            os.remove(f1)
-            assert f1_data == f1_expect, 'MTP file1 wrong data'
-        mtp.get_file_to_file(2, f2)
-        with open(f2, 'rb') as file:
-            f2_data = file.read()
-            os.remove(f2)
-            assert f2_md5_expect == hashlib.md5(f2_data).hexdigest(), 'MTP file2 wrong data'
-        # test send file
-        with open(f3, "wb") as file:
-            # 1524-byte payload + 12-byte MTP header = 3 full 512-byte buffers.
-            # This exercises delivery of the final OUT payload before its ZLP.
-            f3_data = bytes((i % 251) + 1 for i in range(1524))
-            file.write(f3_data)
-            file.close()
-            fid = mtp.send_file_from_file(f3, b'file3')
-            f3_readback = f3 + b'_readback'
-            mtp.get_file_to_file(fid, f3_readback)
-            with open(f3_readback, 'rb') as f:
-                f3_rb_data = f.read()
-                os.remove(f3_readback)
-                assert f3_rb_data == f3_data, 'MTP file3 wrong data'
-            os.remove(f3)
-            mtp.delete_object(fid)
-    finally:
-        mtp.disconnect()
+    script = Path(__file__).resolve().parent / 'mtp_test.py'
+    # 2x, as master's in-process open_mtp_dev used: libmtp-runtime publishes
+    # /dev/libmtp-* only after its SYNCHRONOUS mtp-probe finishes, seconds on a freshly
+    # flashed FS board, and the gio unmount eats part of what is left before the first
+    # probe. Extracting the session into a subprocess halved this by accident (8s/4s),
+    # which fails healthy hardware on the retry.
+    t = 2 * enum_timeout()
+    r = hil_util.run_cmd(
+        f'{shlex.quote(sys.executable)} {shlex.quote(str(script))} --uid {shlex.quote(uid)} --timeout {t}',
+        timeout=t + MTP_SESSION_MARGIN)
+    if r.returncode == 124:
+        # "abandoned", not "killed": a session blocked in a usbfs ioctl (D state) never
+        # receives the SIGKILL -- it lingers until its device path clears, by design
+        raise AssertionError(f'MTP session wedged (abandoned after {t + MTP_SESSION_MARGIN}s; '
+                             f'the session process may linger unkillable in D state)')
+    assert r.returncode == 0, f'MTP session failed (rc {r.returncode}):\n{r.stdout}'
 
 
 def test_device_net_lwip_webserver(board):
     # MAC hard-coded in examples/device/net_lwip_webserver/src/main.c; Linux names the
-    # USB network interface enx<MAC_lowercase_no_colons>. Device IP is 192.168.7.1 and
-    # the example runs an iperf2 TCP server on port 5001 (INCLUDE_IPERF).
+    # iface enx<MAC_lowercase_no_colons>. Device IP 192.168.7.1, iperf2 TCP server on 5001
+    # (INCLUDE_IPERF).
     import socket
     mac_no_colons = '0202846a9600'
     iface = 'enx' + mac_no_colons
     device_ip = '192.168.7.1'
     iperf_port = 5001
 
-    # Wait for the host to get an IPv4 address in the device's subnet (DHCP served by the device).
-    # USB enum + DHCP serve can take longer on the CI HIL hardware than on local — give it 30s.
+    # Wait for an IPv4 address in the device's subnet (it serves DHCP); 30s because USB
+    # enum + DHCP serve is slower on the CI HIL hardware than locally.
     iface_timeout = 30
     deadline = time.monotonic() + iface_timeout
     host_ip = None
@@ -1142,8 +1128,7 @@ def test_device_net_lwip_webserver(board):
         time.sleep(0.5)
     assert host_ip, f'USB net iface {iface} did not come up with 192.168.7.x within {iface_timeout}s'
 
-    # Poll the iperf TCP port until the device is accepting. The net stack comes up a bit
-    # after DHCP completes; iperf server binding isn't instantaneous after reflash.
+    # Poll until the device accepts: the net stack and the iperf bind come up after DHCP.
     deadline = time.monotonic() + enum_timeout()
     last_err = None
     while time.monotonic() < deadline:
@@ -1156,12 +1141,12 @@ def test_device_net_lwip_webserver(board):
             time.sleep(0.3)
     assert last_err is None, f'iperf TCP {device_ip}:{iperf_port} not accepting within {enum_timeout()}s: {last_err}'
 
-    # Throughput: 5-second iperf2 TCP test, CSV output for stable parsing.
-    # iperf2 CSV final summary line: timestamp,src_ip,src_port,dst_ip,dst_port,id,interval,bytes,bps
-    ret = subprocess.run(['iperf', '-c', device_ip, '-t', '5', '-y', 'C'],
-                         capture_output=True, text=True, timeout=30)
-    stderr = ret.stderr.strip()
-    stdout = ret.stdout.strip()
+    # 5-second iperf2 TCP test; -y C for stable parsing (final summary line is
+    # timestamp,src_ip,src_port,dst_ip,dst_port,id,interval,bytes,bps).
+    ret = hil_util.run_cmd(f'iperf -c {device_ip} -t 5 -y C',
+                           timeout=30, split_stderr=True, quiet=True)
+    stderr = (ret.stderr or '').strip()
+    stdout = (ret.stdout or '').strip()
     assert ret.returncode == 0, f'iperf rc={ret.returncode}: stderr={stderr!r} stdout={stdout!r}'
     lines = [l for l in stdout.splitlines() if l]
     assert lines, f'iperf produced no output (rc={ret.returncode}, stderr={stderr!r})'
@@ -1172,19 +1157,16 @@ def test_device_net_lwip_webserver(board):
     mbps = bps / 1e6
     print(f'  iperf {mbps:5.1f} Mbps', end='')
 
-    # Reject implausibly low throughput - a working USB-net link should clear this easily.
     assert mbps >= 1.0, f'iperf throughput too low: {mbps:.2f} Mbps'
 
 
 def test_device_msc_dual_lun(board):
     uid = board['uid']
 
-    # Read README from LUN 0
     data0 = read_disk_file(uid, 0, 'README0.TXT')
     readme0 = b"LUN0: " + MSC_README_TXT
     assert data0 == readme0, f'MSC LUN0 wrong data in README0.TXT\n  expected: {readme0}\n  received: {data0}'
 
-    # Read README from LUN 1
     data1 = read_disk_file(uid, 1, 'README1.TXT')
     readme1 = b"LUN1: " + MSC_README_TXT
     assert data1 == readme1, f'MSC LUN1 wrong data in README1.TXT\n  expected: {readme1}\n  received: {data1}'
@@ -1193,7 +1175,6 @@ def test_device_msc_dual_lun(board):
 def test_device_midi_test(board):
     uid = board['uid']
 
-    # Find MIDI device via /dev/snd/by-id using board UID
     timeout = enum_timeout()
     midi_port = None
     while timeout > 0:
@@ -1211,7 +1192,6 @@ def test_device_midi_test(board):
         timeout -= 1
     assert midi_port is not None, f'MIDI device not found for {uid}'
 
-    # Read MIDI messages and verify note on/off
     import select
     midi_fd = os.open(midi_port, os.O_RDONLY | os.O_NONBLOCK)
     try:
@@ -1246,7 +1226,6 @@ def test_device_midi_test(board):
             i += 1
 
     assert len(notes) >= 2, f'Expected at least 2 MIDI notes, got {len(notes)}'
-    # Verify notes are from the expected sequence
     note_sequence = [
         74, 78, 81, 86, 90, 93, 98, 102, 57, 61, 66, 69, 73, 78, 81, 85,
         88, 92, 97, 100, 97, 92, 88, 85, 81, 78, 74, 69, 66, 62, 57, 62,
@@ -1287,8 +1266,11 @@ def test_device_audio_test_freertos(board):
         raw_path,
     ]
 
-    ret = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
-    assert ret.returncode == 0, f'arecord failed: {ret.stderr.strip() or ret.stdout.strip()}'
+    # run_cmd: ALSA capture from a wedged device blocks in D state (see read_disk_file)
+    ret = hil_util.run_cmd(' '.join(shlex.quote(c) for c in cmd),
+                           timeout=20, split_stderr=True, quiet=True)
+    assert ret.returncode == 0, \
+        f'arecord failed: {(ret.stderr or "").strip() or (ret.stdout or "").strip()}'
 
     try:
         with open(raw_path, 'rb') as f:
@@ -1322,7 +1304,6 @@ def test_device_hid_generic_inout(board):
     uid = board['uid']
     import hid  # cython-hidapi (pip: hidapi, apt: python3-hid)
 
-    # Find HID device by UID (VID=0xCafe)
     timeout = enum_timeout()
     dev = None
     while timeout > 0:
@@ -1339,7 +1320,6 @@ def test_device_hid_generic_inout(board):
     h = hid.device()
     h.open(dev['vendor_id'], dev['product_id'], uid)
     try:
-        # Echo test: send random data and verify echo
         for size in [8, 32, 63]:
             # Report ID (0) + payload, padded to 64 bytes
             payload = bytes([random.randint(1, 255) for _ in range(size)])
@@ -1356,63 +1336,177 @@ def test_device_hid_generic_inout(board):
 
 
 def test_device_usbtest(board):
-    # Run the Linux testusb tier-4 battery (test/hil/usbtest.py) against the enumerated cafe:4010
-    # device; surface the pass count in the report cell ("✅ 30/30", or "❌ 29/30" on a partial).
+    global board_wedged
+    # Runs test/hil/usbtest.py against the cafe:4010 device; the pass count goes in the
+    # report cell ("✅ 30/30", or "❌ 29/30" on a partial).
     uid = board['uid']
 
     def usbtest_enumerated():
-        # match VID:PID too, not just the serial: right after flashing, the previous example's
-        # enumeration (same serial, different PID) can linger and would fail usbtest.py's lookup
-        for f in glob.glob('/sys/bus/usb/devices/*/serial'):
-            d = os.path.dirname(f)
-            try:
-                if (open(f).read().strip().lower() == uid.lower()
-                        and open(os.path.join(d, 'idVendor')).read().strip() == 'cafe'
-                        and open(os.path.join(d, 'idProduct')).read().strip() == '4010'):
-                    return True
-            except OSError:
-                pass
-        return False
+        """True, False, or None when a bounded read did not answer -- absence unproven."""
+        # vid_pid FIRST: right after flashing, the previous example's enumeration (same
+        # serial, different PID) can linger and would fail usbtest.py's lookup -- and
+        # filtering on the two lock-free descriptor fields rules out every other device
+        # on the bus before the one read that can block. usb_scan memoises paths that
+        # already stranded, so one wedged peer cannot spend the blindness budget here.
+        devs, unknown = hil_util.usb_scan(vid_pid=('cafe', '4010'), serial=uid)
+        if devs:
+            return True
+        return None if unknown else False
 
     end = time.monotonic() + enum_timeout()
-    while time.monotonic() < end and not usbtest_enumerated():
+    seen = usbtest_enumerated()
+    while time.monotonic() < end and seen is not True:
         time.sleep(0.2)
+        seen = usbtest_enumerated()
     # fail before usbtest_permit: an absent device would otherwise queue on the battery
     # mutex for minutes behind real batteries just to have usbtest.py report "no device"
-    if not usbtest_enumerated():
+    if seen is not True:
         # 0/30 rather than a bare cell: the battery never ran (30 = standard case count)
-        raise TestFail(f'no cafe:4010 device with serial {uid}',
-                       metric=f'{REPORT_CELL["fail"]} 0/30')
-    # settle: right after flashing the enumeration can bounce once (and on dual-port parts like
-    # CH32V307 the other port's stale usbtest node — same serial and PID — lingers a moment);
-    # running testusb into that gap sees the device drop mid-case
+        raise TestFail(
+            f'no cafe:4010 device with serial {uid}' if seen is False else
+            f'cannot tell whether cafe:4010 {uid} is present: the bounded sysfs reads did '
+            f'not answer{hil_util.sysfs_blind_note()}',
+            metric=f'{REPORT_CELL["fail"]} 0/30')
+    # settle: right after flashing the enumeration can bounce once (and on dual-port parts
+    # the other port's stale node — same serial and PID — lingers), and testusb run into
+    # that gap sees the device drop mid-case
     time.sleep(3)
 
     # --keep-binding is required for concurrent batteries: usbtest.py's cleanup unbinds
-    # EVERY usbtest-bound interface (releasing stale same-PID grabs), which would kill a
-    # peer battery mid-run under USBTEST_PARALLEL > 1; the unbind path has also wedged a
-    # host xHCI (usb_hcd_alloc_bandwidth) on this rig. Leaving bindings is harmless with
-    # unique example PIDs - the next example re-enumerates under a different PID and binds
-    # its normal driver. usbtest_permit budgets USBTEST_PARALLEL batteries per controller.
+    # EVERY usbtest-bound interface, killing a peer battery under USBTEST_PARALLEL > 1, and
+    # that unbind path has also wedged a host xHCI (usb_hcd_alloc_bandwidth) here. Harmless
+    # to leave: the next example enumerates under a different PID.
     script = Path(__file__).resolve().parent / 'usbtest.py'
-    cmd = f'python3 "{script}" --serial "{uid}" --json --keep-binding --timeout 60'
+    # --budget makes the battery a real bound: repeated case timeouts (a FAIL, not a HUNG,
+    # so the battery keeps going) can otherwise spend the whole outer timeout inside the
+    # case loop, leaving the recovery below nothing.
+    cmd = (f'{shlex.quote(sys.executable)} {shlex.quote(str(script))} '
+           f'--serial {shlex.quote(uid)} --json --keep-binding '
+           f'--timeout 60 --budget {USBTEST_BATTERY_BUDGET}')
+    # Post-hang recovery reflashes the DUT through its own probe, NEVER a root-port cycle
+    # (one board reached instead of every fixture under the port; see usb-kernel-recover).
+    # _current_fw is the artifact test_example flashed for THIS test: re-deriving it from
+    # board['name'] reflashes the wrong build on variant-only boards. --outer-timeout lets
+    # usbtest skip a reflash it cannot finish before our run_cmd kill, which would orphan
+    # the flasher (own session) on the probe. Never under --skip-flash -- and say so: a
+    # HUNG case then holds the DUT's usbfs lock for the rest of the run, and a probe reset
+    # is no substitute (the DWC2 pullup survives a core halt).
+    # ...and only when this flasher can DELIVER that reflash past a poisoned node
+    # (hil_flash.convoy_safe). Otherwise the flags cost twice: the delivery adds a SECOND
+    # stray, and the board reserves recovery budget for a path that cannot fire.
+    # The RECOVERY flasher, which may be the roster's optional `flasher_recover` rather
+    # than the primary -- a jlink/stlink board can name an openocd entry that reaches the
+    # same probe convoy-safely without changing how the board is normally flashed.
+    _rec_flasher = hil_flash.recover_flasher(board)
+    recovery = bool(_current_fw and not skip_flash and hil_flash.convoy_safe(_rec_flasher))
+    # ONE bound, computed here and used for BOTH the child's --outer-timeout and our own
+    # run_cmd kill below. Three separate expressions disagreed: --skip-flash appended no
+    # --outer-timeout at all (usbtest reads 0 as "no limit"), and the no-recovery branch
+    # narrowed only the CHILD's view while run_cmd still waited the full reserve -- so a
+    # board that cannot recover held a pool worker AND its battery permit idle for
+    # USBTEST_RECOVERY_BUDGET it had no way to spend, under a usbtest width of 2.
+    outer = USBTEST_BATTERY_BUDGET + (USBTEST_RECOVERY_BUDGET if recovery
+                                      else USBTEST_OVERSHOOT)
+    if _current_fw and skip_flash:
+        print('note: --skip-flash disables usbtest hang recovery; a HUNG case will leave '
+              'the device wedged until it is reflashed', flush=True)
+    elif _current_fw and not recovery:
+        print(f'note: {_rec_flasher["name"]} cannot deliver a reflash past a poisoned '
+              f'usbfs node, so usbtest hang recovery is disabled for {board["name"]}; a '
+              f'HUNG case will leave it wedged for the rest of the run', flush=True)
+    if recovery:
+        # ship the RECOVERY flasher as `flasher`: usbtest.py, recovery_steps and
+        # convoy_safe all read board['flasher'], so substituting here keeps the entire
+        # child side unaware that a second roster entry exists
+        rb = json.dumps({'name': board['name'], 'flasher': _rec_flasher})
+        cmd += f' --recover-board {shlex.quote(rb)} --recover-fw {shlex.quote(_current_fw)}'
+    cmd += f' --outer-timeout {outer}'
+    # The reserve above USBTEST_BATTERY_BUDGET exists because the battery can overrun by
+    # one already-started case, and a hang there needs room for the recovery (whose reflash
+    # is bounded by usbtest.RECOVER_FLASH_TIMEOUT, not HIL_CMD_TIMEOUT). Without it run_cmd
+    # SIGKILLs usbtest.py mid-recovery, losing the JSON and the diagnosis.
     with hil_lock.usbtest_permit(uid):
-        r = hil_flash.run_cmd(cmd, timeout=200)
-    out = hil_flash.cmd_stdout_text(r.stdout)
+        # split_stderr: the battery's final JSON is parsed from stdout, and stderr is the
+        # only detail left when the outer timeout kills the battery before it prints
+        r = hil_util.run_cmd(cmd, timeout=outer, split_stderr=True)
+    out = hil_util.cmd_stdout_text(r.stdout)
     brace = out.find('{')
     try:
+        # brace < 0 would slice from the END ('...rc 0' -> '0' -> int 0, whose subscript
+        # raises TypeError outside the tuple below and loses the diagnosis)
+        if brace < 0:
+            raise ValueError('no JSON object on stdout')
         data = json.loads(out[brace:])
         passed, failed = int(data['passed']), int(data['failed'])
-    except (ValueError, KeyError, json.JSONDecodeError):
-        raise TestFail(f'usbtest did not run: {compact_output(out) or hil_flash.cmd_stdout_text(r.stderr)}',
+    except (ValueError, KeyError, TypeError, json.JSONDecodeError):
+        # compact BOTH, never `or`: a battery SIGKILLed mid-print leaves a truthy JSON
+        # fragment on stdout, so an `or` drops the stderr that explains the failure
+        parts = [compact_output(hil_util.cmd_stdout_text(r.stderr)), compact_output(out)]
+        detail = ' | '.join(p for p in parts if p)
+        # Retryable even on rc 124 (run_cmd's outer kill), though the retry re-pays the
+        # whole budget: 124 only says the timer expired, which a healthy battery can hit
+        # under load, and test_example REFLASHES before each attempt. Where usbtest's
+        # in-band recovery is off (--skip-flash, a flasher failing convoy_safe, a terminal
+        # wedge) that reflash is the only thing left to unpoison the DUT for the boards
+        # that share its controller.
+        # No JSON to read the verdict from, so fall back to the text: a battery SIGKILLed
+        # mid-hang still says HUNG on stdout, and this raise happens BEFORE the latch below
+        # -- which is why the outer-timeout case, the likeliest real wedge, never latched.
+        if 'HUNG' in out:
+            board_wedged = (f'{board["name"]}: usbtest reported a hang and was killed '
+                            f'before it could report a verdict')
+        raise TestFail(f'usbtest did not run: {detail}',
                        metric=f'{REPORT_CELL["fail"]} 0/30')
 
-    total = passed + failed
-    if failed == 0 and total > 0:
+    # A HUNG case that recovery could not clear leaves a D-state holder on this board's
+    # usbfs node. Latch it: the remaining examples would each flash THROUGH that node,
+    # block, survive SIGKILL and add another stray -- turning one wedge into one stray per
+    # remaining example, which is the convoy this branch exists to contain.
+    # The battery's OWN verdict first: `recovery` only says the flags were passed, not that
+    # the reflash worked, so a convoy-safe board whose recovery failed used to come back
+    # unlatched and flash every remaining example through the poisoned node.
+    if data.get('wedged') or (not recovery and 'HUNG' in out):
+        # _rec_flasher, NOT board['flasher']: recovery was decided against recover_flasher()
+        # at the top of this function, and the two diverge as soon as a roster carries the
+        # optional `flasher_recover` key -- naming the wrong one sends the operator to the
+        # wrong probe. The wording stays on what usbtest actually reported ("still wedged"),
+        # because unrecovered_hang is also set by the ambiguous/inconclusive aborts, where
+        # nothing hung and the old text was false on both clauses.
+        board_wedged = (f'{board["name"]}: usbtest reports the device still wedged '
+                        + (f'after a recovery reflash via {_rec_flasher["name"]}' if recovery
+                           else f'and {_rec_flasher["name"]} cannot deliver a recovery reflash'))
+
+    # notrun counts toward the denominator but is NOT a failure: listing cases that never
+    # ran as failures sends a maintainer bisecting one of them.
+    notrun = int(data.get('notrun', 0))
+    total = passed + failed + notrun
+    if board_wedged and failed == 0 and notrun == 0:
+        # Every case passed and the device STILL wedged -- usbtest's inconclusive/ambiguous
+        # abort fires after the last case, so nothing back-fills a BUDGET entry. Reporting
+        # the pass would exit 0 with a D-state holder on the rig and the board absent from
+        # the re-run spec. parsed=True: a retry re-pays the whole battery to re-observe a
+        # wedge, and flashes through the poisoned node to do it.
+        raise TestFail(f'usbtest {passed}/{total} but the device wedged ({board_wedged})',
+                       metric=f'{REPORT_CELL["fail"]} {passed}/{total}', parsed=True)
+    if failed == 0 and notrun == 0 and total > 0:
         return f'{REPORT_CELL["pass"]} {passed}/{total}'
-    bad = [c.get('num') for c in data.get('cases', []) if c.get('status') != 'PASS']
-    raise TestFail(f'usbtest {passed}/{total} (cases failed: {bad})',
-                   metric=f'{REPORT_CELL["fail"]} {passed}/{total}')
+    bad = [c.get('num') for c in data.get('cases', [])
+           if c.get('status') not in ('PASS', 'BUDGET')]
+    why = f'usbtest {passed}/{total}'
+    if bad:
+        why += f' (cases failed: {bad})'
+    if notrun:
+        # the reason is per BUDGET entry: a hang or a device drop also aborts the battery,
+        # and blaming the budget points the maintainer at the wrong thing
+        reasons = {c.get('detail', '') for c in data.get('cases', [])
+                   if c.get('status') == 'BUDGET'}
+        reason = (reasons.pop().replace('not run: ', '') if len(reasons) == 1
+                  else 'the battery stopped early')
+        why += f'; {notrun} case(s) never ran ({reason}), so this says nothing about them'
+    # parsed ONLY when every case ran: an aborted battery (budget expiry, kernel hang, bus
+    # drop) leaves BUDGET entries, and those are exactly what a reflash retry can fix.
+    raise TestFail(why, metric=f'{REPORT_CELL["fail"]} {passed}/{total}',
+                   parsed=(notrun == 0))
 
 
 # -------------------------------------------------------------
@@ -1437,42 +1531,68 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str, st
 
     test_name = f'{variant:40} {example:30} ...'
 
-    # --skip-flash runs whatever is already on the board, so any build counts as present:
-    # only the flashing path needs the artifact this board's flasher actually consumes.
-    # Filtering there too would skip the test as "no binary" over an extension it never uses.
+    # --skip-flash runs whatever is already on the board, so any build counts as present;
+    # filtering by flasher there would skip the test over an extension it never uses.
     fw_name = hil_flash.find_firmware(variant, example,
                                       flasher=None if skip_flash else board['flasher']['name'])
     if fw_name is None:
         log_line(f'{test_name} Skip (no binary)')
         return 0, 'skip', None
+    # usbtest's hang recovery reflashes the exact artifact under test; re-deriving it from
+    # board['name'] breaks on variant-only boards
+    global _current_fw
+    _current_fw = str(fw_name)
 
     if verbose:
         log_line(f'Firmware {fw_name}')
 
-    # flash firmware (unless --skip-flash), then run the test. Both may fail randomly,
-    # retry a few times.
     global _enum_timeout
     start_s = time.time()
     flash_ok = True
     last_err = ''
     last_detail = ''
+    wedge_break = False
     for i in range(max_retry):
+        if board_wedged and i:
+            # The latch is set MID-attempt (a HUNG usbtest whose flasher cannot recover),
+            # so test_board's check between tests is too late for THIS test's own retries:
+            # every further attempt re-flashes into the D-state-held node, blocks, survives
+            # SIGKILL and leaves another stray. The wedge is not something a retry can fix.
+            log_line(f'{test_name}  not retrying: {board_wedged}')
+            # COUNT it. Breaking out here skips the i == max_retry - 1 branch that would
+            # have incremented err_count, so the board rendered a red cell, contributed 0
+            # to the exit status and was omitted from the re-run spec -- a rig left with a
+            # D-state holder published under sys.exit(0). Latent at CI's --retry 1, live
+            # for every local run and for the workflows that pass no -r.
+            wedge_break = True
+            break
         _enum_timeout = ENUM_TIMEOUT if i == 0 else ENUM_TIMEOUT_RETRY
         attempt_out = io.StringIO()
         with redirect_stdout(attempt_out):
             if not skip_flash:
                 with hil_lock.flash_permit(board['uid']):
                     t_flash = time.monotonic()
-                    ret = getattr(hil_flash, f'flash_{board["flasher"]["name"].lower()}')(board, str(fw_name))
+                    try:
+                        ret = getattr(hil_flash,
+                                      f'flash_{board["flasher"]["name"].lower()}')(board, str(fw_name))
+                    except Exception as e:
+                        # A flasher that RAISES (esptool's get_serial_dev when the adapter
+                        # drops off the bus, a missing config.env, an unwritable CWD) would
+                        # propagate out of the worker and abort the whole drain, costing
+                        # every board still in flight.
+                        print(f'flash raised: {type(e).__name__}: {e}', flush=True)
+                        ret = subprocess.CompletedProcess(args='flash', returncode=1,
+                                                          stdout=f'{type(e).__name__}: {e}')
                     if PROFILE:
                         log_line(f'[prof] {variant} {example} flash attempt {i + 1}: '
                                  f'{time.monotonic() - t_flash:.1f}s rc={ret.returncode}')
                     flash_ok = (ret.returncode == 0)
-                    # A wedged RP2040/RP2350 DAP answers nothing and the probe has no reset
-                    # line, so the retry would fail identically; POR it via the Rescue DP
-                    # first. No-op for every other board and every other flash failure.
-                    if not flash_ok and i + 1 < max_retry and \
-                            hil_flash.rescue_openocd(board, hil_flash.cmd_stdout_text(ret.stdout)):
+                    # A wedged RP2040/RP2350 DAP answers nothing and the probe has no
+                    # reset line, so the retry fails identically; POR it via the Rescue DP
+                    # first (no-op otherwise). NOT gated on a remaining attempt: CI HIL jobs
+                    # run --retry 1, and this leaves the DAP POR'd for the jobs that follow.
+                    if not flash_ok and \
+                            hil_flash.rescue_openocd(board, hil_util.cmd_stdout_text(ret.stdout)):
                         log_line(f'{variant} {example}: DAP wedged, rescued via Rescue DP')
             if flash_ok:
                 try:
@@ -1484,7 +1604,6 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str, st
                     else:
                         status = STATUS_OK
                         result_status = 'pass'
-                        # a test may return a string to show in its report cell (e.g. speed)
                         metric = tret if isinstance(tret, str) else None
                     msg = f'{test_name}  {status}'
                     if last_detail:
@@ -1495,9 +1614,20 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str, st
                 except Exception as e:
                     last_err = str(e)
                     last_detail = compact_output(attempt_out.getvalue())
+                    if getattr(e, 'parsed', False):
+                        # a PARSED per-case result (usbtest's "29/30"): retrying re-pays
+                        # the whole battery, inside the fleet's usbtest permit, to
+                        # re-observe a number the JSON already reported. Only that case.
+                        err_count += 1
+                        metric = getattr(e, 'metric', None)
+                        msg = f'{test_name}  {STATUS_FAILED}: {e}'
+                        if last_detail:
+                            msg += f'  {last_detail}'
+                        msg += f'  in {time.time() - start_s:.1f}s'
+                        log_line(msg)
+                        break
                     if i == max_retry - 1:
                         err_count += 1
-                        # a failing test may still carry a metric to show in its cell (e.g. "❌ 29/30")
                         metric = getattr(e, 'metric', None)
                         msg = f'{test_name}  {STATUS_FAILED}: {e}'
                         if last_detail:
@@ -1530,13 +1660,21 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str, st
         msg += f'  in {time.time() - start_s:.1f}s'
         log_line(msg)
 
+    if wedge_break and not err_count:
+        # ONE error for the test, never two: a board that also failed to flash has already
+        # been counted just above. Without this the test returns 0 -- red cell, clean exit
+        # status, absent from the re-run spec.
+        err_count += 1
     return err_count, result_status, metric
 
 
 def build_board(board: Board) -> tuple[str, int]:
     """Build firmware for this board via tools/build.py.
     Honors board config's variant list and build.args defines.
-    Output goes to cmake-build/cmake-build-<variant>/ (tools/build.py layout)."""
+    Output goes to cmake-build/cmake-build-<variant>/ (tools/build.py layout).
+
+    Unbounded on purpose: --build is a local convenience (no CI workflow passes it), so
+    the developer watching the build is the timeout."""
     name = board['name']
     bcfg = cast(BuildCfg, board.get('build', {}))
     extra_defs = bcfg.get('args', [])
@@ -1544,7 +1682,7 @@ def build_board(board: Board) -> tuple[str, int]:
 
     failed = 0
     for v in variants:
-        cmd = [sys.executable, str(hil_flash.TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name]
+        cmd = [sys.executable, str(hil_util.TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name]
         for d in extra_defs:
             cmd += ['-D', d]
         if v['name'] != name:
@@ -1556,8 +1694,19 @@ def build_board(board: Board) -> tuple[str, int]:
         if verbose:
             cmd.append('-v')
             print(f'  + {" ".join(cmd)}')
-        r = subprocess.run(cmd, cwd=hil_flash.TINYUSB_ROOT)
-        if r.returncode != 0:
+        # stdio is inherited so the build STREAMS: a silent buffer is
+        # indistinguishable from a stall.
+        proc = subprocess.Popen(cmd, cwd=hil_util.TINYUSB_ROOT, start_new_session=True)
+        try:
+            rc = proc.wait()
+        except KeyboardInterrupt:
+            # start_new_session means the build never saw the terminal's SIGINT
+            try:
+                os.killpg(proc.pid, signal.SIGKILL)
+            except OSError:
+                proc.kill()
+            raise
+        if rc != 0:
             failed += 1
     return name, failed
 
@@ -1567,28 +1716,29 @@ BOUNDARY_CELL = 'same-PID boundary'
 
 
 def test_board(board: Board) -> tuple[str, int, list[str], list, float]:
+    swept = False
     name = board['name']
     flasher = board['flasher']
 
+    global board_wedged
+    board_wedged = ''
     try:
         _lock_fh = hil_lock.acquire_board_lock(name)
     except RuntimeError as e:
         log_line(f'{name:25} {STATUS_FAILED}: {e}')
-        # visible report row so the ❌ matches the exit code; failed-tests stays
-        # empty so a re-run repeats the whole board (no bogus -bt test filter)
+        # visible report row so the ❌ matches the exit code; failed-tests stays empty so a
+        # re-run repeats the whole board (no bogus -bt filter)
         return name, 1, [], [(name, {'board-locked': 'fail'}, None)], 0.0
     # after the lock: flock wait behind a concurrent run is not board cost
     t_board = time.monotonic()
     try:
-        # default to all tests
         test_list = []
 
         if name in board_test:
             test_list = board_test[name]
         elif len(test_only) > 0:
-            # Explicit -t: filter against the board's capabilities so a device-only
-            # board doesn't try to run host/dual tests (the test functions need a
-            # `dev_attached` entry in the board config that won't exist).
+            # Explicit -t: filter against the board's capabilities, or a device-only board
+            # runs host/dual tests whose `dev_attached` config entry does not exist.
             board_tests = board.get('tests', {})
             if 'only' in board_tests:
                 allowed = set(board_tests['only'])
@@ -1618,7 +1768,7 @@ def test_board(board: Board) -> tuple[str, int, list[str], list, float]:
         err_count = 0
         failed_tests = []
         board_wide_fail = False  # re-run the whole board, not a subset of its tests
-        rows = []  # list of (row_label, {example: status}, duration) — one row per build variant
+        rows = []  # list of (row_label, {example: status}, duration) — one per build variant
         # a -t/-bt filtered run times only a subset; report no duration so an accumulate
         # re-run keeps the previous full-run value
         partial = bool(test_only) or name in board_test
@@ -1627,11 +1777,10 @@ def test_board(board: Board) -> tuple[str, int, list[str], list, float]:
         prev_last = None  # last test of the previous variant: the variant boundary is an adjacency too
         for v in variants:
             vname = v['name']
-            # Shuffle each (board, variant)'s run order — de-synchronizes the worker pool so
-            # usbtest batteries and flash churn spread across the timeline instead of convoying,
-            # and surfaces order-dependent bugs. Seeded for replay (HIL_SHUFFLE_SEED, logged by
-            # main). Unique per-example PIDs make any two different examples re-enumerate; only
-            # the variant boundary can repeat the same example (same PID) — swap it away.
+            # Shuffle each (board, variant)'s run order: spreads batteries and flash churn
+            # across the timeline instead of convoying, and surfaces order-dependent bugs.
+            # Seeded for replay (HIL_SHUFFLE_SEED). Unique per-example PIDs re-enumerate
+            # between examples; only the variant boundary can repeat one.
             run_list = list(test_list)
             if shuffle_seed is not None and len(run_list) > 1:
                 random.Random(f'{shuffle_seed}:{name}:{vname}').shuffle(run_list)
@@ -1639,23 +1788,33 @@ def test_board(board: Board) -> tuple[str, int, list[str], list, float]:
                     run_list[0], run_list[-1] = run_list[-1], run_list[0]
             cells = {}
             if run_list and run_list[0] == prev_last and not skip_flash:
-                # Same example (same PID) still repeats across the boundary: a one-test
-                # list (the common case for a -bt scoped run) leaves nothing to swap
-                # with. Park on board_test first - it disables the board's USB, so the
-                # PID goes away and the next flash must re-enumerate to be seen.
+                # Same example (same PID) still repeats across the boundary (a one-test
+                # -bt run has nothing to swap with). Park on board_test first: it disables
+                # the board's USB, so the next flash must re-enumerate to be seen.
                 t_park = time.monotonic()
-                park_ec, park_status, _ = test_example(board, vname, 'device/board_test')
+                # _should_park, same as the teardown park: this is attempt 0, so
+                # test_example's retry guard does not stop it flashing into a poisoned node
+                park_ec, park_status, _ = (
+                    test_example(board, vname, 'device/board_test') if _should_park(skip_flash)
+                    else (0, 'skip', None))
                 if park_ec or park_status == 'skip':
-                    # Boundary not cleared: the previous variant's device may still be
-                    # enumerated under the same PID, so this variant's tests could pass
-                    # against its firmware. Skip them - a false green proves nothing and
-                    # is worse than a gap - and record the boundary itself as the failure
-                    # (a visible ❌ cell, mirroring the board-lock row above) so the report
-                    # matches the exit code instead of rendering all-green.
-                    why = 'no board_test binary' if park_status == 'skip' else 'park flash failed'
+                    # Boundary not cleared: the previous variant may still be enumerated
+                    # under the same PID, so this variant's tests could pass against ITS
+                    # firmware. Skip them and record the boundary as the failure, so the
+                    # report matches the exit code instead of rendering all-green.
+                    # A 'skip' here has two very different causes: no board_test build, or
+                    # _should_park refusing to flash a WEDGED board. Reporting the latter as
+                    # a missing binary sends the operator hunting a build that exists.
+                    wedge_skip = park_status == 'skip' and bool(board_wedged)
+                    why = ('the board is wedged' if wedge_skip else
+                           'no board_test binary' if park_status == 'skip' else
+                           'park flash failed')
                     log_line(f'{vname:40} {"same-PID boundary":30} {STATUS_FAILED}: not cleared ({why}); '
                              f'skipping {len(run_list)} test(s) on this variant')
-                    err_count += 1
+                    # the wedge already charged its own error through test_device_usbtest;
+                    # charging again would double-count one incident in the exit code
+                    if not wedge_skip:
+                        err_count += 1
                     cells[BOUNDARY_CELL] = 'fail'
                     # blaming run_list[0] would re-run an innocent test that then passes,
                     # leaving the boundary unretested; re-run the whole board instead
@@ -1668,31 +1827,72 @@ def test_board(board: Board) -> tuple[str, int, list[str], list, float]:
                 prev_last = run_list[-1]
             t_variant = time.monotonic()
             for test in run_list:
+                if board_wedged:
+                    # Do NOT flash through a poisoned node: each attempt enumerates into
+                    # it, blocks uninterruptibly and leaves another stray behind. Report
+                    # the skip so the cell is not mistaken for a pass.
+                    cells[test] = f'{REPORT_CELL["skip"]} board wedged'
+                    # ...and re-run the WHOLE board, like the boundary-failure path above:
+                    # these tests never executed, so naming them individually in the .failed
+                    # spec is not enough -- an --accumulate re-run that fixes only the wedged
+                    # test would merge a green cell over it and leave these skips standing
+                    # from the earlier attempt, forever, under a green job.
+                    board_wide_fail = True
+                    continue
                 ec, status, metric = test_example(board, vname, test)
                 err_count += ec
                 cells[test] = metric if metric else status
                 if ec > 0:
                     failed_tests.append(test)
+            if board_wedged:
+                log_line(f'{vname:40} SKIPPING the rest of this board: {board_wedged}; '
+                         f'flashing through the poisoned node would add a stray per test')
             dur = f'{time.monotonic() - t_variant:.0f}s' if run_list and not partial else None
             rows.append((vname, cells, dur))
 
-        # board duration excludes the teardown park-flash below; a partial (filtered)
-        # run reports 0.0 so it never overwrites a cached full-run duration
+        # excludes the teardown park-flash below; a partial (filtered) run reports 0.0 so
+        # it never overwrites a cached full-run duration
         t_total = 0.0 if partial else time.monotonic() - t_board
 
-        # flash board_test last to disable board's usb (skipped when --skip-flash is set);
-        # this is teardown/park, not a test — not recorded in the report
-        if not skip_flash:
+        # park: flash board_test last to disable the board's usb; teardown, not a test,
+        # so it is not recorded in the report.
+        #
+        # NOT on a wedged board: the latch has just skipped every remaining test precisely
+        # because flashing through a D-state-held node blocks, survives SIGKILL and leaves
+        # a stray -- and this park is a flash like any other. test_example's own guard does
+        # not stop it (that one only suppresses RETRIES, and this is attempt 0), so the
+        # containment path would add the very stray it exists to prevent.
+        if _should_park(skip_flash):
             test_example(board, variants[0]['name'], 'device/board_test')
 
-        return name, err_count, [] if board_wide_fail else sorted(set(failed_tests)), rows, t_total
+        # Sweep HERE, not in main()'s finally: maxtasksperchild=1 retires this process as
+        # soon as it returns, reparenting anything it spawned to init and off the pool's
+        # ppid tree, so the main-side sweep walks fresh idle workers and finds nothing.
+        # Measured: 4 tasks, zero overlap, sweep 0, all 4 strays alive.
+        stray = hil_health.kill_own_children()
+        swept = True
+
+        # LAST fields: whether this worker ran out of bounded-read budget, and what it could
+        # not kill. Only the worker can answer either -- the blindness latch is
+        # process-global and this is a separate process -- and the result tuple already
+        # crosses back, so no Manager round-trip.
+        return (name, err_count, [] if board_wide_fail else sorted(set(failed_tests)),
+                rows, t_total, hil_util.sysfs_blind(), stray)
     finally:
+        # A raise skips the sweep above, and maxtasksperchild=1 retires this process
+        # immediately afterwards -- reparenting its flasher to init and erasing the ppid
+        # link, so main's sweep cannot see it either. The count cannot reach the report on
+        # this path (there is no result tuple), but the KILL still frees the probe.
+        if not swept:
+            try:
+                hil_health.kill_own_children()
+            except Exception as se:   # noqa: BLE001 - never mask the original failure
+                print(f'warning: stray sweep failed: {type(se).__name__}: {se}', flush=True)
         if _lock_fh:
             try:
-                # clear our pid record before dropping the flock: this worker
-                # process lives on (pool reuse), so a stale record would make
-                # hil_lock.py's pid-liveness checks report a freed board as
-                # still locked for the rest of the run
+                # clear our pid record before dropping the flock: this worker process
+                # lives on (pool reuse), so a stale record would make hil_lock's
+                # pid-liveness checks report a freed board as locked for the rest of the run
                 _lock_fh.truncate(0)
             except OSError:
                 pass
@@ -1701,10 +1901,9 @@ def test_board(board: Board) -> tuple[str, int, list[str], list, float]:
 
 REPORT_MD = 'hil_report.md'
 REPORT_JSON = 'hil_report.json'
-# controller hints learned from previous runs: uid -> {'name', 'pci', 'duration'}. Only
-# 'pci' is consumed (dispatch order and first-flash budgeting, never battery
-# serialization); name/duration are informational. PCI addresses are boot-stable (bus
-# numbers are not), so the cache survives reboots and only goes stale on re-cabling.
+# controller hints from previous runs: uid -> {'name', 'pci', 'duration'}. Only 'pci' is
+# consumed (dispatch order and first-flash budgeting, never battery serialization). PCI
+# addresses are boot-stable, so the cache survives reboots and goes stale on re-cabling.
 CONTROLLER_CACHE = Path.home() / '.cache' / 'tinyusb-hil' / 'controller_cache.json'
 
 
@@ -1729,8 +1928,8 @@ def render_matrix(rows_all: list) -> str:
     if not seen:
         return 'No tests were run.'
 
-    # metric-bearing columns pinned first (usbtest score, throughput, explorer read speed),
-    # the rest alphabetical by bare test name: stable regardless of the (shuffled) execution order
+    # metric-bearing columns pinned first, the rest alphabetical: stable regardless of the
+    # shuffled execution order
     pinned = ['usbtest', 'cdc_msc_throughput', 'msc_file_explorer', 'msc_file_explorer_freertos']
 
     def col_key(t):
@@ -1761,9 +1960,8 @@ def render_matrix(rows_all: list) -> str:
     sep = '| ' + '-' * board_w + ' | ' + ' | '.join(':' + '-' * (w - 2) + ':' for w in col_w) + ' |'
     body = [line(lbl, vals) for lbl, vals in rows_vals]
 
-    # tally run cells (blank/not-run cells are absent from the dicts). A cell is a bare status
-    # ('pass'/'fail'/'skip') or a metric string that carries its own icon (e.g. "❌ 29/30" is a
-    # fail, "✅ 30/30" / "✅ CDC …" a pass), so classify by the leading icon.
+    # tally run cells (not-run cells are absent from the dicts). A cell is a bare status or
+    # a metric string carrying its own icon ("❌ 29/30"), so classify by the leading icon.
     def cell_kind(v):
         if v == 'fail' or (isinstance(v, str) and v.startswith(REPORT_CELL['fail'])):
             return 'fail'
@@ -1780,7 +1978,119 @@ def render_matrix(rows_all: list) -> str:
     return summary + '\n\n' + '\n'.join([header, sep] + body)
 
 
-def accumulate_report(mret: list, report_dir: Path, fresh: bool, scope: str = '') -> str:
+def _write_failed_spec(failed_fname: Path, report_dir: Path, mret: list) -> None:
+    """Re-run spec: only the failed boards (-b), each restricted to its own failed tests
+    (-bt); a board with failures but no test list re-runs entirely.
+
+    Shared with the pool-guard path, which feeds it the boards that never reported. That
+    path used to leave this unwritten -- and a fresh run has already unlinked it -- so
+    build.yml's "Get re-run spec" step found nothing and the GitHub re-run repeated the
+    whole fleet to find the one board that wedged."""
+    parts = ['--accumulate']
+    for name, err, fts, *_ in mret:
+        if err > 0:
+            parts.append(f'-b {name}')
+            if fts:
+                parts.append(f'-bt {name}:{",".join(fts)}')
+    if len(parts) > 1:  # build-only failures have no boards to re-run
+        report_dir.mkdir(parents=True, exist_ok=True)
+        with failed_fname.open('w') as f:
+            f.write(' '.join(parts))
+    else:
+        failed_fname.unlink(missing_ok=True)
+
+
+class PoolDrainTimeout(MpTimeoutError):
+    """Guard expiry, carrying the rows that DID finish.
+
+    They ride on the exception because the raise is the containment path: losing them here
+    is what map_async did, and what the drain exists to stop.
+    """
+
+    def __init__(self, finished: list):
+        super().__init__()
+        self.finished = finished
+
+
+def drain_pool(it, boards: list, deadline: float, out: list | None = None) -> list:
+    """Collect imap_unordered results against ONE deadline. Returns the finished rows.
+
+    Raises PoolDrainTimeout (carrying those same rows) when the deadline passes with boards
+    still in flight -- the caller keeps them, names only what is missing, and writes a
+    re-run spec covering just those.
+
+    A function, not an inline loop, so the tests can call THIS instead of a copy of it: the
+    loop's previous test built its own ThreadPool and its own drain and asserted on those,
+    so deleting the real one outright kept the suite green.
+    """
+    # `out` is the CALLER's list: a worker that raises something other than a timeout
+    # (get_serial_dev on a dropped adapter, a Manager EOFError) propagates bare, and a
+    # local accumulator would take every finished board with it -- the exact loss the
+    # drain replaced map_async to prevent.
+    mret: list = out if out is not None else []
+    for _ in boards:
+        left = deadline - time.monotonic()
+        if left <= 0:
+            raise PoolDrainTimeout(mret)
+        try:
+            mret.append(it.next(timeout=left))
+        except MpTimeoutError:
+            raise PoolDrainTimeout(mret) from None
+    return mret
+
+
+def _should_park(skip_flash: bool) -> bool:
+    """Flash the teardown park (device/board_test, to switch the DUT's USB off)?
+
+    Not on a wedged board. The latch has just skipped every remaining test precisely
+    because flashing through a D-state-held node blocks, survives SIGKILL and leaves a
+    stray -- and the park is a flash like any other. test_example's own guard does not stop
+    it either: that one only suppresses RETRIES, and the park is always attempt 0. So the
+    containment path would end by adding the very stray it exists to prevent.
+    """
+    return not skip_flash and not board_wedged
+
+
+def _stray_note(mret: list) -> str:
+    """Name the strays the workers could not kill, for the report banner.
+
+    Summed from the result tuples rather than computed in main()'s finally: that finally
+    runs AFTER accumulate_report on both abort paths, so a banner appended there was
+    written to a variable nobody read again.
+    """
+    dirty = [(r[0], r[6]) for r in mret if len(r) > 6 and r[6]]
+    if not dirty:
+        return ''
+    total = sum(n for _, n in dirty)
+    return (f'> **Rig dirty.** {total} process(es) survived SIGKILL and still hold a probe '
+            f'or usbfs node into the next job: '
+            f'{", ".join(f"{b} ({n})" for b, n in dirty)}.\n')
+
+
+def _blind_note(mret: list) -> str:
+    """Name the boards whose worker went blind, for the report banner.
+
+    A blind worker answers SYSFS_UNKNOWN for every attribute, so its "device not found" is
+    "could not tell". That already reaches the log and the per-cell failure text, but the
+    TABLE is what gets quoted -- and a red cell there is read as a broken board. Seen live
+    (run 31794359407): four workers blind, several cells red because of it, and a report
+    that said nothing.
+
+    Per-board, not global: maxtasksperchild=1 gives every board a fresh worker, so a board
+    that ran on a healthy one is not smeared by a neighbour's wedge. Rows synthesised by
+    the timeout path are 5 fields wide and have nothing to report.
+    """
+    blind = [r[0] for r in mret if len(r) > 5 and r[5]]
+    if not blind:
+        return ''
+    return (f'> **Not all verdicts are evidence.** {len(blind)} board(s) ran on a worker '
+            f'that went blind on sysfs -- too many bounded reads stranded on a wedged '
+            f'device -- so "not found" from them means "could not tell": '
+            f'{", ".join(blind)}. See the usb-kernel-recover skill.\n')
+
+
+def accumulate_report(mret: list, report_dir: Path, fresh: bool, scope: str = '',
+                      banner: str = '') -> str:
     """Merge this run's results into hil_report.json in report_dir, then (re)write
     the markdown matrix to hil_report.md. `fresh` (a first run, no --accumulate)
     starts a new report; otherwise a re-run accumulates so boards/tests that
@@ -1788,29 +2098,34 @@ def accumulate_report(mret: list, report_dir: Path, fresh: bool, scope: str = ''
     board filter, if any, so a scoped table is not mistaken for a full one.
     Returns the md."""
     acc = {}  # ordered {row_label: [cells dict, duration str|None]}
+    prior_banner = ''
     jpath = report_dir / REPORT_JSON
     if not fresh and jpath.is_file():
         try:
             saved = json.loads(jpath.read_text())
-            # CI keys the report dir by run id, so the sidecar can only have been
-            # written by an earlier attempt of the same run
+            # CI keys the report dir by run id, so the sidecar is from an earlier attempt
             for entry in saved.get('rows', []):
                 acc[entry['board']] = [dict(entry['cells']), entry.get('duration')]
+            # ... and so is the caveat those cells were collected under. A rerun on a rig
+            # that has since recovered contributes no banner, and the .failed spec reruns
+            # only FAILURES -- so the earlier attempt's passes are never re-earned and
+            # would be published as clean results of a rig that was not.
+            prior_banner = saved.get('banner', '')
         except (ValueError, KeyError, TypeError):
             pass  # corrupt/old sidecar: start fresh
 
-    # merge this run: current cells override prior for boards/tests that ran; a filtered
-    # run reports duration None, keeping the previous full-run value
-    for name, _, _, rows, _ in mret:
+    # current cells override prior for boards/tests that ran; a filtered run reports
+    # duration None, keeping the previous full-run value
+    for name, _, _, rows, *_ in mret:
         if rows and not any('board-locked' in cells for _, cells, _ in rows):
-            # board ran for real this time: clear a stale lock-failure cell
-            # (its row is keyed by board name; test rows may be variant names)
+            # board ran for real: clear a stale lock-failure cell (its row is keyed by
+            # board name; test rows may be variant names)
             stale = acc.get(name)
             if stale is not None:
                 stale[0].pop('board-locked', None)
                 if not stale[0]:
-                    # variant-keyed boards never repopulate the board-name row —
-                    # drop it or it renders as a blank ghost row
+                    # variant-keyed boards never repopulate the board-name row, so drop it
+                    # or it renders as a blank ghost row
                     del acc[name]
         for row_label, cells, dur in rows:
             row = acc.setdefault(row_label, [{}, None])
@@ -1823,18 +2138,100 @@ def accumulate_report(mret: list, report_dir: Path, fresh: bool, scope: str = ''
                 row[1] = dur
 
     report_dir.mkdir(parents=True, exist_ok=True)
+    # by LINE, deduped: attempts repeat the same caveat far more often than they add a new
+    # one, and three copies of the D-state note reads as three incidents
+    seen, merged = set(), []
+    for line in (prior_banner + banner).splitlines():
+        if line.strip() and line not in seen:
+            seen.add(line)
+            merged.append(line)
+    banner = '\n'.join(merged) + '\n' if merged else ''
     jpath.write_text(json.dumps({'rows': [{'board': k, 'cells': c, 'duration': d}
-                                          for k, (c, d) in acc.items()]}, indent=2) + '\n')
+                                          for k, (c, d) in acc.items()],
+                                 'banner': banner}, indent=2) + '\n')
 
     md = render_matrix([(k, c, d) for k, (c, d) in acc.items()])
     if scope:
-        # a scoped run's small table is otherwise indistinguishable from a full one,
-        # and it replaces the previous full table in the sticky PR comment
+        # a scoped run's small table is otherwise indistinguishable from a full one, and
+        # it replaces the previous full table in the sticky PR comment
         md = f'_Scoped run: {scope}. Boards/tests not listed were not run._\n\n' + md
+    # LAST, so it is outermost: a rig-health caveat outranks the table AND the scope note,
+    # and the top of the report is where hil/SKILL.md tells the agent to look for it.
+    if banner:
+        md = banner + '\n' + md
     (report_dir / REPORT_MD).write_text(md + '\n', encoding='utf-8')
     return md
 
 
+# containment paths print through hil_health._p: stdout may already be a dead pipe (a
+# dropped ssh session), and a BrokenPipeError there would skip os._exit
+_p = hil_health._p
+
+
+def _abandon_exit(pool, mgr, abandoned: bool, err_count: int,
+                  report: Path | None = None) -> None:
+    """Free the runner when the pool could not be shut down. Returns only if not abandoned.
+
+    Must run even while an exception is propagating: multiprocessing's atexit handler
+    SIGTERMs its daemon workers (ignored in uninterruptible sleep) and then join()s them
+    with NO timeout, so an abandoned pool plus any raise between the pool's finally and
+    here hangs the interpreter until the job ceiling kills it. Reproduced: rc=124 at 25s
+    with SIGTERM-ignoring workers standing in for D state."""
+    if not abandoned:
+        return
+    try:
+        if sys.exc_info()[0] is not None:
+            # os._exit below discards the traceback, and this is often the only place the
+            # real failure would ever be printed
+            traceback.print_exc()
+    except OSError:
+        pass
+    # Word this on evidence: shutdown_pool also returns False when terminate() RAISES, and
+    # a live worker after terminate() is what distinguishes a wedge from a harness bug.
+    # Count WORKERS only -- _pool_procs appends the Manager, our own healthy child, so
+    # including it made n >= 1 always and the harness-error branch unreachable. It is killed
+    # separately: os._exit skips its finalizer, and orphaned it holds the runner's stdout.
+    n = hil_health.kill_pool_children(pool)
+    hil_health.kill_pool_children(None, mgr)
+    if n:
+        _p(f'HIL worker pool would not terminate ({n} worker(s) still live, '
+           f'uninterruptible); SIGKILLed them and abandoned the rest to free the '
+           f'runner. Boards held by any leaked worker stay locked until the host is '
+           f'power-cycled.', flush=True)
+    else:
+        _p('HIL worker pool shutdown failed but left no live worker behind, so this is '
+           'a harness error rather than a wedged rig -- see the Pool.terminate() '
+           'warning above. Exiting early anyway to free the runner; no board should '
+           'stay locked.', flush=True)
+    # A report already written by accumulate_report says nothing about the abandon, and a
+    # green table under a red job is how an agent ends up pasting it as this run's result.
+    # Prepend the caveat; best-effort, never at the cost of exiting.
+    if report is not None:
+        try:
+            if report.exists():
+                # utf-8 explicitly (the cells are ✅/❌/⚪) and catch ValueError too: a torn
+                # report or a LANG=C locale raises UnicodeDecodeError -- NOT an OSError --
+                # straight past os._exit, stranding the runner.
+                body = report.read_text(encoding='utf-8', errors='replace')
+                # Only when no banner is there yet, searched anywhere in the head rather
+                # than at char 0: write_timeout_report's banner must stay FIRST (its table
+                # is a PREVIOUS attempt's) and it puts the rig-health quote above itself.
+                if '**HIL run ab' not in body[:2000]:
+                    report.write_text(
+                        '**HIL run abandoned: the worker pool would not shut down.** The '
+                        'table below was collected before the abandon; treat board '
+                        'results as unverified.\n\n' + body, encoding='utf-8')
+        except (OSError, ValueError):
+            pass
+    try:
+        sys.stdout.flush()
+    except OSError:
+        pass
+    # Clamped: os._exit takes a status byte, so err_count == 256 would truncate to 0 and
+    # report a failing, abandoned run as green.
+    os._exit(min(err_count, 125) if err_count else 1)
+
+
 def main() -> None:
     """
     Hardware test on specified boards
@@ -1864,14 +2261,21 @@ def main() -> None:
                         help='Per-board test list as BOARD:test1,test2 (overrides -t for that board); repeat for multiple boards')
     parser.add_argument('-B', '--build-dir', default='cmake-build', help='Build folder name (default: cmake-build)')
     parser.add_argument('--build', action='store_true', help='Build firmware for selected boards with cmake before running tests')
-    parser.add_argument('-r', '--retry', type=int, default=3, help='Retry count for failed tests (default: 3)')
+    # default 1, not 3: the pool guard is a FLAT 3600s that does not scale with max_retry,
+    # and one usbtest test at default 3 can burn 1530s of it (510s outer x3) for a single
+    # board. Every CI caller already pins --retry 1; the bare invocations in the hil skill
+    # and hil-validate.js run against the same one-slot rig and used to inherit 3.
+    parser.add_argument('-r', '--retry', type=int, default=1, help='Retry count for failed tests (default: 1)')
     parser.add_argument('-v', '--verbose', action='store_true', help='Verbose output')
     args = parser.parse_args()
+    if args.retry < 1:
+        # 0 would make every test loop body never run: all-red cells, exit 0
+        parser.error('--retry must be >= 1')
 
     config_file = Path(args.config_file)
     boards = args.board
     verbose = args.verbose
-    hil_flash.verbose = args.verbose
+    hil_util.verbose = args.verbose
     test_only = args.test_only
     for entry in args.board_test:
         bname, _, tnames = entry.partition(':')
@@ -1899,6 +2303,32 @@ def main() -> None:
         config_boards = [e for e in config['boards'] if e['name'] in boards]
     config_boards = [e for e in config_boards if e['flasher']['name'] not in args.exclude_flasher
                      and (not args.flasher or e['flasher']['name'] in args.flasher)]
+    if not config_boards:
+        # same reason the unknown -b board exits 1: 'No tests were run.' with rc 0 reads as
+        # a green HIL leg, so a roster edit emptying a leg's filter stops testing silently
+        msg = (f'No boards left after the flasher filter (--flasher '
+               f'{args.flasher or "-"}, --exclude-flasher {args.exclude_flasher or "-"})')
+        print(msg, flush=True)
+        # loud AND leaving evidence: exiting with no report at all lets the PR comment
+        # keep the previous push's stale table under a red job
+        try:
+            rd = Path(os.environ.get('HIL_REPORT_DIR', '.'))
+            rd.mkdir(parents=True, exist_ok=True)
+            (rd / REPORT_MD).write_text(f'**HIL run selected no boards.** {msg}\n',
+                                        encoding='utf-8')
+        except OSError:
+            pass
+        sys.exit(1)
+
+
+    # Before the build: the probe needs nothing from it, and the annotation is more useful
+    # early than after a multi-board cmake build has been paid for.
+    # One line, not a probe: a D-state pid at start-up is a hint for whoever reads a red
+    # cell, never a reason to refuse the run. hil_pool_check does diagnosis.
+    note = hil_health.d_state_note()
+    if note:
+        log_line(f'rig note: {note}')
+    health_banner = f'> **Rig note.** {note}. Not a fault on its own -- a healthy testusb sits in D state for most of every case.\n' if note else ''
 
     build_err = 0
     if args.build:
@@ -1915,26 +2345,24 @@ def main() -> None:
         print(f'Build phase done: {build_err} failed')
         print('-' * 30)
 
-    # HIL report sidecar (hil_report.json/.md) and the .failed re-run spec live in
-    # report_dir (CI keys it by run id, so it persists across run attempts but is
-    # private to one run). A full run starts fresh; a re-run (--accumulate, which
-    # the generated .failed spec always starts with) merges so already-passed
-    # boards/tests are preserved. Clear prior state up front on a fresh run so a
-    # crash mid-run can't leave a stale report or re-run spec for a retry.
-    # -bt alone is not a re-run marker: PR-scoped first attempts pass -bt too.
+    # The report sidecar and the .failed re-run spec live in report_dir (CI keys it by run
+    # id: persistent across attempts, private to one run). A full run starts fresh; a re-run
+    # (--accumulate, which .failed always starts with) merges so already-passed boards
+    # survive. -bt alone is not a re-run marker.
     report_dir = Path(os.environ.get('HIL_REPORT_DIR', '.'))
     failed_fname = report_dir / (config_file.name + '.failed')
     fresh = not args.accumulate
-    if fresh:
-        report_dir.mkdir(parents=True, exist_ok=True)
-        for f in (REPORT_JSON, REPORT_MD):
-            (report_dir / f).unlink(missing_ok=True)
-        failed_fname.unlink(missing_ok=True)
+    # The unlink is DEFERRED to inside the pool try/except below: wiping here leaves
+    # Manager() and Pool() running with the old report gone and no report-writing path
+    # armed, so an EAGAIN/ENOMEM on fork gives CI an EMPTY report dir with no reason.
 
     seed = os.getenv('HIL_SHUFFLE_SEED') or str(int(time.time()))
     log_line(f'test-order shuffle seed: {seed} (HIL_SHUFFLE_SEED={seed} to replay); '
              f'flash/usbtest parallel per controller: {hil_lock.FLASH_PARALLEL}/{hil_lock.USBTEST_PARALLEL}; '
-             f'enum timeout first/retry: {ENUM_TIMEOUT}/{ENUM_TIMEOUT_RETRY}s')
+             f'enum timeout first/retry: {ENUM_TIMEOUT}/{ENUM_TIMEOUT_RETRY}s; '
+             # all three are env-tunable, so a run that dies on the guard is otherwise
+             # unattributable from the log alone
+             f'pool guard: {POOL_TIMEOUT}s')
 
     hints = {}
     try:
@@ -1949,94 +2377,232 @@ def main() -> None:
     config_boards = schedule_boards(config_boards, hints_by_uid)
     log_line('dispatch order: ' + ', '.join(b['name'] for b in config_boards))
 
-    mgr = Manager()
-    cmap = mgr.dict()
-    initargs = (Lock(), seed,
-                [Semaphore(hil_lock.USBTEST_PARALLEL) for _ in range(hil_lock.CONTROLLER_SLOTS)],
-                [Semaphore(hil_lock.FLASH_PARALLEL) for _ in range(hil_lock.CONTROLLER_SLOTS)],
-                cmap, Lock(), hints_by_uid)
-    with Pool(processes=os.cpu_count() or 1, initializer=init_worker, initargs=initargs) as pool:
-        async_ret = pool.map_async(test_board, config_boards)
+    # Bound BEFORE the try so the finally can name them whatever failed: Pool() forks, and
+    # the EAGAIN/ENOMEM the wipe comment below worries about is most likely to come from
+    # that fork -- after a convoy, where every stranded read holds a thread and an fd. Left
+    # outside, an OSError there escaped with mgr LIVE and `pool` unbound, so no report was
+    # written and the interpreter unwound into multiprocessing's unbounded atexit join.
+    pool = mgr = cmap = None
+    # Defined before the pool so _abandon_exit always has a value: a raise before
+    # `err_count = build_err + ...` would turn the containment path into a NameError.
+    err_count = build_err
+    # Fail CLOSED: only a shutdown_pool() that actually returned True clears this, and the
+    # assignment sits at the END of the inner finally, so anything raising before it
+    # (kill_worker_children, a BrokenPipeError from its print) leaves _abandon_exit armed.
+    pool_abandoned = True
+    # BEFORE Manager()/Pool(), not inside the try: hil_ci.sh reuses a persistent REMOTE_DIR
+    # and scp's the report back unconditionally, so if a fork failure (OSError/EAGAIN right
+    # after a convoy -- the case this whole block guards) skipped the wipe, the finally's
+    # _abandon_exit would prepend "HIL run abandoned" to the PREVIOUS run's table and
+    # publish last night's board results as this run's. Nothing is live yet here, so an
+    # OSError from the wipe itself just exits with its traceback -- it cannot strand the
+    # interpreter in multiprocessing's unbounded atexit join, which is what deferring it
+    # was protecting against.
+    if fresh:
+        report_dir.mkdir(parents=True, exist_ok=True)
+        for f in (REPORT_JSON, REPORT_MD):
+            (report_dir / f).unlink(missing_ok=True)
+        failed_fname.unlink(missing_ok=True)
+    try:
+        mgr = Manager()
+        cmap = mgr.dict()
+        initargs = (Lock(), seed,
+                    hil_lock.make_permit_sems(Semaphore, hil_lock.USBTEST_PARALLEL),
+                    hil_lock.make_permit_sems(Semaphore, hil_lock.FLASH_PARALLEL),
+                    cmap, Lock(), hints_by_uid)
+        # maxtasksperchild=1: the sysfs blindness latch is process-global and permanent
+        # (no decrement anywhere -- see hil_util.SYSFS_STUCK_MAX), so a worker that goes
+        # blind on ONE wedged board would report 0/30 and "probe missing" for the 2-3
+        # healthy boards it picked up afterwards. A fresh worker per board confines the
+        # damage to the board that caused it; the extra fork is noise against a
+        # flash+test cycle.
+        pool = Pool(processes=os.cpu_count() or 1, initializer=init_worker,
+                    initargs=initargs, maxtasksperchild=1)
+        # OUTER: encloses the pool block too, not just the reporting below. An exception
+        # escaping async_ret.get() (a worker exception, a Ctrl-C) runs the pool finally and
+        # then propagates straight out of main(); with _abandon_exit in a sibling try it
+        # was never reached.
         try:
-            mret = async_ret.get(timeout=POOL_TIMEOUT)
-        except MpTimeoutError:
-            pool.terminate()
-            pool.join()
-            raise RuntimeError(f'HIL worker pool timed out after {POOL_TIMEOUT}s')
+            # imap_unordered, NOT map_async: map_async is all-or-nothing, so a guard expiry
+            # threw away every board that had already finished -- up to a worker-width of
+            # completed rig time -- and left the re-run spec unwritten, so CI re-tested all
+            # ~26 boards to find the one that wedged. Draining as results arrive keeps what
+            # finished and names only what was still in flight.
+            it = pool.imap_unordered(test_board, config_boards)
+            mret = []
+            deadline = time.monotonic() + POOL_TIMEOUT
+            try:
+                mret = drain_pool(it, config_boards, deadline, out=mret)
+            except MpTimeoutError as te:
+                mret = te.finished
+                stuck = [b['name'] for b in config_boards
+                         if b['name'] not in {r[0] for r in mret}]
+                # The re-run spec FIRST and before the raise: a fresh run already unlinked
+                # it, so leaving it unwritten is what made the GitHub re-run repeat the
+                # whole fleet. Only the boards that never reported go in it.
+                _write_failed_spec(failed_fname, report_dir,
+                                   [(n, 1, [], None, 0) for n in stuck]
+                                   + [r for r in mret if r[1] > 0])
+                # Then the report, with the rows that DID finish, before anything that can
+                # block. Then RAISE into the ONE containment path: the inner finally runs
+                # the ordered sweep (kill_worker_children BEFORE terminate, or a reaped
+                # worker's flasher reparents out of reach), the outer one os._exit's.
+                banner = (f'**HIL run abandoned: worker pool timed out after '
+                          f'{POOL_TIMEOUT}s.** {len(mret)} board(s) below finished and '
+                          f'are this run\'s; {len(stuck)} never reported and are NOT in '
+                          f'the table: {", ".join(stuck)}. Re-run covers those.\n')
+                try:
+                    accumulate_report(mret, report_dir, fresh, '',
+                                      health_banner + _blind_note(mret)
+                                      + _stray_note(mret) + banner)
+                except Exception as rerr:  # noqa: BLE001 - the raise below must still happen
+                    # FALL BACK, do not just warn: accumulate_report can raise on an
+                    # unwritable/root-owned report dir or a torn JSON, and _abandon_exit
+                    # only PREPENDS to a report that exists. Without this the artifact
+                    # upload finds nothing (if-no-files-found: ignore) and the sticky PR
+                    # comment keeps the previous push's green table under a red job.
+                    print(f'warning: partial report failed: {type(rerr).__name__}: {rerr}; '
+                          f'falling back to the board list', flush=True)
+                    try:
+                        hil_health.write_timeout_report(
+                            report_dir, [b for b in config_boards
+                                         if b['name'] in stuck], POOL_TIMEOUT, REPORT_MD,
+                            prefix=health_banner)
+                    except Exception as re2:  # noqa: BLE001
+                        print(f'warning: fallback report failed too: '
+                              f'{type(re2).__name__}: {re2}', flush=True)
+                _p(f'HIL worker pool timed out after {POOL_TIMEOUT}s; sweeping and '
+                   f'shutting it down (abandoning it if a worker is unkillable)',
+                   flush=True)
+                raise RuntimeError(f'HIL worker pool timed out after {POOL_TIMEOUT}s')
+            except Exception as e:
+                # A worker RAISED -- e.g. a flasher adapter dropping off the bus makes
+                # get_serial_dev raise in the worker's flash section, which no per-test
+                # handler guards. Same treatment as the timeout path: the drain means
+                # `mret` already holds every board that finished, so keep those rows and
+                # name only the ones still in flight. (Under map_async they were all lost,
+                # which is what the old banner here claimed.)
+                done = {r[0] for r in mret}
+                stuck = [b['name'] for b in config_boards if b['name'] not in done]
+                _write_failed_spec(failed_fname, report_dir,
+                                   [(n, 1, [], None, 0) for n in stuck]
+                                   + [r for r in mret if r[1] > 0])
+                banner = (f'**HIL run aborted: a worker raised {type(e).__name__}: {e}.** '
+                          f'{len(mret)} board(s) below finished and are this run\'s; '
+                          f'{len(stuck)} did not report: {", ".join(stuck)}.\n')
+                try:
+                    accumulate_report(mret, report_dir, fresh, '',
+                                      health_banner + _blind_note(mret)
+                                      + _stray_note(mret) + banner)
+                except Exception as re2:  # noqa: BLE001 - the raise below must still happen
+                    print(f'warning: partial report failed: {type(re2).__name__}: {re2}',
+                          flush=True)
+                raise
 
-        err_count = build_err + sum(e[1] for e in mret)
-        # generate the re-run spec if anything failed: run ONLY the failed boards (-b),
-        # each restricted to its own failed tests (-bt); a board with failures but no
-        # test list (e.g. board-locked) re-runs entirely. --accumulate preserves the
-        # already-passed cells in the report.
-        parts = ['--accumulate']
-        for name, err, fts, _, _ in mret:
-            if err > 0:
-                parts.append(f'-b {name}')
-                if fts:
-                    parts.append(f'-bt {name}:{",".join(fts)}')
-        if len(parts) > 1:  # build-only failures have no boards to re-run
-            report_dir.mkdir(parents=True, exist_ok=True)
-            with failed_fname.open('w') as f:
-                f.write(' '.join(parts))
-        else:
-            failed_fname.unlink(missing_ok=True)
+            err_count = build_err + sum(e[1] for e in mret)
+            _write_failed_spec(failed_fname, report_dir, mret)
+        finally:
+            # Not `with Pool(...)`: its __exit__ joins the workers unbounded, hanging on
+            # any worker in uninterruptible sleep. shutdown_pool bounds the same terminate()
+            # by a grace period, so the pool is NOT cleanly closed/joined when it returns
+            # False. Record the outcome but never exit here: the report below is the only
+            # record of a run that otherwise passed.
+            #
+            # Same ordering as the timeout path: what the workers spawned must be
+            # snapshotted and killed while its parent is alive, or terminate() reparents it
+            # out of reach.
+            #
+            # Both calls must stay guarded: a raise here skips accumulate_report(), so a run
+            # whose boards ALL passed publishes an empty report dir -- and both can raise
+            # for reasons unrelated to the results. pool_abandoned stays fail-CLOSED, so
+            # _abandon_exit still arms.
+            try:
+                # Still worth running for the TIMEOUT path, where the workers are
+                # genuinely stuck mid-task and their children are still reachable through
+                # the pool's ppid tree. On the normal path every worker has already swept
+                # its own (kill_own_children) and retired, so this finds nothing.
+                #
+                # No banner from here: this finally runs AFTER accumulate_report on both
+                # abort paths, so anything appended to health_banner now is written to a
+                # variable nobody reads again. The report gets its count from the result
+                # tuples instead, via _stray_note.
+                hil_health.kill_worker_children(pool, mgr)
+            except Exception as e:
+                print(f'warning: worker-child sweep failed: {type(e).__name__}: {e}',
+                      flush=True)
+            try:
+                pool_abandoned = not hil_health.shutdown_pool(pool)
+            except Exception as e:
+                print(f'warning: pool shutdown failed: {type(e).__name__}: {e}', flush=True)
 
-    # refresh controller hints: pci resolved this run, plus board durations when the
-    # full test list ran (a -t/-bt filtered run would understate the board's real cost)
-    try:
-        if PROFILE:
-            # debug snapshot of the run's live uid->PCI / PCI->slot resolutions
-            report_dir.mkdir(parents=True, exist_ok=True)
-            with (report_dir / 'hil_profile_ctrl.json').open('w') as f:
-                json.dump(dict(cmap), f, indent=1, sort_keys=True)
-        uid_of = {b['name']: b['uid'] for b in config['boards']}
-        for name, _, _, _, dur in mret:
-            uid = uid_of.get(name)
-            if uid is None:
-                continue
-            h = dict(hints.get(uid) or {})
-            h['name'] = name  # informational: cache is keyed by uid
-            h['pci'] = cmap.get(f'uid:{uid}') or h.get('pci')
-            if dur > 0:  # test_board reports 0.0 for filtered (partial) runs
-                h['duration'] = round(dur, 1)
-            hints[uid] = h
-        # merge-on-write: another HIL job (e.g. the esp split) may have finished since
-        # our startup read - re-read and overlay only this run's boards so its entries
-        # survive, then replace atomically so a concurrent reader never sees a torn file
-        merged = {}
+        # refresh controller hints: pci resolved this run, plus durations from full runs
+        # only (a filtered run would understate the board's real cost)
         try:
-            with CONTROLLER_CACHE.open() as f:
-                cur = json.load(f)
-            if isinstance(cur, dict):
-                merged = {k: v for k, v in cur.items() if isinstance(v, dict)}
-        except (OSError, ValueError):
-            pass
-        merged.update({uid_of[n]: hints[uid_of[n]] for n, *_ in mret if n in uid_of})
-        CONTROLLER_CACHE.parent.mkdir(parents=True, exist_ok=True)
-        tmp = CONTROLLER_CACHE.with_suffix('.json.tmp')
-        with tmp.open('w') as f:
-            json.dump(merged, f, indent=1, sort_keys=True)
-        tmp.replace(CONTROLLER_CACHE)
-    except OSError as e:
-        print(f'warning: cannot persist controller hints to {CONTROLLER_CACHE}: {e}')
+            if PROFILE:
+                # debug snapshot of the run's live uid->PCI / PCI->slot resolutions
+                report_dir.mkdir(parents=True, exist_ok=True)
+                with (report_dir / 'hil_profile_ctrl.json').open('w') as f:
+                    json.dump(dict(cmap), f, indent=1, sort_keys=True)
+            uid_of = {b['name']: b['uid'] for b in config['boards']}
+            for name, _, _, _, dur, *_ in mret:
+                uid = uid_of.get(name)
+                if uid is None:
+                    continue
+                h = dict(hints.get(uid) or {})
+                h['name'] = name  # informational: cache is keyed by uid
+                h['pci'] = cmap.get(f'uid:{uid}') or h.get('pci')
+                if dur > 0:  # test_board reports 0.0 for filtered (partial) runs
+                    h['duration'] = round(dur, 1)
+                hints[uid] = h
+            # merge-on-write: another HIL job (e.g. the esp split) may have finished since
+            # our startup read, so overlay only this run's boards and replace atomically
+            merged = {}
+            try:
+                with CONTROLLER_CACHE.open() as f:
+                    cur = json.load(f)
+                if isinstance(cur, dict):
+                    merged = {k: v for k, v in cur.items() if isinstance(v, dict)}
+            except (OSError, ValueError):
+                pass
+            merged.update({uid_of[n]: hints[uid_of[n]] for n, *_ in mret if n in uid_of})
+            CONTROLLER_CACHE.parent.mkdir(parents=True, exist_ok=True)
+            tmp = CONTROLLER_CACHE.with_suffix('.json.tmp')
+            with tmp.open('w') as f:
+                json.dump(merged, f, indent=1, sort_keys=True)
+            tmp.replace(CONTROLLER_CACHE)
+        except Exception as e:
+            # Deliberately broad, and it must stay that way: this best-effort refresh makes
+            # Manager proxy RPCs that raise EOFError / BrokenPipeError / RemoteError when
+            # the Manager child has died, none of them OSErrors -- an OSError-only guard let
+            # those skip accumulate_report(). Nothing here is worth the report.
+            print(f'warning: cannot persist controller hints to {CONTROLLER_CACHE}: '
+                  f'{type(e).__name__}: {e}')
 
-    # board x test result matrix -> hil_report.md (accumulates across re-runs) + stdout
-    # -b/-bt in play means a filtered run (PR selection or a re-run spec): say so in the
-    # report, which otherwise looks exactly like a full run that happened to be small
-    scoped = sorted(set(args.board) | set(board_test))
-    scope = f'{len(scoped)} board(s) — {", ".join(scoped)}' if scoped else ''
-    report = accumulate_report(mret, report_dir, fresh, scope)
-    print()
-    print(report)
-    print(f'\nReport written to {(report_dir / REPORT_MD).resolve()}')
 
-    duration = time.time() - duration
-    print()
-    print("-" * 30)
-    print(f'Total failed: {err_count} in {duration:.1f}s')
-    print("-" * 30)
-    sys.exit(err_count)
+        # board x test result matrix -> hil_report.md (accumulates across re-runs) + stdout.
+        # -b/-bt means a filtered run (PR selection or a re-run spec): say so, or the report
+        # looks exactly like a full run that happened to be small
+        scoped = sorted(set(args.board) | set(board_test))
+        scope = f'{len(scoped)} board(s) — {", ".join(scoped)}' if scoped else ''
+        report = accumulate_report(mret, report_dir, fresh, scope,
+                                   health_banner + _blind_note(mret)
+                                   + _stray_note(mret))
+        print()
+        print(report)
+        print(f'\nReport written to {(report_dir / REPORT_MD).resolve()}')
+
+        duration = time.time() - duration
+        print()
+        print("-" * 30)
+        print(f'Total failed: {err_count} in {duration:.1f}s')
+        print("-" * 30)
+    finally:
+        # In the finally, not after: any raise above (accumulate_report sits outside the
+        # OSError handler) would skip the abandon path and unwind into multiprocessing's
+        # unbounded atexit join, hanging the runner.
+        _abandon_exit(pool, mgr, pool_abandoned, err_count, report_dir / REPORT_MD)
+    # Same clamp: exit status is a byte either way, so 256 failures would report green.
+    sys.exit(min(err_count, 125))
 
 
 if __name__ == '__main__':