feat: Claude Code multi-agent dev/test harness for TinyUSB

Add worker agents (builder, port-dev, driver-reviewer, hil-operator,
pr-monitor), deterministic workflows (validate, fanout-dev, driver-review,
hil-validate, full-check, pr-babysit) and a /pre-pr gate skill, so sessions
can fan build/test/review/PR-triage work out to tiered subagents. pr-babysit
drives a PR to green: triage CI + bot reviews, fix validated findings, verify,
push, and reply-to + resolve each inline review thread (fixed or refuted).

Replace the stop-the-runner HIL discipline with per-board flock locks:
test/hil/board_lock.py plus a fail-open guard in hil_test.py let CI and dev
sessions share the rig per board (locked boards fail fast and re-run;
HIL_NO_BOARD_LOCK=1 is a user-authorized bypass). The actions-runner is
never stopped.

Design spec, implementation plan, and real-rig smoke evidence under
docs/superpowers/.

Co-Authored-By: Claude Fable 5 <[email protected]>
Claude-Session: https://claude.ai/code/session_01Rn1AN5DsTdFhRwhugfgKZi

2 files changed, 293 insertions, 51 deletions

diff --git a/test/hil/board_lock.py b/test/hil/board_lock.py
new file mode 100755
index 000000000..feee2b849
--- /dev/null
+++ b/test/hil/board_lock.py
@@ -0,0 +1,192 @@
+#!/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, so stale locks are
+impossible (/tmp also clears on reboot).
+
+Usage:
+  board_lock.py hold BOARD [BOARD...] --reason TEXT
+  board_lock.py hold --all [--config test/hil/tinyusb.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 some live process currently holds the flock."""
+    path = lock_path(board)
+    if not os.path.exists(path):
+        return False
+    with open(path) as f:
+        try:
+            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            fcntl.flock(f, fcntl.LOCK_UN)
+            return False
+        except OSError:
+            return True
+
+
+def cmd_hold(boards, reason):
+    os.makedirs(LOCK_DIR, exist_ok=True)
+    already = [b for b in boards if is_locked(b)]
+    if already:
+        for b in already:
+            print(f'ERROR: {b} already locked: {read_info(b)}', file=sys.stderr)
+        return 1
+    # 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
+        print('ERROR: holder failed to acquire locks (lost a race?)', 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)
+    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)
+    signal.signal(signal.SIGTERM, lambda *_: os._exit(0))
+    while True:
+        signal.pause()
+
+
+def cmd_release(boards):
+    pids = set()
+    for b in boards:
+        if not is_locked(b):
+            continue
+        info = read_info(b) or {}
+        if info.get('pid'):
+            pids.add(info['pid'])
+    for holder in sorted(pids):
+        try:
+            os.kill(holder, signal.SIGTERM)
+            print(f'released holder pid {holder}')
+        except ProcessLookupError:
+            pass
+    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 0
+
+
+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='test/hil/tinyusb.json')
+    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_test.py b/test/hil/hil_test.py
index e66c86e56..979c784ab 100755
--- a/test/hil/hil_test.py
+++ b/test/hil/hil_test.py
@@ -58,6 +58,47 @@ import ctypes
 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:
+        return None  # odd lock dir (perms, path collision): proceed unlocked
+    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
+
+
 ENUM_TIMEOUT = 15
 
 STATUS_OK = "\033[32mOK\033[0m"
@@ -1659,63 +1700,72 @@ def test_board(board: Board) -> tuple[str, int, list[str], list]:
     name = board['name']
     flasher = board['flasher']
 
-    # default to all tests
-    test_list = []
+    try:
+        _lock_fh = acquire_board_lock(name)
+    except RuntimeError as e:
+        log_line(f'{name:25} {STATUS_FAILED}: {e}')
+        return name, 1, [], []
+    try:
+        # default to all tests
+        test_list = []
 
-    if name in board_test:
-        test_list = board_test[name]
-    elif len(test_only) > 0:
-        # Explicit -t: filter against the board's capabilities so a device-only
-        # board doesn't try to run host/dual tests (the test functions need a
-        # `dev_attached` entry in the board config that won't exist).
-        board_tests = board.get('tests', {})
-        if 'only' in board_tests:
-            allowed = set(board_tests['only'])
-            test_list = [t for t in test_only if t in allowed]
-        else:
-            for t in test_only:
-                category = t.split('/', 1)[0]
-                if board_tests.get(category) is True:
-                    test_list.append(t)
-    else:
-        if 'tests' in board:
-            board_tests = board['tests']
-            if board_tests.get('device') is True:
-                test_list += list(device_tests)
-            if board_tests.get('dual') is True:
-                test_list += dual_tests
-            if board_tests.get('host') is True:
-                test_list += host_test
+        if name in board_test:
+            test_list = board_test[name]
+        elif len(test_only) > 0:
+            # Explicit -t: filter against the board's capabilities so a device-only
+            # board doesn't try to run host/dual tests (the test functions need a
+            # `dev_attached` entry in the board config that won't exist).
+            board_tests = board.get('tests', {})
             if 'only' in board_tests:
-                test_list = board_tests['only']
-            if 'skip' in board_tests:
-                for skip in board_tests['skip']:
-                    if skip in test_list:
-                        test_list.remove(skip)
-                        log_line(f'{name:25} {skip:30} ... Skip')
+                allowed = set(board_tests['only'])
+                test_list = [t for t in test_only if t in allowed]
+            else:
+                for t in test_only:
+                    category = t.split('/', 1)[0]
+                    if board_tests.get(category) is True:
+                        test_list.append(t)
+        else:
+            if 'tests' in board:
+                board_tests = board['tests']
+                if board_tests.get('device') is True:
+                    test_list += list(device_tests)
+                if board_tests.get('dual') is True:
+                    test_list += dual_tests
+                if board_tests.get('host') is True:
+                    test_list += host_test
+                if 'only' in board_tests:
+                    test_list = board_tests['only']
+                if 'skip' in board_tests:
+                    for skip in board_tests['skip']:
+                        if skip in test_list:
+                            test_list.remove(skip)
+                            log_line(f'{name:25} {skip:30} ... Skip')
 
-    err_count = 0
-    failed_tests = []
-    rows = []  # list of (row_label, {example: status}) — one row per build variant
-    variants = board.get('variant') or [{'name': name, 'flags': ''}]
+        err_count = 0
+        failed_tests = []
+        rows = []  # list of (row_label, {example: status}) — one row per build variant
+        variants = board.get('variant') or [{'name': name, 'flags': ''}]
 
-    for v in variants:
-        vname = v['name']
-        cells = {}
-        for test in test_list:
-            ec, status, metric = test_example(board, vname, test)
-            err_count += ec
-            cells[test] = metric if metric else status
-            if ec > 0:
-                failed_tests.append(test)
-        rows.append((vname, cells))
+        for v in variants:
+            vname = v['name']
+            cells = {}
+            for test in test_list:
+                ec, status, metric = test_example(board, vname, test)
+                err_count += ec
+                cells[test] = metric if metric else status
+                if ec > 0:
+                    failed_tests.append(test)
+            rows.append((vname, cells))
 
-    # flash board_test last to disable board's usb (skipped when --skip-flash is set);
-    # this is teardown/park, not a test — not recorded in the report
-    if not skip_flash:
-        test_example(board, variants[0]['name'], 'device/board_test')
+        # flash board_test last to disable board's usb (skipped when --skip-flash is set);
+        # this is teardown/park, not a test — not recorded in the report
+        if not skip_flash:
+            test_example(board, variants[0]['name'], 'device/board_test')
 
-    return name, err_count, sorted(set(failed_tests)), rows
+        return name, err_count, sorted(set(failed_tests)), rows
+    finally:
+        if _lock_fh:
+            _lock_fh.close()
 
 
 REPORT_MD = 'hil_report.md'