hil: split hil_test.py into hil_lock/hil_flash, add pool_check, update rig probes (#3794)
test/hil: add board-pool health check, split hil_test into focused modules (#3794)
Add test/hil/hil_pool_check.py: per-board rig health scan — probe presence,
light-example flash (dfu_runtime; device_info + serial check for host-only
boards), uid re-enumeration, safe recovery (probe authorized-toggle, board
reset), verified board_test re-park, USB topology report, and a markdown
summary table. Missing firmware is built on the spot (tools/build.py, idf.py
for espressif, one get_deps retry); row statuses: ok, flash-failed, failed,
locked. Board locks are always respected, never bypassed.
Refactor hil_test.py into hil_lock.py (flock protocol, controller permits,
hold/release/status CLI; replaces board_lock.py) and hil_flash.py (flashers,
find_firmware, run_cmd). Update WCH probe uids and the board roster in
tinyusb.json; add the hil-pool-check skill.
diff --git a/.claude/agents/hil-operator.md b/.claude/agents/hil-operator.md
index d19eca0..6f04f6d 100644
--- a/.claude/agents/hil-operator.md
+++ b/.claude/agents/hil-operator.md
@@ -7,7 +7,7 @@
You operate physical USB test hardware. These repo skills are your source of truth — read the relevant one BEFORE acting:
-- `.claude/skills/hil/SKILL.md` — run `hostname` first (host `ci` = local mode with `test/hil/tinyusb.json`; host `htpc` = local `local.json` or remote via `test/hil/hil_ci.sh`); the board lock protocol; exact `hil_test.py` invocations.
+- `.claude/skills/hil/SKILL.md` — run `hostname` first (host `ci` = local mode with `test/hil/tinyusb.json`; host `tusb` = local mode with `test/hil/hfp.json`; any other host (dev PC) = local `local.json` or remote via `test/hil/hil_ci.sh`); the board lock protocol; exact `hil_test.py` invocations.
- `.claude/skills/usb-kernel-recover/SKILL.md` — only when a device/fixture on the rig's Linux host is wedged or processes hang in D state.
- `.claude/skills/usb-kernel-debug/SKILL.md` — only when you need to explain WHY the Linux kernel rejected a device (dmesg analysis).
@@ -18,12 +18,12 @@
- `python3 test/hil/hil_test.py ...` runs: do NOT pre-hold those boards — `hil_test.py` self-locks each board for its flash+test and would fail fast with `board locked` against your own hold.
- ANY other hardware action (JLinkExe/openocd/GDB, manual flash, usbtest.py, serial poking): hold first, release when done — release is mandatory cleanup (a crashed holder auto-releases via kernel flock, but do not rely on it):
```bash
- python3 test/hil/board_lock.py hold <board...> --reason "<task>"
+ python3 test/hil/hil_lock.py hold <board...> --reason "<task>"
# ... hardware work ...
- python3 test/hil/board_lock.py release <board...>
+ python3 test/hil/hil_lock.py release <board...>
```
-- Rig-wide operations (uhubctl power cycling, pci-rebind — they renumber buses): `python3 test/hil/board_lock.py hold --all --reason "<why>"` first.
-- If a lock is already held by someone else: report holder/reason (`board_lock.py status`) — never force, never kill the holder. If the holder's reason is `hil_test.py`, that is a concurrent CI job mid-test on the board: waiting a few minutes and retrying once is appropriate when your task allows; otherwise return the holder info so the orchestrator can ask the user.
+- Rig-wide operations (uhubctl power cycling, pci-rebind — they renumber buses): `python3 test/hil/hil_lock.py hold --all --reason "<why>"` first.
+- If a lock is already held by someone else: report holder/reason (`hil_lock.py status`) — never force, never kill the holder. If the holder's reason is `hil_test.py`, that is a concurrent CI job mid-test on the board: waiting a few minutes and retrying once is appropriate when your task allows; otherwise return the holder info so the orchestrator can ask the user.
- You cannot ask the user anything. Bypassing a lock (`HIL_NO_BOARD_LOCK=1`, or proceeding with manual hardware work despite a held lock) is allowed ONLY when your prompt explicitly states the user authorized forcing.
## Hard rules
diff --git a/.claude/agents/target-debugger.md b/.claude/agents/target-debugger.md
index 246ad16..655b5f5 100644
--- a/.claude/agents/target-debugger.md
+++ b/.claude/agents/target-debugger.md
@@ -46,7 +46,7 @@
## Lock discipline
-- Hold the board lock for the WHOLE session (`board_lock.py hold <board>
+- Hold the board lock for the WHOLE session (`hil_lock.py hold <board>
--reason "target debug: <bug>"`). Multi-hour holds are fine; never stop the
actions-runner. Locks held by others: report holder/reason, never force
unless your prompt states the user authorized it.
diff --git a/.claude/skills/etm-trace/SKILL.md b/.claude/skills/etm-trace/SKILL.md
index a200b14..9e25057 100644
--- a/.claude/skills/etm-trace/SKILL.md
+++ b/.claude/skills/etm-trace/SKILL.md
@@ -43,7 +43,7 @@
capture script uses automation port **19201**, never an interactive Ozone's
19200.
- Hold the board lock (see the `hil` skill):
- `python3 test/hil/board_lock.py hold <board> --reason "etm capture"`.
+ `python3 test/hil/hil_lock.py hold <board> --reason "etm capture"`.
- Committed `hw/bsp/**/ozone/*.jdebug` are the maintainer's interactive
projects — automation never opens them (Ozone rewrites project files); the
script generates a throwaway project.
diff --git a/.claude/skills/hil-pool-check/SKILL.md b/.claude/skills/hil-pool-check/SKILL.md
new file mode 100644
index 0000000..49d252f
--- /dev/null
+++ b/.claude/skills/hil-pool-check/SKILL.md
@@ -0,0 +1,63 @@
+---
+name: hil-pool-check
+description: Use when asked for a pool check or board/probe health scan on a TinyUSB HIL rig, when probes or boards are offline or fail to flash, after rig maintenance, reboot, or re-cabling, or before starting a HIL test campaign.
+---
+
+# HIL Pool Check (board/probe health)
+
+Health-scan the HIL board pool with `test/hil/hil_pool_check.py`: per board it checks the flash
+probe is on the USB bus, flashes a light example (`device/dfu_runtime`; host-only boards get
+`host/device_info`, verified by serial output), waits for the board's uid to re-enumerate,
+applies safe per-device recovery (probe authorized-toggle, board reset), re-parks with
+`board_test`, and prints a summary table plus a USB topology report. Flags and details: `--help`
+and the module docstring.
+
+**REQUIRED BACKGROUND:** the `hil` skill owns config-by-hostname selection (run `hostname`
+first) and the board-lock protocol. Locked boards are reported 🔒 locked and skipped — never
+waited on, never bypassed; a needed build peeks the lock first. A CI worker reaching a board the
+pool check holds fails it as "board locked" — prefer running between CI runs.
+
+## A "pool check" means the full check
+
+A request for a "pool check" means the DEFAULT full check below. Use `--scan-only` only when the
+user explicitly asks for a quick look, or when you have VERIFIED a CI sweep is mid-run right now
+(`python3 test/hil/hil_lock.py status` shows `hil_test.py` holders) — "CI might be running" is not
+that predicate: the full check is already lock-safe (CI-held boards report 🔒 locked and are never
+touched), so an unconfirmed suspicion is no reason to downgrade. In either scan case say which
+mode ran and why; never silently substitute the scan for the full check.
+
+```bash
+python3 test/hil/hil_pool_check.py # full check: ~10 s + ~1-2 s/board with firmware built;
+ # first run on an unbuilt tree takes minutes (it builds)
+python3 test/hil/hil_pool_check.py --scan-only # USB presence only, <1 s, no locks/flashing/building
+python3 test/hil/hil_pool_check.py -b BOARD [-b …] # subset; may name boards-skip (parked) entries
+
+# from a dev PC, against the ci rig (bash -lc: flashers like STM32_Programmer_CLI live in ~/bin):
+ssh ci.lan 'bash -lc "cd ~/code/tinyusb && python3 test/hil/hil_pool_check.py"'
+```
+
+## Notes
+
+Missing firmware is **built on the spot** — never skipped (`--no-build` opts out; those boards
+then report `flash-failed`). Builds need the family env, exported on the rig in
+`~/.profile`/`~/.bashrc`: `PICO_SDK_PATH` for rp2040/rp2350 (`~/code/pico/pico-sdk`), the
+ESP-IDF env (`get-idf`) for espressif — which also needs `esptool` on PATH (pip's
+`~/.local/bin/esptool`; a non-login shell may lack it — run via `bash -lc`). An explicit `-B` is
+searched exclusively for *existing* firmware; builds still land in `cmake-build/` and are noted
+`built <example>`. Espressif boards park too when the IDF env is present. A first run on an
+unbuilt tree builds for many minutes: use a command timeout ≥ 30 min and NEVER cancel early — a
+killed run leaves detached cmake/ninja children still writing to `cmake-build/`.
+
+Statuses: `ok` (flashed and verified; in `--scan-only` it only means the probe is present),
+`flash-failed` (firmware delivery failed: probe missing, build failed, flasher error, silent
+no-op, park unverified), `failed` (check ran but did not verify), `locked` (flock held;
+untouched). Exit code = `flash-failed` + `failed` (clamped at 125); `locked` and scan-only rows
+are *unverified*, not healthy — read the footer, not just `$?`. A `⚠ pid … source says …` note
+means stale firmware or a silent flash no-op (J-Link lore); a device off the bus entirely needs
+the usb-kernel-recover skill or a physical replug.
+
+## Reporting
+
+The user-facing answer to a pool check IS the tool's summary table: paste the complete per-board
+table (and footer counts) verbatim — never truncate rows or reduce it to a prose digest like
+"27/27 healthy"; at most one line of commentary below it.
diff --git a/.claude/skills/hil/SKILL.md b/.claude/skills/hil/SKILL.md
index 2aeaa10..6c4d3a8 100644
--- a/.claude/skills/hil/SKILL.md
+++ b/.claude/skills/hil/SKILL.md
@@ -1,22 +1,22 @@
---
name: hil
-description: Use when running TinyUSB Hardware-in-the-Loop (HIL) tests on physical boards, debugging HIL failures, or copying firmware to the ci.lan test rig. Covers per-host config selection (htpc uses local.json, ci uses tinyusb.json), local execution on either htpc or ci, remote execution over SSH from htpc, and debugging tips.
+description: Use when running TinyUSB Hardware-in-the-Loop (HIL) tests on physical boards, debugging HIL failures, or copying firmware to the ci.lan test rig. Covers per-host config selection (infra rigs ci/tusb use tinyusb.json/hfp.json, any dev PC uses local.json), local and remote execution, the board-lock protocol, and debugging tips. For board/probe health scans ("pool check") use the hil-pool-check skill.
---
# Hardware-in-the-Loop (HIL) Testing
-Run TinyUSB HIL tests on real boards. **Run `hostname` first** — it tells you which host you are on, which determines the default config and whether remote mode is possible.
+Run TinyUSB HIL tests on real boards. **Run `hostname` first** — it tells you which host you are on, which determines the default config and whether remote mode is possible. Rule of thumb: only `ci` and `tusb` are infra rigs; **any other hostname is a dev PC** and uses `local.json`.
-| Host | Local config | Remote (SSH → ci.lan)? |
-|----------------------------|--------------------------------------|------------------------------------------------------|
-| `htpc` (dev PC) | `test/hil/local.json` | yes (large pool, `test/hil/tinyusb.json`) |
-| `ci` (the rig) | `test/hil/tinyusb.json` (large pool) | no — can't SSH to htpc, and boards are already local |
-| `hifiphile` (external rig) | `test/hil/hfp.json` | no outbound SSH to htpc/ci; SSH-reachable FROM both |
+| Host | Local config | Remote (SSH → ci.lan)? |
+|-----------------------------------|--------------------------------------|--------------------------------------------------------|
+| `ci` (the rig) | `test/hil/tinyusb.json` (large pool) | no — boards are already local |
+| `tusb` (hifiphile's external rig) | `test/hil/hfp.json` | no outbound SSH to dev PCs/ci; SSH-reachable FROM both |
+| anything else (a dev PC) | `test/hil/local.json` | yes (large pool, `test/hil/tinyusb.json`) |
-Default to **local**. Use **remote** only when on `htpc` and the user says `remote`/`ci.lan`. Never attempt remote on `ci`.
+Default to **local**. Use **remote** only when on a dev PC and the user says `remote`/`ci.lan`. Never attempt remote on `ci`.
-`hifiphile` is an external rig (hosted by maintainer hifiphile), exercised by the GitHub CI
-`hil-tinyusb (hfp.json)` matrix job — **never run HIL against it unless the user explicitly asks.**
+`tusb` (ssh alias `hifiphile`) is an external rig (hosted by maintainer hifiphile), exercised by the
+GitHub CI `hil-tinyusb (hfp.json)` matrix job — **never run HIL against it unless the user explicitly asks.**
## Board locks — the CI runner keeps running
@@ -26,30 +26,35 @@
- For hardware work outside `hil_test.py` (JLink/GDB, manual flashing, `usbtest.py`, serial poking), hold the lock first:
```bash
-python3 test/hil/board_lock.py hold BOARD [BOARD...] --reason "why"
+python3 test/hil/hil_lock.py hold BOARD [BOARD...] --reason "why"
# ... hardware work ...
-python3 test/hil/board_lock.py release BOARD [BOARD...]
+python3 test/hil/hil_lock.py release BOARD [BOARD...]
```
- Never pre-hold boards you are about to run `hil_test.py` on — it self-locks and would treat your own hold as a conflict.
-- Rig-wide operations (uhubctl power cycling, pci-rebind — bus renumbering) affect every board: `board_lock.py hold --all --reason "..."` first.
-- `board_lock.py status` lists holders. Locks auto-release when the holder process dies (kernel flock); `/tmp` clears on reboot.
+- Rig-wide operations (uhubctl power cycling, pci-rebind — bus renumbering) affect every board: `hil_lock.py hold --all --reason "..."` first.
+- `hil_lock.py status` lists holders. Locks auto-release when the holder process dies (kernel flock); `/tmp` clears on reboot.
- Forcing past a lock: `HIL_NO_BOARD_LOCK=1 python3 test/hil/hil_test.py ...` bypasses the guard without killing the holder. Only with the user's explicit go-ahead — they accept the risk of colliding with whatever holds the board.
+## Pool check (board/probe health)
+
+Board/probe health scanning (`test/hil/hil_pool_check.py`) has its own skill: **hil-pool-check**.
+Use it before a HIL campaign, after rig maintenance/reboot, or when boards fail to flash.
+
## Prerequisites
-Examples must be built for the target board(s) — see CLAUDE.md "Build" → "All examples for a board" (produces `examples/cmake-build-<board>/`). `-B examples` points `hil_test.py` at that parent folder.
+Examples must be built for the target board(s) — see CLAUDE.md "Build" → "All examples for a board" (produces `examples/cmake-build-<board>/`). `-B examples` points `hil_test.py` at that parent folder. (This applies to `hil_test.py`; `hil_pool_check.py` builds its own missing firmware.)
## Arguments
- **Board:** `-b BOARD_NAME` for one board; omit to run all boards in the config.
- **Pass-through:** `-v`, `-r N`, etc. forwarded unchanged.
-If `local.json` is missing on `htpc`, ask the user to supply one (only fall back to `tinyusb.json` if told to).
+If `local.json` is missing on a dev PC, ask the user to supply one (only fall back to `tinyusb.json` if told to).
## Local execution
-Set `CONFIG` from `hostname` first (`test/hil/local.json` on htpc, `test/hil/tinyusb.json` on ci):
+Set `CONFIG` from `hostname` first (`test/hil/local.json` on a dev PC, `test/hil/tinyusb.json` on ci, `test/hil/hfp.json` on tusb):
```bash
CONFIG=test/hil/local.json # on ci use: CONFIG=test/hil/tinyusb.json
@@ -61,7 +66,7 @@
python3 test/hil/hil_test.py -b stm32f723disco -B examples "$CONFIG"
```
-## Remote execution (htpc → ci.lan only)
+## Remote execution (dev PC → ci.lan only)
`test/hil/hil_ci.sh` handles dir setup, scp of test scripts, rsync of firmware (`.elf`/`.bin`/`.hex`), and runs `hil_test.py` on `ci.lan` with `tinyusb.json`:
@@ -81,4 +86,7 @@
## Reporting
-Show the output, summarize pass/fail per board. On failure, retry with `-v`; if that's not enough, add temporary debug prints to `hil_test.py`.
+The user-facing answer to a HIL run IS the tool's summary table: paste the complete per-board
+table (and footer counts) verbatim — never truncate rows or reduce it to a prose digest; at most
+one line of commentary below it. On failure, retry with `-v`; if that's not enough, add temporary
+debug prints to `hil_test.py`.
diff --git a/.claude/skills/pre-pr/SKILL.md b/.claude/skills/pre-pr/SKILL.md
index 3f062db..4283834 100644
--- a/.claude/skills/pre-pr/SKILL.md
+++ b/.claude/skills/pre-pr/SKILL.md
@@ -24,7 +24,7 @@
## 3. HIL boards
-- `hilBoards` = chosen boards that are on the rig roster. This host must be able to reach the rig (per `.claude/skills/hil/SKILL.md`: host `ci` = local, `htpc` = remote). If none qualify, run software-only.
+- `hilBoards` = chosen boards that are on the rig roster. This host must be able to reach the rig (per `.claude/skills/hil/SKILL.md`: host `ci`/`tusb` = local, any other host (dev PC) = remote). If none qualify, run software-only.
## 4. Launch
diff --git a/.claude/skills/target-debug/SKILL.md b/.claude/skills/target-debug/SKILL.md
index 5165a76..28678c3 100644
--- a/.claude/skills/target-debug/SKILL.md
+++ b/.claude/skills/target-debug/SKILL.md
@@ -30,9 +30,9 @@
actions-runner (see the `hil` skill for the full lock protocol):
```bash
-python3 test/hil/board_lock.py hold <board> --reason "target debug: <bug>"
+python3 test/hil/hil_lock.py hold <board> --reason "target debug: <bug>"
# ... instrument / build / flash / capture / GDB ...
-python3 test/hil/board_lock.py release <board>
+python3 test/hil/hil_lock.py release <board>
```
Board → probe mapping: `test/hil/tinyusb.json` — `flasher.name` is the probe
@@ -44,7 +44,7 @@
`hw/bsp/<family>/boards/<board>/board.cmake` (or `board.mk`); family via
`ls -d hw/bsp/*/boards/<board>`.
- Run on the host that owns the probe — config `test/hil/tinyusb.json` on ci,
- `local.json` on htpc (`hil` skill).
+ `test/hil/hfp.json` on tusb, `local.json` on any other host (dev PC) (`hil` skill).
- Espressif boards (S3/P4): different toolchain, probe model, and PHY
constraints entirely — read `esp-target-debug` first.
diff --git a/.claude/skills/usb-kernel-recover/SKILL.md b/.claude/skills/usb-kernel-recover/SKILL.md
index 9e45641..ea5931c 100644
--- a/.claude/skills/usb-kernel-recover/SKILL.md
+++ b/.claude/skills/usb-kernel-recover/SKILL.md
@@ -57,7 +57,7 @@
It bounces **every fixture under that root port** — on ci that is up to 25
devices. Hold the affected boards' locks first if you can, but note
-`board_lock.py` uses `LOCK_EX | LOCK_NB` and so fails immediately when CI already
+`hil_lock.py` uses `LOCK_EX | LOCK_NB` and so fails immediately when CI already
holds them; there is no wait-for-lock. When CI is mid-run you are choosing
between bouncing its fixtures and leaving the bus wedged for everything. The
automated path in `usbtest.py` takes no locks at all and accepts that collateral
diff --git a/.idea/codeStyles/Project.xml b/.idea/codeStyles/Project.xml
new file mode 100644
index 0000000..35c56fc
--- /dev/null
+++ b/.idea/codeStyles/Project.xml
@@ -0,0 +1,10 @@
+<component name="ProjectCodeStyleConfiguration">
+ <code_scheme name="Project" version="173">
+ <RiderCodeStyleSettings>
+ <option name="/Default/CodeStyle/CodeFormatting/CppClangFormat/EnableClangFormatSupport/@EntryValue" value="true" type="bool" />
+ </RiderCodeStyleSettings>
+ <clangFormatSettings>
+ <option name="ENABLED" value="true" />
+ </clangFormatSettings>
+ </code_scheme>
+</component>
\ No newline at end of file
diff --git a/.idea/codeStyles/codeStyleConfig.xml b/.idea/codeStyles/codeStyleConfig.xml
new file mode 100644
index 0000000..79ee123
--- /dev/null
+++ b/.idea/codeStyles/codeStyleConfig.xml
@@ -0,0 +1,5 @@
+<component name="ProjectCodeStyleConfiguration">
+ <state>
+ <option name="USE_PER_PROJECT_SETTINGS" value="true" />
+ </state>
+</component>
\ No newline at end of file
diff --git a/.idea/hil-pool-check.iml b/.idea/hil-pool-check.iml
new file mode 100644
index 0000000..4c94235
--- /dev/null
+++ b/.idea/hil-pool-check.iml
@@ -0,0 +1,2 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module classpath="CIDR" type="CPP_MODULE" version="4" />
\ No newline at end of file
diff --git a/.idea/modules.xml b/.idea/modules.xml
new file mode 100644
index 0000000..063c94f
--- /dev/null
+++ b/.idea/modules.xml
@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+ <component name="ProjectModuleManager">
+ <modules>
+ <module fileurl="file://$PROJECT_DIR$/.idea/hil-pool-check.iml" filepath="$PROJECT_DIR$/.idea/hil-pool-check.iml" />
+ </modules>
+ </component>
+</project>
\ No newline at end of file
diff --git a/docs/superpowers/plans/2026-07-28-hil-test-split.md b/docs/superpowers/plans/2026-07-28-hil-test-split.md
new file mode 100644
index 0000000..6b85289
--- /dev/null
+++ b/docs/superpowers/plans/2026-07-28-hil-test-split.md
@@ -0,0 +1,355 @@
+# hil_test.py Split Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Split `test/hil/hil_test.py` (2370 ln) into a test-focused core plus `hil_lock.py` (board locks + controller permits + operator CLI, superseding `board_lock.py`) and `hil_flash.py` (run_cmd + flash backends + firmware/serial lookup), with no behavior change.
+
+**Architecture:** Pure code motion per `docs/superpowers/specs/2026-07-28-hil-test-refactor-design.md`. Import graph: `hil_test` → {`hil_lock`, `hil_flash`}; helpers import nothing local. Call sites use module-qualified names (`hil_lock.flash_permit(...)`), never wildcard mirroring.
+
+**Tech Stack:** Python 3.11+ (existing `TypedDict`/`NotRequired` usage), stdlib only in the helpers (fcntl, json, glob, multiprocessing objects passed in).
+
+## Global Constraints
+
+- Work in worktree `.claude/worktrees/hil-test-split` (branch `claude/hil-test-split`); never touch the primary checkout.
+- Behavior-preserving: `hil_test.py` CLI args, log lines, report format, lock/permit semantics, flash behavior all byte-identical. The ONLY user-visible change is the CLI filename `board_lock.py` → `hil_lock.py`.
+- Moved functions are moved **verbatim** — no reformatting, no comment editing, no "improvements". A diff of a moved function's body against its old self must be empty.
+- Commit messages: imperative, scoped, no Co-Authored-By/Claude-Session trailers.
+- Every commit leaves the tree working: `python3 -m py_compile` clean on all touched modules, and `python3 .claude/skills/hil/pool_check.py --scan-only` exits 0 (safe on the rig: scan-only takes no locks, flashes nothing).
+- Hardware steps (Task 4) run on the `ci` rig only, from this worktree, and rely on the tools' own board flocks — never pre-hold boards you are about to run `hil_test.py`/`pool_check.py` on.
+
+---
+
+### Task 1: Create hil_flash.py; repoint hil_test + pool_check flash call sites
+
+**Files:**
+- Create: `test/hil/hil_flash.py`
+- Modify: `test/hil/hil_test.py` (delete moved code; add import; qualify call sites)
+- Modify: `.claude/skills/hil/pool_check.py` (flash-related imports)
+- Modify: `test/hil/hil_ci.sh` (scp list)
+
+**Interfaces:**
+- Produces (used by Tasks 2-4): module `hil_flash` with `CMD_TIMEOUT`, `run_cmd(cmd, cwd=None, timeout=CMD_TIMEOUT)`, `cmd_stdout_text(out)`, `OPENCOD_ADI_PATH`, `TINYUSB_ROOT`, `flash_jlink/reset_jlink`, `flash_stlink/reset_stlink`, `flash_stflash/reset_stflash`, `flash_openocd/reset_openocd`, `flash_openocd_wch/reset_openocd_wch`, `flash_openocd_adi/reset_openocd_adi`, `flash_wlink_rs/reset_wlink_rs`, `flash_esptool/reset_esptool`, `flash_uniflash/reset_uniflash`, `flash_lm4flash/reset_lm4flash`, `find_firmware(variant, example)`, `get_serial_dev(id, vendor_str, product_str, ifnum)`, module globals `build_dir = 'cmake-build'`, `verbose = False`.
+
+- [ ] **Step 1: Create `test/hil/hil_flash.py`**
+
+Header (new code), then the moved blocks verbatim:
+
+```python
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+# Firmware flashing for the TinyUSB HIL rig: run_cmd, one flash_*/reset_* pair per
+# flasher type (dispatched by config name via getattr), find_firmware, and the
+# fixture serial-port resolver get_serial_dev (here, not hil_test: flash_esptool
+# needs it and helpers must not import hil_test).
+# Callers set module globals `build_dir` and `verbose` (hil_test.main from argparse,
+# pool_check directly) exactly as they set hil_test's globals today.
+
+import glob
+import json
+import os
+import signal
+import subprocess
+import sys
+from pathlib import Path
+
+verbose = False
+build_dir = 'cmake-build'
+```
+
+Then MOVE (cut from `hil_test.py`, paste unchanged, in this order):
+1. `CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))` (from the constants block; leave `POOL_TIMEOUT`/`SERIAL_*_TIMEOUT` in hil_test)
+2. `def cmd_stdout_text(out)`
+3. `OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi'` and `TINYUSB_ROOT = Path(__file__).resolve().parents[2]`
+4. `def get_serial_dev(id, vendor_str, product_str, ifnum)`
+5. `def run_cmd(cmd, cwd=None, timeout=CMD_TIMEOUT)`
+6. All ten `flash_*`/`reset_*` pairs listed in Interfaces, in current file order
+7. `def find_firmware(variant, example)`
+
+- [ ] **Step 2: Delete the moved code from `hil_test.py` and qualify call sites**
+
+In `hil_test.py`: add `import hil_flash` under the existing imports; delete the moved definitions and the `build_dir = 'cmake-build'` global (line ~165) plus `global build_dir` in `main`. Repoint every use, all module-qualified:
+- `globals()[f'flash_{...}']` → `getattr(hil_flash, f'flash_{...}')` (1 site, in `test_example`)
+- `globals()[f'reset_{...}']` → `getattr(hil_flash, f'reset_{...}')` (3 sites: `test_host_device_info`, `test_host_cdc_msc_hid`, `test_host_msc_file_explorer`)
+- bare `run_cmd(` → `hil_flash.run_cmd(` ; `cmd_stdout_text(` → `hil_flash.cmd_stdout_text(` ; `find_firmware(` → `hil_flash.find_firmware(` ; `get_serial_dev(` → `hil_flash.get_serial_dev(` ; `TINYUSB_ROOT` → `hil_flash.TINYUSB_ROOT` (in `build_board`, `CONTROLLER_CACHE` stays hil_test-local)
+- In `main()`: `build_dir = args.build_dir` → `hil_flash.build_dir = args.build_dir`; where `verbose` is set, add `hil_flash.verbose = args.verbose` (hil_test keeps its own `verbose` for test-side prints)
+- `run_cmd`'s `elif verbose:` branch now reads `hil_flash.verbose` (it moved with the function — verify it references the module-local name, not hil_test's)
+
+Find every remaining call site mechanically:
+
+Run: `grep -nE 'run_cmd|cmd_stdout_text|find_firmware|get_serial_dev|flash_[a-z]|reset_[a-z]|TINYUSB_ROOT|OPENCOD' test/hil/hil_test.py | grep -v hil_flash`
+Expected: only hits inside comments/strings and the `reset_{flasher}` dispatch f-strings already qualified.
+
+- [ ] **Step 3: Repoint pool_check's flash imports**
+
+In `.claude/skills/hil/pool_check.py`: add `import hil_flash` next to `import hil_test`; replace `hil_test.find_firmware` → `hil_flash.find_firmware` (3 sites), `hil_test.cmd_stdout_text` → `hil_flash.cmd_stdout_text`, `hil_test.get_serial_dev` → `hil_flash.get_serial_dev`, `hil_test.TINYUSB_ROOT` → `hil_flash.TINYUSB_ROOT`, `hil_test.build_dir` → `hil_flash.build_dir` (2 sites incl. `main`'s assignment), `hil_test.verbose = args.verbose` → `hil_flash.verbose = args.verbose`, `getattr(hil_test, f'flash_...')`/`getattr(hil_test, f'reset_...')` → `getattr(hil_flash, ...)` (4 sites). Keep `import hil_test` and the pymtp shim for now (locks still live there; removed in Task 2).
+
+- [ ] **Step 4: Add hil_flash.py to the hil_ci.sh scp list**
+
+```bash
+scp -q "$ROOT_DIR/test/hil/hil_test.py" \
+ "$ROOT_DIR/test/hil/hil_flash.py" \
+ "$ROOT_DIR/test/hil/pymtp.py" \
+ "$CONFIG" \
+ "$REMOTE:$REMOTE_DIR/test/hil/"
+```
+
+- [ ] **Step 5: Verify**
+
+Run: `python3 -m py_compile test/hil/hil_flash.py test/hil/hil_test.py .claude/skills/hil/pool_check.py && python3 test/hil/hil_test.py --help >/dev/null && python3 .claude/skills/hil/pool_check.py --scan-only`
+Expected: compiles; help prints nothing to stderr; scan-only prints the table and exits 0.
+
+- [ ] **Step 6: Commit**
+
+```bash
+git add test/hil/hil_flash.py test/hil/hil_test.py test/hil/hil_ci.sh .claude/skills/hil/pool_check.py
+git commit -m "hil: extract flashing into hil_flash.py"
+```
+
+---
+
+### Task 2: Create hil_lock.py core (flock protocol + controller permits); repoint hil_test + pool_check
+
+**Files:**
+- Create: `test/hil/hil_lock.py`
+- Modify: `test/hil/hil_test.py`
+- Modify: `.claude/skills/hil/pool_check.py`
+- Modify: `test/hil/hil_ci.sh`
+
+**Interfaces:**
+- Produces: module `hil_lock` with `BOARD_LOCK_DIR`, `CI_REASON = 'hil_test.py'`, `lock_path(board)`, `flock_nb(board)`, `write_record(fh, reason)`, `clear_record(fh)`, `read_record(board)`, `acquire_board_lock(board, reason=CI_REASON)`, `FLASH_PARALLEL`, `USBTEST_PARALLEL`, `CONTROLLER_SLOTS`, `controller_of(uid)`, `controller_slot(pci)`, `controller_permit`, `flash_permit(uid)`, `usbtest_permit(uid)`, `init_scheduling(b_sems, f_sems, cmap, cmeta, hints, log_fn=None)`.
+
+- [ ] **Step 1: Create `test/hil/hil_lock.py` with the flock core**
+
+New code (the protocol, factored from today's three copies — `board_lock.py` `cmd_hold`/`read_info`, `hil_test.acquire_board_lock`, pool_check `lock_board`; behavior identical to `hil_test.acquire_board_lock` for the acquire path):
+
+```python
+#!/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; it supersedes board_lock.py.
+"""
+import argparse
+import fcntl
+import glob
+import json
+import os
+import re
+import select
+import signal
+import sys
+import time
+
+BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks'
+CI_REASON = 'hil_test.py' # release-protected holder tag (release refuses to kill it)
+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) -> None:
+ """Best-effort holder record; 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
+
+
+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
+```
+
+Then MOVE `acquire_board_lock` from `hil_test.py` verbatim, with exactly two mechanical edits: signature becomes `def acquire_board_lock(board_name, reason=CI_REASON):` and the record-write dict's `'reason': 'hil_test.py'` becomes `'reason': reason`. Do NOT rewrite its body in terms of `flock_nb` — on conflict it reads holder info from the still-open handle before closing, which `flock_nb` (closes on conflict) cannot provide; the fail-open warning text and RuntimeError message must survive character-for-character.
+
+- [ ] **Step 2: Move the controller-permit block into `hil_lock.py`**
+
+MOVE verbatim from `hil_test.py`: the scheduling comment block + `FLASH_PARALLEL`, `USBTEST_PARALLEL`, `CONTROLLER_SLOTS`, the five module globals (`usbtest_sems`, `flash_sems`, `controller_map`, `controller_meta`, `controller_hints`), `controller_of`, `controller_slot`, `controller_permit`, `flash_permit`, `usbtest_permit`. Two mechanical adaptations:
+- add at module scope `log = print` and a setter, replacing the two `log_line(...)` calls inside `controller_of`/`controller_permit` with `log(...)`:
+
+```python
+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
+```
+
+- `PROFILE` inside `controller_permit` now resolves to hil_lock's own module constant (defined in Step 1).
+
+- [ ] **Step 3: Repoint `hil_test.py`**
+
+Add `import hil_lock`. Delete the moved lock + permit code and the five globals. `init_worker` keeps its exact signature and initargs; its body sets the hil_test globals it still owns (`print_lock`, `shuffle_seed`) and forwards the rest:
+
+```python
+def init_worker(lock, seed, b_mutexes, f_sems, cmap, cmeta, hints_by_uid):
+ global print_lock, shuffle_seed
+ print_lock = lock
+ shuffle_seed = seed
+ hil_lock.init_scheduling(b_mutexes, f_sems, cmap, cmeta, hints_by_uid, log_fn=log_line)
+```
+
+Qualify remaining uses: `acquire_board_lock(name)` → `hil_lock.acquire_board_lock(name)` (in `test_board`), `flash_permit(` → `hil_lock.flash_permit(`, `usbtest_permit(` → `hil_lock.usbtest_permit(`, and `main()`'s startup log line + Semaphore construction read `hil_lock.FLASH_PARALLEL`/`hil_lock.USBTEST_PARALLEL`/`hil_lock.CONTROLLER_SLOTS`. `controller_map` reads in the hint-persistence block of `main` use the Manager dict it already holds locally (`cmap`) — no hil_lock global access there; verify.
+
+- [ ] **Step 4: Repoint pool_check to hil_lock and drop its private copies + hil_test import**
+
+In `pool_check.py`: replace `lock_board`/`unlock_board` bodies with the shared core —
+
+```python
+import hil_lock
+
+def lock_board(name: str):
+ try:
+ fh = hil_lock.flock_nb(name)
+ except OSError:
+ info = hil_lock.read_record(name)
+ return json.dumps(info) if info else 'unknown holder'
+ hil_lock.write_record(fh, 'pool_check')
+ return fh
+
+
+def unlock_board(fh) -> None:
+ hil_lock.clear_record(fh)
+ fh.close()
+```
+
+(Behavior note: `lock_board` currently returns the raw record text; JSON-dumping the parsed record is equivalent for display. `hil_lock.BOARD_LOCK_DIR` replaces `hil_test.BOARD_LOCK_DIR`; `os.makedirs(...)` call stays, now on `hil_lock.BOARD_LOCK_DIR`.) Then delete `import hil_test` and the pymtp stub block (`try: import pymtp ... sys.modules['pymtp'] = ...`) — pool_check now imports only `hil_lock` + `hil_flash`.
+
+Run: `grep -n 'hil_test' .claude/skills/hil/pool_check.py`
+Expected: only the docstring mention of the protocol/history, no code references (update the docstring's "imports test/hil/hil_test.py" line to name hil_lock/hil_flash).
+
+- [ ] **Step 5: Add hil_lock.py to the hil_ci.sh scp list** (same block as Task 1 Step 4, one more line: `"$ROOT_DIR/test/hil/hil_lock.py" \`)
+
+- [ ] **Step 6: Verify**
+
+Run: `python3 -m py_compile test/hil/hil_lock.py test/hil/hil_test.py .claude/skills/hil/pool_check.py && python3 test/hil/hil_test.py --help >/dev/null && python3 .claude/skills/hil/pool_check.py --scan-only`
+Expected: clean compile, working scan table, exit 0.
+
+- [ ] **Step 7: Commit**
+
+```bash
+git add test/hil/hil_lock.py test/hil/hil_test.py test/hil/hil_ci.sh .claude/skills/hil/pool_check.py
+git commit -m "hil: extract board locks and controller permits into hil_lock.py"
+```
+
+---
+
+### Task 3: Absorb board_lock.py CLI into hil_lock.py; delete board_lock.py; rename in docs
+
+**Files:**
+- Modify: `test/hil/hil_lock.py` (append CLI)
+- Delete: `test/hil/board_lock.py`
+- Modify: `.claude/skills/hil/SKILL.md`, `.claude/agents/hil-operator.md`, `.claude/agents/target-debugger.md`, `.claude/skills/etm-trace/SKILL.md`, `.claude/skills/usb-kernel-recover/SKILL.md`, `.claude/skills/target-debug/SKILL.md`
+
+**Interfaces:**
+- Produces: `python3 test/hil/hil_lock.py hold|release|status` — identical subcommands, flags, output, and exit codes to today's `board_lock.py`.
+
+- [ ] **Step 1: Move the CLI from `board_lock.py` into `hil_lock.py`**
+
+MOVE verbatim to the end of `hil_lock.py`: `boards_from_config`, `is_locked`, `cmd_hold`, `cmd_release`, `cmd_status`, `main()`, and the `if __name__ == '__main__':` guard. Mechanical adaptations only:
+- `LOCK_DIR` → `BOARD_LOCK_DIR` (all sites), `lock_path` already exists (delete the duplicate), `read_info` → `read_record` (all sites; delete the duplicate definition)
+- `cmd_hold`'s holder loop body (the open/flock/json.dump block) becomes `fh = flock_nb(b)` + `write_record(fh, reason)` inside the existing try/except OSError
+- `_bow_out`'s per-handle truncate loop becomes `clear_record(h)` per handle
+- `cmd_release`'s probe uses `flock_nb(b)` in a try/except OSError (held → existing record/victim logic, with the literal `'hil_test.py'` comparison becoming `CI_REASON`); the free-path truncate becomes `clear_record(fh)`
+- `main()`'s module docstring reference for `--help` text: keep the usage lines, updating the tool name to `hil_lock.py`
+
+Then delete `test/hil/board_lock.py` (`git rm test/hil/board_lock.py`).
+
+- [ ] **Step 2: Rename `board_lock.py` → `hil_lock.py` in the six live docs**
+
+Run: `cd <worktree> && sed -i 's/board_lock\.py/hil_lock.py/g' .claude/skills/hil/SKILL.md .claude/agents/hil-operator.md .claude/agents/target-debugger.md .claude/skills/etm-trace/SKILL.md .claude/skills/usb-kernel-recover/SKILL.md .claude/skills/target-debug/SKILL.md`
+Then: `grep -rn 'board_lock' .claude/ test/ --include='*.md' --include='*.py' --include='*.sh'`
+Expected: zero hits outside `docs/superpowers/` history (which stays untouched).
+
+- [ ] **Step 3: Verify CLI behavior end-to-end**
+
+```bash
+python3 test/hil/hil_lock.py status # expect: no locks (or current holders)
+python3 test/hil/hil_lock.py hold stm32f072disco --reason "split test" &
+sleep 1
+python3 test/hil/hil_lock.py status # expect: stm32f072disco: {... 'reason': 'split test' ...}
+python3 test/hil/hil_lock.py hold stm32f072disco --reason "rival" || echo "conflict OK" # expect: ERROR ... locked + conflict OK
+python3 test/hil/hil_lock.py release stm32f072disco # expect: released holder pid NNN
+python3 test/hil/hil_lock.py status # expect: no locks
+```
+
+Also verify CI-holder protection: create a fake record `echo '{"pid": 1, "reason": "hil_test.py"}' > /tmp/tinyusb-hil-locks/faketest.lock` — since pid 1 holds no flock, `release faketest` must clear the stale record without printing the mid-test error; then `rm -f /tmp/tinyusb-hil-locks/faketest.lock`.
+
+- [ ] **Step 4: Commit**
+
+```bash
+git add -A test/hil .claude
+git commit -m "hil: fold board_lock CLI into hil_lock.py, retire board_lock.py"
+```
+
+---
+
+### Task 4: Rig verification + pre-commit
+
+**Files:** none new (fixes only if verification fails)
+
+- [ ] **Step 1: pool_check flash path on one board**
+
+Run: `python3 .claude/skills/hil/pool_check.py -b stm32f407disco`
+Expected: `✅ dfu_runtime ✅ cafe:...`, exit 0.
+
+- [ ] **Step 2: Capture a pre-refactor baseline report**
+
+Run: `cd /home/hathach/code/tinyusb && python3 test/hil/hil_test.py -b stm32f407disco -B examples test/hil/tinyusb.json && cp hil_report.md /tmp/claude-1000/-home-hathach-code-tinyusb/*/scratchpad/hil_report_master.md`
+(Primary checkout = pre-refactor code but same rig/config; its working tree already carries the new probe uids.)
+
+- [ ] **Step 3: Run the same board from the worktree and diff the report shape**
+
+Run: `cd .claude/worktrees/hil-test-split && python3 test/hil/hil_test.py -b stm32f407disco -B /home/hathach/code/tinyusb/examples test/hil/tinyusb.json && diff <(sed 's/[0-9.]*s//g;s/[0-9.]* [kMG]B\/s//g' hil_report.md) <(sed 's/[0-9.]*s//g;s/[0-9.]* [kMG]B\/s//g' /tmp/claude-1000/-home-hathach-code-tinyusb/*/scratchpad/hil_report_master.md)`
+Expected: empty diff after stripping timings/speeds. Note: `-B` accepts the absolute path so the worktree run reuses the primary checkout's built firmware; `find_firmware` resolves `TINYUSB_ROOT/<build_dir>` and an absolute `-B` overrides relative rooting — if it does not (Path join semantics), instead symlink `ln -s /home/hathach/code/tinyusb/examples/cmake-build-stm32f407disco examples/cmake-build-stm32f407disco` in the worktree and use `-B examples`.
+
+- [ ] **Step 4: pre-commit + final grep hygiene**
+
+Run: `pre-commit run --files test/hil/hil_test.py test/hil/hil_lock.py test/hil/hil_flash.py test/hil/hil_ci.sh .claude/skills/hil/pool_check.py $(git diff --name-only HEAD~3 -- '*.md')`
+Expected: all hooks pass.
+
+- [ ] **Step 5: Commit any verification fixes**
+
+```bash
+git add -A && git commit -m "hil: post-split verification fixes" # only if Steps 1-4 required changes
+```
diff --git a/docs/superpowers/specs/2026-07-28-hil-test-refactor-design.md b/docs/superpowers/specs/2026-07-28-hil-test-refactor-design.md
new file mode 100644
index 0000000..3cd202d
--- /dev/null
+++ b/docs/superpowers/specs/2026-07-28-hil-test-refactor-design.md
@@ -0,0 +1,141 @@
+# hil_test.py refactor: test core + infra helpers
+
+**Date:** 2026-07-28
+**Branch:** `claude/hil-test-split` (based on `claude/hil-pool-check`, which adds `pool_check.py`)
+
+## Motivation
+
+`test/hil/hil_test.py` is 2370 lines mixing five concerns: board-lock protocol, per-controller
+scheduling permits, flash/reset backends, the actual per-example tests, and orchestration/report/CLI.
+The lock protocol additionally exists in three copies (`hil_test.py`, `board_lock.py`,
+`.claude/skills/hil/pool_check.py`), which has already produced drift (pool_check's copy lacks
+hil_test's fail-open and error guards). Splitting the infrastructure out makes `hil_test.py`
+test-focused and gives external tools (pool_check) one canonical import for locks, permits, and
+flashing.
+
+## Goal / non-goals
+
+**Goal:** behavior-preserving code motion. `hil_test.py`'s CLI, arguments, output, report format,
+and runtime behavior stay byte-identical. One deliberate user-visible change: the operator lock CLI
+moves from `board_lock.py` to `hil_lock.py` (same subcommands, same behavior); `board_lock.py` is
+deleted.
+
+**Non-goals (explicit follow-ups, not this change):**
+- The 15 pool_check findings from the 2026-07-28 code review (exception isolation, park-on-failure,
+ espressif coverage, probe-recovery criterion, etc.).
+- pool_check adopting `flash_permit` controller budgeting (enabled by this split).
+- Any change to lock semantics, permit widths, flash behavior, or test logic.
+
+## Resulting layout (`test/hil/`)
+
+| File | ~Lines | Role |
+|---|---|---|
+| `hil_test.py` | 1600 | tests + orchestration + report + CLI (unchanged interface) |
+| `hil_lock.py` (new) | 420 | board-lock protocol + controller permits + operator CLI |
+| `hil_flash.py` (new) | 250 | `run_cmd` + flash/reset backends + `find_firmware` |
+| `board_lock.py` | deleted | superseded by `hil_lock.py` |
+
+Import graph: `hil_test` → {`hil_lock`, `hil_flash`}; the helpers import nothing local (no cycles).
+`pool_check.py` imports all three.
+
+## hil_lock.py
+
+Docstring states the scope: board locks + controller flash/battery permits; the CLI manages board
+locks only (permits are in-process semaphores with no CLI meaning).
+
+**Flock core** (protocol defined once; moved from `board_lock.py`/`hil_test.py`):
+- `BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks'`, `lock_path(board)`
+- `CI_REASON = 'hil_test.py'` — the release-protected holder tag (release refuses to kill it)
+- `flock_nb(board) -> fh` — `os.open(O_RDWR|O_CREAT, 0o666)` **without O_TRUNC** (a losing racer
+ must not wipe the winner's record), `fdopen('r+')`, `LOCK_EX|LOCK_NB`; raises `OSError` when held
+- `write_record(fh, reason)` — truncate+seek+`json.dump({pid, reason, since})`+flush
+- `clear_record(fh)` — truncate(0), swallow OSError (records stay truthful on release)
+- `read_record(board) -> dict | None` — today's `board_lock.read_info`
+- `acquire_board_lock(board, reason=CI_REASON) -> fh | None` — today's `hil_test.acquire_board_lock`
+ with a `reason` parameter: `HIL_NO_BOARD_LOCK=1` bypass, fail-open with warning on lock-dir
+ OSError, `RuntimeError` carrying holder info on conflict
+
+**Controller permits** (moved verbatim from `hil_test.py`):
+- `FLASH_PARALLEL`, `USBTEST_PARALLEL`, `CONTROLLER_SLOTS` (env-overridable as today)
+- `controller_of(uid)`, `controller_slot(pci)`, `controller_permit`, `flash_permit(uid)`,
+ `usbtest_permit(uid)`
+- Per-worker globals (`usbtest_sems`, `flash_sems`, `controller_map`, `controller_meta`,
+ `controller_hints`) set by a new `init_scheduling(sems, fsems, cmap, cmeta, hints)` hook that
+ `hil_test.init_worker` calls from the Pool initializer. `controller_permit`'s PROFILE logging
+ calls back through a module-level `log = print`-style hook that `hil_test` points at `log_line`
+ during `init_scheduling` (keeps helpers free of hil_test imports). The `PROFILE` env flag
+ (`HIL_PROFILE=1`) is read independently in `hil_lock` at import, same derivation as today.
+
+**Operator CLI** (moved verbatim from `board_lock.py`): `hold`/`release`/`status` subcommands with
+the daemon-holder machinery (double-fork, setsid, stdio detach, success pipe, SIGTERM bow-out),
+release policy (probe the flock; protect `CI_REASON` holders; SIGTERM other recorded pids),
+`is_locked` pid-liveness, `--all`/`--config` roster handling. The hold/release/status internals
+switch to the flock-core helpers above; observable behavior unchanged.
+
+## hil_flash.py
+
+Moved verbatim from `hil_test.py`:
+- `CMD_TIMEOUT` (env-overridable), `run_cmd(cmd, cwd, timeout)`, `cmd_stdout_text(out)`
+- `OPENCOD_ADI_PATH`, `TINYUSB_ROOT`
+- All backends: `flash_jlink`/`reset_jlink`, `flash_stlink`/`reset_stlink`,
+ `flash_stflash`/`reset_stflash`, `flash_openocd`/`reset_openocd`,
+ `flash_openocd_wch`/`reset_openocd_wch`, `flash_openocd_adi`/`reset_openocd_adi`,
+ `flash_wlink_rs`/`reset_wlink_rs`, `flash_esptool`/`reset_esptool`,
+ `flash_uniflash`/`reset_uniflash`, `flash_lm4flash`/`reset_lm4flash`
+- `find_firmware(variant, example)`
+- `get_serial_dev(id, vendor_str, product_str, ifnum)` — moves here (not hil_test) because
+ `flash_esptool` calls it; keeping it test-side would create a helper→hil_test import cycle.
+ Tests call `hil_flash.get_serial_dev`.
+- Module globals `build_dir = 'cmake-build'` and `verbose = False`, set by callers exactly as the
+ `hil_test` globals are today (`hil_test.main` sets them from argparse; pool_check sets them
+ directly). `run_cmd`'s verbose echo reads `hil_flash.verbose`.
+
+Dispatch in callers stays string-based: `getattr(hil_flash, f'flash_{flasher["name"].lower()}')`.
+
+## hil_test.py (what remains)
+
+Config TypedDicts (`Board`, `FlasherCfg`, …), device-node lookup except `get_serial_dev`
+(`get_disk_dev`, `get_hid_dev`, `get_alsa_capture_dev`, `open_serial_dev`, `serial_write_all`,
+`read_disk_file`, `open_mtp_dev`, `get_printer_dev`/`open_printer_dev`), enum-timeout globals +
+`wait_until`,
+`log_line`/print-lock, `compact_output`, all `test_*` functions, test lists, `test_example`,
+`build_board`, `test_board`, report rendering/accumulation, `main`. Call sites use explicit
+module-qualified names (`hil_lock.flash_permit(...)`, `hil_flash.run_cmd(...)`) so provenance is
+greppable; no `from … import *`-style mirroring.
+
+`init_worker` keeps its signature (Pool initargs unchanged) and forwards the scheduling state to
+`hil_lock.init_scheduling(...)`.
+
+## Consumer updates (same commit)
+
+- **`.claude/skills/hil/pool_check.py`** — drop its private `lock_board`/`unlock_board` in favor of
+ `hil_lock.flock_nb` + `write_record(fh, 'pool_check')` (+ `clear_record` on release; deliberately NOT `acquire_board_lock`, whose HIL_NO_BOARD_LOCK bypass and fail-open behavior pool_check must not inherit); import
+ flashers/`find_firmware`/`get_serial_dev`/`cmd_stdout_text`/`TINYUSB_ROOT`/`build_dir` from
+ `hil_flash`; `BOARD_LOCK_DIR` references move to `hil_lock`. pool_check then imports **only**
+ `hil_lock` + `hil_flash` (no `hil_test`), so its `pymtp` stub shim is deleted — that shim existed
+ solely because importing `hil_test` pulls in libmtp.
+- **`test/hil/hil_ci.sh`** — the scp list is currently `hil_test.py`, `pymtp.py`, `$CONFIG`; add
+ `hil_lock.py` and `hil_flash.py` (hil_test cannot even import without them). `board_lock.py` was
+ never in the list.
+- **Docs rename `board_lock.py` → `hil_lock.py`** (live docs only): `.claude/skills/hil/SKILL.md`,
+ `.claude/agents/hil-operator.md`, `.claude/agents/target-debugger.md`,
+ `.claude/skills/etm-trace/SKILL.md`, `.claude/skills/usb-kernel-recover/SKILL.md`,
+ `.claude/skills/target-debug/SKILL.md`. Historical `docs/superpowers/{plans,specs}` stay as
+ records.
+- **CI workflow** — untouched (invokes `hil_test.py` CLI only).
+
+## Verification
+
+1. `python3 -m py_compile` on all three modules + pool_check.
+2. `hil_lock.py hold/status/release` interplay: hold, conflicting hold, status listing, release,
+ protection of a `CI_REASON` record, stale-record cleanup.
+3. `pool_check.py --scan-only`, then a single flash board (e.g. `-b stm32f407disco`).
+4. Full `hil_test.py -b stm32f407disco -B examples tinyusb.json` on the rig; compare the report
+ row and log shape against a pre-refactor run.
+5. `pre-commit run` on all touched files.
+
+## Sequencing
+
+Lands on top of `claude/hil-pool-check`. After merge, fix the pool_check review findings as a
+separate change on the new module boundaries, and update agent-memory references to
+`board_lock.py`.
diff --git a/test/hil/board_lock.py b/test/hil/board_lock.py
deleted file mode 100755
index c35e137..0000000
--- a/test/hil/board_lock.py
+++ /dev/null
@@ -1,254 +0,0 @@
-#!/usr/bin/env python3
-"""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:
- board_lock.py hold BOARD [BOARD...] --reason TEXT
- board_lock.py hold --all [--config CONFIG.json] --reason TEXT
- board_lock.py release BOARD [BOARD...] | release --all
- board_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.
-"""
-import argparse
-import fcntl
-import json
-import os
-import select
-import signal
-import sys
-import time
-
-LOCK_DIR = '/tmp/tinyusb-hil-locks'
-
-
-def lock_path(board: str) -> str:
- return os.path.join(LOCK_DIR, f'{board}.lock')
-
-
-def boards_from_config(config: str) -> list:
- try:
- with open(config) as f:
- return [b['name'] for b in json.load(f)['boards']]
- except (OSError, ValueError, KeyError) as e:
- print(f'ERROR: cannot read board roster {config}: {e}', file=sys.stderr)
- sys.exit(1)
-
-
-def read_info(board: str):
- try:
- with open(lock_path(board)) as f:
- return json.load(f)
- except (OSError, ValueError):
- return None
-
-
-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_info(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(LOCK_DIR, exist_ok=True)
- # No pre-check: the holder's own LOCK_NB flock is the only authority — a
- # recorded pid may be stale or recycled (e.g. a live hil_test.py worker
- # that already released this board's flock but not its record).
- # The holder signals success through this pipe. A generic is_locked()
- # poll would be fooled by a RIVAL invocation's flock — only the holder
- # itself 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_info(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:
- # O_RDWR without O_TRUNC: never truncate before the flock is
- # held — a losing racer must not wipe the winner's holder info.
- fd = os.open(lock_path(b), os.O_RDWR | os.O_CREAT, 0o666)
- fh = os.fdopen(fd, 'r+')
- fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
- 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()
- 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_info/status stay truthful
- # (the kernel drops the flocks themselves on exit either way)
- for h in handles:
- try:
- h.truncate(0)
- except OSError:
- pass
- 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_info(b) or {}
- pid = info.get('pid')
- if info.get('reason') == 'hil_test.py':
- print(f'ERROR: {b} is mid-test by hil_test.py (pid {pid}) — not killing a '
- 'CI run; 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
- try:
- fh.truncate(0)
- except OSError:
- pass
- 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(LOCK_DIR):
- print('no locks')
- return 0
- any_locked = False
- for fn in sorted(os.listdir(LOCK_DIR)):
- if not fn.endswith('.lock'):
- continue
- b = fn[:-5]
- if is_locked(b):
- any_locked = True
- print(f'{b}: {read_info(b)}')
- if not any_locked:
- print('no locks')
- return 0
-
-
-def main():
- ap = argparse.ArgumentParser(description=__doc__,
- 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.abspath(__file__)),
- 'tinyusb.json'),
- help='board roster JSON (default: tinyusb.json beside 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(LOCK_DIR) if fn.endswith('.lock')]
- if os.path.isdir(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/hil_ci.sh b/test/hil/hil_ci.sh
index 3ec9079..3384b4e 100644
--- a/test/hil/hil_ci.sh
+++ b/test/hil/hil_ci.sh
@@ -44,15 +44,22 @@
ssh "$REMOTE" bash -s -- "$REMOTE_DIR" <<'REMOTE'
set -e
rm -rf -- "$1"
-mkdir -p -- "$1/test/hil" "$1/examples"
+# .claude path: usbtest.py's HUNG recovery resolves usb_recover.sh relative to the
+# staged repo root — without it, recovery ENOENTs and the wedge is left in place
+mkdir -p -- "$1/test/hil" "$1/examples" "$1/.claude/skills/usb-kernel-recover/scripts"
REMOTE
# Copy HIL test script and config
echo "==> Copying test scripts"
scp -q "$ROOT_DIR/test/hil/hil_test.py" \
+ "$ROOT_DIR/test/hil/hil_flash.py" \
+ "$ROOT_DIR/test/hil/hil_lock.py" \
+ "$ROOT_DIR/test/hil/usbtest.py" \
"$ROOT_DIR/test/hil/pymtp.py" \
"$CONFIG" \
"$REMOTE:$REMOTE_DIR/test/hil/"
+scp -q "$ROOT_DIR/.claude/skills/usb-kernel-recover/scripts/usb_recover.sh" \
+ "$REMOTE:$REMOTE_DIR/.claude/skills/usb-kernel-recover/scripts/"
# Copy only firmware binaries (elf/bin/hex) plus esptool metadata
# (config.env + flash_args needed by the esptool flasher), preserving structure
diff --git a/test/hil/hil_flash.py b/test/hil/hil_flash.py
new file mode 100755
index 0000000..8142580
--- /dev/null
+++ b/test/hil/hil_flash.py
@@ -0,0 +1,293 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+# Firmware flashing for the TinyUSB HIL rig: run_cmd, one flash_*/reset_* pair per
+# flasher type (dispatched by config name via getattr), find_firmware, and the
+# fixture serial-port resolver get_serial_dev (here, not hil_test: flash_esptool
+# needs it and helpers must not import hil_test).
+# Callers set module globals `build_dir` and `verbose` (hil_test.main from argparse,
+# pool_check directly) exactly as they set hil_test's globals today.
+#
+# from __future__ import annotations (below): some moved function signatures use
+# type hints (Any, Board) not defined in this module; postponed evaluation (PEP
+# 563) keeps those as unevaluated strings so the verbatim-moved defs still load.
+
+from __future__ import annotations
+
+import glob
+import json
+import os
+import signal
+import subprocess
+from pathlib import Path
+
+verbose = False
+build_dir = 'cmake-build'
+
+CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))
+
+# flasher names (dispatch key, board['flasher']['name'].lower()) whose reset_* is a no-op
+RESET_NOOP = {'esptool', 'lm4flash', 'stflash', 'uniflash'}
+
+# extra parents find_firmware ALSO searches after build_dir. Empty by default so
+# hil_test's -B stays authoritative (a board missing there must report "Skip (no
+# binary)", never silently flash a stale binary from another tree); pool_check
+# opts in to cover both standard layouts.
+EXTRA_BUILD_DIRS: list = []
+
+
+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)
+
+
+# -------------------------------------------------------------
+# Path
+# -------------------------------------------------------------
+OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi'
+TINYUSB_ROOT = Path(__file__).resolve().parents[2]
+
+# 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]
+
+
+# -------------------------------------------------------------
+# Flashing firmware
+# -------------------------------------------------------------
+def run_cmd(cmd: str, cwd: str | None = None, timeout: int = CMD_TIMEOUT) -> subprocess.CompletedProcess:
+ popen_kwargs = {
+ 'cwd': cwd,
+ 'shell': True,
+ 'stdout': subprocess.PIPE,
+ 'stderr': subprocess.STDOUT,
+ 'text': True,
+ 'encoding': 'utf-8',
+ 'errors': 'replace',
+ }
+ if os.name != 'nt':
+ # 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, _ = p.communicate(timeout=timeout)
+ r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out)
+ except subprocess.TimeoutExpired as ex:
+ if os.name != 'nt':
+ try:
+ os.killpg(p.pid, signal.SIGKILL)
+ except ProcessLookupError:
+ pass
+ else:
+ p.kill()
+ try:
+ out, _ = p.communicate(timeout=10)
+ except subprocess.TimeoutExpired: # unkillable (e.g. D-state on wedged USB)
+ out = None
+ timeout_out = ex.stdout or out or b''
+ title = f'COMMAND TIMEOUT ({timeout}s): {cmd}'
+ print()
+ if os.getenv('CI'):
+ print(f"::group::{title}")
+ print(cmd_stdout_text(timeout_out))
+ print(f"::endgroup::")
+ else:
+ print(title)
+ print(cmd_stdout_text(timeout_out))
+ return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out)
+
+ if r.returncode != 0:
+ title = f'COMMAND FAILED: {cmd}'
+ print()
+ if os.getenv('CI'):
+ print(f"::group::{title}")
+ print(cmd_stdout_text(r.stdout))
+ print(f"::endgroup::")
+ else:
+ print(title)
+ print(cmd_stdout_text(r.stdout))
+ elif verbose:
+ print(cmd)
+ print(cmd_stdout_text(r.stdout))
+ return r
+
+
+def flash_jlink(board: Board, firmware: str) -> subprocess.CompletedProcess:
+ flasher = board['flasher']
+ script = ['halt', 'r', f'loadfile {firmware}.elf', 'r', 'go', 'exit']
+ f_jlink = Path(f'{board["name"]}_{Path(firmware).name}.jlink')
+ with f_jlink.open('w') as f:
+ f.writelines(f'{s}\n' for s in script)
+ ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
+ f_jlink.unlink(missing_ok=True)
+ return ret
+
+
+def reset_jlink(board: Board) -> subprocess.CompletedProcess:
+ flasher = board['flasher']
+ script = ['halt', 'r', 'go', 'exit']
+ f_jlink = Path(f'{board["name"]}_reset.jlink')
+ if not f_jlink.exists():
+ with f_jlink.open('w') as f:
+ f.writelines(f'{s}\n' for s in script)
+ ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
+ return ret
+
+
+def flash_stlink(board, firmware):
+ flasher = board['flasher']
+ return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --write {firmware}.elf --go')
+
+
+def reset_stlink(board):
+ flasher = board['flasher']
+ return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --rst --go')
+
+def flash_stflash(board, firmware):
+ flasher = board['flasher']
+ ret = run_cmd(f'st-flash --serial {flasher["uid"]} write {firmware}.bin 0x8000000')
+ return ret
+
+
+def reset_stflash(board):
+ flasher = board['flasher']
+ return subprocess.CompletedProcess(args=['dummy'], returncode=0)
+
+
+def flash_openocd(board, firmware):
+ flasher = board['flasher']
+ ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
+ f'{flasher["args"]} -c "init; halt; program {firmware}.elf verify; reset; exit"')
+ return ret
+
+
+def reset_openocd(board):
+ flasher = board['flasher']
+ ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
+ f'{flasher["args"]} -c "init; reset run; exit"')
+ return ret
+
+
+def flash_openocd_wch(board, firmware):
+ flasher = board['flasher']
+ ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "telnet_port disabled" '
+ f'-c "adapter serial {flasher["uid"]}" {flasher.get("args", "")} -c "program {firmware}.elf reset exit"')
+ return ret
+
+
+def reset_openocd_wch(board):
+ flasher = board['flasher']
+ ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "telnet_port disabled" '
+ f'-c "adapter serial {flasher["uid"]}" {flasher.get("args", "")} -c "init; reset run; exit"')
+ return ret
+
+
+def flash_openocd_adi(board: Board, firmware: str) -> subprocess.CompletedProcess:
+ flasher = board['flasher']
+ openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
+ tcl_dir = OPENCOD_ADI_PATH / 'tcl'
+ ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
+ f'{flasher["args"]} -c "program {firmware}.elf reset exit"')
+ return ret
+
+
+def reset_openocd_adi(board: Board) -> subprocess.CompletedProcess:
+ flasher = board['flasher']
+ openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
+ tcl_dir = OPENCOD_ADI_PATH / 'tcl'
+ ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
+ f'{flasher["args"]} -c "program reset exit"')
+ return ret
+
+
+def flash_wlink_rs(board, firmware):
+ flasher = board['flasher']
+ # wlink use index for probe selection and lacking usb serial support
+ ret = run_cmd(f'wlink flash {firmware}.elf')
+ return ret
+
+
+def reset_wlink_rs(board):
+ flasher = board['flasher']
+ # wlink use index for probe selection and lacking usb serial support
+ ret = run_cmd(f'wlink reset')
+ return ret
+
+
+def flash_esptool(board: Board, firmware: str) -> subprocess.CompletedProcess:
+ flasher = board['flasher']
+ port = get_serial_dev(flasher["uid"], None, None, 0)
+ fw_dir = Path(f'{firmware}.bin').parent
+ with (fw_dir / 'config.env').open() as f:
+ idf_target = json.load(f)['IDF_TARGET']
+ with (fw_dir / 'flash_args').open() as f:
+ flash_args = f.read().strip().replace('\n', ' ')
+ command = (f'esptool --chip {idf_target} -p {port} {flasher["args"]} '
+ f'--before=default_reset --after=hard_reset write_flash {flash_args}')
+ ret = run_cmd(command, cwd=str(fw_dir))
+ return ret
+
+
+def reset_esptool(board):
+ flasher = board['flasher']
+ return subprocess.CompletedProcess(args=['dummy'], returncode=0)
+
+
+def flash_uniflash(board, firmware):
+ flasher = board['flasher']
+ ret = run_cmd(f'dslite.sh {flasher["args"]} -f {firmware}.hex')
+ return ret
+
+
+def reset_uniflash(board):
+ flasher = board['flasher']
+ return subprocess.CompletedProcess(args=['dummy'], returncode=0)
+
+
+def flash_lm4flash(board, firmware):
+ # TI Tiva-C / Stellaris ICDI: lightweight lm4flash, resets and runs after write
+ flasher = board['flasher']
+ ret = run_cmd(f'lm4flash -s {flasher["uid"]} {flasher["args"]} {firmware}.bin')
+ return ret
+
+
+def reset_lm4flash(board):
+ # lm4flash has no reset-only mode; it resets+runs on flash, so reset is a no-op
+ flasher = board['flasher']
+ return subprocess.CompletedProcess(args=['dummy'], returncode=0)
+
+
+def find_firmware(variant: str, example: str, roots: list | None = None):
+ """Locate a built example's firmware base path (no extension) under
+ <build_dir>/cmake-build-<variant>/<example>/, then under EXTRA_BUILD_DIRS
+ (empty unless the caller opts in — see its comment). `roots` overrides that
+ search list entirely for one call (e.g. to find a build just produced by
+ tools/build.py in its fixed cmake-build/ layout without widening the global
+ policy). Accepts the single-config layout (firmware directly in the example
+ dir) or Ninja Multi-Config (a per-config subdir like RelWithDebInfo/).
+ Returns the base Path, or None if not built."""
+ base = Path(example).name
+ for bd in dict.fromkeys(roots if roots is not None else [build_dir, *EXTRA_BUILD_DIRS]):
+ fw_dir = TINYUSB_ROOT / bd / f'cmake-build-{variant}' / example
+ if not fw_dir.is_dir():
+ continue
+ for cand in [fw_dir / base, fw_dir / 'RelWithDebInfo' / base,
+ *(p.with_suffix('') for p in sorted(fw_dir.glob(f'*/{base}.elf')))]:
+ if cand.with_suffix('.elf').exists() or cand.with_suffix('.bin').exists():
+ return cand
+ return None
diff --git a/test/hil/hil_lock.py b/test/hil/hil_lock.py
new file mode 100755
index 0000000..e570da1
--- /dev/null
+++ b/test/hil/hil_lock.py
@@ -0,0 +1,479 @@
+#!/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 glob
+import json
+import os
+import re
+import select
+import signal
+import sys
+import time
+
+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 on it like board_lock.py did (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.
+# - uPD720201 cards need their latest firmware (>= 2.0.2.6; RAM-uploaded, reloads every
+# power cycle): ROM firmware dies under battery + re-enumeration churn, and usbtest.py
+# refuses the unlink-stress cases on it.
+# - widths (profiled 2026-07-13/14): wall time 22.2/14.3/12.5/10.8 min at usbtest width
+# 1/2/3/4, plateau after; flash width beyond 8 only adds flasher-hub contention;
+# battery case failures start at 12/8 (bandwidth stretch on shared leaf-hub uplinks).
+# - a marginal DUT port bouncing during concurrent batteries can wedge/kill a uPD720201
+# ("xHCI host not responding to stop endpoint command"): fix the port/cable or pull
+# the board, don't lower the widths (2026-07-16: every death traced to one board's port).
+FLASH_PARALLEL = int(os.getenv('HIL_FLASH_PARALLEL', '8'))
+USBTEST_PARALLEL = int(os.getenv('HIL_USBTEST_PARALLEL', '4'))
+CONTROLLER_SLOTS = 12 # lock slots; controllers are assigned to slots on first sight
+usbtest_sems = None # CONTROLLER_SLOTS semaphores: per-slot usbtest-battery permits
+flash_sems = None # CONTROLLER_SLOTS semaphores: 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 if the device
+ is not enumerated (e.g. parked in board_test firmware with USB off). 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
+ for f in glob.glob('/sys/bus/usb/devices/*/serial'):
+ d = os.path.dirname(f)
+ try:
+ if open(f).read().strip().lower() != uid.lower():
+ continue
+ bus = int(open(os.path.join(d, 'busnum')).read())
+ root = os.path.realpath(f'/sys/bus/usb/devices/usb{bus}')
+ 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]
+ except (OSError, ValueError):
+ continue
+ 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
+
+
+class controller_permit:
+ """Context manager: one permit from `sems` on the board's controller slot. If the
+ controller is unknown, fail closed: take one permit from EVERY slot, in order, so the
+ operation respects the budget wherever it might land. `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
+ if sems is None:
+ return
+ pci = controller_of(uid)
+ if pci is None and not warn_unknown:
+ # last-run cabling hint, flash budgeting only: a mis-budgeted flash is harmless,
+ # but a battery must never trust a stale hint (it could stack two batteries on
+ # one controller). In practice only a board's first flash lands here - batteries
+ # assert enumeration before taking their permit.
+ pci = controller_hints.get(uid)
+ if pci is None and warn_unknown:
+ log(f'warning: cannot resolve {uid} to a host controller; '
+ 'taking a permit on every slot (over-serialized)')
+ self.slots = [controller_slot(pci)] if pci else list(range(CONTROLLER_SLOTS))
+
+ def __enter__(self):
+ if self.slots:
+ t0 = time.monotonic()
+ taken = []
+ try:
+ for s in self.slots:
+ self.sems[s].acquire()
+ taken.append(s)
+ # stays inside the try: if this raises (e.g. broken stdout), the permits
+ # must be released - a failed __enter__ never gets its __exit__
+ 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.slots):
+ self.sems[s].release()
+ 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 — a
+ # recorded pid may be stale or recycled (e.g. a live hil_test.py worker
+ # that already released this board's flock but not its record).
+ # The holder signals success through this pipe. A generic is_locked()
+ # poll would be fooled by a RIVAL invocation's flock — only the holder
+ # itself 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.abspath(__file__)),
+ 'tinyusb.json'),
+ help='board roster JSON (default: tinyusb.json beside 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/hil_pool_check.py b/test/hil/hil_pool_check.py
new file mode 100644
index 0000000..6328421
--- /dev/null
+++ b/test/hil/hil_pool_check.py
@@ -0,0 +1,1013 @@
+#!/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/ beside hil_lock.py and hil_flash.py, which it imports; 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 subprocess
+import sys
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(Path(__file__).resolve().parent)) # for import-as-module callers
+
+import hil_lock
+import hil_flash
+
+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()
+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, named 'usbN' with no dash, are
+ excluded: their fabricated PCI-address 'serial' and slow autosuspend-wake read
+ cost 6-7s/scan on this rig). Keyed by busport, not serial: a serial can be
+ shared by two different devices (e.g. an Espressif USB-Serial-JTAG bridge and
+ the cafe TinyUSB device it flashes derive both from the same MAC) — collapsing
+ them into one dict slot would silently drop whichever lost the race."""
+ found = {}
+ for f in glob.glob('/sys/bus/usb/devices/*-*/serial'):
+ d = os.path.dirname(f)
+ busport = os.path.basename(d)
+ try:
+ sn = open(f).read().strip().lower()
+ vidpid = f'{open(d + "/idVendor").read().strip()}:{open(d + "/idProduct").read().strip()}'
+ found[busport] = {'serial': sn, 'vidpid': vidpid, 'ino': os.stat(d + '/').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) sharing 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 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 as locked and skipped — never waited on,
+ and there is deliberately 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; a cross-uid setup
+ # would need flock_nb to distinguish the two
+ 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
+ # can't show us and release can't recognize the protected holder — bail out
+ 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:
+ r = subprocess.run(['sudo', '-n', 'true'], capture_output=True)
+ except OSError: # sudo not installed (bare dev PC/container): recovery off, not fatal
+ 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 (observed to flake while the toggle worked).
+ J-Links respond with a full disconnect and can stay off the bus for >8 s."""
+ pre = find_usb(uid)
+ try:
+ # bounded: 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
+ subprocess.run(['sudo', '-n', str(USB_RECOVER), 'authorized', busport],
+ capture_output=True, text=True, timeout=30)
+ except subprocess.TimeoutExpired:
+ 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):
+ 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 base
+ path to flash. 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)
+ 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
+ RuntimeError when a bridge's /dev/serial/by-id node vanishes, config.env
+ FileNotFoundError, .jlink script OSError) and an exception must not skip the
+ caller's retry/recovery ladder. Returns (returncode, error line)."""
+ try:
+ ret = fn(*fn_args)
+ if ret.returncode == 0:
+ return 0, ''
+ err = flash_error_line(hil_flash.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 flash attempt afterward, regardless of
+ whether the replug is confirmed — some probes (WCH-Link, ST-Link, CP210x,
+ picoprobe) leave their sysfs kobject intact 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 — skip the toggle
+ 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 (get-idf)'
+ 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."""
+ import serial
+ try:
+ port = hil_flash.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 issuing the reset: pyserial's open-time flush is long past,
+ # so this drops the pre-reset CDC backlog (which must not count as life)
+ # while keeping the board's post-reset boot banner, which prints while the
+ # reset tool is still tearing down and would be eaten by a post-reset flush
+ ser.reset_input_buffer()
+ if do_reset:
+ getattr(hil_flash, f'reset_{board["flasher"]["name"].lower()}')(board)
+ # collect the WHOLE window and judge content, 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. Early-exit once non-board_test output proves
+ # a real example is talking.
+ 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: fresh board_test hello
+ # (park verification) vs any non-board_test output (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: -T target, -D per build.args, variant defines/flags,
+ --build-name), or idf.py directly for espressif (tools/build.py's esp branch
+ ignores -T and builds everything; variant flags travel as -DCFLAGS_CLI, the
+ same channel tools/build.py uses). Bounded and process-group-killed via
+ run_cmd; 600 s: a first configure+build of an SDK-heavy family (pico, nrf,
+ esp) exceeds the old 300. Builds normally run pre-lock (pick_example / the
+ pre-park ensure), 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
+ build's 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 board.get('build', {}).get('args', []) + 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_flash.run_cmd(shlex.join(cmd), cwd=str(hil_flash.TINYUSB_ROOT),
+ timeout=600).returncode
+ cmd = [sys.executable, str(hil_flash.TINYUSB_ROOT / 'tools' / 'build.py'),
+ '-b', name, '-T', Path(example).name,
+ '-j', str(max(1, (os.cpu_count() or _jobs) // _jobs))]
+ for d in board.get('build', {}).get('args', []):
+ cmd += ['-D', d]
+ 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_flash.run_cmd(shlex.join(cmd), cwd=str(hil_flash.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 build attempt per
+ (variant, example) per run, success or failure — memoized in _builds, so a
+ repeat call (park, under the held flock) resolves instantly even when an
+ exclusive -B hides the fresh cmake-build/ artifact from the global search."""
+ fw = hil_flash.find_firmware(variant, example)
+ 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 (get-idf)')
+ 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 already-present shared deps (lib/fatfs's ffconf.h dance), so
+ # it must exclude every in-flight build, not just other get_deps calls:
+ # it drains ALL build slots before running.
+ with _deps_lock:
+ for _ in range(_jobs):
+ _build_sem.acquire()
+ try:
+ r = hil_flash.run_cmd(shlex.join([sys.executable, str(hil_flash.TINYUSB_ROOT / 'tools' / 'get_deps.py'),
+ '-b', board['name']]),
+ cwd=str(hil_flash.TINYUSB_ROOT), timeout=600)
+ finally:
+ for _ in range(_jobs):
+ _build_sem.release()
+ if r.returncode != 0:
+ note.append('get_deps failed')
+ bd = hil_flash.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
+ # tools/build.py and the idf.py invocation above always write to cmake-build/:
+ # look there too even when an explicit -B narrowed the global search — this is
+ # OUR fresh build, not a stale-candidate fallback
+ fw = hil_flash.find_firmware(variant, example,
+ roots=[hil_flash.build_dir, 'cmake-build'])
+ _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')
+ 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 here; a missing build is built on the spot further
+ # down (after a lock peek), except in scan/no-build modes — and never for a
+ # missing probe (nothing could be flashed anyway)
+ 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, not just the probe's: the enumerated device
+ # (with busport), off-bus (normal when parked in board_test), or n/a for
+ # host-only boards whose uid never enumerates
+ 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: that check DID run, 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') is None)
+ if need_example or need_bt:
+ # builds are long and run BEFORE locking (park must never hold the flock
+ # through a build); peek the lock 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-image 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 firmware is on it (the parked board_test echoes and
+ # prints a periodic hello 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): re-park while the
+ # board 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 (a fast park drops it immediately)
+ 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 already started board_test);
+ # POSITIVE marker: its hello must appear — stale example output may still
+ # drain from the probe bridge's FIFO alongside it and 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:
+ # board never enumerated this run: uid-disappearance can't distinguish 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 (arm-none-eabi-gcc + esptool
+ # in ~/.local/bin, STM32_Programmer_CLI in ~/bin) which non-login shells may
+ # lack — 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 with a concurrent hil_test.py run yet (would need
+ # a file-lock budget in hil_lock; hil_test uses in-process semaphores) — keep 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_flash.verbose = args.verbose
+ if args.build_dir is None:
+ # default mode: search both standard layouts (cmake-build/ from tools/build.py
+ # + ESP-IDF, examples/ from manual builds). An EXPLICIT -B is exclusive — the
+ # caller named an artifact tree, so a miss must report, not silently flash an
+ # older build from elsewhere. hil_test's -B is likewise untouched by this.
+ 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_flash'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]
+ widths = [max(len(h), *(len(c[i]) for c in cells)) if cells else len(h)
+ for i, h in enumerate(headers)]
+ line = lambda vals: '| ' + ' | '.join(v.ljust(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')
+ sys.exit(min(counts['flash-failed'] + counts['failed'], 125))
+
+
+if __name__ == '__main__':
+ main()
diff --git a/test/hil/hil_test.py b/test/hil/hil_test.py
index 80d1e18..71e85f5 100755
--- a/test/hil/hil_test.py
+++ b/test/hil/hil_test.py
@@ -46,10 +46,9 @@
import select
import sys
import time
-import signal
from contextlib import redirect_stdout
from pathlib import Path
-from typing import Any, TypedDict, NotRequired, cast
+from typing import TypedDict, NotRequired, cast
import serial
import subprocess
@@ -58,6 +57,9 @@
import multiprocessing
from multiprocessing import TimeoutError as MpTimeoutError
+import hil_flash
+import hil_lock
+
# Raw Lock/Semaphore objects passed via Pool initargs are inheritable only under the fork
# start method (spawn/forkserver pickle them and fail at Pool creation) — pin it so a
# future interpreter default change cannot break the run at startup.
@@ -68,51 +70,6 @@
from pymtp import MTP
import string
-# --- per-board dev-session locks (see test/hil/board_lock.py) ------------
-BOARD_LOCK_DIR = '/tmp/tinyusb-hil-locks'
-
-def acquire_board_lock(board_name):
- """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 board_lock.py / 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': 'hil_test.py',
- 'since': time.strftime('%Y-%m-%dT%H:%M:%S%z')}, fh)
- fh.flush()
- except OSError:
- pass
- return fh
-
-
# Enumeration wait budget. The first attempt gets ENUM_TIMEOUT; retry attempts get the
# shorter ENUM_TIMEOUT_RETRY - the board was just re-flashed again, and a device that is
# going to enumerate shows up within a few seconds, so a failing test costs ~3-5x a
@@ -162,41 +119,16 @@
PROFILE = os.environ.get('HIL_PROFILE') == '1' # timestamped logs + permit/flash timing + ctrl-map dump
test_only = []
board_test = {}
-build_dir = 'cmake-build'
skip_flash = False
print_lock = None
shuffle_seed = None # per-run seed for the per-board test-order shuffle (HIL_SHUFFLE_SEED to replay)
-# 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.
-# - uPD720201 cards need their latest firmware (>= 2.0.2.6; RAM-uploaded, reloads every
-# power cycle): ROM firmware dies under battery + re-enumeration churn, and usbtest.py
-# refuses the unlink-stress cases on it.
-# - widths (profiled 2026-07-13/14): wall time 22.2/14.3/12.5/10.8 min at usbtest width
-# 1/2/3/4, plateau after; flash width beyond 8 only adds flasher-hub contention;
-# battery case failures start at 12/8 (bandwidth stretch on shared leaf-hub uplinks).
-# - a marginal DUT port bouncing during concurrent batteries can wedge/kill a uPD720201
-# ("xHCI host not responding to stop endpoint command"): fix the port/cable or pull
-# the board, don't lower the widths (2026-07-16: every death traced to one board's port).
-FLASH_PARALLEL = int(os.getenv('HIL_FLASH_PARALLEL', '8'))
-USBTEST_PARALLEL = int(os.getenv('HIL_USBTEST_PARALLEL', '4'))
-CONTROLLER_SLOTS = 12 # lock slots; controllers are assigned to slots on first sight
-usbtest_sems = None # CONTROLLER_SLOTS semaphores: per-slot usbtest-battery permits
-flash_sems = None # CONTROLLER_SLOTS semaphores: 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)
-
def init_worker(lock, seed, b_mutexes, f_sems, cmap, cmeta, hints_by_uid):
- global print_lock, shuffle_seed, usbtest_sems, flash_sems, controller_map, controller_meta, controller_hints
+ global print_lock, shuffle_seed
print_lock = lock
shuffle_seed = seed
- usbtest_sems = b_mutexes
- flash_sems = f_sems
- controller_map = cmap
- controller_meta = cmeta
- controller_hints = hints_by_uid
+ hil_lock.init_scheduling(b_mutexes, f_sems, cmap, cmeta, hints_by_uid, log_fn=log_line)
def log_line(msg: str) -> None:
@@ -210,108 +142,6 @@
print(msg, file=out, flush=True)
-# -------------------------------------------------------------
-# Per-controller scheduling
-# -------------------------------------------------------------
-def controller_of(uid: str):
- """Resolve a DUT uid to its root host controller's PCI address, or None if the device
- is not enumerated (e.g. parked in board_test firmware with USB off). 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
- for f in glob.glob('/sys/bus/usb/devices/*/serial'):
- d = os.path.dirname(f)
- try:
- if open(f).read().strip().lower() != uid.lower():
- continue
- bus = int(open(os.path.join(d, 'busnum')).read())
- root = os.path.realpath(f'/sys/bus/usb/devices/usb{bus}')
- 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]
- except (OSError, ValueError):
- continue
- 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
-
-
-class controller_permit:
- """Context manager: one permit from `sems` on the board's controller slot. If the
- controller is unknown, fail closed: take one permit from EVERY slot, in order, so the
- operation respects the budget wherever it might land. `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
- if sems is None:
- return
- pci = controller_of(uid)
- if pci is None and not warn_unknown:
- # last-run cabling hint, flash budgeting only: a mis-budgeted flash is harmless,
- # but a battery must never trust a stale hint (it could stack two batteries on
- # one controller). In practice only a board's first flash lands here - batteries
- # assert enumeration before taking their permit.
- pci = controller_hints.get(uid)
- if pci is None and warn_unknown:
- log_line(f'warning: cannot resolve {uid} to a host controller; '
- 'taking a permit on every slot (over-serialized)')
- self.slots = [controller_slot(pci)] if pci else list(range(CONTROLLER_SLOTS))
-
- def __enter__(self):
- if self.slots:
- t0 = time.monotonic()
- taken = []
- try:
- for s in self.slots:
- self.sems[s].acquire()
- taken.append(s)
- # stays inside the try: if this raises (e.g. broken stdout), the permits
- # must be released - a failed __enter__ never gets its __exit__
- if PROFILE and time.monotonic() - t0 > 1.0:
- log_line(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.slots):
- self.sems[s].release()
- 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)
-
-
def compact_output(raw: str) -> str:
if not raw:
return ''
@@ -365,47 +195,16 @@
class HilConfig(TypedDict):
boards: list[Board]
-CMD_TIMEOUT = int(os.getenv('HIL_CMD_TIMEOUT', '180'))
POOL_TIMEOUT = int(os.getenv('HIL_POOL_TIMEOUT', '4200')) # usbtest batteries are serialized fleet-wide, lengthening the tail
SERIAL_READ_TIMEOUT = float(os.getenv('HIL_SERIAL_READ_TIMEOUT', '5'))
SERIAL_WRITE_TIMEOUT = float(os.getenv('HIL_SERIAL_WRITE_TIMEOUT', '10'))
-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)
-
-
MSC_README_TXT = \
b"This is tinyusb's MassStorage Class demo.\r\n\r\n\
If you find any bugs or get any questions, feel free to file an\r\n\
issue at github.com/hathach/tinyusb"
-# -------------------------------------------------------------
-# Path
-# -------------------------------------------------------------
-OPENCOD_ADI_PATH = Path.home() / 'app' / 'openocd_adi'
-TINYUSB_ROOT = Path(__file__).resolve().parents[2]
-
-# 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]
-
-
# get usb disk by id
def get_disk_dev(id, vendor_str, lun):
return f'/dev/disk/by-id/usb-{vendor_str}_Mass_Storage_{id}-0:{lun}'
@@ -530,214 +329,12 @@
# -------------------------------------------------------------
-# Flashing firmware
-# -------------------------------------------------------------
-def run_cmd(cmd: str, cwd: str | None = None, timeout: int = CMD_TIMEOUT) -> subprocess.CompletedProcess:
- popen_kwargs = {
- 'cwd': cwd,
- 'shell': True,
- 'stdout': subprocess.PIPE,
- 'stderr': subprocess.STDOUT,
- 'text': True,
- 'encoding': 'utf-8',
- 'errors': 'replace',
- }
- if os.name != 'nt':
- popen_kwargs['preexec_fn'] = os.setsid
-
- p = subprocess.Popen(cmd, **popen_kwargs)
- try:
- out, _ = p.communicate(timeout=timeout)
- r = subprocess.CompletedProcess(args=cmd, returncode=p.returncode, stdout=out)
- except subprocess.TimeoutExpired as ex:
- if os.name != 'nt':
- try:
- os.killpg(p.pid, signal.SIGKILL)
- except ProcessLookupError:
- pass
- else:
- p.kill()
- out, _ = p.communicate()
- timeout_out = ex.stdout or out or b''
- title = f'COMMAND TIMEOUT ({timeout}s): {cmd}'
- print()
- if os.getenv('CI'):
- print(f"::group::{title}")
- print(cmd_stdout_text(timeout_out))
- print(f"::endgroup::")
- else:
- print(title)
- print(cmd_stdout_text(timeout_out))
- return subprocess.CompletedProcess(args=cmd, returncode=124, stdout=timeout_out)
-
- if r.returncode != 0:
- title = f'COMMAND FAILED: {cmd}'
- print()
- if os.getenv('CI'):
- print(f"::group::{title}")
- print(cmd_stdout_text(r.stdout))
- print(f"::endgroup::")
- else:
- print(title)
- print(cmd_stdout_text(r.stdout))
- elif verbose:
- print(cmd)
- print(cmd_stdout_text(r.stdout))
- return r
-
-
-def flash_jlink(board: Board, firmware: str) -> subprocess.CompletedProcess:
- flasher = board['flasher']
- script = ['halt', 'r', f'loadfile {firmware}.elf', 'r', 'go', 'exit']
- f_jlink = Path(f'{board["name"]}_{Path(firmware).name}.jlink')
- with f_jlink.open('w') as f:
- f.writelines(f'{s}\n' for s in script)
- ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
- f_jlink.unlink(missing_ok=True)
- return ret
-
-
-def reset_jlink(board: Board) -> subprocess.CompletedProcess:
- flasher = board['flasher']
- script = ['halt', 'r', 'go', 'exit']
- f_jlink = Path(f'{board["name"]}_reset.jlink')
- if not f_jlink.exists():
- with f_jlink.open('w') as f:
- f.writelines(f'{s}\n' for s in script)
- ret = run_cmd(f'JLinkExe -USB {flasher["uid"]} {flasher["args"]} -if swd -JTAGConf -1,-1 -speed auto -NoGui 1 -ExitOnError 1 -CommandFile {f_jlink}')
- return ret
-
-
-def flash_stlink(board, firmware):
- flasher = board['flasher']
- return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --write {firmware}.elf --go')
-
-
-def reset_stlink(board):
- flasher = board['flasher']
- return run_cmd(f'STM32_Programmer_CLI --connect port=swd sn={flasher["uid"]} --rst --go')
-
-def flash_stflash(board, firmware):
- flasher = board['flasher']
- ret = run_cmd(f'st-flash --serial {flasher["uid"]} write {firmware}.bin 0x8000000')
- return ret
-
-
-def reset_stflash(board):
- flasher = board['flasher']
- return subprocess.CompletedProcess(args=['dummy'], returncode=0)
-
-
-def flash_openocd(board, firmware):
- flasher = board['flasher']
- ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
- f'{flasher["args"]} -c "init; halt; program {firmware}.elf verify; reset; exit"')
- return ret
-
-
-def reset_openocd(board):
- flasher = board['flasher']
- ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "adapter serial {flasher["uid"]}" '
- f'{flasher["args"]} -c "init; reset run; exit"')
- return ret
-
-
-def flash_openocd_wch(board, firmware):
- flasher = board['flasher']
- ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "telnet_port disabled" '
- f'-c "adapter serial {flasher["uid"]}" {flasher.get("args", "")} -c "program {firmware}.elf reset exit"')
- return ret
-
-
-def reset_openocd_wch(board):
- flasher = board['flasher']
- ret = run_cmd(f'openocd -c "tcl_port disabled" -c "gdb_port disabled" -c "telnet_port disabled" '
- f'-c "adapter serial {flasher["uid"]}" {flasher.get("args", "")} -c "init; reset run; exit"')
- return ret
-
-
-def flash_openocd_adi(board: Board, firmware: str) -> subprocess.CompletedProcess:
- flasher = board['flasher']
- openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
- tcl_dir = OPENCOD_ADI_PATH / 'tcl'
- ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
- f'{flasher["args"]} -c "program {firmware}.elf reset exit"')
- return ret
-
-
-def reset_openocd_adi(board: Board) -> subprocess.CompletedProcess:
- flasher = board['flasher']
- openocd = OPENCOD_ADI_PATH / 'src' / 'openocd'
- tcl_dir = OPENCOD_ADI_PATH / 'tcl'
- ret = run_cmd(f'{openocd} -c "adapter serial {flasher["uid"]}" -s {tcl_dir} '
- f'{flasher["args"]} -c "program reset exit"')
- return ret
-
-
-def flash_wlink_rs(board, firmware):
- flasher = board['flasher']
- # wlink use index for probe selection and lacking usb serial support
- ret = run_cmd(f'wlink flash {firmware}.elf')
- return ret
-
-
-def reset_wlink_rs(board):
- flasher = board['flasher']
- # wlink use index for probe selection and lacking usb serial support
- ret = run_cmd(f'wlink reset')
- return ret
-
-
-def flash_esptool(board: Board, firmware: str) -> subprocess.CompletedProcess:
- flasher = board['flasher']
- port = get_serial_dev(flasher["uid"], None, None, 0)
- fw_dir = Path(f'{firmware}.bin').parent
- with (fw_dir / 'config.env').open() as f:
- idf_target = json.load(f)['IDF_TARGET']
- with (fw_dir / 'flash_args').open() as f:
- flash_args = f.read().strip().replace('\n', ' ')
- command = (f'esptool --chip {idf_target} -p {port} {flasher["args"]} '
- f'--before=default_reset --after=hard_reset write_flash {flash_args}')
- ret = run_cmd(command, cwd=str(fw_dir))
- return ret
-
-
-def reset_esptool(board):
- flasher = board['flasher']
- return subprocess.CompletedProcess(args=['dummy'], returncode=0)
-
-
-def flash_uniflash(board, firmware):
- flasher = board['flasher']
- ret = run_cmd(f'dslite.sh {flasher["args"]} -f {firmware}.hex')
- return ret
-
-
-def reset_uniflash(board):
- flasher = board['flasher']
- return subprocess.CompletedProcess(args=['dummy'], returncode=0)
-
-
-def flash_lm4flash(board, firmware):
- # TI Tiva-C / Stellaris ICDI: lightweight lm4flash, resets and runs after write
- flasher = board['flasher']
- ret = run_cmd(f'lm4flash -s {flasher["uid"]} {flasher["args"]} {firmware}.bin')
- return ret
-
-
-def reset_lm4flash(board):
- # lm4flash has no reset-only mode; it resets+runs on flash, so reset is a no-op
- flasher = board['flasher']
- return subprocess.CompletedProcess(args=['dummy'], returncode=0)
-
-
-# -------------------------------------------------------------
# Tests: dual
# -------------------------------------------------------------
def test_dual_host_info_to_device_cdc(board):
uid = board['uid']
declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
- port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
+ port = hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
ser = open_serial_dev(port)
ser.timeout = 0.1
@@ -785,12 +382,12 @@
flasher = board['flasher']
declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
- port = get_serial_dev(flasher["uid"], None, None, 0)
+ port = hil_flash.get_serial_dev(flasher["uid"], None, None, 0)
ser = open_serial_dev(port)
ser.timeout = 0.1
# reset device since we can miss the first line
- ret = globals()[f'reset_{flasher["name"].lower()}'](board)
+ ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
assert ret.returncode == 0, 'Failed to reset device'
# read until all expected devices are enumerated
@@ -864,12 +461,12 @@
if not cdc_devs and not msc_devs:
return 'skipped'
- port = get_serial_dev(flasher["uid"], None, None, 0)
+ port = hil_flash.get_serial_dev(flasher["uid"], None, None, 0)
ser = open_serial_dev(port)
ser.timeout = 0.1
# reset device to catch mount messages
- ret = globals()[f'reset_{flasher["name"].lower()}'](board)
+ ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
assert ret.returncode == 0, 'Failed to reset device'
# Wait for all expected mount messages
@@ -957,12 +554,12 @@
if not msc_devs:
return 'skipped'
- port = get_serial_dev(flasher["uid"], None, None, 0)
+ port = hil_flash.get_serial_dev(flasher["uid"], None, None, 0)
ser = open_serial_dev(port)
ser.timeout = 0.1
# reset device to catch mount messages
- ret = globals()[f'reset_{flasher["name"].lower()}'](board)
+ ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
assert ret.returncode == 0, 'Failed to reset device'
# Wait for MSC mount (Disk Size message)
@@ -1051,8 +648,8 @@
def test_device_cdc_dual_ports(board):
uid = board['uid']
port = [
- get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0),
- get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 2)
+ hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0),
+ hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 2)
]
ser = [open_serial_dev(p) for p in port]
@@ -1091,7 +688,7 @@
def test_device_cdc_msc(board):
uid = board['uid']
# CDC Echo test
- port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
+ port = hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
ser = open_serial_dev(port)
def rand_ascii(length):
@@ -1140,7 +737,7 @@
assert timeout > 0, f'Disk {dev} not found'
# Wait for CDC tty enumeration
- tty = get_serial_dev(uid, 'TinyUSB', 'Throughput', 0)
+ tty = hil_flash.get_serial_dev(uid, 'TinyUSB', 'Throughput', 0)
timeout = enum_timeout()
while timeout > 0:
if os.path.exists(tty):
@@ -1159,8 +756,8 @@
pass
# Put tty in raw mode so dd sees pure binary throughput.
- rs = run_cmd(f'timeout 30 stty -F {tty} raw -echo')
- assert rs.returncode == 0, f'stty failed: {cmd_stdout_text(rs.stdout)}'
+ rs = hil_flash.run_cmd(f'timeout 30 stty -F {tty} raw -echo')
+ assert rs.returncode == 0, f'stty failed: {hil_flash.cmd_stdout_text(rs.stdout)}'
# Payload aim: ~5 s per direction at FS (~830 kB/s), much less at HS.
msc_count = 2 if is_fs else 16 # bs=1M
@@ -1168,21 +765,21 @@
tmp_file = f'/tmp/cdc_msc_tp_{uid}.bin'
- rw = run_cmd(f'timeout 30 dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1')
- assert rw.returncode == 0, f'CDC dd write failed: {cmd_stdout_text(rw.stdout)}'
- cdc_w = parse_speed(cmd_stdout_text(rw.stdout))
+ rw = hil_flash.run_cmd(f'timeout 30 dd if=/dev/zero of={tty} bs=64K count={cdc_count} 2>&1')
+ assert rw.returncode == 0, f'CDC dd write failed: {hil_flash.cmd_stdout_text(rw.stdout)}'
+ cdc_w = parse_speed(hil_flash.cmd_stdout_text(rw.stdout))
- rr = run_cmd(f'timeout 30 dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1')
- assert rr.returncode == 0, f'CDC dd read failed: {cmd_stdout_text(rr.stdout)}'
- cdc_r = parse_speed(cmd_stdout_text(rr.stdout))
+ rr = hil_flash.run_cmd(f'timeout 30 dd if={tty} of=/dev/null bs=64K count={cdc_count} iflag=fullblock 2>&1')
+ assert rr.returncode == 0, f'CDC dd read failed: {hil_flash.cmd_stdout_text(rr.stdout)}'
+ cdc_r = parse_speed(hil_flash.cmd_stdout_text(rr.stdout))
- rmr = run_cmd(f'dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1')
- assert rmr.returncode == 0, f'MSC dd read failed: {cmd_stdout_text(rmr.stdout)}'
- msc_r = parse_speed(cmd_stdout_text(rmr.stdout))
+ rmr = hil_flash.run_cmd(f'dd if={dev} of={tmp_file} bs=1M count={msc_count} iflag=direct 2>&1')
+ assert rmr.returncode == 0, f'MSC dd read failed: {hil_flash.cmd_stdout_text(rmr.stdout)}'
+ msc_r = parse_speed(hil_flash.cmd_stdout_text(rmr.stdout))
- rmw = run_cmd(f'dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1')
- assert rmw.returncode == 0, f'MSC dd write failed: {cmd_stdout_text(rmw.stdout)}'
- msc_w = parse_speed(cmd_stdout_text(rmw.stdout))
+ rmw = hil_flash.run_cmd(f'dd if={tmp_file} of={dev} bs=1M count={msc_count} oflag=direct 2>&1')
+ assert rmw.returncode == 0, f'MSC dd write failed: {hil_flash.cmd_stdout_text(rmw.stdout)}'
+ msc_w = parse_speed(hil_flash.cmd_stdout_text(rmw.stdout))
try:
os.remove(tmp_file)
@@ -1213,8 +810,8 @@
deadline = time.monotonic() + enum_timeout()
found = False
while time.monotonic() < deadline:
- ret = run_cmd(f'dfu-util -l')
- stdout = cmd_stdout_text(ret.stdout)
+ ret = hil_flash.run_cmd(f'dfu-util -l')
+ stdout = hil_flash.cmd_stdout_text(ret.stdout)
if f'serial="{uid}"' in stdout and 'Found DFU: [cafe:400b]' in stdout:
found = True
break
@@ -1232,10 +829,10 @@
except OSError:
pass
- ret = run_cmd(f'dfu-util -S {uid} -a 0 -U {f_dfu0}')
+ ret = hil_flash.run_cmd(f'dfu-util -S {uid} -a 0 -U {f_dfu0}')
assert ret.returncode == 0, 'Upload failed'
- ret = run_cmd(f'dfu-util -S {uid} -a 1 -U {f_dfu1}')
+ ret = hil_flash.run_cmd(f'dfu-util -S {uid} -a 1 -U {f_dfu1}')
assert ret.returncode == 0, 'Upload failed'
with open(f_dfu0) as f:
@@ -1254,8 +851,8 @@
deadline = time.monotonic() + enum_timeout()
found = False
while time.monotonic() < deadline:
- ret = run_cmd(f'dfu-util -l')
- stdout = cmd_stdout_text(ret.stdout)
+ ret = hil_flash.run_cmd(f'dfu-util -l')
+ stdout = hil_flash.cmd_stdout_text(ret.stdout)
if f'serial="{uid}"' in stdout and 'Found Runtime: [cafe:400c]' in stdout:
found = True
break
@@ -1291,7 +888,7 @@
uid = board['uid']
# Wait for CDC port and printer device
- cdc_port = get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
+ cdc_port = hil_flash.get_serial_dev(uid, 'TinyUSB', "TinyUSB_Device", 0)
ser = open_serial_dev(cdc_port)
lp_dev = open_printer_dev(uid, 'TinyUSB', 'TinyUSB_Device', 2)
@@ -1731,15 +1328,15 @@
# its normal driver. usbtest_permit budgets USBTEST_PARALLEL batteries per controller.
script = Path(__file__).resolve().parent / 'usbtest.py'
cmd = f'python3 "{script}" --serial "{uid}" --json --keep-binding --timeout 60'
- with usbtest_permit(uid):
- r = run_cmd(cmd, timeout=200)
- out = cmd_stdout_text(r.stdout)
+ with hil_lock.usbtest_permit(uid):
+ r = hil_flash.run_cmd(cmd, timeout=200)
+ out = hil_flash.cmd_stdout_text(r.stdout)
brace = out.find('{')
try:
data = json.loads(out[brace:])
passed, failed = int(data['passed']), int(data['failed'])
except (ValueError, KeyError, json.JSONDecodeError):
- raise TestFail(f'usbtest did not run: {compact_output(out) or cmd_stdout_text(r.stderr)}',
+ raise TestFail(f'usbtest did not run: {compact_output(out) or hil_flash.cmd_stdout_text(r.stderr)}',
metric=f'{REPORT_CELL["fail"]} 0/30')
total = passed + failed
@@ -1788,21 +1385,6 @@
]
-def find_firmware(variant: str, example: str):
- """Locate a built example's firmware base path (no extension) under
- cmake-build-<variant>/<example>/. Accepts the single-config layout (firmware
- directly in the example dir) or Ninja Multi-Config (a per-config subdir like
- RelWithDebInfo/). Returns the base Path, or None if not built."""
- fw_dir = TINYUSB_ROOT / build_dir / f'cmake-build-{variant}' / example
- base = Path(example).name
- if fw_dir.is_dir():
- for cand in [fw_dir / base, fw_dir / 'RelWithDebInfo' / base,
- *(p.with_suffix('') for p in sorted(fw_dir.glob(f'*/{base}.elf')))]:
- if cand.with_suffix('.elf').exists() or cand.with_suffix('.bin').exists():
- return cand
- return None
-
-
def test_example(board: Board, variant: str, example: str) -> tuple[int, str, str | None]:
"""
Test example firmware
@@ -1820,7 +1402,7 @@
test_name = f'{variant:40} {example:30} ...'
- fw_name = find_firmware(variant, example)
+ fw_name = hil_flash.find_firmware(variant, example)
if fw_name is None:
log_line(f'{test_name} Skip (no binary)')
return 0, 'skip', None
@@ -1840,9 +1422,9 @@
attempt_out = io.StringIO()
with redirect_stdout(attempt_out):
if not skip_flash:
- with flash_permit(board['uid']):
+ with hil_lock.flash_permit(board['uid']):
t_flash = time.monotonic()
- ret = globals()[f'flash_{board["flasher"]["name"].lower()}'](board, str(fw_name))
+ ret = getattr(hil_flash, f'flash_{board["flasher"]["name"].lower()}')(board, str(fw_name))
if PROFILE:
log_line(f'[prof] {variant} {example} flash attempt {i + 1}: '
f'{time.monotonic() - t_flash:.1f}s rc={ret.returncode}')
@@ -1917,7 +1499,7 @@
failed = 0
for v in variants:
- cmd = [sys.executable, str(TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name]
+ cmd = [sys.executable, str(hil_flash.TINYUSB_ROOT / 'tools' / 'build.py'), '-b', name]
for d in extra_defs:
cmd += ['-D', d]
if v['name'] != name:
@@ -1929,7 +1511,7 @@
if verbose:
cmd.append('-v')
print(f' + {" ".join(cmd)}')
- r = subprocess.run(cmd, cwd=TINYUSB_ROOT)
+ r = subprocess.run(cmd, cwd=hil_flash.TINYUSB_ROOT)
if r.returncode != 0:
failed += 1
return name, failed
@@ -1940,7 +1522,7 @@
flasher = board['flasher']
try:
- _lock_fh = acquire_board_lock(name)
+ _lock_fh = hil_lock.acquire_board_lock(name)
except RuntimeError as e:
log_line(f'{name:25} {STATUS_FAILED}: {e}')
# visible report row so the ❌ matches the exit code; failed-tests stays
@@ -2033,7 +1615,7 @@
try:
# clear our pid record before dropping the flock: this worker
# process lives on (pool reuse), so a stale record would make
- # board_lock.py's pid-liveness checks report a freed board as
+ # hil_lock.py's pid-liveness checks report a freed board as
# still locked for the rest of the run
_lock_fh.truncate(0)
except OSError:
@@ -2174,7 +1756,6 @@
global verbose
global test_only
global board_test
- global build_dir
global max_retry
global skip_flash
@@ -2204,13 +1785,14 @@
config_file = Path(args.config_file)
boards = args.board
verbose = args.verbose
+ hil_flash.verbose = args.verbose
test_only = args.test_only
for entry in args.board_test:
bname, _, tnames = entry.partition(':')
if not bname or not tnames:
parser.error(f'invalid --board-test value: {entry!r} (expected BOARD:test1,test2)')
board_test[bname] = [t for t in tnames.split(',') if t]
- build_dir = args.build_dir
+ hil_flash.build_dir = args.build_dir
max_retry = args.retry
skip_flash = args.skip_flash
@@ -2234,8 +1816,8 @@
build_err = 0
if args.build:
- if build_dir != 'cmake-build':
- print(f'warning: --build writes into cmake-build/, but -B is {build_dir!r}; '
+ if hil_flash.build_dir != 'cmake-build':
+ print(f'warning: --build writes into cmake-build/, but -B is {hil_flash.build_dir!r}; '
f'tests will not find the freshly built firmware')
print('-' * 30)
print(f'Build phase: {len(config_boards)} board(s)')
@@ -2264,7 +1846,7 @@
seed = os.getenv('HIL_SHUFFLE_SEED') or str(int(time.time()))
log_line(f'test-order shuffle seed: {seed} (HIL_SHUFFLE_SEED={seed} to replay); '
- f'flash/usbtest parallel per controller: {FLASH_PARALLEL}/{USBTEST_PARALLEL}; '
+ f'flash/usbtest parallel per controller: {hil_lock.FLASH_PARALLEL}/{hil_lock.USBTEST_PARALLEL}; '
f'enum timeout first/retry: {ENUM_TIMEOUT}/{ENUM_TIMEOUT_RETRY}s')
hints = {}
@@ -2283,8 +1865,8 @@
mgr = Manager()
cmap = mgr.dict()
initargs = (Lock(), seed,
- [Semaphore(USBTEST_PARALLEL) for _ in range(CONTROLLER_SLOTS)],
- [Semaphore(FLASH_PARALLEL) for _ in range(CONTROLLER_SLOTS)],
+ [Semaphore(hil_lock.USBTEST_PARALLEL) for _ in range(hil_lock.CONTROLLER_SLOTS)],
+ [Semaphore(hil_lock.FLASH_PARALLEL) for _ in range(hil_lock.CONTROLLER_SLOTS)],
cmap, Lock(), hints_by_uid)
with Pool(processes=os.cpu_count() or 1, initializer=init_worker, initargs=initargs) as pool:
async_ret = pool.map_async(test_board, config_boards)
diff --git a/test/hil/tinyusb.json b/test/hil/tinyusb.json
index 8316dbc..8f321ba 100644
--- a/test/hil/tinyusb.json
+++ b/test/hil/tinyusb.json
@@ -464,7 +464,7 @@
},
"flasher": {
"name": "openocd_wch",
- "uid": "EBCA8F0670AF",
+ "uid": "A76D8F062C2A",
"args": "-f target/wch-riscv.cfg"
}
},
@@ -513,32 +513,13 @@
},
"flasher": {
"name": "openocd_wch",
- "uid": "7FD88F0604B5",
+ "uid": "57468F06DC03",
"args": "-f target/wch-riscv.cfg"
}
},
{
- "name": "nrf5340dk",
- "uid": "78E60E166B5F88BE",
- "tests": {
- "device": true,
- "host": false,
- "dual": false,
- "skip": ["device/cdc_msc_freertos", "device/audio_test_freertos"],
- "comment": "board new to HIL: FreeRTOS examples hardfault (UFSR=INVPC) at first task launch on the CM33_NTZ port - pre-existing upstream issue, non-FreeRTOS examples and usbtest pass; fix separately"
- },
- "flasher": {
- "name": "jlink",
- "uid": "001050076405",
- "args": "-device NRF5340_XXAA_APP"
- }
- }
- ],
- "boards-skip": [
- {
"name": "mimxrt1064_evk",
"uid": "BAE96FB95AFA6DBB8F00005002001200",
- "comment-skip": "device-port cable degraded from enum drops to killing the uPD720201 mid-battery (2026-07-17); replace the cable, verify enum, then move back",
"tests": {
"device": true,
"host": true,
@@ -568,7 +549,6 @@
{
"name": "nrf54lm20dk",
"uid": "899C3DE5B0F4D5CA",
- "comment-skip": "J-Link probe fails most flashes (2026-07-16); replug/repair the probe, then move back",
"tests": {
"device": true,
"host": false,
@@ -597,7 +577,9 @@
"uid": "000831915224",
"args": "-device R7FA6M5BH"
}
- },
+ }
+ ],
+ "boards-skip": [
{
"name": "ra8m1_ek",
"uid": "797D142D36345030364E1737922E4B4E",
