summaryrefslogtreecommitdiff
path: root/test/hil/helper
diff options
context:
space:
mode:
Diffstat (limited to 'test/hil/helper')
-rw-r--r--test/hil/helper/__init__.py4
-rw-r--r--test/hil/helper/hil_health.py338
-rwxr-xr-xtest/hil/helper/hil_lock.py524
-rw-r--r--test/hil/helper/hil_pool_check.py1062
-rw-r--r--test/hil/helper/hil_report.py578
-rw-r--r--test/hil/helper/hil_util.py571
6 files changed, 3077 insertions, 0 deletions
diff --git a/test/hil/helper/__init__.py b/test/hil/helper/__init__.py
new file mode 100644
index 000000000..a080a2f55
--- /dev/null
+++ b/test/hil/helper/__init__.py
@@ -0,0 +1,4 @@
+# Marks helper/ as a REGULAR package. Without this it is only a PEP 420 namespace portion,
+# and a regular package named `helper` anywhere on sys.path wins over it even though
+# test/hil is sys.path[0] -- one transitive pip install would break every HIL entry point
+# at import. `helper` is a real distribution name on PyPI.
diff --git a/test/hil/helper/hil_health.py b/test/hil/helper/hil_health.py
new file mode 100644
index 000000000..d78d0f220
--- /dev/null
+++ b/test/hil/helper/hil_health.py
@@ -0,0 +1,338 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+"""Shutting a wedged HIL run down: kill what the workers spawned.
+
+A device whose usbfs node is held by a D-state process cannot be freed -- SIGKILL is not
+delivered in uninterruptible sleep -- so the goal is never to fix the rig from here. It is
+to free the runner's single job slot and leave a report naming what survived, instead of
+letting the job sit until GitHub cancels it with nothing to show.
+
+Deliberately shallow. We SIGKILL the process groups the workers spawned, wait a grace,
+and report whoever is still alive; we do not re-scan groups, prove pid ownership or
+escalate through sudo. A root-owned survivor is named in the report for hil_pool_check
+and the usb-kernel-recover skill to deal with -- signalling a pid we cannot prove is ours
+is the worse failure, and the job ceiling backstops whatever this misses.
+
+Everything here is stdlib-only and reads /proc unprivileged (dmesg is restricted on the
+rig), which keeps it importable -- and testable -- on a bare runner.
+"""
+import os
+import signal
+import threading
+import time
+from pathlib import Path
+
+PROC = Path('/proc')
+
+
+# How long to let a SIGKILL land before calling a process a survivor. Generous enough to
+# cover scheduling delay on a loaded rig, short enough that a fleet-wide sweep stays quick.
+CONFIRM_KILL_GRACE = 2.0
+
+
+def _p(*args, **kwargs) -> None:
+ # These run on the free-the-runner path, where stdout can already be a dead pipe (a
+ # dropped ssh session). An unguarded print would raise BrokenPipeError out of
+ # hil_test's inner finally, skipping shutdown_pool AND the report writing.
+ try:
+ print(*args, **kwargs)
+ except (OSError, ValueError):
+ # ValueError, not just OSError: printing to a CLOSED stream raises
+ # "ValueError: I/O operation on closed file", and both hil_pool_check and
+ # hil_test redirect stdout into a StringIO that can be closed under us. Escaping
+ # here skips shutdown_pool/kill_pool_children/os._exit -- stranding the runner,
+ # the exact failure this wrapper exists to prevent.
+ pass
+
+
+def _state(pid_dir: Path) -> str:
+ """The state letter from /proc/<pid>/stat. comm can contain ')', so the field is
+ located from the right rather than by splitting."""
+ # bytes, not read_text(): read_text decodes with the LOCALE encoding, so under LANG=C
+ # (systemd services, self-hosted runners) a non-ASCII comm raises UnicodeDecodeError
+ # and the entry silently vanishes from the scan.
+ stat = (pid_dir / 'stat').read_bytes()
+ return chr(stat[stat.rindex(b')') + 2])
+
+
+def _pids():
+ """/proc pid entries. Yields nothing rather than raising if /proc is unreadable."""
+ try:
+ entries = list(PROC.iterdir())
+ except OSError:
+ return
+ for entry in entries:
+ if entry.name.isdigit():
+ yield entry
+
+
+def d_state_note() -> str:
+ """Pids in uninterruptible sleep, for the report. Never aborts, never blocks.
+
+ A D-state process at start-up is NOT a fault on its own -- a healthy in-flight testusb
+ looks exactly like this, and the rig supports a dev run alongside CI. It is a hint for
+ whoever reads a red cell below. Diagnosis proper is hil_pool_check and the
+ usb-kernel-recover skill; this is one line, not a probe."""
+ stuck = []
+ for d in PROC.glob('[0-9]*'):
+ try:
+ if _state(d) == 'D':
+ stuck.append(d.name)
+ except (OSError, ValueError, IndexError):
+ pass # raced with exit, or /proc is restricted: not our problem here
+ if not stuck:
+ return ''
+ return (f'{len(stuck)} process(es) in D state when this run started: '
+ f'{sorted(stuck)[:10]}')
+
+
+def shutdown_pool(pool, grace: float = 30) -> bool:
+ """terminate() a worker Pool without ever blocking forever.
+
+ multiprocessing joins its workers unbounded (util.py _exit_function terminate()s the
+ daemonic ones, then calls p.join() -- no timeout -- on every remaining active child,
+ CPython 3.13.5), and a worker in uninterruptible sleep never
+ reaps -- so terminate() itself hangs, taking the runner's only job slot with it. False
+ when the pool refuses to die within `grace` (the caller must then abandon it); a
+ terminate() that *raises* counts as failure too, the pool being just as alive."""
+ outcome = {}
+
+ def _term():
+ try:
+ pool.terminate()
+ outcome['ok'] = True
+ except BaseException as e: # noqa: BLE001 - any failure means the pool is still up
+ # Say what happened: Pool._terminate_pool really can raise (CPython:
+ # AssertionError 'Cannot have cache with result_handler not alive'), and a
+ # swallowed one is indistinguishable from an unkillable D-state worker.
+ outcome['err'] = e
+ _p(f'warning: Pool.terminate() raised {type(e).__name__}: {e}', flush=True)
+
+ t = threading.Thread(target=_term, daemon=True)
+ t.start()
+ t.join(grace)
+ # Decide on the thread, not the dict: _term may set outcome['ok'] after join(grace)
+ # expired, reporting a merely-slow terminate as success on one read and abandoned on
+ # another. Still inside terminate() == not shut down.
+ if t.is_alive():
+ return False
+ return outcome.get('ok', False)
+
+
+def child_procs(pids) -> dict:
+ """{ancestor pid in `pids`: [(descendant pid, its pgid), ...]}, from ONE walk of /proc.
+
+ DESCENDANTS, not direct children: a worker's usbtest.py spawns its recovery flasher
+ through run_cmd (own session), so it is a GRANDCHILD that a direct-child sweep misses
+ and a kill mid-recovery would orphan on the probe. pgid comes back too because the two
+ kinds of child need different signals (see kill_pool_children)."""
+ wanted = set(pids)
+ by_parent: dict = {} # ppid -> [(pid, pgid), ...] for EVERY process
+ for entry in _pids():
+ try:
+ stat = (entry / 'stat').read_bytes()
+ except OSError:
+ continue # exited between the scan and the read, or not readable
+ # comm (field 2) is parenthesised and may contain spaces and ')' -- so split only
+ # what follows the LAST ')': state, ppid, pgrp, ...
+ try:
+ fields = stat[stat.rindex(b')') + 2:].split()
+ ppid, pgid = int(fields[1]), int(fields[2])
+ except (ValueError, IndexError):
+ continue # truncated or unparsable stat line
+ by_parent.setdefault(ppid, []).append((int(entry.name), pgid))
+ out: dict = {}
+ for root in wanted:
+ todo = list(by_parent.get(root, []))
+ while todo:
+ pid, pgid = todo.pop()
+ out.setdefault(root, []).append((pid, pgid))
+ todo += by_parent.get(pid, [])
+ return out
+
+
+def _pool_procs(pool, extra) -> list:
+ """The pool's worker Process objects, plus each extra's own process.
+
+ Manager() runs in its own child process and inherits the same descriptors as the
+ workers, so leaving it behind defeats the point: os._exit skips its finalizer."""
+ procs = list(getattr(pool, '_pool', []) or [])
+ for e in extra:
+ procs.append(getattr(e, '_process', e))
+ return procs
+
+
+
+
+def kill_worker_children(pool, *extra) -> int:
+ """SIGKILL what the pool's workers spawned; returns how many SURVIVED.
+
+ For the TIMEOUT path only. On the normal path each worker has already run
+ kill_own_children() and retired (maxtasksperchild=1), so this walks fresh idle workers
+ and finds nothing -- measured: 4 tasks, zero overlap with the pool at sweep time.
+
+ Call it BEFORE shutdown_pool(): terminate() reaps the (interruptible) worker and its
+ flasher is reparented to init, erasing the ppid link this matches on. Signalling the
+ worker's own group instead cannot work -- a forked pool worker inherits OUR group
+ (CPython 3.13.5 multiprocessing never setsid/setpgid) and run_cmd gives every flasher
+ a session of its own.
+
+ TWO passes because our SIGKILL can fail an in-flight flash and the worker then retries
+ in a fresh session, which one /proc snapshot misses. `seen` stops a pid signalled in
+ pass 1 being confirmed twice.
+ """
+ seen: set = set()
+ total = 0
+ for i in range(2):
+ if i:
+ time.sleep(0.5)
+ procs = _pool_procs(pool, extra)
+ total += _kill_kids(
+ child_procs(getattr(p, 'pid', None) for p in procs if p is not None), seen)
+ return total
+
+
+def kill_own_children() -> int:
+ """SIGKILL what THIS process spawned. Returns how many survived.
+
+ For the worker to call before it returns. maxtasksperchild=1 retires it the moment the
+ task ends, reparenting its children to init, so main()'s sweep walks fresh idle workers
+ and finds nothing (measured over 4 tasks: zero overlap, sweep 0, 4 strays alive).
+ Inside the worker the ppid link is still live.
+ """
+ return _kill_kids(child_procs([os.getpid()]), set())
+
+
+def _kill_kids(kids: dict, seen: set) -> int:
+ """SIGKILL every pid in a ppid-tree snapshot; return how many survived.
+
+ Every pid here is a DESCENDANT of a process we own, so it is ours by construction -- no
+ argv identity check, because we never signal anything we did not discover through our
+ own ppid tree.
+ """
+ try:
+ own = os.getpgid(0)
+ except OSError:
+ own = None # cannot tell our own group apart: never killpg, signal pids only
+ touched: list = []
+ for children in kids.values():
+ for cpid, cpgid in children:
+ if cpid in seen:
+ continue # a previous pass already signalled it
+ seen.add(cpid)
+ try:
+ if own is not None and cpgid != own:
+ # A run_cmd child: its own session, so one killpg also reaps what it
+ # spawned. Recorded because killpg cannot report a partial kill.
+ os.killpg(cpgid, signal.SIGKILL)
+ else:
+ # Shares our group (a plain subprocess.run), so killpg would take
+ # down the whole run -- it is signalled by pid in _kill_and_confirm.
+ pass
+ touched.append(cpid)
+ except PermissionError:
+ # NOT "already gone": the signal did not land, so this pid MUST still be
+ # confirmed, or the one case this handler exists for (an all-root session:
+ # the sudo wrapper died, its root members did not) is the one case that
+ # never reaches the report.
+ touched.append(cpid)
+ except ProcessLookupError:
+ pass # already gone
+ except OSError:
+ pass
+ # Both paths need confirming: a killpg'd flasher and a same-group mtype blocked on a
+ # wedged device are both in D state, and os.kill reported success on either.
+ denied = _kill_and_confirm(touched)
+ if denied:
+ _p(f'warning: could not kill {sorted(denied)}; they still hold whatever they '
+ f'had open (probe, usbfs node) into the next job', flush=True)
+ # SURVIVORS, not the count we signalled: the caller needs to know the rig is dirty for
+ # the next job, and a killpg is counted once per child sharing the group anyway.
+ return len(denied)
+
+
+def _kill_and_confirm(pids) -> list:
+ """SIGKILL every pid, then return those STILL alive after ONE grace window.
+
+ SIGKILL is QUEUED, not delivered, for a task in uninterruptible sleep -- and testusb
+ waits in a plain wait_for_completion() with no timeout (v6.12.96 usbtest.c:1404;
+ usb_sg_wait, message.c:765), so that is the normal state of a healthy in-flight case
+ too. os.kill returning success proves nothing; only the recheck does. It is also
+ asynchronous, so probing immediately reports a process we just killed as a survivor
+ (measured: 11 of 20 plain `sleep`s with no grace).
+
+ Signal all, then poll the set against ONE shared deadline: per-pid windows made this
+ scale with stray count, minutes on a convoy. A pid we cannot signal is reported, never
+ sudo-killed.
+ """
+ pending = []
+ for pid in pids:
+ try:
+ os.kill(pid, signal.SIGKILL)
+ except ProcessLookupError:
+ continue # already gone
+ except OSError:
+ pass # EPERM (root-owned): it stays, and the poll below reports it
+ pending.append(pid)
+
+ deadline = time.monotonic() + CONFIRM_KILL_GRACE
+ while True:
+ alive = []
+ for pid in pending:
+ try:
+ os.kill(pid, 0)
+ except ProcessLookupError:
+ continue # ESRCH: genuinely gone
+ except OSError:
+ pass # EPERM: it exists; the state check decides
+ try:
+ # a ZOMBIE answers kill(pid, 0) too: dead, merely unreaped. Not a survivor.
+ if _state(PROC / str(pid)) == 'Z':
+ continue
+ except (OSError, ValueError, IndexError):
+ continue # unreadable: assume gone rather than cry wolf
+ alive.append(pid)
+ pending = alive
+ if not pending or time.monotonic() >= deadline:
+ return pending # outlasted SIGKILL: D state, or not ours to kill
+ time.sleep(0.02)
+
+
+def kill_pool_children(pool, *extra) -> int:
+ """SIGKILL the pool's worker processes themselves. Returns how many are STILL ALIVE
+ after the grace -- not how many were signalled.
+
+ Survivors, not signals: the caller turns this number into "power-cycle the host", so
+ counting signals would send someone to a hypervisor over workers that all died.
+
+ Call after a shutdown_pool() that returned False, and after kill_worker_children().
+ A D-state worker ignores SIGKILL, but every other worker dies and drops the inherited
+ descriptors -- a survivor holds the runner's stdout pipe open and the runner waits for
+ EOF even after we exit, so the early exit would not free the job slot."""
+ killed_procs: list = []
+ for proc in _pool_procs(pool, extra):
+ try:
+ # Process.kill(), never a raw pid: multiprocessing's _send_signal re-checks
+ # `self.returncode is None` first, so once shutdown_pool's thread has reaped a
+ # worker this is a no-op instead of signalling a pid the OS may have recycled.
+ # os.pidfd_open(proc.pid) is worse: it skips that guard entirely.
+ if proc is None or not proc.is_alive():
+ continue
+ proc.kill()
+ killed_procs.append(proc)
+ except (OSError, AttributeError, ValueError):
+ continue # already reaped, never started, or not a real process
+ # Re-check the Process objects, never the pids collected a moment ago: shutdown_pool's
+ # thread is STILL join()ing workers, so a pid killed here can be reaped and RECYCLED
+ # before _kill_and_confirm signals it -- and on EPERM that escalates to `sudo -n kill
+ # -9 <stale pid>`, killing an unrelated ROOT process as the last act before os._exit.
+ killed_pids = []
+ for proc in killed_procs:
+ try:
+ if proc.is_alive() and proc.pid is not None:
+ killed_pids.append(proc.pid)
+ except (OSError, AttributeError, ValueError):
+ continue
+ # SIGKILL is asynchronous and a D-state task ignores it: only a confirmed survivor
+ # justifies the caller's power-cycle wording
+ return len(_kill_and_confirm(killed_pids)) if killed_pids else 0
diff --git a/test/hil/helper/hil_lock.py b/test/hil/helper/hil_lock.py
new file mode 100755
index 000000000..91f05ca86
--- /dev/null
+++ b/test/hil/helper/hil_lock.py
@@ -0,0 +1,524 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+"""Board locks + controller permits for the TinyUSB HIL rig.
+
+Board locks are kernel flocks in BOARD_LOCK_DIR arbitrating hardware access
+between dev sessions and CI's hil_test.py (never stop the actions-runner).
+Controller permits are in-process semaphores budgeting flashes and usbtest
+batteries per host controller; they have no CLI meaning. The CLI below
+(hold/release/status) manages board locks only.
+"""
+import argparse
+import fcntl
+import json
+import os
+import re
+import select
+import signal
+import sys
+import time
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # helper/ scripts import via the test/hil root
+from helper import hil_util
+
+BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks'
+CI_REASON = 'hil_test.py' # release-protected holder tag (release refuses to kill it)
+PROTECTED_REASONS = {CI_REASON, 'pool_check'} # cmd_release refuses to SIGTERM these holders
+PROFILE = os.environ.get('HIL_PROFILE') == '1'
+
+
+def lock_path(board: str) -> str:
+ return os.path.join(BOARD_LOCK_DIR, f'{board}.lock')
+
+
+def flock_nb(board: str):
+ """Open-or-create the lock file WITHOUT truncating (a losing racer must not
+ wipe the winner's record) and take LOCK_EX|LOCK_NB. Returns the open handle;
+ raises OSError when the flock is held elsewhere (handle already closed)."""
+ fd = os.open(lock_path(board), os.O_RDWR | os.O_CREAT, 0o666)
+ fh = os.fdopen(fd, 'r+')
+ try:
+ fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ except OSError:
+ fh.close()
+ raise
+ return fh
+
+
+def write_record(fh, reason: str) -> bool:
+ """Holder record; the flock itself is already held. Returns False on a write failure:
+ acquire_board_lock stays best-effort (the flock is the authority), but cmd_hold aborts
+ -- a hold whose record is missing is invisible to status/release."""
+ try:
+ fh.truncate(0)
+ fh.seek(0)
+ json.dump({'pid': os.getpid(), 'reason': reason,
+ 'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh)
+ fh.flush()
+ return True
+ except OSError:
+ return False
+
+
+def clear_record(fh) -> None:
+ """Clear our record before dropping the flock so records stay truthful."""
+ try:
+ fh.truncate(0)
+ except OSError:
+ pass
+
+
+def read_record(board: str):
+ try:
+ with open(lock_path(board)) as f:
+ return json.load(f)
+ except (OSError, ValueError):
+ return None
+
+
+# --- per-board dev-session locks ------------------------------------------
+def acquire_board_lock(board_name, reason=CI_REASON):
+ """Take this board's flock for the duration of its flash+test.
+ Returns an open file handle (keep it referenced; closing releases it),
+ or None when HIL_NO_BOARD_LOCK=1 or the lock dir is unusable (fail-open:
+ locking must never break a test run by itself).
+ Raises RuntimeError only when another session holds the board."""
+ import fcntl
+ if os.environ.get('HIL_NO_BOARD_LOCK') == '1':
+ return None # user-authorized bypass — see hil skill
+ try:
+ os.makedirs(BOARD_LOCK_DIR, exist_ok=True)
+ fd = os.open(os.path.join(BOARD_LOCK_DIR, f'{board_name}.lock'),
+ os.O_RDWR | os.O_CREAT, 0o666)
+ fh = os.fdopen(fd, 'r+')
+ except OSError as e:
+ # odd lock dir (perms, path collision): proceed unlocked, but say so —
+ # a silent fail-open is indistinguishable from the intentional bypass
+ print(f'warning: board lock unavailable for {board_name} ({e}); proceeding unlocked',
+ flush=True)
+ return None
+ try:
+ fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ except OSError:
+ try:
+ info = fh.read(500).strip()
+ except (OSError, UnicodeDecodeError):
+ info = ''
+ fh.close()
+ raise RuntimeError(f'board locked: {info or "unknown holder"}')
+ # announce ourselves so the other side's conflict message is truthful;
+ # best-effort — the flock itself is already held
+ try:
+ fh.truncate(0)
+ fh.seek(0)
+ json.dump({'pid': os.getpid(), 'reason': reason,
+ 'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh)
+ fh.flush()
+ except OSError:
+ pass
+ return fh
+
+
+# Per-host-controller concurrency (see controller_of/controller_slot below): a usbtest
+# battery saturates its DUT's host controller, so batteries and flashes are budgeted per
+# controller. The 4/2 defaults trade ~3.5 min on the usbtest leg for bandwidth margin on
+# the shared leaf-hub uplinks, where battery case failures were observed from 12/8
+# (profiled 2026-07-13/14: 22.2/14.3/12.5/10.8 min at usbtest width 1/2/3/4, plateau
+# after). Raise per run via HIL_FLASH_PARALLEL/HIL_USBTEST_PARALLEL.
+# - uPD720201 cards need firmware >= 2.0.2.6 (RAM-uploaded, reloads every power cycle):
+# the ROM firmware dies under battery + re-enumeration churn.
+# - a marginal DUT port bouncing during concurrent batteries can kill a uPD720201 ("xHCI
+# host not responding to stop endpoint command"): fix the port/cable or pull the board
+# -- lowering the widths does not fix a bad port (2026-07-16, every death).
+FLASH_PARALLEL = hil_util.pos_int_env('HIL_FLASH_PARALLEL', 4)
+USBTEST_PARALLEL = hil_util.pos_int_env('HIL_USBTEST_PARALLEL', 2)
+CONTROLLER_SLOTS = 12 # lock slots; controllers are assigned to slots on first sight
+# Bound on ONE permit wait. Generous: a real queue behind a slow board is normal,
+# and this only has to beat the pool guard so a leaked permit cannot consume it.
+PERMIT_TIMEOUT = hil_util.pos_int_env('HIL_PERMIT_TIMEOUT', 900)
+# CONTROLLER_SLOTS + 1 entries each, built by make_permit_sems: UNKNOWN_SLOT indexes the
+# extra one. Sized to CONTROLLER_SLOTS instead, the first unresolved board IndexErrors
+# inside a pool worker -- which now surfaces through drain_pool as a worker-raise (the
+# finished boards survive), but still loses this board and aborts the run.
+usbtest_sems = None # per-slot usbtest-battery permits
+flash_sems = None # per-slot flash permits
+controller_map = None # shared dict: 'pci:<addr>' -> slot, 'uid:<uid>' -> pci addr cache
+controller_meta = None # guards slot assignment in controller_map
+controller_hints = {} # static uid -> pci from the last run's cache (read-only per worker)
+
+
+log = print # hil_test.init_worker points this at log_line via init_scheduling
+
+
+def init_scheduling(b_sems, f_sems, cmap, cmeta, hints, log_fn=None):
+ """Install per-worker scheduling state (called from hil_test.init_worker)."""
+ global usbtest_sems, flash_sems, controller_map, controller_meta, controller_hints, log
+ usbtest_sems, flash_sems = b_sems, f_sems
+ controller_map, controller_meta, controller_hints = cmap, cmeta, hints
+ if log_fn is not None:
+ log = log_fn
+
+
+# -------------------------------------------------------------
+# Per-controller scheduling
+# -------------------------------------------------------------
+def controller_of(uid: str):
+ """Resolve a DUT uid to its root host controller's PCI address, or None when it cannot
+ be resolved — the device is not enumerated (e.g. parked in board_test firmware with USB
+ off), or sysfs would not answer. Successful resolutions are cached — cabling does not
+ change mid-run. Dual-port parts (e.g. CH32V307 usbhs/usbfs variants) share one uid and
+ one cache entry: budgeting is only exact when both ports sit on the same controller
+ (true on this rig)."""
+ if controller_map is None:
+ return None
+ cached = controller_map.get(f'uid:{uid}')
+ if cached:
+ return cached
+ # vid='cafe' first: the target is always a TinyUSB DUT, and the VID is a lock-free
+ # descriptor field. Without it this reads every probe's and hub's `serial` -- the one
+ # attribute served under device_lock -- so a wedged peer would block us here.
+ devs = hil_util.usb_scan(vid='cafe', serial=uid)
+ for dev in devs:
+ busnum = hil_util.read_sysfs(os.path.join(dev['dir'], 'busnum'))
+ if busnum is None:
+ continue
+ try:
+ root = os.path.realpath(f'/sys/bus/usb/devices/usb{int(busnum)}')
+ except ValueError:
+ continue
+ m = re.findall(r'[0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9a-f]', root)
+ if m:
+ controller_map[f'uid:{uid}'] = m[-1]
+ return m[-1]
+ return None
+
+
+def controller_slot(pci: str) -> int:
+ """Map a controller PCI address to a lock slot (assigned on first sight)."""
+ key = f'pci:{pci}'
+ with controller_meta:
+ slot = controller_map.get(key)
+ if slot is None:
+ slot = controller_map.get('nslots', 0)
+ if slot >= CONTROLLER_SLOTS:
+ slot = 0 # more controllers than slots: overflow shares slot 0 (safe, over-serialized)
+ else:
+ controller_map['nslots'] = slot + 1
+ controller_map[key] = slot
+ return slot
+
+
+# Unresolved boards budget in a slot of their OWN, one past the real ones, and that slot
+# holds exactly ONE permit whatever the per-controller width is. Neither neighbour works:
+# a permit on every slot (the old fail-closed rule) serialized the whole fleet the moment
+# one board could not be resolved, while a full private budget let unknown boards run a second
+# controller's worth of batteries on top of the resolved ones -- doubling the load on
+# whichever physical controller they actually sit on, which is the saturation the
+# uPD720201 deaths above are attributed to. Width 1 caps the over-subscription at +1.
+UNKNOWN_SLOT = CONTROLLER_SLOTS
+
+
+def make_permit_sems(semaphore, width: int) -> list:
+ """One semaphore per controller slot at `width`, plus the unknown bucket at 1."""
+ return [semaphore(width) for _ in range(CONTROLLER_SLOTS)] + [semaphore(1)]
+
+
+class controller_permit:
+ """Context manager: one permit from `sems` on the board's controller slot. An
+ unresolved controller budgets in UNKNOWN_SLOT, which admits one at a time: unresolved
+ boards serialize against each other, never against the whole rig, and never add a
+ second full budget to a controller. `warn_unknown` logs that fallback (used by
+ usbtest, where the device is expected to be enumerated by the caller)."""
+ def __init__(self, sems, uid: str, warn_unknown: bool = False):
+ self.sems = sems
+ self.slots = None
+ self.uid = uid
+ # what __enter__ actually ACQUIRED. Not the same as self.slots: a bounded acquire
+ # that times out is skipped on purpose, and releasing it anyway would add a permit
+ # that was never taken -- multiprocessing semaphores are unbounded, so the width
+ # grows for the rest of the run, on the controller throttle that exists to keep
+ # concurrent batteries from killing the uPD720201 xHCI.
+ self.taken: list = []
+ if sems is None:
+ return
+ # Hint FIRST for flash budgeting: a mis-budgeted flash is harmless, and the board
+ # is usually parked in board_test with USB off at this point, so controller_of
+ # cannot resolve it anyway -- it just walks the whole bus to say so, once per
+ # flash permit (~14 examples x ~21 boards a leg), each walk spawning a bounded
+ # reader per device. usbtest still resolves for real (warn_unknown), and by then
+ # the DUT is enumerated, so that walk succeeds and caches.
+ pci = None if warn_unknown else controller_hints.get(uid)
+ if pci is None:
+ pci = controller_of(uid)
+ if pci is None and warn_unknown:
+ log(f'warning: cannot resolve {uid} to a host controller; '
+ f'budgeting it in the unknown bucket')
+ self.slots = [controller_slot(pci) if pci else UNKNOWN_SLOT]
+
+ def __enter__(self):
+ if self.slots:
+ t0 = time.monotonic()
+ taken = self.taken = []
+ try:
+ for s in self.slots:
+ # BOUNDED. multiprocessing semaphores are NOT released when a holder
+ # dies, and the pool sweep SIGKILLs workers -- so a permit lost that
+ # way would block every later worker on this controller forever, and
+ # boards unrelated to the wedge would burn the whole pool guard. On
+ # expiry proceed over-subscribed and say so: a slower controller is a
+ # far better failure than a hung run.
+ if not self.sems[s].acquire(timeout=PERMIT_TIMEOUT):
+ log(f'warning: waited {PERMIT_TIMEOUT}s for a permit on slot {s} '
+ f'(uid {self.uid}); a holder probably died without releasing '
+ f'it -- proceeding over-subscribed')
+ continue
+ taken.append(s)
+ # inside the try: a failed __enter__ never gets its __exit__, so a raise
+ # here (e.g. broken stdout) must still release the permits
+ if PROFILE and time.monotonic() - t0 > 1.0:
+ log(f'[prof] permit wait {time.monotonic() - t0:.1f}s '
+ f'(uid {self.uid}, slots {self.slots})')
+ except BaseException:
+ for s in reversed(taken):
+ self.sems[s].release()
+ raise
+ return self
+
+ def __exit__(self, *exc):
+ if self.slots:
+ for s in reversed(self.taken):
+ self.sems[s].release()
+ self.taken = []
+ return False
+
+
+def flash_permit(uid: str) -> controller_permit:
+ return controller_permit(flash_sems, uid)
+
+
+def usbtest_permit(uid: str) -> controller_permit:
+ return controller_permit(usbtest_sems, uid, warn_unknown=True)
+
+
+# --- operator CLI (hold/release/status) ------------------------------------
+def boards_from_config(config: str) -> list:
+ """All board names, INCLUDING boards-skip: `hold --all` guards rig-wide
+ operations, and parked boards can still be touched (pool_check -b names them
+ explicitly), so a rig-wide hold that skipped them would leave a gap."""
+ try:
+ with open(config) as f:
+ cfg = json.load(f)
+ return [b['name'] for b in cfg['boards'] + cfg.get('boards-skip', [])]
+ except (OSError, ValueError, KeyError) as e:
+ print(f'ERROR: cannot read board roster {config}: {e}', file=sys.stderr)
+ sys.exit(1)
+
+
+def is_locked(board: str) -> bool:
+ """True if the recorded holder process is still alive.
+
+ Deliberately never touches the flock: even a momentary probe lock would
+ make a concurrent acquirer's LOCK_NB attempt fail spuriously. The flock
+ taken by acquirers themselves stays the only authority."""
+ info = read_record(board)
+ pid = info.get('pid') if isinstance(info, dict) else None
+ if not isinstance(pid, int) or pid <= 0:
+ return False
+ try:
+ os.kill(pid, 0)
+ except ProcessLookupError:
+ return False
+ except PermissionError:
+ return True # alive but owned by another user (e.g. the CI runner)
+ return True
+
+
+def cmd_hold(boards, reason):
+ os.makedirs(BOARD_LOCK_DIR, exist_ok=True)
+ # No pre-check: the holder's own LOCK_NB flock is the only authority, since a recorded
+ # pid may be stale or recycled. The holder signals success through this pipe because a
+ # generic is_locked() poll would be fooled by a RIVAL invocation's flock — only the
+ # holder knows whether it won every board.
+ r_fd, w_fd = os.pipe()
+ pid = os.fork()
+ if pid > 0:
+ os.close(w_fd)
+ os.waitpid(pid, 0) # reap intermediate child
+ ready, _, _ = select.select([r_fd], [], [], 10)
+ ok = bool(ready) and os.read(r_fd, 1) == b'1'
+ os.close(r_fd)
+ if ok:
+ print(f'held: {", ".join(boards)}')
+ return 0
+ for b in boards:
+ info = read_record(b)
+ if info:
+ print(f'ERROR: {b} locked: {info}', file=sys.stderr)
+ print('ERROR: holder failed to acquire locks', file=sys.stderr)
+ return 1
+ # intermediate child: detach, then spawn the actual holder
+ os.setsid()
+ if os.fork() > 0:
+ os._exit(0)
+ # holder (grandchild): acquire all flocks, signal the parent, sleep until killed
+ os.close(r_fd)
+ # Keep the success pipe clear of fds 0-2: invoked with stdio closed, os.pipe() can
+ # land there and the dup2 loop below would clobber it.
+ if w_fd <= 2:
+ w_fd = fcntl.fcntl(w_fd, fcntl.F_DUPFD, 3)
+ # Detach stdio: a `hold` whose output is captured must see EOF when the front-end
+ # exits — the immortal holder must not keep that pipe open.
+ devnull = os.open(os.devnull, os.O_RDWR)
+ for std_fd in (0, 1, 2):
+ os.dup2(devnull, std_fd)
+ if devnull > 2:
+ os.close(devnull)
+ try:
+ handles = []
+ for b in boards:
+ fh = flock_nb(b)
+ if not write_record(fh, reason):
+ raise OSError(f'cannot write holder record for {b}')
+ handles.append(fh)
+ except OSError:
+ try:
+ os.write(w_fd, b'0')
+ except OSError:
+ pass
+ os._exit(1) # lost a race; parent reports the failure
+ os.write(w_fd, b'1')
+ os.close(w_fd)
+
+ def _bow_out(*_):
+ # clear the records before dying so read_record/status stay truthful (the kernel
+ # drops the flocks themselves on exit either way)
+ for h in handles:
+ clear_record(h)
+ os._exit(0)
+
+ signal.signal(signal.SIGTERM, _bow_out)
+ while True:
+ signal.pause()
+
+
+def cmd_release(boards):
+ rc = 0
+ victims = set()
+ for b in boards:
+ try:
+ fd = os.open(lock_path(b), os.O_RDWR)
+ except OSError:
+ continue # no lock file (or another user's): nothing we can release
+ fh = os.fdopen(fd, 'r+')
+ try:
+ fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ except OSError:
+ # flock genuinely held — never SIGTERM on a mere pid record: the pid may be
+ # recycled, or a live worker that already moved on.
+ fh.close()
+ info = read_record(b) or {}
+ pid = info.get('pid')
+ reason = info.get('reason')
+ if reason in PROTECTED_REASONS:
+ print(f'ERROR: {b} is mid-test by {reason} (pid {pid}) — not killing it; '
+ 'wait for it to finish', file=sys.stderr)
+ rc = 1
+ elif isinstance(pid, int) and pid > 0:
+ victims.add(pid)
+ else:
+ print(f'ERROR: {b} is held but its record is unreadable', file=sys.stderr)
+ rc = 1
+ continue
+ # flock was free: only a stale record remained — clear it
+ clear_record(fh)
+ fh.close()
+ for holder in sorted(victims):
+ try:
+ os.kill(holder, signal.SIGTERM)
+ print(f'released holder pid {holder}')
+ except ProcessLookupError:
+ pass
+ except PermissionError:
+ print(f'ERROR: holder pid {holder} belongs to another user — cannot signal it',
+ file=sys.stderr)
+ rc = 1
+ time.sleep(0.3)
+ still = [b for b in boards if is_locked(b)]
+ if still:
+ print(f'ERROR: still locked: {", ".join(still)}', file=sys.stderr)
+ return 1
+ return rc
+
+
+def cmd_status():
+ if not os.path.isdir(BOARD_LOCK_DIR):
+ print('no locks')
+ return 0
+ any_locked = False
+ for fn in sorted(os.listdir(BOARD_LOCK_DIR)):
+ if not fn.endswith('.lock'):
+ continue
+ b = fn[:-5]
+ if is_locked(b):
+ any_locked = True
+ print(f'{b}: {read_record(b)}')
+ if not any_locked:
+ print('no locks')
+ return 0
+
+
+_CLI_USAGE = """Per-board advisory locks for the HIL rig.
+
+Arbitrates board access between dev sessions and CI's hil_test.py without
+stopping the actions-runner. Locks are kernel flocks: the kernel releases
+them automatically when the holder process dies, and holders clear their
+lock-file record on release so records stay truthful (/tmp also clears on
+reboot).
+
+Usage:
+ hil_lock.py hold BOARD [BOARD...] --reason TEXT
+ hil_lock.py hold --all [--config CONFIG.json] --reason TEXT
+ hil_lock.py release BOARD [BOARD...] | release --all
+ hil_lock.py status
+
+A holder process holds ALL boards given in one `hold` call; releasing any of
+them kills that holder and releases all of its boards.
+"""
+
+
+def main():
+ ap = argparse.ArgumentParser(description=_CLI_USAGE,
+ formatter_class=argparse.RawDescriptionHelpFormatter)
+ sub = ap.add_subparsers(dest='cmd', required=True)
+ p_hold = sub.add_parser('hold')
+ p_hold.add_argument('boards', nargs='*')
+ p_hold.add_argument('--all', action='store_true')
+ p_hold.add_argument('--config',
+ default=os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+ 'tinyusb.json'),
+ help='board roster JSON (default: tinyusb.json in test/hil, one level above this script)')
+ p_hold.add_argument('--reason', required=True)
+ p_rel = sub.add_parser('release')
+ p_rel.add_argument('boards', nargs='*')
+ p_rel.add_argument('--all', action='store_true')
+ sub.add_parser('status')
+ a = ap.parse_args()
+ if a.cmd == 'hold':
+ boards = boards_from_config(a.config) if a.all else a.boards
+ if not boards:
+ ap.error('no boards given (name boards or use --all)')
+ sys.exit(cmd_hold(boards, a.reason))
+ if a.cmd == 'release':
+ if a.all:
+ boards = ([fn[:-5] for fn in os.listdir(BOARD_LOCK_DIR) if fn.endswith('.lock')]
+ if os.path.isdir(BOARD_LOCK_DIR) else [])
+ else:
+ boards = a.boards
+ if not boards:
+ ap.error('no boards given (name boards or use --all)')
+ sys.exit(cmd_release(boards))
+ sys.exit(cmd_status())
+
+
+if __name__ == '__main__':
+ main()
diff --git a/test/hil/helper/hil_pool_check.py b/test/hil/helper/hil_pool_check.py
new file mode 100644
index 000000000..d98b92bd4
--- /dev/null
+++ b/test/hil/helper/hil_pool_check.py
@@ -0,0 +1,1062 @@
+#!/usr/bin/env python3
+"""Quick HIL pool health check.
+
+For every board in the rig's HIL config: is the flash probe on the USB bus, does a
+light example flash, and does the board's USB device (uid) come back up? Missing
+firmware is BUILT on the spot (tools/build.py, idf.py for espressif; one get_deps
+retry) — never skipped; --no-build opts out. Applies only per-device-safe recovery
+(probe authorized-toggle, board reset/re-flash) and prints a markdown summary
+table. Row statuses: ok (flashed and verified; under --scan-only: probe present —
+the scan checks presence only), flash-failed (firmware delivery failed: probe
+missing, build failed, flasher error, silent flash no-op, park not verified),
+failed (the check ran but did not verify: flashed with no enumeration/serial, or
+the check itself errored), locked (board flock held by another process —
+reported, never waited on or bypassed).
+
+Config is picked by hostname unless given: ci -> tinyusb.json, tusb (hifiphile
+rig) -> hfp.json, anything else is a dev PC -> local.json.
+
+Lives in test/hil/helper/ beside hil_lock.py; imports it and hil_flash; board
+recovery uses the repo's .claude/skills/usb-kernel-recover/scripts/usb_recover.sh.
+"""
+
+import argparse
+import io
+import json
+import glob
+import os
+import re
+import shlex
+import shutil
+import socket
+import sys
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # hil_flash + the helper package
+import hil_flash
+from helper import hil_lock, hil_util
+
+REPO_ROOT = hil_util.TINYUSB_ROOT
+USB_RECOVER = REPO_ROOT / '.claude' / 'skills' / 'usb-kernel-recover' / 'scripts' / 'usb_recover.sh'
+SEEN_CACHE = Path.home() / '.cache' / 'tinyusb-hil' / 'pool_seen.json'
+CONFIG_BY_HOST = {'ci': 'tinyusb.json', 'tusb': 'hfp.json'} # anything else: dev PC -> local.json
+
+# light-example preference; first built wins
+DEVICE_CANDIDATES = ['device/dfu_runtime', 'device/cdc_msc', 'device/cdc_msc_freertos',
+ 'device/hid_composite_freertos', 'device/cdc_dual_ports']
+HOST_CANDIDATES = ['host/device_info', 'host/cdc_msc_hid', 'host/msc_file_explorer_freertos']
+
+ENUM_WAIT = 12 # s, uid wait after flash
+ENUM_WAIT_RETRY = 8 # s, uid wait after a recovery reset/re-flash
+SERIAL_WAIT = 6 # s, host-board serial-output wait
+
+print_mutex = threading.Lock()
+_STRANDED_WARNED = False # scan_usb's caveat: once per process, not once per poll
+t0 = time.monotonic()
+
+
+def say(msg: str) -> None:
+ with print_mutex:
+ print(f'[{time.monotonic() - t0:6.1f}s] {msg}', file=sys.__stdout__, flush=True)
+
+
+def scan_usb() -> dict:
+ """busport -> {'serial', 'vidpid', 'ino'} for every enumerated USB device. Only
+ <bus>-<port>[.<port>...] dirs match; root hubs ('usbN', no dash) are excluded because
+ their 'serial' is a fabricated PCI address, and including them measured 6-7s/scan slower
+ (an observation; NOT an autosuspend wake -- that read is cached and does no I/O).
+ Keyed by busport, not serial: two devices can share a serial (an Espressif
+ USB-Serial-JTAG bridge and the cafe device it flashes both derive it from the same
+ MAC), and one dict slot would silently drop whichever lost the race."""
+ found = {}
+ # usb_scan's `serial` read is bounded by default (see hil_util.read_sysfs) -- this tool
+ # has no pool guard behind it and is run exactly when a device is suspected wedged. A
+ # device that will not answer is simply absent from the table; the footer says so.
+ devs = hil_util.usb_scan()
+ # ONCE per process, at SCAN time, not only in the footer: this tool prints rows as it
+ # goes over minutes, so a board dropped from the scan says "probe MISSING" within
+ # seconds while the only qualification would arrive after the final counts -- and an
+ # operator acting on the streaming output, or a run cut short by ^C, never sees it.
+ global _STRANDED_WARNED
+ if hil_util.sysfs_stranded() and not _STRANDED_WARNED:
+ _STRANDED_WARNED = True
+ say('WARNING: a bounded sysfs read gave up; rows below that say a probe or board '
+ 'is missing may be this scan losing sight of healthy hardware. Find the '
+ 'wedged device (usb-kernel-recover) and re-run.')
+ for dev in devs:
+ try:
+ found[dev['busport']] = {
+ 'serial': dev['serial'].lower(),
+ 'vidpid': f"{dev['vid']}:{dev['pid']}",
+ 'ino': os.stat(dev['dir'] + '/').st_ino}
+ except OSError:
+ continue
+ return found
+
+
+def find_usb(uid: str, devs: dict | None = None):
+ """Locate a flasher probe by uid, excluding VID cafe (TinyUSB DUT firmware): a
+ probe's uid can coincidentally equal its DUT's (Espressif USB-Serial-JTAG
+ bridges derive both from the same MAC), and the DUT is never the probe.
+
+ J-Link zero-pads numeric serials (681295394 -> 000681295394): an all-digit uid
+ matches an all-digit serial only when that serial equals the uid zero-padded to
+ the serial's own length (leading zeros only) — never when the zero-stripped uid
+ is empty, so a placeholder serial (metro_m4_express's probe legitimately reports
+ '123456') can't be mistaken for an unrelated device."""
+ devs = devs if devs is not None else scan_usb()
+ u = uid.lower()
+ candidates = [(bp, dev) for bp, dev in devs.items() if not dev['vidpid'].startswith('cafe:')]
+ for bp, dev in candidates:
+ if dev['serial'] == u:
+ return bp, dev['vidpid'], dev['ino']
+ stripped = u.lstrip('0')
+ if u.isdigit() and stripped:
+ for bp, dev in candidates:
+ s = dev['serial']
+ if s.isdigit() and s == stripped.zfill(len(s)):
+ return bp, dev['vidpid'], dev['ino']
+ return None
+
+
+def find_device(uid: str, pid: str | None):
+ """Board-online check: TinyUSB device (idVendor cafe) with this uid, optionally
+ PID-pinned. VID cafe keeps an Espressif USB-Serial-JTAG (303a) that shares the MAC
+ serial from false-passing."""
+ for busport, dev in scan_usb().items():
+ if (dev['serial'] == uid.lower() and dev['vidpid'].startswith('cafe:')
+ and (pid is None or dev['vidpid'].endswith(pid))):
+ return busport, dev['vidpid'], dev['ino']
+ return None
+
+
+def wait_device(uid: str, pid: str | None, old_ino, budget: float):
+ """Wait for the board's device with a NEW sysfs inode (flash resets the MCU, so a
+ genuine flash must re-enumerate; the inode is the re-enumeration marker)."""
+ deadline = time.monotonic() + budget
+ while time.monotonic() < deadline:
+ hit = find_device(uid, pid)
+ if hit and hit[2] != old_ino:
+ return hit
+ time.sleep(0.5)
+ return None
+
+
+def lock_board(name: str):
+ """Nonblocking flock per hil_lock.py protocol. Returns the handle, or a str with the
+ holder's info when the board is locked elsewhere. Board locks are ALWAYS respected: a
+ held board is reported and skipped, never waited on, and there is no bypass here."""
+ os.makedirs(hil_lock.BOARD_LOCK_DIR, exist_ok=True)
+ try:
+ fh = hil_lock.flock_nb(name)
+ except OSError:
+ # NB: conflates a held flock with open() failures (EACCES/EROFS/ENOSPC) — benign
+ # while everything on the rig runs as one uid
+ info = hil_lock.read_record(name)
+ return json.dumps(info) if info else 'unknown holder'
+ if not hil_lock.write_record(fh, 'pool_check'):
+ # an invisible lock (flock held, no record) is worse than no lock: status cannot
+ # show us and release cannot recognize the protected holder
+ hil_lock.clear_record(fh)
+ fh.close()
+ return 'ERROR: holder record write failed (lock dir unwritable?)'
+ return fh
+
+
+def unlock_board(fh) -> None:
+ hil_lock.clear_record(fh)
+ fh.close()
+
+
+def can_recover() -> bool:
+ if not USB_RECOVER.is_file():
+ return False
+ try:
+ # run_cmd, not subprocess.run: run's post-timeout reap is an UNBOUNDED wait(), and
+ # our kill bounces off a setuid-root sudo with EPERM, leaving communicate() on a
+ # pipe that never closes. run_cmd killpgs, escalates through sudo, reaps bounded.
+ r = hil_util.run_cmd('sudo -n true', timeout=10, quiet=True)
+ except OSError: # sudo not installed
+ return False
+ return r.returncode == 0
+
+
+def recover_probe(uid: str, busport: str) -> bool:
+ """Soft-replug an enumerated-but-wedged probe: deauthorize+reauthorize (no VBUS cut,
+ touches only this device). Success = the probe re-enumerated (new sysfs inode), not the
+ helper's exit code, which flakes while the toggle works. J-Links respond with a full
+ disconnect and can stay off the bus for >8 s."""
+ pre = find_usb(uid)
+ # Bounded through run_cmd (same reason as can_recover): the sysfs authorized store can
+ # block in D state on a wedged device, and this runs while the board's release-
+ # PROTECTED flock is held -- a hang here would lock the board until the host reboots.
+ cmd = ' '.join(shlex.quote(a) for a in
+ ['sudo', '-n', str(USB_RECOVER), 'authorized', busport])
+ if hil_util.run_cmd(cmd, timeout=30, quiet=True).returncode == 124:
+ return False
+ deadline = time.monotonic() + 20
+ while time.monotonic() < deadline:
+ post = find_usb(uid)
+ if post and (pre is None or post[2] != pre[2]):
+ return True
+ time.sleep(0.5)
+ return False
+
+
+def resolve_variant(board: dict, example: str, note: list | None = None) -> str:
+ """Build-dir variant name for `example`: the first of the board's variants with
+ already-built firmware, falling back to the board name. Notes the pick when it
+ differs from the board name (e.g. nanoch32v203's build dir is variant
+ 'nanoch32v203-fsdev', not the board name)."""
+ name = board['name']
+ for v in board.get('variant') or [{'name': name}]:
+ vn = v['name']
+ if hil_flash.find_firmware(vn, example, flasher=board['flasher']['name']):
+ if vn != name and note is not None and f'variant: {vn}' not in note:
+ note.append(f'variant: {vn}')
+ return vn
+ return name
+
+
+def pick_example(board: dict, note: list, build_missing: bool = True):
+ """(example, kind, variant, fw) with built firmware for this board; kind is
+ 'device' (uid check) or 'host' (serial-output check); variant is the resolved
+ build-dir variant that has it (see resolve_variant); fw is the firmware path to
+ flash, extension included. When nothing is built and build_missing is set (the default —
+ never skip a board for lack of a build), the preferred candidate is built on
+ the spot via ensure_fw."""
+ tests = board.get('tests', {})
+ only = tests.get('only', [])
+ skip = set(tests.get('skip', [])) # config's known-broken examples: never pick one
+ is_device = tests.get('device') or any(t.startswith('device/') for t in only)
+ if is_device:
+ cand = DEVICE_CANDIDATES + [t for t in only if t.startswith('device/') and t != 'device/usbtest']
+ kind = 'device'
+ else:
+ cand = HOST_CANDIDATES + [t for t in only if t.startswith('host/')]
+ kind = 'host'
+ for ex in dict.fromkeys(cand):
+ if ex in skip:
+ continue
+ variant = resolve_variant(board, ex, note)
+ fw = hil_flash.find_firmware(variant, ex, flasher=board['flasher']['name'])
+ if fw:
+ return ex, kind, variant, fw
+ if not build_missing:
+ return None, kind, None, None
+ # nothing built anywhere: build the preferred candidate (an only-list board
+ # must get one of its own examples — dfu_runtime etc. may not even configure)
+ pref = [c for c in dict.fromkeys(cand) if c not in skip and (not only or c in only)]
+ if not pref:
+ return None, kind, None, None
+ variant = (board.get('variant') or [{'name': board['name']}])[0]['name']
+ for ex in pref[:2]: # the second candidate covers a preferred example that fails to build
+ fw = ensure_fw(board, variant, ex, note)
+ if fw:
+ return ex, kind, variant, fw
+ return None, kind, None, None
+
+
+_pid_cache: dict[str, str | None] = {}
+
+
+def get_expected_pid(example: str) -> str | None:
+ """USB_PID for `example`'s device descriptor (examples/<example>/src/
+ usb_descriptors.c, '#define USB_PID 0x....'), lowercased and without the 0x
+ prefix to match sysfs idProduct. Cached per example; None (also cached) when
+ the file or define isn't there — host examples have no usb_descriptors.c, and
+ the caller must stay quiet rather than false-warn."""
+ if example not in _pid_cache:
+ pid = None
+ try:
+ text = (REPO_ROOT / 'examples' / example / 'src' / 'usb_descriptors.c').read_text()
+ # optional parens as in tools/check_example_pids.py's parser
+ m = re.search(r'#define\s+USB_PID\s+\(?\s*(0x[0-9a-fA-F]+)', text)
+ if m:
+ pid = m.group(1)[2:].lower()
+ except OSError:
+ pass
+ _pid_cache[example] = pid
+ return _pid_cache[example]
+
+
+def call_flasher(fn, *fn_args) -> tuple[int, str]:
+ """Run a hil_flash flash_*/reset_* backend, normalizing raises to a failure: several
+ backends raise instead of returning nonzero (get_serial_dev when a bridge's
+ /dev/serial/by-id node vanishes, a missing config.env, a .jlink script OSError), and an
+ exception must not skip the caller's retry/recovery ladder. Returns (rc, error line)."""
+ try:
+ ret = fn(*fn_args)
+ if ret.returncode == 0:
+ return 0, ''
+ err = flash_error_line(hil_util.cmd_stdout_text(ret.stdout))
+ return ret.returncode, err or f'rc={ret.returncode}'
+ except Exception as e:
+ return -1, repr(e)[:90]
+
+
+def flash(board: dict, fw, allow_recovery: bool, probe_port: str, note: list) -> bool:
+ """Flash the resolved firmware with one retry; on repeated failure soft-replug the
+ probe and always make one final attempt afterward, confirmed replug or not — some
+ probes (WCH-Link, ST-Link, CP210x, picoprobe) keep their sysfs kobject across an
+ authorized toggle instead of dropping off the bus. Returns True on success.
+
+ `fw` comes from pick_example: a re-resolve here would use the global search policy and
+ miss a firmware ensure_fw just built into cmake-build/ under an exclusive -B."""
+ fn = getattr(hil_flash, f'flash_{board["flasher"]["name"].lower()}')
+ for attempt in range(3):
+ if attempt == 2:
+ if not (allow_recovery and probe_port):
+ return False
+ cur = find_usb(board['flasher']['uid'])
+ if cur is None:
+ # probe gone from the bus: its old busport may now hold an UNRELATED device
+ # (bus renumbering) and the helper only checks occupancy, so toggling would
+ # deauthorize an innocent fixture
+ note.append('probe vanished before toggle')
+ else:
+ say(f'{board["name"]:26} recovery: replugging probe {cur[0]} (authorized toggle)')
+ if recover_probe(board['flasher']['uid'], cur[0]):
+ note.append('probe replugged')
+ time.sleep(2) # udev recreates /dev/serial/by-id symlinks after re-enumeration
+ else:
+ note.append('probe toggle unconfirmed')
+ rc, err = call_flasher(fn, board, str(fw))
+ if rc == 0:
+ return True
+ if rc == 127: # flasher binary missing: retries/probe recovery can't fix env
+ note.append(f'flasher tool missing ({err}) — esptool needs the ESP-IDF env '
+ f'(. "$IDF_PATH/export.sh")'
+ if board['flasher']['name'].lower() == 'esptool' else
+ f'flasher tool missing: {err}')
+ return False
+ if attempt == 0:
+ say(f'{board["name"]:26} flash retry: {err}')
+ else:
+ note.append(f'flash: {err}')
+ return False
+
+
+def flash_error_line(out: str) -> str:
+ """Most informative line of a failed flash's output: last error-looking line,
+ else the last non-empty one."""
+ lines = [l.strip() for l in out.splitlines() if l.strip()]
+ for l in reversed(lines):
+ if any(k in l.lower() for k in ('error', 'fail', 'unknown', 'cannot', 'timeout',
+ 'no valid', 'not found', 'unable')):
+ return l[:90]
+ return lines[-1][:90] if lines else ''
+
+
+def check_host_serial(board: dict, do_reset: bool = True, want_hello: bool = False) -> bytes | None:
+ """Host-only boards never enumerate their uid (their USB port is the host side);
+ aliveness = output on the flasher's UART bridge after a reset. A probe byte is
+ written each poll so an echo-only firmware (board_test) also answers. Returns
+ the first output chunk (b'' when silent, None when the port is absent/drops) so
+ the caller can also judge WHAT answered — see boardtest_output().
+
+ do_reset=False listens to the firmware as-is: used right after a flash whose
+ own reset already started it — a second openocd/JLink session back-to-back on
+ the same probe can fail transiently and leave the target halted.
+
+ "logger": "rtt" boards have no VCOM: the same check runs over the probe's RTT
+ console instead. The reset happens BEFORE the console opens (it owns the probe),
+ which also zeroes the .bss ring — so pre-reset backlog cannot count as life, and
+ without a reset Commander delivers the boot burst the preceding flash left."""
+ if board.get('logger') == 'rtt':
+ if do_reset:
+ # a failed reset leaves the previous run's ring intact: attaching anyway would
+ # score stale output as life, so bail to host_alive's board_test reflash ladder
+ rc, err = call_flasher(getattr(hil_flash, f'reset_{board["flasher"]["name"].lower()}'), board)
+ if rc:
+ say(f'{board["name"]:26} reset failed: {err}')
+ return None
+ try:
+ ser = hil_util.JlinkRtt(board, timeout=0.3)
+ except hil_util.RttError as e:
+ say(f'{board["name"]:26} no RTT console: {e}')
+ return None
+ try:
+ data = b''
+ deadline = time.monotonic() + SERIAL_WAIT
+ while time.monotonic() < deadline:
+ ser.write(b'U')
+ data += ser.read(256)
+ # JLinkExe's banner arrives whether or not the target is alive --
+ # judged unfiltered it scores a dead board 'alive'. Same shared filter
+ # as test_host_device_info; complete_only drops a trailing partial
+ # line, so a banner FRAGMENT split by this read boundary cannot count
+ # as target output either.
+ td = hil_util.strip_banner(data, complete_only=True)
+ if want_hello:
+ if b'Hello from TinyUSB' in td:
+ return td
+ elif td and not boardtest_output(td):
+ return td
+ return hil_util.strip_banner(data)
+ except hil_util.RttError:
+ return None # console died mid-poll (server exited, probe dropped)
+ finally:
+ ser.close()
+ import serial
+ try:
+ port = hil_util.get_serial_dev(board['flasher']['uid'], None, None, 0)
+ ser = serial.Serial(port, baudrate=115200, timeout=0.3, write_timeout=1)
+ except Exception as e:
+ say(f'{board["name"]:26} no flasher serial port: {e}')
+ return None
+ try:
+ # flush BEFORE the reset: this drops the pre-reset CDC backlog (which must not
+ # count as life) while keeping the post-reset boot banner, which prints while the
+ # reset tool is still tearing down and a post-reset flush would eat
+ ser.reset_input_buffer()
+ if do_reset:
+ getattr(hil_flash, f'reset_{board["flasher"]["name"].lower()}')(board)
+ # judge the WHOLE window, not the first chunk: the probe's CDC bridge has its own
+ # FIFO, so stale pre-flash output (e.g. board_test hellos) can arrive after our
+ # host-side flush and must not decide the verdict alone.
+ data = b''
+ deadline = time.monotonic() + SERIAL_WAIT
+ while time.monotonic() < deadline:
+ try:
+ ser.write(b'U')
+ data += ser.read(256)
+ except serial.SerialTimeoutException:
+ pass
+ except serial.SerialException:
+ return None # port dropped mid-poll (bridge re-enumerating)
+ # early-exit on the caller's positive signal (board_test hello for park
+ # verification, any non-board_test output for example liveness): stale
+ # bridge-FIFO backlog of the OTHER kind must not end the window
+ if want_hello:
+ if b'Hello from TinyUSB' in data:
+ return data
+ elif data and not boardtest_output(data):
+ return data
+ return data
+ finally:
+ ser.close()
+
+
+def boardtest_output(data: bytes) -> bool:
+ """True when (non-empty) serial output is recognizably ONLY board_test's: its
+ periodic HELLO_STR and echoes of our b'U' pokes, nothing else. Any residue
+ beyond that (an example banner, log lines) proves other firmware is talking,
+ however much stale board_test backlog surrounds it. Used as a negative
+ identity marker — after flashing a host example, board_test-only chatter
+ means the flash silently didn't take (the host analog of the PID check)."""
+ residue = data.replace(b'Hello from TinyUSB', b'')
+ for junk in (b'U', b'\r', b'\n'):
+ residue = residue.replace(junk, b'')
+ return len(residue) == 0
+
+
+def build_example(board: dict, variant: str, example: str) -> int:
+ """Build one example for this board: tools/build.py (same invocation shape as
+ hil_test.build_board), or idf.py directly for espressif (tools/build.py's esp branch
+ ignores -T and builds everything; variant flags travel as -DCFLAGS_CLI, the channel
+ tools/build.py uses). Bounded and process-group-killed via run_cmd; 600 s covers a
+ first configure+build of an SDK-heavy family (pico, nrf, esp). Builds normally run
+ pre-lock, so a board flock is not held here except on rare recovery paths. Per-build
+ compile parallelism is capped at cpu/-j so -j concurrent builds cannot swamp sibling
+ workers' verification windows. Returns the returncode (127 = ESP-IDF env missing)."""
+ name = board['name']
+ variants = board.get('variant') or [{'name': name}]
+ vcfg = next((v for v in variants if v['name'] == variant), variants[0])
+ if board['flasher']['name'].lower() == 'esptool':
+ if not shutil.which('idf.py'):
+ return 127 # ESP-IDF env not sourced in this shell
+ # -B keyed off the VARIANT so ensure_fw's post-build lookup finds it
+ cmd = ['idf.py', '-C', f'examples/{example}',
+ '-B', f'cmake-build/cmake-build-{vcfg["name"]}/{example}',
+ '-G', 'Ninja', f'-DBOARD={name}', 'build']
+ for d in vcfg.get('defines', []):
+ cmd.insert(-1, f'-D{d}')
+ if vcfg.get('flags'):
+ cmd.insert(-1, f'-DCFLAGS_CLI={vcfg["flags"]}')
+ # the IDF component manager writes examples/<ex>/dependencies.lock in the
+ # SOURCE tree (idf.py -B relocates only the build dir), so concurrent esp
+ # builds of one example for different targets corrupt each other's solve
+ with _esp_lock, _build_sem:
+ return hil_util.run_cmd(shlex.join(cmd), cwd=str(hil_util.TINYUSB_ROOT),
+ timeout=600).returncode
+ cmd = [sys.executable, str(hil_util.TINYUSB_ROOT / 'tools' / 'build.py'),
+ '-b', name, '-T', Path(example).name,
+ '-j', str(max(1, (os.cpu_count() or _jobs) // _jobs))]
+ if vcfg['name'] != name:
+ cmd += ['--build-name', vcfg['name']]
+ for d in vcfg.get('defines', []):
+ cmd += ['-D', d]
+ for tok in vcfg.get('flags', '').split():
+ cmd += [f'--cflag={tok}']
+ with _build_sem:
+ return hil_util.run_cmd(shlex.join(cmd), cwd=str(hil_util.TINYUSB_ROOT),
+ timeout=600).returncode
+
+
+_deps_lock = threading.Lock() # one get_deps at a time (it also drains _build_sem)
+_esp_lock = threading.Lock() # idf.py mutates source-tree dependencies.lock per example
+_no_build = False # --no-build: ensure_fw never invokes a build
+_jobs = 4 # mirrors -j; set in main before the pool starts
+_build_sem = threading.BoundedSemaphore(4) # build slots; get_deps drains ALL (exclusive)
+_builds: dict = {} # (variant, example) -> (fw|None, reason): one attempt per run
+
+
+def ensure_fw(board: dict, variant: str, example: str, note: list):
+ """Firmware for `example`, building it when absent — never skip a board for lack of a
+ build (--no-build opts out). One retry with deps fetched and the CMake caches dropped
+ when the first build fails (fresh checkouts lack the family deps; a cache configured
+ in a broken env poisons every later attempt). Returns the firmware path, or None with
+ the failure noted. Call BEFORE taking the board lock: builds are long. One attempt per
+ (variant, example) per run, memoized in _builds, so a repeat call (park, under the
+ held flock) resolves instantly even when an exclusive -B hides the fresh artifact."""
+ fw = hil_flash.find_firmware(variant, example, flasher=board['flasher']['name'])
+ if fw:
+ return fw
+ key, base = (variant, example), Path(example).name
+ if key in _builds:
+ return _builds[key][0]
+ if _no_build:
+ _builds[key] = (None, 'disabled')
+ note.append(f'build skipped (--no-build): {base}')
+ return None
+ rc = build_example(board, variant, example)
+ if rc == 127 and board['flasher']['name'].lower() == 'esptool':
+ _builds[key] = (None, 'no-env')
+ note.append(f'cannot build {base}: ESP-IDF env missing '
+ f'(. "$IDF_PATH/export.sh")')
+ return None
+ if rc == 124: # hung build: a deps/cache retry cannot cure it, don't double the stall
+ _builds[key] = (None, 'timeout')
+ note.append(f'build timeout: {base}')
+ return None
+ if rc != 0:
+ # retry once with deps fetched and the CMake caches dropped (cache only — a tree
+ # wipe would destroy every other example's firmware). get_deps git-resets shared
+ # deps that are already present, so it drains ALL build slots first.
+ with _deps_lock:
+ for _ in range(_jobs):
+ _build_sem.acquire()
+ try:
+ r = hil_util.run_cmd(shlex.join([sys.executable, str(hil_util.TINYUSB_ROOT / 'tools' / 'get_deps.py'),
+ '-b', board['name']]),
+ cwd=str(hil_util.TINYUSB_ROOT), timeout=600)
+ finally:
+ for _ in range(_jobs):
+ _build_sem.release()
+ if r.returncode != 0:
+ note.append('get_deps failed')
+ bd = hil_util.TINYUSB_ROOT / 'cmake-build' / f'cmake-build-{variant}'
+ # esp configures one level deeper (<variant>/<example>/): wipe both layouts
+ for d in (bd, bd / example):
+ shutil.rmtree(d / 'CMakeFiles', ignore_errors=True)
+ (d / 'CMakeCache.txt').unlink(missing_ok=True)
+ rc = build_example(board, variant, example)
+ if rc != 0:
+ _builds[key] = (None, 'fail')
+ note.append(f'build failed: {base}')
+ return None
+ # both build paths write to cmake-build/, so look there even when an explicit -B
+ # narrowed the global search — this is OUR fresh build, not a stale fallback
+ fw = hil_flash.find_firmware(variant, example,
+ roots=[hil_flash.build_dir, 'cmake-build'],
+ flasher=board['flasher']['name'])
+ _builds[key] = (fw, 'ok' if fw else 'no-fw')
+ note.append(f'built {base}' if fw else f'build produced no firmware: {base}')
+ return fw
+
+
+def ensure_board_test(board: dict, variant: str, note: list):
+ """board_test firmware for parking, building it if absent (via ensure_fw).
+ Espressif included — tools/build.py builds board_test for that family too;
+ the build just needs the ESP-IDF env (127 → noted, park is then skipped)."""
+ fw = hil_flash.find_firmware(variant, 'device/board_test', flasher=board['flasher']['name'])
+ if fw:
+ return fw
+ variants = board.get('variant') or [{'name': board['name']}]
+ if not any(v['name'] == variant for v in variants):
+ variant = variants[0]['name']
+ return ensure_fw(board, variant, 'device/board_test', note)
+
+
+def verdict(row: dict, ok: bool) -> str:
+ """Row status for a verification result, preserving a 'flash-failed' a deeper
+ layer already recorded (silent flash no-op, board_test delivery failure)."""
+ return 'ok' if ok else ('flash-failed' if row['status'] == 'flash-failed' else 'failed')
+
+
+def host_alive(board: dict, note: list, row: dict, flashed_example: bool = False) -> bool:
+ """Serial aliveness with recovery: silent -> (build and) flash board_test (it
+ hellos every second and echoes) -> recheck. Also cures a silent flash no-op
+ that left the board crashed.
+
+ With flashed_example=True (a host example was just flashed), board_test-shaped
+ output FAILS the check: the parked image still talking means the example flash
+ silently didn't take — the host analog of the device path's PID check.
+
+ Side effect: delivery-class failures (silent no-op, board_test build/flash
+ failure) set row['status'] = 'flash-failed' so verdict() preserves the cause;
+ the caller derives the final status from the return value via verdict()."""
+ data = check_host_serial(board)
+ if data:
+ if flashed_example and boardtest_output(data):
+ note.append('board_test output after example flash: silent flash no-op')
+ row['status'] = 'flash-failed'
+ return False
+ return True
+ variant = resolve_variant(board, 'device/board_test', note)
+ fw = ensure_board_test(board, variant, note)
+ if fw is None:
+ note.append('serial silent; board_test unavailable')
+ row['status'] = 'flash-failed'
+ return False
+ say(f'{board["name"]:26} recovery: serial silent, flashing board_test')
+ rc, err = call_flasher(getattr(hil_flash, f'flash_{board["flasher"]["name"].lower()}'), board, str(fw))
+ if rc != 0:
+ note.append(f'serial silent; board_test flash failed: {err}')
+ row['status'] = 'flash-failed'
+ return False
+ if not check_host_serial(board):
+ return False
+ if flashed_example:
+ # board_test talking proves the BOARD is alive, but the just-flashed
+ # example never produced serial — that verification still fails
+ note.append('example silent; board alive via board_test reflash')
+ return False
+ note.append('recovered via board_test reflash')
+ return True
+
+
+def device_recover_and_check(board: dict, example: str, variant: str, old_ino, note: list, row: dict, seen: dict) -> bool:
+ """Wait for the flashed board's uid to re-enumerate; on timeout, try one board
+ reset (skipped for flashers with no hardware reset — see hil_flash.RESET_NOOP,
+ it would just burn the wait) and wait again.
+
+ The PID policy is deliberately asymmetric. Pre-reset, the re-enumeration was
+ caused by the flash itself, so a PID mismatch most likely means the build dir
+ is stale (the flash DID write what find_firmware found) — warn, don't fail —
+ UNLESS the firmware was built this very run: then 'stale build' is impossible
+ and the mismatch can only be a silent flash no-op, which fails. Post-reset,
+ the re-enumeration proves nothing about the flash (the reset alone explains
+ it), so a mismatch is treated as a silent flash no-op and fails; an unknown
+ expected PID scores ok with a 'pid unverified' note in both paths."""
+ name = board['name']
+ expected_pid = get_expected_pid(example)
+ built_this_run = _builds.get((variant, example), (None, ''))[1] == 'ok'
+
+ def seen_hit(hit):
+ seen[board['uid']] = {'name': name, 'busport': hit[0], 'when': time.strftime('%Y-%m-%d %H:%M')}
+
+ hit = wait_device(board['uid'], None, old_ino, ENUM_WAIT)
+ if hit:
+ if expected_pid is not None and not hit[1].endswith(expected_pid):
+ if built_this_run:
+ row['device'] = f'❌ {hit[1]}'
+ note.append(f'pid {hit[1]}, this run built {expected_pid}: silent flash no-op')
+ row['status'] = 'flash-failed'
+ return False
+ note.append(f'⚠ pid {hit[1]}, source says {expected_pid}: stale build or silent flash no-op')
+ elif expected_pid is None:
+ note.append('pid unverified')
+ row['device'] = f'✅ {hit[1]}'
+ seen_hit(hit)
+ return True
+
+ flasher_name = board['flasher']['name'].lower()
+ if flasher_name in hil_flash.RESET_NOOP:
+ note.append(f'no hardware reset available for {flasher_name}')
+ row['device'] = '❌ not enumerated'
+ return False
+
+ say(f'{name:26} recovery: uid not up, resetting board')
+ rc, err = call_flasher(getattr(hil_flash, f'reset_{flasher_name}'), board)
+ if rc != 0:
+ note.append(f'reset failed: {err}')
+ hit = wait_device(board['uid'], None, old_ino, ENUM_WAIT_RETRY)
+ if not hit:
+ row['device'] = '❌ not enumerated'
+ note.append('reset did not help')
+ return False
+ if expected_pid is None:
+ row['device'] = f'✅ {hit[1]}'
+ note.append('reset recovered (pid unverified)')
+ seen_hit(hit)
+ return True
+ if hit[1].endswith(expected_pid):
+ row['device'] = f'✅ {hit[1]}'
+ note.append('reset recovered')
+ seen_hit(hit)
+ return True
+ row['device'] = f'❌ {hit[1]}'
+ note.append(f'reset recovered wrong pid, expected {expected_pid}: silent flash no-op')
+ row['status'] = 'flash-failed'
+ return False
+
+
+def check_board(board: dict, args, allow_recovery: bool, seen: dict) -> dict:
+ name = board['name']
+ row = {'name': name, 'probe': '❌ missing', 'flash': '–', 'device': '–', 'note': [], 'status': 'failed'}
+ note = row['note']
+
+ probe = find_usb(board['flasher']['uid'])
+ if probe:
+ row['probe'] = f'✅ {probe[0]}'
+ seen[board['flasher']['uid']] = {'name': f'{name} probe', 'busport': probe[0],
+ 'when': time.strftime('%Y-%m-%d %H:%M')}
+ else:
+ last = seen.get(board['flasher']['uid'])
+ note.append(f'probe last seen {last["busport"]} {last["when"]}' if last
+ else 'probe never seen by pool_check')
+ say(f'{name:26} probe MISSING ({board["flasher"]["name"]} {board["flasher"]["uid"]})')
+
+ # existing firmware only; a missing build is built further down (after a lock peek),
+ # except in scan/no-build modes and never for a missing probe
+ example, kind, variant, fw = pick_example(board, note, build_missing=False)
+ if kind == 'host':
+ note.append('host-only board')
+
+ if args.scan_only:
+ hit = find_device(board['uid'], None)
+ # report the BOARD's usb state too: enumerated (with busport), off-bus (normal
+ # when parked in board_test), or n/a for host-only boards
+ if hit:
+ row['device'] = f'✅ {hit[1]} @{hit[0]}'
+ elif kind == 'host':
+ row['device'] = '– n/a (host-only)'
+ else:
+ row['device'] = '⚫ off bus (parked?)'
+ # scan verifies probe presence only, so probe present is ok; a missing probe means
+ # no firmware could be delivered → flash-failed
+ row['status'] = 'ok' if probe else 'flash-failed'
+ if probe:
+ say(f'{name:26} probe ✅ {probe[0]}' + (f' device {hit[1]}' if hit else ''))
+ return row
+ if not probe:
+ row['status'] = 'flash-failed'
+ return row
+
+ bt_variant = resolve_variant(board, 'device/board_test', note)
+ need_example = example is None and not args.no_build
+ # board_test is also host_alive's recovery image, so host boards pre-build it
+ # even under --no-park; --no-build gates EVERY build, board_test included
+ need_bt = (not args.no_build
+ and (not args.no_park or kind == 'host')
+ and hil_flash.find_firmware(bt_variant, 'device/board_test',
+ flasher=board['flasher']['name']) is None)
+ if need_example or need_bt:
+ # builds are long and run BEFORE locking (park must never hold the flock through
+ # one); peek first so minutes of building are not wasted on — or a rebuilt tree
+ # swapped under — a board CI holds right now
+ peek = lock_board(name)
+ if isinstance(peek, str):
+ if peek.startswith('ERROR:'): # environment failure, not a held lock
+ row['flash'] = '❌ lock'
+ row['status'] = 'failed'
+ else:
+ row['flash'] = '🔒 locked'
+ row['status'] = 'locked'
+ note.append(peek)
+ say(f'{name:26} locked: {peek}')
+ return row
+ unlock_board(peek)
+ if need_example:
+ example, kind, variant, fw = pick_example(board, note, build_missing=True)
+ if need_bt and (example is not None or kind == 'host'):
+ # skip the park build when the example build already failed on a device board:
+ # the row returns before any flash/park could use it
+ ensure_board_test(board, bt_variant, note)
+
+ if example is None:
+ if not any(n.startswith(('build failed', 'build timeout', 'build produced',
+ 'build skipped', 'cannot build')) for n in note):
+ note.append('no firmware built')
+ if kind != 'host':
+ row['status'] = 'flash-failed'
+ say(f'{name:26} probe ✅ {probe[0]} (no firmware to flash)')
+ return row
+ # host-only board: aliveness is still checkable without flashing — reset and listen
+ # to whatever is on it (parked board_test echoes and hellos on the flasher UART)
+
+ lk = lock_board(name)
+ if isinstance(lk, str):
+ if lk.startswith('ERROR:'): # environment failure, not a held lock
+ row['flash'] = '❌ lock'
+ row['status'] = 'failed'
+ else:
+ row['flash'] = '🔒 locked'
+ row['status'] = 'locked'
+ note.append(lk)
+ say(f'{name:26} locked: {lk}')
+ return row
+ try:
+ if example is None: # host-only without firmware: UART-only aliveness check
+ ok = host_alive(board, note, row)
+ row['device'] = '✅ serial out' if ok else '❌ no serial out'
+ row['status'] = verdict(row, ok)
+ say(f'{name:26} – {row["device"]} (existing firmware)')
+ return row
+
+ pre = find_device(board['uid'], None)
+ old_ino = pre[2] if pre else None
+
+ try:
+ if not flash(board, fw, allow_recovery, probe[0], note):
+ row['flash'] = f'❌ {Path(example).name}'
+ row['status'] = 'flash-failed'
+ say(f'{name:26} flash FAILED ({example})')
+ return row
+ row['flash'] = f'✅ {Path(example).name}'
+
+ if kind == 'host':
+ ok = host_alive(board, note, row, flashed_example=True)
+ row['device'] = '✅ serial out' if ok else '❌ no serial out'
+ else:
+ ok = device_recover_and_check(board, example, variant, old_ino, note, row, seen)
+ row['status'] = verdict(row, ok)
+ say(f'{name:26} {row["flash"]} {row["device"]}')
+ return row
+ finally:
+ # teardown for EVERY path that attempted a flash (a failed programmer op can
+ # still have erased/half-written the target), while the lock is still held
+ if not args.no_park:
+ park_board(board, kind, row, note)
+ finally:
+ unlock_board(lk)
+
+
+def park_board(board: dict, kind: str, row: dict, note: list) -> None:
+ """Re-park with board_test, building it if absent (ensure_board_test), and
+ VERIFY it took: board_test never enumerates USB, so a device board's cafe
+ device must drop off the bus, and a host board must answer with board_test's
+ own output — a rc=0 park that changed nothing (silent no-op) must not pass.
+ A board left unparked marks an ok row flash-failed (never downgrading a
+ 'failed' verify verdict — that is the more diagnostic signal), with one
+ exception: an espressif board without the ESP-IDF env cannot build
+ board_test — noted, not a board fault."""
+ # capture BEFORE the park flash: uid-disappearance only verifies the park if the
+ # device was on the bus to begin with
+ on_bus_before = kind != 'host' and find_device(board['uid'], None) is not None
+ variant = resolve_variant(board, 'device/board_test', note)
+ fw = ensure_board_test(board, variant, note)
+ if fw is None:
+ if any(n.startswith('cannot build board_test') for n in note):
+ note.append('park skipped (no ESP-IDF env)')
+ else:
+ # --no-build disables builds, not parking (--no-park is that opt-out):
+ # a board left running a USB-active image is unparked either way
+ note.append('unparked: board_test not built (--no-build)'
+ if any(n.startswith('build skipped (--no-build): board_test') for n in note)
+ else 'unparked: board_test unavailable (build failed/timed out)')
+ if row['status'] == 'ok':
+ row['status'] = 'flash-failed'
+ return
+ rc, err = call_flasher(getattr(hil_flash, f'flash_{board["flasher"]["name"].lower()}'),
+ board, str(fw))
+ if rc != 0:
+ note.append(f'park flash failed: {err}')
+ if row['status'] == 'ok':
+ row['status'] = 'flash-failed'
+ return
+ if kind == 'host':
+ # no second reset (the park flash's own reset started board_test); POSITIVE
+ # marker: its hello must appear, and stale bridge-FIFO output alongside it is not
+ # disqualifying
+ data = check_host_serial(board, do_reset=False, want_hello=True)
+ if not (data and b'Hello from TinyUSB' in data):
+ note.append('park unverified: no board_test output')
+ if row['status'] == 'ok':
+ row['status'] = 'flash-failed'
+ return
+ if not on_bus_before:
+ # never enumerated this run: uid-disappearance cannot tell a verified park from a
+ # silent no-op — say so instead of passing vacuously
+ note.append('park unverified (device already off bus)')
+ return
+ deadline = time.monotonic() + 6
+ while time.monotonic() < deadline:
+ if find_device(board['uid'], None) is None:
+ return
+ time.sleep(0.5)
+ note.append('park unverified: device still enumerated')
+ if row['status'] == 'ok':
+ row['status'] = 'flash-failed'
+
+
+def check_board_safe(board: dict, args, allow_recovery: bool, seen: dict) -> dict:
+ """Isolate one board's exceptions: a crashing worker must not discard every
+ other board's row, the table, the topology, and the seen-cache write."""
+ try:
+ return check_board(board, args, allow_recovery, seen)
+ except Exception as e:
+ name = board.get('name', '?')
+ say(f'{name:26} INTERNAL ERROR: {e!r}')
+ return {'name': name, 'probe': '–', 'flash': '–', 'device': '❌ error',
+ 'note': [repr(e)[:120]], 'status': 'failed'}
+
+
+def controller_summary() -> list[str]:
+ """USB topology: controller (PCI addr, vendor) -> bus -> root-port subtree device
+ counts (hubs included, interfaces/root hubs not). Bus numbers renumber every boot;
+ PCI addresses and root-port numbers are stable."""
+ vendor_names = {'0x1022': 'AMD', '0x1912': 'Renesas', '0x8086': 'Intel', '0x1b21': 'ASMedia'}
+ ctrl = {}
+ for root in glob.glob('/sys/bus/usb/devices/usb*'):
+ bus = int(os.path.basename(root)[3:])
+ m = re.findall(r'[0-9a-f]{4}:[0-9a-f]{2}:[0-9a-f]{2}\.[0-9a-f]', os.path.realpath(root))
+ pci = m[-1] if m else '?'
+ c = ctrl.setdefault(pci, {'vendor': '?', 'buses': {}})
+ subtrees = {}
+ for d in glob.glob(f'/sys/bus/usb/devices/{bus}-*'):
+ b = os.path.basename(d)
+ if ':' in b:
+ continue
+ subtrees[b.split('.')[0]] = subtrees.get(b.split('.')[0], 0) + 1
+ c['buses'][bus] = subtrees
+ try:
+ vid = open(f'/sys/bus/pci/devices/{pci}/vendor').read().strip()
+ c['vendor'] = vendor_names.get(vid, vid)
+ except OSError:
+ pass
+
+ lines = []
+ for pci, c in sorted(ctrl.items()):
+ lines.append(f'{pci} ({c["vendor"]})')
+ for bus, subtrees in sorted(c['buses'].items()):
+ detail = ' '.join(f'{k}: {n} dev' for k, n in
+ sorted(subtrees.items(), key=lambda i: int(i[0].split('-')[1])))
+ lines.append(f' bus {bus}: {sum(subtrees.values())} devices'
+ + (f' {detail}' if detail else ''))
+ return lines
+
+
+def main() -> None:
+ # toolchain/flasher CLIs live in the user bin dirs, which non-login shells may lack --
+ # the same PATH shim hil_ci.sh applies on the remote side
+ for d in (Path.home() / 'bin', Path.home() / '.local' / 'bin'):
+ if d.is_dir() and str(d) not in os.environ.get('PATH', '').split(os.pathsep):
+ os.environ['PATH'] = f'{d}{os.pathsep}{os.environ.get("PATH", "")}'
+
+ parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+ parser.add_argument('config', nargs='?', help='HIL config json (default: by hostname)')
+ parser.add_argument('-b', '--board', action='append', default=[], help='only these boards')
+ parser.add_argument('-B', '--build-dir', default=None,
+ help='firmware parent dir, searched EXCLUSIVELY when given '
+ '(default: examples, plus cmake-build as fallback)')
+ parser.add_argument('--scan-only', action='store_true',
+ help='USB presence scan only: no locks, no flashing')
+ parser.add_argument('--no-build', action='store_true',
+ help='do not build missing firmware (default: build the light example on the spot)')
+ parser.add_argument('--no-park', action='store_true',
+ help='leave the light example running (default: park with board_test)')
+ # no cross-process flash budget against a concurrent hil_test.py run (its semaphores
+ # are in-process), so keep this modest
+ parser.add_argument('-j', '--jobs', type=int, default=4)
+ parser.add_argument('-v', '--verbose', action='store_true')
+ args = parser.parse_args()
+ global _no_build, _jobs, _build_sem
+ _no_build = args.no_build
+ _jobs = max(1, args.jobs)
+ _build_sem = threading.BoundedSemaphore(_jobs)
+
+ host = socket.gethostname()
+ cfg_name = args.config or CONFIG_BY_HOST.get(host, 'local.json')
+ cfg_path = Path(cfg_name)
+ if not cfg_path.exists():
+ cfg_path = REPO_ROOT / 'test' / 'hil' / cfg_name
+ if not cfg_path.exists():
+ sys.exit(f'config not found: {cfg_name} (host {host}; dev PCs need test/hil/local.json)')
+ with cfg_path.open() as f:
+ config = json.load(f)
+
+ boards = list(config['boards']) # boards-skip (parked hardware) is not scanned by default
+ if args.board:
+ boards += config.get('boards-skip', []) # explicitly named parked boards are fair game
+ unknown = set(args.board) - {b['name'] for b in boards}
+ if unknown:
+ sys.exit(f'board(s) not in {cfg_path.name}: {", ".join(sorted(unknown))}')
+ boards = [b for b in boards if b['name'] in args.board]
+
+ hil_flash.build_dir = args.build_dir or 'examples'
+ hil_util.verbose = args.verbose
+ if args.build_dir is None:
+ # default mode: search both standard layouts (cmake-build/ from tools/build.py and
+ # ESP-IDF, examples/ from manual builds). An EXPLICIT -B stays exclusive: the caller
+ # named an artifact tree, so a miss must report rather than flash an older build.
+ hil_flash.EXTRA_BUILD_DIRS = ['cmake-build', 'examples']
+ allow_recovery = not args.scan_only and can_recover()
+ seen = {}
+ try:
+ loaded = json.loads(SEEN_CACHE.read_text())
+ if isinstance(loaded, dict): # tolerate a torn/hand-edited cache
+ seen = {k: v for k, v in loaded.items() if isinstance(v, dict)}
+ except (OSError, ValueError):
+ pass
+
+ roots = ' + '.join(dict.fromkeys([hil_flash.build_dir, *hil_flash.EXTRA_BUILD_DIRS]))
+ say(f'pool check: host {host}, config {cfg_path.name}, {len(boards)} boards, '
+ f'{"scan-only" if args.scan_only else f"flash via {{{roots}}}/cmake-build-<board>"}'
+ f'{"" if allow_recovery or args.scan_only else ", recovery unavailable (no sudo -n / usb_recover.sh)"}')
+
+ if args.verbose:
+ rows = [check_board_safe(b, args, allow_recovery, seen) for b in boards]
+ else:
+ with io.StringIO() as spool, ThreadPoolExecutor(max_workers=args.jobs) as pool:
+ sys.stdout = spool # silence hil_util.run_cmd's COMMAND FAILED dumps; say() uses __stdout__
+ try:
+ rows = list(pool.map(lambda b: check_board_safe(b, args, allow_recovery, seen), boards))
+ finally:
+ sys.stdout = sys.__stdout__
+
+ try:
+ SEEN_CACHE.parent.mkdir(parents=True, exist_ok=True)
+ tmp = SEEN_CACHE.with_suffix('.json.tmp')
+ tmp.write_text(json.dumps(seen, indent=1, sort_keys=True) + '\n')
+ tmp.replace(SEEN_CACHE) # atomic: a killed run can't tear the cache
+ except OSError:
+ pass
+
+ status_mark = {'ok': '✅ ok', 'flash-failed': '❌ flash-failed', 'failed': '❌ failed',
+ 'locked': '🔒 locked'}
+ headers = ['Board', 'Probe', 'Flash', 'Device', 'Status', 'Note']
+ cells = [[r['name'], r['probe'], r['flash'], r['device'],
+ status_mark.get(r['status'], r['status']), '; '.join(r['note'])] for r in rows]
+ # display_width, not len(): ✅ / ❌ / 🔒 / ⚠ are one character and two columns, so
+ # len() pads every row holding one a column short of the header rule
+ _w = hil_util.display_width
+ widths = [max(_w(h), *(_w(c[i]) for c in cells)) if cells else _w(h)
+ for i, h in enumerate(headers)]
+ line = lambda vals: ('| ' + ' | '.join(hil_util.pad(v, w)
+ for v, w in zip(vals, widths)) + ' |')
+ print()
+ print(line(headers))
+ print('|' + '|'.join('-' * (w + 2) for w in widths) + '|')
+ for c in cells:
+ print(line(c))
+
+ print('\nUSB topology (controller → root-port subtree):')
+ for line in controller_summary():
+ print(f' {line}')
+
+ counts = {'ok': 0, 'flash-failed': 0, 'failed': 0, 'locked': 0}
+ for r in rows:
+ counts[r.get('status', 'failed')] += 1
+ print(f'\n{counts["ok"]} ok · {counts["flash-failed"]} flash-failed · {counts["failed"]} failed '
+ f'· {counts["locked"]} locked · in {time.monotonic() - t0:.0f}s')
+ if hil_util.sysfs_stranded():
+ # Without this the table is the worst kind of wrong: a device whose `serial` never
+ # answered is absent from the scan, which prints as "probe MISSING"/"off bus" for
+ # hardware that is physically present -- during exactly the incident this tool is
+ # run to diagnose, and it sends the operator to power-cycle a healthy rig.
+ print('WARNING: at least one sysfs read did not answer within '
+ f'{hil_util.SYSFS_READ_GRACE:.0f}s, so rows above that say a probe or board '
+ f'is missing may be this tool losing sight of healthy hardware rather than '
+ f'absent hardware. Find the wedged device (see the usb-kernel-recover '
+ f'skill) and re-run before acting on the table.')
+ sys.exit(min(counts['flash-failed'] + counts['failed'], 125))
+
+
+if __name__ == '__main__':
+ main()
diff --git a/test/hil/helper/hil_report.py b/test/hil/helper/hil_report.py
new file mode 100644
index 000000000..c93c8e6a1
--- /dev/null
+++ b/test/hil/helper/hil_report.py
@@ -0,0 +1,578 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+"""The HIL report document: one owner for hil_report.json and hil_report.md.
+
+The markdown IS a rendering of the sidecar -- every writer goes through render_report(), so
+a table can never contain something the JSON does not. This module owns the whole life of
+that document: the cell vocabulary, the one classifier both artifacts share, rendering, the
+writers, and the fold to one machine-readable verdict per board.
+
+Dual-mode by design: imported as `helper.hil_report` by hil_test.py, and run as a script by
+the operator (see .claude/agents/hil-operator.md). A script run puts test/hil/helper on
+sys.path rather than test/hil, so this module imports no sibling helper at all --
+_p and the width helpers below are defined locally for that reason.
+"""
+import argparse
+import json
+import sys
+import unicodedata
+from pathlib import Path
+
+
+def _w(s: str) -> int:
+ """Terminal COLUMNS, not characters. Every status mark in REPORT_CELL is one Python
+ character and TWO columns wide, so len() pads a cell holding one a column short and
+ the pipes drift out of line with the header rule for the whole table.
+
+ Local, like _p above and for the same reason: this module is also run as a script, and
+ under PYTHONSAFEPATH=1 a sibling import dies before argparse runs. hil_util carries the
+ same pair for callers that can import it.
+ """
+ return sum(2 if unicodedata.east_asian_width(c) in 'WF' else 1 for c in s)
+
+
+def _pad(s: str, width: int, center: bool = False) -> str:
+ """str.ljust/center, measured in display columns. See _w."""
+ room = max(0, width - _w(s))
+ if not center:
+ return s + ' ' * room
+ left = room // 2
+ return ' ' * left + s + ' ' * (room - left)
+
+
+def _p(*args, **kwargs) -> None:
+ """Print that cannot raise. Defined here rather than imported from hil_health: this
+ module is ALSO run as a script (hil-operator.md invokes it by path), and under
+ PYTHONSAFEPATH=1 -- which the suite's own MTP fixtures set -- sys.path[0] is not the
+ script dir, so any sibling import dies before argparse runs. Five lines beat that."""
+ try:
+ print(*args, **kwargs)
+ except (OSError, ValueError):
+ # ValueError too: printing to a CLOSED stream raises "I/O operation on closed
+ # file", and escaping here skips the containment path's os._exit.
+ pass
+
+REPORT_MD = 'hil_report.md'
+REPORT_JSON = 'hil_report.json'
+# The status vocabulary, shared by the code that WRITES a cell (hil_test's test runners) and
+# the code that reads one back (cell_state). One dict, so the human's table and the agent's
+# verdict cannot drift apart.
+REPORT_CELL = {'pass': '✅', 'fail': '❌', 'skip': '⚪'}
+BOUNDARY_CELL = 'same-PID boundary'
+LOCKED_CELL = 'board-locked'
+# A pseudo-test column, not a real one: write_timeout_report marks the boards that were
+# still dispatched when the pool guard fired. accumulate_report clears it on a retry.
+POOL_TIMEOUT_CELL = 'pool-timeout'
+# The other way a board can fail to report: the pool did not expire, a worker RAISED. Same
+# shape, different cause, and naming the cause is the whole point of the column -- a board
+# marked pool-timeout by an abort that never timed out sends the reader after the guard.
+RUN_ABORTED_CELL = 'run-aborted'
+
+
+def _load(report_dir: Path) -> tuple:
+ """(doc, readable) for the sidecar, coerced to the canonical shape.
+
+ hil_ci.sh uploads a sidecar as the --accumulate merge base, so a non-conforming one is
+ reachable from OUTSIDE the harness -- and every writer here runs on a path where a
+ TypeError costs the whole report. Coerce once, at the boundary, instead of guarding
+ each use: `banner: null` used to kill a fully successful run with a traceback and no
+ artifact at all, and `cells: null` sent write_timeout_report down its fallback so a
+ board that ate the whole pool guard was published as a pass.
+
+ `readable` is False only when a sidecar EXISTS but could not be parsed, or is absent --
+ both mean its rows are unrecoverable, which callers use to avoid destroying a markdown
+ that may still hold them."""
+ jpath = report_dir / REPORT_JSON
+ if not jpath.is_file():
+ return {'rows': [], 'banner': '', 'scope': '', 'caveat': ''}, False
+ try:
+ raw = json.loads(jpath.read_text())
+ if not isinstance(raw, dict):
+ raise ValueError('sidecar is not an object')
+ except (OSError, ValueError, TypeError):
+ return {'rows': [], 'banner': '', 'scope': '', 'caveat': ''}, False
+ rows = []
+ # isinstance, not `or []`: a sidecar with `rows: 1` iterates an int and raises outside
+ # the parse handler above.
+ for r in (raw.get('rows') if isinstance(raw.get('rows'), list) else []):
+ if not isinstance(r, dict) or 'board' not in r:
+ continue
+ cells = r.get('cells')
+ dur = r.get('duration')
+ # VALUES as well as keys: render_matrix does REPORT_CELL.get(v, v), which raises
+ # TypeError on an unhashable value, and cell_state does v.startswith. A non-str
+ # cell is corrupt, and dropping it renders blank -- "not run" -- which is the
+ # honest reading. Coercing it to str would make it classify as a PASS.
+ rows.append({'board': str(r['board']),
+ 'cells': {str(k): v for k, v in cells.items() if isinstance(v, str)}
+ if isinstance(cells, dict) else {},
+ 'duration': dur if isinstance(dur, str) else None})
+ text = lambda k: raw[k] if isinstance(raw.get(k), str) else ''
+ return {'rows': rows, 'banner': text('banner'), 'scope': text('scope'),
+ 'caveat': text('caveat')}, True
+
+
+def cell_state(v) -> str:
+ """'pass' | 'fail' | 'skip' for one report cell.
+
+ THE classifier -- the markdown tally and the per-board verdict both call this, so they
+ cannot disagree. 'fail' or a fail-icon prefix is a failure, 'skip' or a skip-icon prefix
+ is a skip, and EVERYTHING ELSE is a pass. That last arm is load-bearing: a passing test
+ may return a plain metric string ('480.0 MBps') that lands in the cell unprefixed, while
+ failures are guaranteed marked -- TestFail's docstring pins that its metric is
+ icon-prefixed precisely so render and tally treat it as a failure. Classifying unknown
+ shapes as fail here would publish a green table as a red verdict.
+
+ isinstance-guarded: cells are usually str but a caller may hand over None or a number,
+ and .startswith on those raises inside a report writer that must not raise."""
+ if v == 'fail' or (isinstance(v, str) and v.startswith(REPORT_CELL['fail'])):
+ return 'fail'
+ if v == 'skip' or (isinstance(v, str) and v.startswith(REPORT_CELL['skip'])):
+ return 'skip'
+ return 'pass'
+
+
+def render_matrix(rows_all: list) -> str:
+ """Render rows (list of (row_label, {example: status}, duration)) as an aligned
+ markdown matrix: columns = tests (bare names) centered, boards left-aligned,
+ per-row duration as the trailing column."""
+ seen = set()
+ for _, cells, _ in rows_all:
+ seen.update(cells)
+ if not seen:
+ return 'No tests were run.'
+
+ # 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):
+ name = t.rsplit('/', 1)[-1]
+ return (pinned.index(name) if name in pinned else len(pinned), name, t)
+
+ columns = sorted(seen, key=col_key)
+ headers = [c.rsplit('/', 1)[-1] for c in columns] + ['duration'] # bare example names
+
+ def cell(cells, col):
+ v = cells.get(col)
+ if v is None:
+ return ''
+ return REPORT_CELL.get(v, v) # status symbol, or a metric string (e.g. speed) verbatim
+
+ rows_vals = [(lbl, [cell(cells, c) for c in columns] + [dur or ''])
+ for lbl, cells, dur in rows_all]
+ board_hdr = 'Board'
+ # display_width, not len(): the ✅/❌/⚪ marks are one character and two columns
+ board_w = max([_w(board_hdr)] + [_w(lbl) for lbl, _ in rows_vals])
+ col_w = [max([_w(h)] + [_w(vals[i]) for _, vals in rows_vals])
+ for i, h in enumerate(headers)]
+
+ def line(label, values):
+ padded = [_pad(label, board_w)] + [_pad(v, w, center=True)
+ for v, w in zip(values, col_w)]
+ return '| ' + ' | '.join(padded) + ' |'
+
+ header = line(board_hdr, headers)
+ sep = '| ' + '-' * board_w + ' | ' + ' | '.join(':' + '-' * (w - 2) + ':' for w in col_w) + ' |'
+ body = [line(lbl, vals) for lbl, vals in rows_vals]
+
+ # 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 --
+ # through cell_state, the same call the per-board verdict makes.
+ kinds = [cell_state(v) for _, cells, _ in rows_all for v in cells.values()]
+ failed = kinds.count('fail')
+ skipped = kinds.count('skip')
+ passed = kinds.count('pass')
+ summary = (f'**{REPORT_CELL["pass"]} {passed} passed · {REPORT_CELL["fail"]} {failed} failed · '
+ f'{REPORT_CELL["skip"]} {skipped} skipped · blank not run**')
+
+ return summary + '\n\n' + '\n'.join([header, sep] + body)
+
+
+def render_report(doc: dict) -> str:
+ """The markdown IS a rendering of the sidecar. Every writer goes through here, so a
+ table can never contain something the JSON does not."""
+ # .get throughout, not subscripts: mark_report_abandoned renders a sidecar it did NOT
+ # write (hil_ci.sh reuses a persistent REMOTE_DIR, so it may be an older version's or
+ # a torn one) on the way to os._exit, and a KeyError there is not in its handler --
+ # it would unwind into multiprocessing's unbounded join and hang the runner it is
+ # trying to free. Same reason summarize() below reads cells as `r.get('cells') or {}`.
+ md = render_matrix([(r.get('board', '?'), r.get('cells') or {}, r.get('duration'))
+ for r in doc.get('rows') or [] if isinstance(r, dict)])
+ if doc.get('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
+ md = f'_Scoped run: {doc["scope"]}. Boards/tests not listed were not run._\n\n' + md
+ # banner, then caveat: a rig-health caveat outranks the table AND the scope note, and an
+ # abandon notice outranks even that -- the top of the report is where hil/SKILL.md tells
+ # the agent to look
+ if doc.get('banner'):
+ md = doc['banner'] + '\n' + md
+ if doc.get('caveat'):
+ md = doc['caveat'] + '\n' + md
+ return md
+
+
+def write_report(report_dir: Path, doc: dict) -> None:
+ """Write both artifacts from one document.
+
+ RAISES on failure, deliberately: every caller is on a path whose own handler exists to
+ report exactly this (write_timeout_report's _p warning, hil_test's fallback-of-the-
+ fallback). Swallowing OSError here made both of those dead code, so an unwritable or
+ root-owned report dir produced no artifact AND no message.
+
+ Renders BEFORE writing anything: committing the JSON first and then raising in
+ render_report left a sidecar saying "abandoned" beside a markdown still reading as a
+ clean green table -- the one invariant this module exists to hold."""
+ md = render_report(doc) + '\n'
+ report_dir.mkdir(parents=True, exist_ok=True)
+ (report_dir / REPORT_JSON).write_text(json.dumps(doc, indent=2) + '\n')
+ (report_dir / REPORT_MD).write_text(md, encoding='utf-8')
+
+
+def _abandon_notice(why: str) -> str:
+ # Wording is a CONTRACT: .claude/skills/hil/SKILL.md pins this banner as the case where
+ # "the table below IS this run's ... Report the results AND the abandonment". Calling
+ # the table partial would send the reading agent to re-run boards that already passed.
+ return (f'**HIL run abandoned: {why}** The table below was collected before the '
+ f'abandon; treat board results as unverified.\n')
+
+
+def _already_abandoned(doc: dict) -> bool:
+ """Whether THIS attempt already recorded how it ended.
+
+ `caveat` only. It used to check `banner` too, because hil_test.py folded its abandon
+ notices in there -- but banner is carried across an --accumulate retry by design, so a
+ stale notice from an earlier attempt silenced a genuinely new abandon and the run's own
+ failure went unrecorded. banner now carries rig HEALTH (which describes the conditions
+ the cells were collected under, and so must persist); caveat carries the run's OUTCOME
+ (which must not)."""
+ return '**HIL run ab' in doc.get('caveat', '')
+
+
+def _stamp_markdown(report_dir: Path, notice: str) -> None:
+ """Last line of defence: prepend the notice to the markdown itself.
+
+ pr_comment.yml cats only hil_report.md, so a path that gives up here publishes a clean
+ green table under an abandoned, non-zero job. Master did this unconditionally."""
+ mpath = report_dir / REPORT_MD
+ if not mpath.is_file():
+ return
+ # errors='replace' and catch ValueError: a torn report or a LANG=C locale raises
+ # UnicodeDecodeError -- NOT an OSError -- straight past os._exit.
+ body = mpath.read_text(encoding='utf-8', errors='replace')
+ if '**HIL run ab' not in body[:2000]:
+ mpath.write_text(notice + '\n' + body, encoding='utf-8')
+
+
+def mark_report_abandoned(report_dir: Path, why: str) -> None:
+ """Stamp an existing report as abandoned, in BOTH artifacts.
+
+ Best-effort and silent: this runs while the interpreter is being torn down, and an
+ exception here hangs the process in multiprocessing's unbounded join()."""
+ notice = _abandon_notice(why)
+ try:
+ doc, readable = _load(report_dir)
+ if readable:
+ if _already_abandoned(doc):
+ return # whoever got there first wins, WRITE included
+ doc['caveat'] = notice
+ write_report(report_dir, doc)
+ return
+ except (OSError, ValueError, TypeError, AttributeError):
+ pass # fall through -- a failure here must not cost the stamp entirely
+ # Unreadable sidecar, or the document write failed. Either way the markdown is what
+ # the PR comment reads, so stamp it directly rather than giving up.
+ try:
+ _stamp_markdown(report_dir, notice)
+ except (OSError, ValueError, TypeError, AttributeError):
+ pass
+
+
+def mark_report_no_boards(report_dir: Path, msg: str, fresh: bool = True) -> None:
+ """Record that the board filters intersected to nothing.
+
+ `fresh` mirrors hil_test's own flag, because this runs BEFORE the fresh wipe: without
+ it a fresh run whose filter emptied re-published the PREVIOUS run's green rows under
+ this run's red job -- the stale-table failure it exists to prevent. An --accumulate run
+ keeps them, since nothing this attempt did invalidates them."""
+ try:
+ doc, _ = _load(report_dir)
+ if not fresh and _already_abandoned(doc):
+ # SKILL.md gives the two notices OPPOSITE rules, and an abandon outranks a
+ # filter that matched nothing -- do not overwrite the record of a failed run.
+ # Only while ACCUMULATING, though: this runs before the fresh wipe, so guarding
+ # a fresh run would leave the previous attempt's rows AND its abandon notice
+ # published as this run's.
+ return
+ # A fresh run carries NOTHING from the prior sidecar -- rows, banner and scope
+ # alike, matching accumulate_report, which builds from an empty prior when fresh.
+ # Resetting only rows republished a stale rig-health note and a stale scope line
+ # under this run's notice, from a leftover or uploaded sidecar.
+ prior = {'rows': [], 'banner': '', 'scope': ''} if fresh else doc
+ write_report(report_dir, {'rows': prior['rows'], 'banner': prior['banner'],
+ 'scope': prior['scope'],
+ 'caveat': f'**HIL run selected no boards.** {msg}\n'})
+ except (OSError, ValueError, TypeError, AttributeError):
+ pass # loud on stdout already; the exit code is what the job reads
+
+
+def accumulate_report(mret: list, report_dir: Path, fresh: bool, scope: str = '',
+ banner: str = '', caveat: str = '') -> str:
+ """Merge this run's results into json in report_dir, then (re)write
+ the markdown matrix to md. `fresh` (a first run, no --accumulate)
+ starts a new report; otherwise a re-run accumulates so boards/tests that
+ already passed are preserved while re-run cells are updated. `scope` names the
+ board filter, if any, so a scoped table is not mistaken for a full one.
+ Returns the md.
+
+ `mret` is hil_test.py's worker-result shape (name, err, fts, rows, ...), so this one
+ function knows something about its caller that the rest of the module does not. Folding
+ mret into rows could live in hil_test and only the merge here, but that would rewrite
+ the subtle parts -- stale board-locked clearing, BOUNDARY_CELL dropping, duration=None
+ preservation -- for a tidier seam. Data-shape coupling, not an import cycle."""
+ # ONE canonical load: a sidecar reaching here may have been uploaded by hil_ci.sh as
+ # the merge base, so it is untrusted input. `banner` carries forward -- it describes
+ # the conditions the earlier cells were collected under, and the .failed spec re-runs
+ # only FAILURES so those passes are never re-earned. `caveat` does NOT: it records how
+ # a RUN ENDED, and this attempt has not ended yet. Carrying it made a clean retry
+ # publish "HIL run abandoned" over a run where nothing was abandoned.
+ prior = {'rows': [], 'banner': ''}
+ if not fresh:
+ prior, _ = _load(report_dir)
+ acc = {r['board']: [dict(r['cells']), r['duration']] for r in prior['rows']}
+ prior_banner = prior['banner']
+
+ # 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(LOCKED_CELL in cells for _, cells, _ in rows):
+ # 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(LOCKED_CELL, None)
+ # and the pool-timeout mark: write_timeout_report stamps it on a board that
+ # never reported, and update() below MERGES, so without this a board that
+ # passed clean on the retry kept a red cell for ever.
+ stale[0].pop(POOL_TIMEOUT_CELL, None)
+ stale[0].pop(RUN_ABORTED_CELL, None)
+ if not stale[0]:
+ # 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])
+ # a row that ran is no longer pool-timed-out, whatever it is keyed by
+ row[0].pop(POOL_TIMEOUT_CELL, None)
+ row[0].pop(RUN_ABORTED_CELL, None)
+ # the boundary cell is only ever written on failure, so a re-run of this
+ # variant that cleared the boundary must drop the previous attempt's ❌
+ if BOUNDARY_CELL not in cells:
+ row[0].pop(BOUNDARY_CELL, None)
+ row[0].update(cells)
+ if dur is not None:
+ 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 ''
+ doc = {'rows': [{'board': k, 'cells': c, 'duration': d} for k, (c, d) in acc.items()],
+ 'banner': banner, 'scope': scope, 'caveat': caveat}
+ # through write_report, not hand-rolled: writing the JSON and only then rendering is
+ # the ordering write_report exists to forbid -- a render failure left the sidecar ahead
+ # of the markdown, which is the one invariant this module holds.
+ write_report(report_dir, doc)
+ return render_report(doc)
+
+
+def _write_stuck_over_prior_md(report_dir: Path, doc: dict) -> None:
+ """Sidecar unrecoverable: rebuild it from the stuck rows alone, but leave the
+ markdown's existing table beneath the caveat rather than throwing real results away.
+
+ The one place the md-is-a-rendering-of-the-json invariant is deliberately suspended,
+ because there is no readable json left for it to be a rendering of."""
+ try:
+ prior = (report_dir / REPORT_MD).read_text(encoding='utf-8')
+ except (OSError, ValueError):
+ prior = ''
+ # Say so explicitly: those rows exist only as rendered text, so no later --accumulate
+ # can merge them back. Claiming the sidecar represents them would be false.
+ note = ('_The table below is a previous attempt\'s rendered output. The sidecar could '
+ 'not be read, so those rows are NOT in it and will not survive another run._\n')
+ head = (doc['banner'] + '\n' if doc['banner'] else '') + doc['caveat'] + '\n' + note
+ body = prior if prior.strip() else render_matrix(
+ [(r['board'], r['cells'], r['duration']) for r in doc['rows']])
+ report_dir.mkdir(parents=True, exist_ok=True)
+ (report_dir / REPORT_JSON).write_text(json.dumps(doc, indent=2) + '\n')
+ (report_dir / REPORT_MD).write_text(head + '\n' + body, encoding='utf-8')
+
+
+def write_timeout_report(report_dir: Path, boards, secs: int,
+ banner: str = '', prefix: str = '',
+ cell: str = POOL_TIMEOUT_CELL) -> None:
+ """Leave a report behind when the worker pool has to be abandoned.
+
+ map_async is all-or-nothing, so a timeout loses every per-board result and the report
+ dir would stay empty with no reason for the failure. Any prior attempt's rows are kept
+ and each stuck board is marked with a POOL_TIMEOUT_CELL beside them.
+
+ `prefix` is the preflight rig-health verdict and goes to the BANNER, where rig health
+ lives and where an --accumulate retry carries it forward; the abandon notice goes to
+ the caveat, which does not carry. Folding both into the caveat is what made a clean
+ retry report an abandonment that had not happened."""
+ try:
+ # names INSIDE the try: a roster entry that is not a dict raises here, and outside
+ # it that escaped and stranded the runner.
+ names = [b.get('name', '?') if isinstance(b, dict) else '?' for b in boards]
+ caveat = banner or (
+ f'**HIL run abandoned: worker pool timed out after {secs}s.**\n\n'
+ f'No per-board results could be collected for this attempt. Rows other than '
+ f'the {cell} cells below are from an earlier attempt. Boards '
+ f'dispatched:\n\n' + '\n'.join(f'- {n}' for n in names) + '\n')
+ doc, readable = _load(report_dir)
+ rows = doc['rows']
+ by_board = {r['board']: r for r in rows}
+ for name in names:
+ row = by_board.get(name)
+ if row is None:
+ rows.append({'board': name, 'cells': {cell: 'fail'},
+ 'duration': None})
+ else:
+ # _load guarantees `cells` is a dict, so a null-cells row from an uploaded
+ # sidecar can no longer send this down the fallback and publish a board
+ # that ate the whole pool guard as a pass.
+ row['cells'][cell] = 'fail'
+ out = {'rows': rows, 'scope': doc['scope'], 'caveat': caveat,
+ 'banner': ((doc['banner'] + prefix) if prefix not in doc['banner']
+ else doc['banner'])}
+ if not readable and (report_dir / REPORT_MD).is_file():
+ # `readable` covers ABSENT as well as torn: an absent sidecar beside an intact
+ # markdown used to re-render from the stuck row alone and destroy real results.
+ _write_stuck_over_prior_md(report_dir, out)
+ return
+ write_report(report_dir, out)
+ except Exception as e: # noqa: BLE001
+ # Deliberately broad: this is the first statement of the pool-abandon path, so ANY
+ # escape skips kill_pool_children and os._exit and strands the runner.
+ _p(f'warning: cannot write {REPORT_MD} to {report_dir}: {e}', flush=True)
+ try:
+ # Same wording as above and the same guarded name extraction -- the fallback
+ # used to re-derive b.get("name") outside any try and raise identically, so a
+ # malformed roster left NO artifact at all.
+ names = [b.get('name', '?') if isinstance(b, dict) else '?' for b in boards]
+ head = (prefix + '\n' if prefix else '') + (banner or (
+ f'**HIL run abandoned: worker pool timed out after {secs}s.**\n\n'
+ f'No per-board results could be collected for this attempt, so the table '
+ f'below (if any) is from an earlier one. Boards dispatched:\n\n'
+ + '\n'.join(f'- {n}' for n in names) + '\n'))
+ try:
+ prior = (report_dir / REPORT_MD).read_text(encoding='utf-8')
+ except (OSError, ValueError):
+ prior = ''
+ report_dir.mkdir(parents=True, exist_ok=True)
+ (report_dir / REPORT_MD).write_text(
+ head + (f'\n{prior}' if prior else ''), encoding='utf-8')
+ except Exception as e2: # noqa: BLE001
+ _p(f'warning: fallback {REPORT_MD} write failed too: {e2}', flush=True)
+
+
+def variants_of(cfg: dict, board: str) -> list:
+ for b in cfg.get('boards', []):
+ if b['name'] == board:
+ return [v['name'] for v in (b.get('variant') or [])] or [board]
+ return [board]
+
+
+def summarize(cfg: dict, boards: list, report: dict) -> dict:
+ # .get, not a subscript: this is the one reader an agent's verdict depends on, and a
+ # row without 'board' used to kill the CLI with a traceback and no results at all --
+ # hil-validate.js then reports every board as "hil-operator returned no entry".
+ rows = {r['board']: r.get('cells') or {}
+ for r in (report.get('rows') or [])
+ if isinstance(r, dict) and 'board' in r}
+ owner = {v['name']: b['name'] for b in cfg.get('boards', [])
+ for v in (b.get('variant') or [])}
+ results = []
+ for board in boards:
+ names = variants_of(cfg, board)
+ mine = {n: rows[n] for n in names if n in rows}
+ # a variant name that is neither declared nor prefixed cannot be attributed; the
+ # `<board>-` fallback only helps ad-hoc builds, it is not the primary path. It must
+ # also never steal a row DECLARED by another board: a declared variant need not start
+ # with its own board's name, so it may happen to start with this board's name plus '-'.
+ mine.update({n: c for n, c in rows.items()
+ if n.startswith(f'{board}-') and n not in mine
+ and owner.get(n, board) == board})
+ # the BOARD-name row too: hil_test writes lock contention and pool timeouts keyed
+ # by board name, but variants_of returns only DECLARED variant names -- and
+ # nanoch32v203 / ch32v307v_r1_1v0 declare none equal to their board name. Without
+ # this those rows are invisible, so a lock held by concurrent CI is published as a
+ # hardware FAIL and hil-validate.js never retries it.
+ if board in rows and board not in mine:
+ mine[board] = rows[board]
+ if not mine:
+ results.append({'board': board, 'ran': False, 'pass': False, 'locked': False,
+ 'detail': 'no report row for this board'})
+ continue
+ # a wedge outranks lock contention: `locked` short-circuits `detail` below, so a
+ # stale board-locked cell from an earlier attempt used to mask the pool-timeout
+ # cell the retry added -- publishing a board that hung the rig as LOCKED, which
+ # hil-validate.js then RE-RUNS, paying another pool guard on it. RUN_ABORTED_CELL
+ # is written by the same _abort_report path for a board the guard never reached,
+ # and must outrank it for the same reason.
+ wedged = any(POOL_TIMEOUT_CELL in cells or RUN_ABORTED_CELL in cells
+ for cells in mine.values())
+ locked = not wedged and any(LOCKED_CELL in cells for cells in mine.values())
+ bad = []
+ for vname, cells in sorted(mine.items()):
+ for test, val in sorted(cells.items()):
+ if test == LOCKED_CELL:
+ continue
+ if cell_state(val) == 'fail':
+ bad.append(f'{vname} {test}: {val}')
+ ok = not bad and not locked
+ if locked:
+ detail = 'held by another holder; not flashed'
+ elif bad:
+ detail = '; '.join(bad)
+ else:
+ detail = f'{len(mine)} variant(s), {sum(len(c) for c in mine.values())} cell(s) ok'
+ results.append({'board': board, 'ran': True, 'pass': ok, 'locked': locked,
+ 'detail': detail})
+ # `caveat` too: an abandoned or no-boards run says so THERE, and this JSON is all
+ # an agent gets -- leaving it in the sidecar puts it back where only a human looks.
+ return {'results': results, 'banner': report.get('banner', ''),
+ 'caveat': report.get('caveat', '')}
+
+
+def main() -> int:
+ ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+ ap.add_argument('config_file')
+ ap.add_argument('-b', '--board', action='append', default=[],
+ help='boards to report on; default: every board in the config')
+ ap.add_argument('--report-dir', default='.', help=f'where {REPORT_JSON} lives (default: cwd)')
+ a = ap.parse_args()
+
+ cfg = json.loads(Path(a.config_file).read_text())
+ boards = a.board or [b['name'] for b in cfg.get('boards', [])]
+ jpath = Path(a.report_dir) / REPORT_JSON
+ if not jpath.is_file():
+ print(f'error: {jpath} not found -- did hil_test.py run in this directory?',
+ file=sys.stderr)
+ return 1
+ # through _load, like every writer: feeding raw JSON to summarize left the one reader an
+ # agent's verdict depends on crashing on the malformed sidecars the writers tolerate.
+ doc, _ = _load(Path(a.report_dir))
+ json.dump(summarize(cfg, boards, doc), sys.stdout, indent=2)
+ print()
+ return 0
+
+if __name__ == '__main__':
+ sys.exit(main())
diff --git a/test/hil/helper/hil_util.py b/test/hil/helper/hil_util.py
new file mode 100644
index 000000000..6f84c143d
--- /dev/null
+++ b/test/hil/helper/hil_util.py
@@ -0,0 +1,571 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+# Bottom layer of the HIL harness: the bounded command runner plus the shared helpers and
+# data every other module needs. Stays stdlib-only; its one local dependency is
+# tools/rtt.py (the RTT console, loaded by path below) -- everything
+# else imports this, including the unit tests on GitHub's bare runner; never import them
+# from here. Callers set the module global `verbose`.
+
+from __future__ import annotations
+
+import glob
+import os
+import signal
+import subprocess
+import unicodedata
+import threading
+import sys
+from pathlib import Path
+from typing import Any
+
+
+# -------------------------------------------------------------
+# HIL example test lists, shared by hil_test.py (runner) and ci_select.py (PR-diff
+# selector). Run order is shuffled per board (see test_board); every example carries a
+# unique hardcoded idProduct (see its usb_descriptors.c).
+# -------------------------------------------------------------
+
+# device tests
+device_tests = [
+ 'device/cdc_dual_ports',
+ 'device/cdc_msc',
+ 'device/dfu',
+ 'device/cdc_msc_throughput',
+ 'device/audio_test_freertos',
+ 'device/dfu_runtime',
+ 'device/cdc_msc_freertos',
+ 'device/hid_boot_interface',
+ 'device/msc_dual_lun',
+ 'device/hid_generic_inout',
+ 'device/printer_to_cdc',
+ 'device/midi_test',
+ 'device/mtp',
+ 'device/usbtest', # cafe:4010, unique PID; runs the Linux testusb tier-4 battery via usbtest.py
+ # 'device/net_lwip_webserver', # disabled for PR #3605: USB net iface enum is flaky on the CI HIL host
+]
+
+dual_tests = [
+ 'dual/host_info_to_device_cdc',
+]
+
+host_test = [
+ 'host/cdc_msc_hid',
+ 'host/msc_file_explorer',
+ 'host/msc_file_explorer_freertos',
+ 'host/device_info',
+]
+
+verbose = False
+
+def pos_int_env(name: str, default: int) -> int:
+ # One parsing policy for every HIL_* knob: a bare int() crashes every run at import
+ # on a malformed value, and 0/negative silently removes the bound the knob enforces.
+ try:
+ v = int(os.getenv(name, str(default)))
+ except ValueError:
+ print(f'warning: {name} is not an integer; using {default}',
+ file=sys.stderr, flush=True)
+ return default
+ if v <= 0:
+ print(f'warning: {name}={v} is not usable; using {default}',
+ file=sys.stderr, flush=True)
+ return default
+ return v
+
+
+def pos_float_env(name: str, default: float) -> float:
+ try:
+ v = float(os.getenv(name, str(default)))
+ except ValueError:
+ print(f'warning: {name} is not a number; using {default}',
+ file=sys.stderr, flush=True)
+ return default
+ # float() accepts 'inf'/'nan': an infinite serial timeout is an unbounded read, the
+ # very thing these knobs exist to prevent, and nan fails every comparison silently
+ if not (v > 0 and v < float('inf')):
+ print(f'warning: {name}={v} is not usable; using {default}',
+ file=sys.stderr, flush=True)
+ return default
+ return v
+
+
+CMD_TIMEOUT = pos_int_env('HIL_CMD_TIMEOUT', 180)
+# Post-SIGKILL reap, spent ON TOP of a run_cmd timeout whenever the child has to be killed.
+# A caller budgeting several bounded steps must add one of these PER STEP, or its own outer
+# bound fires mid-step -- for a flasher, orphaning it on the probe.
+REAP_GRACE = 10
+
+TINYUSB_ROOT = Path(__file__).resolve().parents[3] # test/hil/helper/ -> repo root
+
+
+def display_width(s: str) -> int:
+ """Terminal COLUMNS, not characters.
+
+ The status marks the reports use -- ✅ ❌ ⚪ ⚠ 🔒 -- are one Python character and TWO
+ columns wide. Measuring with len() pads every cell containing one a column short, so
+ the pipes drift out of line against the header rule for the whole table.
+ """
+ return sum(2 if unicodedata.east_asian_width(c) in 'WF' else 1 for c in s)
+
+
+def pad(s: str, width: int, center: bool = False) -> str:
+ """str.ljust/center, measured in display columns. See display_width."""
+ room = max(0, width - display_width(s))
+ if not center:
+ return s + ' ' * room
+ left = room // 2
+ return ' ' * left + s + ' ' * (room - left)
+
+
+def cmd_stdout_text(out: Any) -> str:
+ if out is None:
+ return ''
+ if isinstance(out, bytes):
+ return out.decode('utf-8', errors='ignore')
+ return str(out)
+
+
+def _banner_body(out: Any, err: Any) -> str:
+ # split_stderr callers keep the diagnostic in stderr — a banner of stdout alone
+ # would be blank exactly when something went wrong
+ body = cmd_stdout_text(out)
+ err_text = cmd_stdout_text(err)
+ if err_text:
+ body = f'{body}\n{err_text}' if body else err_text
+ return body
+
+
+# Shared with compact_output's stripper in hil_test: duplicated literals let the two
+# layers drift and reintroduce literal marker noise mid-row in the GitHub log.
+GROUP_MARK, ENDGROUP_MARK = '::group::', '::endgroup::'
+
+
+def strip_workflow_markers(line: str) -> str:
+ # run_cmd only ever emits markers at line start; mid-line is not a real case.
+ return line.removeprefix(GROUP_MARK).removeprefix(ENDGROUP_MARK)
+
+
+def _ci_log_groups() -> bool:
+ # GitHub folds ::group::/::endgroup:: only at line start of the JOB's real stdout; a
+ # pool worker's capture is compacted into one row line, where they render literally.
+ return bool(os.getenv('CI')) and sys.stdout is sys.__stdout__
+
+
+def _print_banner(title: str, out: Any, err: Any) -> None:
+ print()
+ if _ci_log_groups():
+ print(f'{GROUP_MARK}{title}')
+ print(_banner_body(out, err))
+ print(ENDGROUP_MARK)
+ else:
+ print(title)
+ print(_banner_body(out, err))
+
+
+SYSFS_READ_GRACE = 2.0 # default bound on one attribute read; see read_sysfs
+
+# path -> the kernfs inode the node had when its bounded read gave up. Keyed by INODE, not
+# by path alone: a busport does not change when a board returns to the same physical port,
+# so a path-only blacklist outlives the wedge -- hil_pool_check resets or reflashes the
+# board, wait_device polls that busport for the new inode, and the scan it polls through
+# would never look at the device again. A re-enumeration destroys the kernfs node and makes
+# a new one, so a CHANGED inode is the all-clear. os.stat is safe on a wedged device: it
+# does not call ->show(), so it cannot block on the lock the reader is stuck behind.
+_stranded: dict = {}
+_strand_hits: dict = {} # path -> how many times it has stranded, ever
+_refused: set = set() # paths answered None WITHOUT reading, once past _STRAND_MAX
+_strand_lock = threading.Lock()
+_ever_stranded = False
+
+# Each strand costs a thread AND an fd for the life of the process -- on sysfs the open()
+# SUCCEEDS and only the read blocks. Two ceilings, because they bound different things:
+#
+# _PATH_STRAND_MAX -- a device that FLAPS while still wedged re-enumerates, clears the
+# inode memo, and strands again. Per path, so one sick board cannot leak without bound.
+# After this many it stays memoised whatever its inode says.
+# _STRAND_MAX -- a whole-process backstop against RLIMIT_NOFILE or the thread ceiling,
+# which would raise inside a worker and lose every board's result. Counted PER PATH, not
+# per reader: hil_pool_check runs four poll threads over one bus, and counting each
+# reader let four threads on ONE wedged device spend four credits between them. With
+# per-path counting a 27-board rig cannot approach this.
+_PATH_STRAND_MAX = 4
+_STRAND_MAX = 64
+
+
+def sysfs_stranded() -> bool:
+ """True once any bounded read has given up, and it STAYS true.
+
+ A sticky, process-wide fact, so it answers exactly one question: "could anything in
+ this process's output be the tool losing sight of healthy hardware?" -- which is what
+ hil_pool_check's footer needs. It canNOT answer "is THIS device unreadable" for a
+ caller deciding what a single missing device means; use path_stranded() for that.
+ """
+ return _ever_stranded
+
+
+def strand_note() -> str:
+ """Suffix for an absence claim, so "not found" never reads as proven absence.
+
+ Lives here because every caller that can say "not found" needs the same sentence, and
+ the one that had to re-invent it got missed: a wedged-but-enumerated printer was
+ reported as an enumeration failure, sending a maintainer after firmware.
+ """
+ return (' (a bounded sysfs read gave up, so "not found" here means "could not tell"'
+ ' -- see the usb-kernel-recover skill)') if sysfs_stranded() else ''
+
+
+def path_stranded(path: str) -> bool:
+ """Whether THIS attribute is currently memoised as unreadable.
+
+ The per-device question sysfs_stranded() cannot answer. usbtest uses it to tell a DUT
+ whose `serial` is held under device_lock from one that genuinely left the bus, because
+ the difference decides whether it performs driver-registry writes that take the
+ UNINTERRUPTIBLE device_lock.
+ """
+ with _strand_lock:
+ return path in _stranded or path in _refused
+
+
+def read_sysfs(path: str, timeout: float = SYSFS_READ_GRACE) -> str | None:
+ """A sysfs attribute's value, or None when it did not answer.
+
+ BOUNDED BY DEFAULT, and it has to be. `serial` is served by usb_string_attr, which
+ takes usb_lock_device_interruptible (v6.12.96 sysfs.c:141-143) -- the same lock a
+ wedged usbfs ioctl holds. Every OTHER attribute the harness reads (idVendor, idProduct,
+ bcdDevice, busnum, devnum, speed) is a lock-free sysfs_emit from a cached field and
+ cannot block.
+
+ "Only the wedged board's own worker pays" is FALSE, which is why the bound is not
+ opt-in: usb_scan reads `serial` on every device matching the VID to find the one it
+ wants, so resolving MY board touches every peer's locked attribute. hil_lock's
+ controller_of does that from controller_permit, on essentially every board -- one
+ wedged DUT would stall every worker, not one. hil_pool_check has no guard at all.
+
+ A give-up reads as None, the same as unreadable: there is no third value and no
+ per-attribute blindness. The memo is keyed by inode so the cost stays on the device
+ that is actually wedged; path_stranded() tells a caller which device that was.
+ """
+ with _strand_lock:
+ was = _stranded.get(path)
+ stuck_for_good = _strand_hits.get(path, 0) >= _PATH_STRAND_MAX
+ budget_spent = len(_stranded) >= _STRAND_MAX
+ if was is not None:
+ try:
+ if os.stat(path).st_ino == was:
+ return None # same kernfs node, still wedged
+ except OSError:
+ pass # gone: let the read below report it
+ if stuck_for_good:
+ return None # flapped too many times; see _PATH_STRAND_MAX
+ with _strand_lock:
+ _stranded.pop(path, None) # a different inode is the all-clear
+ elif budget_spent:
+ # see _STRAND_MAX. Recorded, not just returned: usbtest fails CLOSED on
+ # path_stranded() before the lock-taking cleanup, and a path we declined to read
+ # is exactly the case it must not be told is readable-and-absent.
+ with _strand_lock:
+ _refused.add(path)
+ return None
+
+ # BEFORE the read, not after: a node that re-enumerates DURING the grace would
+ # otherwise have its brand-new HEALTHY inode recorded as the wedged one, and only a
+ # second re-enumeration could ever clear it. If it cannot be stat'd there is no key to
+ # memoise against, so the path is simply re-read next time -- the open fails fast.
+ try:
+ ino = os.stat(path).st_ino
+ except OSError:
+ ino = None
+ out: dict = {}
+
+ def _read():
+ try:
+ with open(path) as f:
+ out['v'] = f.read().strip()
+ except (OSError, ValueError):
+ pass
+
+ t = threading.Thread(target=_read, daemon=True)
+ t.start()
+ t.join(timeout)
+ # `out` FIRST: a reader can deposit its value and still be alive for a moment
+ # afterwards, and counting that as a strand blacklists a healthy attribute forever
+ if 'v' in out:
+ # a path that answered is not refused any more: _refused feeds path_stranded(),
+ # and a stale entry makes usbtest read a LATER genuine disconnect as "cannot tell"
+ with _strand_lock:
+ _refused.discard(path)
+ if t.is_alive() and 'v' not in out:
+ global _ever_stranded
+ announce = False
+ if ino is None:
+ # the pre-read stat lost a race the open then won -- the node was replaced
+ # between them. Re-stat now: the reader is blocked on whatever node exists,
+ # so this is the key it is stuck on. Without a key nothing is memoised and
+ # every later poll starts another permanent thread and fd for this path.
+ try:
+ ino = os.stat(path).st_ino
+ except OSError:
+ pass
+ with _strand_lock:
+ _ever_stranded = True
+ if ino is not None:
+ first = path not in _stranded # count the PATH once, not each reader
+ _stranded[path] = ino
+ if first:
+ _strand_hits[path] = _strand_hits.get(path, 0) + 1
+ announce = len(_stranded) == _STRAND_MAX
+ else:
+ _refused.add(path) # unkeyable: at least do not vouch for it
+ if announce:
+ print(f'warning: {_STRAND_MAX} devices have unreadable sysfs attributes; '
+ f'refusing to start more bounded readers, so later reads answer None '
+ f'without looking. Find the wedged device (usb-kernel-recover skill).',
+ file=sys.stderr, flush=True)
+ return None
+ return out.get('v')
+
+
+def usb_scan(vid_pid=None, serial=None, vid=None, timeout=SYSFS_READ_GRACE) -> list:
+ """Enumerated USB devices matching the filters: [{busport, dir, vid, pid, serial}].
+
+ Three rules, one implementation for every caller:
+
+ * Root hubs excluded (glob `*-*`): no DUT is one, and scans including them measured
+ seconds slower (observation, no mechanism -- the "autosuspend wake" explanation was
+ wrong; usb_string_attr reads a cached string, sysfs.c:141-143).
+ * idVendor/idProduct first: lock-free `sysfs_emit` from udev->descriptor
+ (sysfs.c:688-705), so they rule out nearly every device for free.
+ * `serial` LAST and BOUNDED: it is the only attribute here served under the device
+ lock, so it is the only one that can block. Filtering on the lock-free pair first
+ keeps most devices out of it, but a scan for ONE board still reads the serial of
+ every peer that shares its VID -- so the bound is what stops one wedged DUT from
+ stalling every caller (see read_sysfs).
+ """
+ out = []
+ for d in glob.glob('/sys/bus/usb/devices/*-*'):
+ # `in`, not endswith: an interface is '<busport>:<cfg>.<ifnum>' (2-4:1.0), which
+ # CONTAINS the colon rather than ending with it. Screening them out here is worth
+ # real time -- they were 31 of 44 matches on this rig.
+ if ':' in os.path.basename(d):
+ continue
+ try:
+ with open(os.path.join(d, 'idVendor')) as f:
+ dev_vid = f.read().strip()
+ with open(os.path.join(d, 'idProduct')) as f:
+ dev_pid = f.read().strip()
+ except OSError:
+ continue # vanished mid-walk, or not a device dir: a fact, not unknown
+ if vid_pid is not None and (dev_vid, dev_pid) != tuple(vid_pid):
+ continue # ruled out for free, without touching the locked attribute
+ if vid is not None and dev_vid != vid:
+ continue # same, for callers that know the VID but not the PID
+ sn = read_sysfs(os.path.join(d, 'serial'), timeout)
+ if sn is None:
+ continue # no serial attribute
+ if serial is not None and sn.lower() != serial.lower():
+ continue
+ out.append({'busport': os.path.basename(d), 'dir': d,
+ 'vid': dev_vid, 'pid': dev_pid, 'serial': sn})
+ return out
+
+
+def _close_pipes(p: subprocess.Popen) -> None:
+ """Close OUR ends of an abandoned child's pipes. Never raises."""
+ for pipe in (p.stdout, p.stderr, p.stdin):
+ try:
+ if pipe is not None:
+ pipe.close()
+ except OSError:
+ pass
+
+
+def run_alongside(argv: list, work, timeout: int) -> subprocess.CompletedProcess:
+ """Run `argv` alongside `work()`, which runs in THIS thread, then reap it -- bounded.
+
+ The read-while-we-write shape run_cmd cannot express: the caller needs the child
+ RUNNING while it does something else. Everything else about the contract is run_cmd's
+ -- own session, killpg, bounded reap, our pipe ends closed, rc 124 on the kill.
+
+ A PROCESS, not a thread: an abandoned thread keeps the fd, and usblp_open returns
+ -EBUSY while usblp->used (v6.12.96 usblp.c), so every later open in this long-lived
+ worker would read as a wedged device. A killed process takes its fd with it.
+
+ stdout is captured as BYTES and kept CLEAN -- a caller byte-compares it against the
+ payload it sent, so a single stderr byte (a PYTHONWARNINGS chirp, a sitecustomize
+ print, a .pth deprecation from a venv) would read as USB data corruption. stderr gets
+ its own pipe; communicate() drains both, so the split cannot deadlock.
+ `work` runs even if the child dies immediately -- the caller's own asserts decide.
+ """
+ p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ start_new_session=True)
+
+ def _reap() -> subprocess.CompletedProcess:
+ try:
+ out, err = p.communicate(timeout=timeout)
+ return subprocess.CompletedProcess(argv, p.returncode, out, err)
+ except subprocess.TimeoutExpired:
+ try:
+ os.killpg(p.pid, signal.SIGKILL)
+ except OSError:
+ p.kill()
+ try:
+ out, err = p.communicate(timeout=REAP_GRACE)
+ except subprocess.TimeoutExpired:
+ # Outlasted SIGKILL: uninterruptible, still holding whatever it opened.
+ # Abandoned like any other stray -- but as a real child in its own
+ # session, so the containment sweep FINDS it (child_procs walks the ppid
+ # tree) and the report names it. That is the whole difference from a
+ # blocked thread, which no sweep can see and no signal can reach.
+ out, err = b'', b''
+ _close_pipes(p) # our own fds must not leak either
+ return subprocess.CompletedProcess(argv, 124, out, err)
+
+ try:
+ work()
+ except BaseException:
+ # Reap first so the child never outlives us, then let the caller's error through.
+ # A `return` inside a `finally` would SWALLOW it -- an assert in `work` would
+ # vanish and the caller would compare data it never finished sending.
+ _reap()
+ raise
+ return _reap()
+
+
+# The RTT console implementation lives in tools/rtt.py (importable classes + CLI,
+# stdlib-only, harness-critical — see its module docstring). Loaded by file path so
+# no sys.path entry for tools/ can shadow other imports; re-exported here so the
+# harness keeps addressing hil_util.JlinkRtt.
+import importlib.util as _ilu
+
+_rtt_path = TINYUSB_ROOT / 'tools' / 'rtt.py'
+if not _rtt_path.exists():
+ # name the real cause: a bare FileNotFoundError out of an exec_module here reads
+ # as a harness bug, when the actual problem is an incompletely staged tree
+ raise ImportError(f'{_rtt_path} is missing — the RTT console lives there and the '
+ f'harness depends on it; stage it alongside test/hil (hil_ci.sh does)')
+_rtt_spec = _ilu.spec_from_file_location('tinyusb_tools_rtt', _rtt_path)
+_rtt = _ilu.module_from_spec(_rtt_spec)
+sys.modules[_rtt_spec.name] = _rtt # registered: RttError must be picklable across the fork Pool
+_rtt_spec.loader.exec_module(_rtt)
+JlinkRtt = _rtt.JlinkRtt
+OpenocdRtt = _rtt.OpenocdRtt
+RttError = _rtt.RttError
+RTT_BANNER_RE = _rtt.RTT_BANNER_RE
+strip_banner = _rtt.strip_banner
+
+
+def _cmd_label(cmd) -> str:
+ """A one-line name for a banner. An argv whose payload is a `python3 -c` program would
+ otherwise dump the whole body into the CI log, where run_cmd's banners are already the
+ noisiest thing in a failing row."""
+ if isinstance(cmd, str):
+ return cmd
+ parts = [a if len(a) <= 60 else f'<{len(a)}-char program>' for a in cmd]
+ return ' '.join(parts)
+
+
+def run_cmd(cmd: str | list, cwd: str | None = None, timeout: int | None = None,
+ binary: bool = False, split_stderr: bool = False,
+ quiet: bool = False) -> subprocess.CompletedProcess:
+ """Bounded subprocess: own session, killpg on expiry, rc 124 when it had to be killed.
+
+ `cmd` is a shell STRING or an argv LIST. argv exists for a program that cannot survive
+ a trip through the shell -- a multi-line `python3 -c` body -- which is how the harness
+ runs a library call that no in-process bound can contain. A daemon thread cannot bound
+ a C call that holds the GIL, so for those the child process IS the bound.
+ """
+ if timeout is None:
+ timeout = CMD_TIMEOUT
+ # binary: raw bytes (text mode's errors='replace' mangles non-UTF-8 file content).
+ # split_stderr: keep stderr out of stdout, for callers that parse stdout. quiet: no
+ # COMMAND FAILED banner, for retry loops that report failures themselves (timeouts
+ # still print: a killed child is always noteworthy).
+ popen_kwargs = {
+ 'cwd': cwd,
+ # a list goes straight to execve; only a string needs a shell to parse it
+ 'shell': isinstance(cmd, str),
+ 'stdout': subprocess.PIPE,
+ 'stderr': subprocess.PIPE if split_stderr else subprocess.STDOUT,
+ }
+ if not binary:
+ popen_kwargs.update({'text': True, 'encoding': 'utf-8', 'errors': 'replace'})
+ # C-level setsid, same process-group semantics as preexec_fn=os.setsid but safe when
+ # called from threads (pool_check runs flashes from a thread pool)
+ popen_kwargs['start_new_session'] = True
+
+ p = subprocess.Popen(cmd, **popen_kwargs)
+ try:
+ out, err = p.communicate(timeout=timeout)
+ r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out, stderr=err)
+ except subprocess.TimeoutExpired as ex:
+ try:
+ os.killpg(p.pid, signal.SIGKILL)
+ except OSError:
+ # ProcessLookupError: already gone. PermissionError: an all-root group refuses
+ # the group kill -- letting either escape would skip the bounded reap, the pipe
+ # close and the rc-124 return this handler exists for.
+ pass
+ try:
+ out, err = p.communicate(timeout=REAP_GRACE)
+ except subprocess.TimeoutExpired:
+ # Something in the group outlived SIGKILL: D state (truly unkillable), or
+ # root-owned because sudo FORKS rather than execs, so the wrapper dies and its
+ # root child does not. Abandon it and let the report name it; the harness never
+ # sudo-kills its way out. Our ends of its pipes must not leak, though: a pool
+ # worker lives for the whole run, so every wedged command would cost it two fds.
+ out, err = None, None
+ _close_pipes(p)
+ # prefer the post-kill buffers (supersets of the exception's), falling back to ex.*
+ # when the child was unkillable. TimeoutExpired carries BYTES even for a text-mode
+ # Popen, so the fallbacks must be decoded or a text-mode caller gets bytes exactly
+ # when the child wedged in D state.
+ def _typed(v):
+ if not binary and isinstance(v, bytes):
+ return v.decode('utf-8', errors='replace')
+ return v
+
+ timeout_out = _typed(out or ex.stdout) or (b'' if binary else '')
+ # ...and never None: with split_stderr the SUCCESS path always yields a str/bytes,
+ # so a caller that does `r.stderr.strip()` works everywhere except the timeout --
+ # the one path it was written for. Without split_stderr stderr stays None, as on
+ # the success path (it was merged into stdout).
+ timeout_err = _typed(err if err is not None else ex.stderr)
+ if split_stderr and timeout_err is None:
+ timeout_err = b'' if binary else ''
+ _print_banner(f'COMMAND TIMEOUT ({timeout}s): {_cmd_label(cmd)}', timeout_out, timeout_err)
+ return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out, stderr=timeout_err)
+ except BaseException:
+ # BaseException, not Exception (as in CPython's own subprocess.run):
+ # KeyboardInterrupt is the case that matters, and start_new_session put the child in
+ # its OWN group, so it never got the terminal's SIGINT -- without this, Ctrl-C
+ # leaves the flasher or testusb holding the probe and its usbfs node. Kill and
+ # close, never wait: this path must not add a hang of its own.
+ try:
+ os.killpg(p.pid, signal.SIGKILL)
+ except OSError:
+ pass
+ _close_pipes(p)
+ raise
+
+ if r.returncode != 0 and not quiet:
+ _print_banner(f'COMMAND FAILED: {_cmd_label(cmd)}', r.stdout, r.stderr)
+ elif verbose:
+ print(cmd)
+ print(cmd_stdout_text(r.stdout))
+ return r
+
+
+# get usb serial by id
+def get_serial_dev(id, vendor_str, product_str, ifnum):
+ if vendor_str and product_str:
+ # known vendor and product
+ vendor_str = vendor_str.replace(' ', '_')
+ product_str = product_str.replace(' ', '_')
+ return f'/dev/serial/by-id/usb-{vendor_str}_{product_str}_{id}-if{ifnum:02d}'
+ else:
+ # just use id: mostly for cp210x/ftdi flasher
+ pattern = f'/dev/serial/by-id/usb-*_{id}-if*'
+ port_list = glob.glob(pattern)
+ if len(port_list) == 0:
+ raise RuntimeError(f'No serial device found for {pattern}')
+ return port_list[0]
generated by cgit v1.3.1 (git 2.54.0) at 2026-09-02 18:38:59 +0000