feat(bench): add firmware gate so the suite knows what it is testing

Today's runs executed against whatever firmware happened to be on the
board (an old April build at first) and nothing checked. For a
qualification tool that is unacceptable: before any test starts the suite
now establishes, and can control, exactly which build it is testing.

New px4bench/firmware.py: .px4 metadata parsing and validation (fields
verified against Tools/px_mkfw.py and px4_uploader.py), board identity
via ver all over the nsh shell, flashing through Tools/px4_uploader.py
with streamed output and a hard timeout, HW-arch to build-target
inference by enumerating boards/, an early wrong-board check against
firmware.prototype board_id, and the named firmware_identity check
(git-hash prefix match against the artifact).

run_bench_suite.py preflight supports four firmware sources, all
converging on the same flash + verify path: keep what is on the board
(--any-firmware), flash a local file (--firmware), build the inferred
target from this source tree (--build, --target, --build-timeout), or
download a GitHub release artifact (--release TAG|latest, via gh;
asset naming verified against the v1.17.0 release). --expect-hash
verifies without flashing. Flags are mutually exclusive; with none
given, a TTY gets an operator menu covering the same four sources, and
automation exits with an error before touching the board. The detected
identity is printed, written to firmware.json in the suite report dir,
and stamped into every test's report dir via PX4BENCH_FIRMWARE_INFO.

If the board's mavlink is wedged the uploader cannot soft-reboot it into
the bootloader (hit live today); after repeated reboot attempts the gate
prints an operator instruction to replug USB so the uploader catches the
bootloader at power-on.

sih/flight_mission.py gains a verify-only --expect-hash and stamps the
firmware identity into its report dir; it never flashes.

Signed-off-by: Ramon Roche <mrpollo@gmail.com>

Tools/bench_test/README.md

@@ -79,20 +79,70 @@ first; it holds the serial port.
 ## Quick start
 
 ```
-# full non-interactive bench suite, single USB link
-./run_bench_suite.py /dev/tty.usbmodem01
+# flash a known build, verify it, then run the full bench suite
+./run_bench_suite.py /dev/tty.usbmodem01 --firmware build/px4_fmu-v6xrt_default/px4_fmu-v6xrt_default.px4
 
 # with a telemetry radio as a second link (enables link_forwarding and
 # alternating mission uploads)
-./run_bench_suite.py /dev/tty.usbmodem01 /dev/tty.usbserial-RADIO
+./run_bench_suite.py /dev/tty.usbmodem01 /dev/tty.usbserial-RADIO --expect-hash 0c000d59
 
 # operator-assisted USB re-enumeration test (not part of the suite)
 ./bench/usb_replug.py /dev/tty.usbmodem01
 
 # simulated flight on the FMU (reconfigures the board, run separately)
-./sih/flight_mission.py /dev/tty.usbmodem01
+./sih/flight_mission.py /dev/tty.usbmodem01 --expect-hash 0c000d59
 ```
 
+## Firmware gate
+
+Qualification means knowing what you tested. Before any test starts the
+suite connects, reads the board identity (`ver all`: PX4 git-hash, PX4
+version, HW arch, OS version), prints it, writes it to `firmware.json` in
+the suite report dir, and stamps it into every test's report dir. What
+happens next depends on exactly one of five mutually exclusive flags,
+covering four firmware sources:
+
+```
+# 1. keep what is on the board (explicit opt-out, still stamped)
+./run_bench_suite.py /dev/tty.usbmodem01 --any-firmware
+
+# 2. flash a local .px4, verify the flashed identity, then test
+./run_bench_suite.py /dev/tty.usbmodem01 --firmware path/to/px4_fmu-v6xrt_default.px4
+
+# 3. build from this source tree, flash, verify (target inferred from the
+#    connected board via its HW arch; pass --target if ambiguous)
+./run_bench_suite.py /dev/tty.usbmodem01 --build
+
+# 4. download a GitHub release artifact (needs the gh CLI), flash, verify
+./run_bench_suite.py /dev/tty.usbmodem01 --release v1.17.0
+./run_bench_suite.py /dev/tty.usbmodem01 --release latest
+
+# verify only, no flash: assert the board already runs a given hash
+./run_bench_suite.py /dev/tty.usbmodem01 --expect-hash 0c000d59
+```
+
+All flash paths converge: parse the .px4 metadata (refusing unparseable
+files), check its `board_id` against the connected board early (wrong-board
+images fail before flashing; the uploader enforces it again against the
+bootloader), flash via `Tools/px4_uploader.py`, wait for re-enumeration and
+a heartbeat, then re-read `ver all` and compare the git hash against the
+artifact's identity (named check `firmware_identity`, prefix match). A
+mismatch after flashing aborts the suite.
+
+Interactive use: with none of the flags on a TTY, the suite shows the
+detected identity and asks: continue on the current firmware, flash a local
+.px4, build the inferred target from this tree, download a release, or
+abort. In automation (stdin not a TTY) the gate refuses to guess and exits
+with an error listing the flags; CI must always state what it is testing.
+
+Wedged-board note: a board whose mavlink is hung cannot soft-reboot into
+the bootloader, so the uploader sits in its reboot-request loop. The gate
+detects this and prints an operator instruction to unplug and replug USB;
+the uploader then catches the bootloader at power-on.
+
+`sih/flight_mission.py` accepts `--expect-hash` as a verify-only gate and
+stamps the identity into its report dir; it never flashes.
+
 ## Why pymavlink
 
 The bench tests exercise the raw MAVLink protocol surface on purpose: param

Tools/bench_test/px4bench/__init__.py

@@ -109,11 +109,26 @@ class Reporter:
         return 0 if failed == 0 else 1
 
 
+FIRMWARE_INFO_ENV = 'PX4BENCH_FIRMWARE_INFO'
+
+
 def make_report_dir(base='bench_reports', test_name=''):
-    """Create and return a timestamped report directory."""
+    """Create and return a timestamped report directory.
+
+    When the suite preflight has established the firmware identity it
+    exports it via PX4BENCH_FIRMWARE_INFO (a JSON string); stamp it into
+    every report dir so each test result is traceable to a build.
+    """
     stamp = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%SZ')
     path = os.path.join(base, '{}_{}'.format(stamp, test_name) if test_name else stamp)
     os.makedirs(path, exist_ok=True)
+    info = os.environ.get(FIRMWARE_INFO_ENV)
+    if info:
+        try:
+            with open(os.path.join(path, 'firmware.json'), 'w') as f:
+                f.write(info.rstrip('\n') + '\n')
+        except OSError:
+            pass
     return path
 
 
@@ -361,16 +376,17 @@ def wait_device_back(device, timeout=45, pattern=None):
     return None, None
 
 
-def reboot_and_reconnect(mav, conn_str, baud=DEFAULT_BAUD, timeout=60):
-    """Reboot the autopilot and reconnect.
+def wait_reconnect(conn_str, baud=DEFAULT_BAUD, timeout=60, start=None):
+    """Wait for a rebooted/re-enumerated board and reconnect.
 
-    Returns (new_mav_connection, elapsed_seconds).
-    Raises TimeoutError with a description of what stalled.
+    For a serial device: waits for the node to vanish and return (the node
+    name may change across enumeration), lets the CDC ACM interface settle,
+    then retries connect() until a heartbeat arrives. Returns
+    (new_mav_connection, elapsed_seconds). Raises TimeoutError with a
+    description of what stalled.
     """
-    start = time.monotonic()
-    send_reboot(mav)
-    time.sleep(0.5)
-    mav.close()
+    if start is None:
+        start = time.monotonic()
 
     if is_serial_device(conn_str):
         gone = wait_device_gone(conn_str, timeout=15)
@@ -399,6 +415,19 @@ def reboot_and_reconnect(mav, conn_str, baud=DEFAULT_BAUD, timeout=60):
         timeout, last_err))
 
 
+def reboot_and_reconnect(mav, conn_str, baud=DEFAULT_BAUD, timeout=60):
+    """Reboot the autopilot and reconnect.
+
+    Returns (new_mav_connection, elapsed_seconds).
+    Raises TimeoutError with a description of what stalled.
+    """
+    start = time.monotonic()
+    send_reboot(mav)
+    time.sleep(0.5)
+    mav.close()
+    return wait_reconnect(conn_str, baud=baud, timeout=timeout, start=start)
+
+
 def add_connection_args(parser, dual_link=False):
     """Standard CLI surface shared by every test in the suite."""
     parser.add_argument('connection',

Tools/bench_test/px4bench/firmware.py

@@ -0,0 +1,393 @@
+"""Firmware gate: know exactly which firmware is on the board under test.
+
+A qualification run against an unknown build is worthless. These helpers
+let the suite establish the board's identity (`ver all` over the nsh
+shell), parse and validate .px4 firmware files, flash via
+Tools/px4_uploader.py, build a target from this source tree, download a
+released artifact from GitHub, and verify the flashed identity against the
+expectation.
+
+Verified facts this module relies on:
+- .px4 files are JSON with board_id / version / git_identity written by
+  Tools/px_mkfw.py (lines 51-101); recent builds also carry the full
+  git_hash. Tools/px4_uploader.py requires image / board_id / image_size /
+  image_maxsize (line 354) and compares board_id against the bootloader's
+  board_type.
+- `ver all` prints 'PX4 git-hash:', 'PX4 version:', 'HW arch:', and
+  'OS version:' lines (src/systemcmds/ver/ver.cpp:164-250).
+- HW arch is the build target uppercased with '-' replaced by '_'
+  (cmake/px4_config.cmake:108-109), so the reverse mapping is resolved by
+  enumerating boards/<vendor>/<model>.
+- boards/<vendor>/<model>/firmware.prototype carries the numeric board_id,
+  enabling a wrong-board check before flashing.
+- Release assets are named <vendor>_<model>_default.px4 (verified against
+  the v1.17.0 release listing).
+"""
+
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import threading
+import time
+
+from . import (DEFAULT_BAUD, MavlinkShell, is_serial_device, wait_reconnect)
+
+GITHUB_REPO = 'PX4/PX4-Autopilot'
+FLASH_TIMEOUT_S = 600.0
+BUILD_TIMEOUT_S = 1800.0
+DOWNLOAD_TIMEOUT_S = 300.0
+REBOOT_HINT_AFTER_ATTEMPTS = 3
+
+
+def repo_root():
+    """PX4-Autopilot checkout root, derived from this file's location."""
+    return os.path.dirname(os.path.dirname(os.path.dirname(
+        os.path.dirname(os.path.abspath(__file__)))))
+
+
+def parse_px4_metadata(path):
+    """Parse a .px4 firmware file (JSON) and validate its metadata.
+
+    Returns a dict with git_identity, git_hash (may be None on old files),
+    board_id, version, summary, description, image_size, image_maxsize.
+    Raises ValueError with a clear message if the file does not parse or
+    required fields are missing.
+    """
+    try:
+        with open(path) as f:
+            desc = json.load(f)
+    except OSError as e:
+        raise ValueError('cannot read {}: {}'.format(path, e))
+    except ValueError as e:
+        raise ValueError('{} is not a valid .px4 file (not JSON): {}'.format(path, e))
+    if not isinstance(desc, dict):
+        raise ValueError('{} is not a valid .px4 file (not a JSON object)'.format(path))
+
+    missing = [k for k in ('git_identity', 'board_id', 'version', 'image') if k not in desc]
+    if missing:
+        raise ValueError('{} is missing .px4 metadata fields: {}'.format(
+            path, ', '.join(missing)))
+
+    return {k: desc.get(k) for k in (
+        'git_identity', 'git_hash', 'board_id', 'version',
+        'summary', 'description', 'image_size', 'image_maxsize')}
+
+
+def expected_hash_from_metadata(meta):
+    """Best hash expectation from .px4 metadata.
+
+    Prefer the full git_hash; fall back to the short hash after '-g' in the
+    git_identity describe string (e.g. v1.18.0-alpha1-592-g0c000d596a4).
+    """
+    if meta.get('git_hash'):
+        return str(meta['git_hash'])
+    m = re.search(r'-g([0-9a-fA-F]+)$', str(meta.get('git_identity') or ''))
+    if m:
+        return m.group(1)
+    return str(meta.get('git_identity') or '')
+
+
+def hashes_match(board_hash, expected_prefix):
+    """Prefix comparison in either direction (metadata hashes are short)."""
+    a = str(board_hash).strip().lower()
+    b = str(expected_prefix).strip().lower()
+    if not a or not b:
+        return False
+    return a.startswith(b) or b.startswith(a)
+
+
+def board_identity(mav, timeout=15):
+    """Read the board's identity via `ver all` over the nsh shell.
+
+    Returns (identity_dict, None) or (None, error_string). The dict has
+    git_hash, version, hw_arch, os_version (whichever parsed) plus the raw
+    output.
+    """
+    shell = MavlinkShell(mav)
+    if not shell.open(timeout=5):
+        return None, 'nsh shell did not respond within 5s'
+    out, timed_out = shell.run('ver all', timeout=timeout)
+    shell.close()
+    if timed_out:
+        return None, "'ver all' stalled (no completion within {}s)".format(timeout)
+
+    ident = {'raw': out}
+    prefixes = (('PX4 git-hash:', 'git_hash'),
+                ('PX4 version:', 'version'),
+                ('HW arch:', 'hw_arch'),
+                ('OS version:', 'os_version'))
+    for line in out.splitlines():
+        stripped = line.strip()
+        for prefix, key in prefixes:
+            if stripped.startswith(prefix) and key not in ident:
+                ident[key] = stripped[len(prefix):].strip()
+    if 'git_hash' not in ident:
+        return None, "could not parse 'PX4 git-hash:' from ver all output " \
+                     '(got {} bytes)'.format(len(out))
+    return ident, None
+
+
+def format_identity(ident):
+    """One-line-per-field operator-facing summary of a board identity."""
+    lines = []
+    for key, label in (('git_hash', 'PX4 git-hash'), ('version', 'PX4 version'),
+                       ('hw_arch', 'HW arch'), ('os_version', 'OS version')):
+        if ident.get(key):
+            lines.append('  {:<14} {}'.format(label + ':', ident[key]))
+    return '\n'.join(lines)
+
+
+def stamp_identity(report_dir, identity, extra=None):
+    """Write firmware.json into a report dir so every result is traceable."""
+    data = {k: v for k, v in identity.items() if k != 'raw'}
+    if extra:
+        data.update(extra)
+    path = os.path.join(report_dir, 'firmware.json')
+    try:
+        with open(path, 'w') as f:
+            json.dump(data, f, indent=2, sort_keys=True)
+            f.write('\n')
+    except OSError as e:
+        print('[INFO] could not write {}: {}'.format(path, e), flush=True)
+    return path
+
+
+def verify_identity(report, mav, expected_hash_prefix):
+    """Named check 'firmware_identity': board git hash vs the expectation.
+
+    Returns (ok, identity_or_None).
+    """
+    ident, err = board_identity(mav)
+    if ident is None:
+        report.fail('firmware_identity', err)
+        return False, None
+    ok = hashes_match(ident['git_hash'], expected_hash_prefix)
+    report.check('firmware_identity',
+                 ok,
+                 'board reports {} , expected {}'.format(
+                     ident['git_hash'], expected_hash_prefix))
+    return ok, ident
+
+
+def list_board_targets(root=None):
+    """Map build target -> board dir for every boards/<vendor>/<model>."""
+    root = root or repo_root()
+    targets = {}
+    boards_dir = os.path.join(root, 'boards')
+    if not os.path.isdir(boards_dir):
+        return targets
+    for vendor in sorted(os.listdir(boards_dir)):
+        vdir = os.path.join(boards_dir, vendor)
+        if not os.path.isdir(vdir):
+            continue
+        for model in sorted(os.listdir(vdir)):
+            mdir = os.path.join(vdir, model)
+            if os.path.isdir(mdir):
+                targets['{}_{}'.format(vendor, model)] = mdir
+    return targets
+
+
+def infer_build_target(hw_arch, root=None):
+    """Reverse the HW arch string to a build target by enumerating boards/.
+
+    cmake/px4_config.cmake uppercases the target and turns '-' into '_' to
+    produce the HW arch, so match by applying the same transform. Returns
+    (target, None) on a unique match, (None, reason) otherwise.
+    """
+    if not hw_arch:
+        return None, 'board did not report a HW arch'
+    matches = [t for t in list_board_targets(root)
+               if t.upper().replace('-', '_') == hw_arch.strip().upper()]
+    if len(matches) == 1:
+        return matches[0], None
+    if not matches:
+        return None, 'no boards/ entry maps to HW arch {!r}'.format(hw_arch)
+    return None, 'HW arch {!r} is ambiguous: {}'.format(hw_arch, ', '.join(matches))
+
+
+def board_id_for_target(target, root=None):
+    """Numeric board_id from boards/<vendor>/<model>/firmware.prototype."""
+    mdir = list_board_targets(root).get(target)
+    if mdir is None:
+        return None
+    proto = os.path.join(mdir, 'firmware.prototype')
+    try:
+        with open(proto) as f:
+            return json.load(f).get('board_id')
+    except (OSError, ValueError):
+        return None
+
+
+def check_board_id(meta, target, root=None):
+    """Early wrong-board check: .px4 board_id vs the connected board's target.
+
+    Returns (ok, detail). ok is None when the check cannot be performed
+    (unknown target or no prototype); the uploader still enforces board_id
+    against the bootloader at flash time.
+    """
+    if not target:
+        return None, 'board target unknown; deferring board_id check to the uploader'
+    proto_id = board_id_for_target(target, root)
+    if proto_id is None:
+        return None, 'no firmware.prototype board_id for {}; deferring to the uploader'.format(target)
+    if meta['board_id'] == proto_id:
+        return True, 'board_id {} matches {}'.format(meta['board_id'], target)
+    return False, 'firmware board_id {} does not match {} (board_id {})'.format(
+        meta['board_id'], target, proto_id)
+
+
+def _stream_uploader(proc, state, interactive):
+    """Echo uploader output; hint the operator if the reboot loop spins.
+
+    A board with a wedged mavlink cannot be soft-rebooted into the
+    bootloader (hit live on the bench); the fix is a USB replug so the
+    uploader catches the bootloader at power-on.
+    """
+    for line in proc.stdout:
+        line = line.rstrip('\n')
+        print('  [uploader] {}'.format(line), flush=True)
+        if 'Attempting reboot' in line:
+            state['reboot_attempts'] += 1
+            if (interactive and not state['hinted']
+                    and state['reboot_attempts'] >= REBOOT_HINT_AFTER_ATTEMPTS):
+                state['hinted'] = True
+                print('\n  OPERATOR: the board is not responding to the soft '
+                      'reboot request (wedged mavlink cannot reboot itself).\n'
+                      '  UNPLUG and REPLUG the USB cable now; the uploader '
+                      'will catch the bootloader at power-on.\n', flush=True)
+
+
+def flash_firmware(px4_path, connection, baud=DEFAULT_BAUD, interactive=True,
+                   timeout=FLASH_TIMEOUT_S, mav=None):
+    """Flash a .px4 via Tools/px4_uploader.py and reconnect.
+
+    Closes the given mav connection first (the uploader needs the port),
+    streams uploader output with an overall timeout, then waits for USB
+    re-enumeration plus a fresh heartbeat. Returns (new_mav, None) or
+    (None, error_string).
+    """
+    if not is_serial_device(connection):
+        return None, 'flashing requires a serial device connection, got {!r}'.format(connection)
+
+    if mav is not None:
+        try:
+            mav.close()
+        except Exception:
+            pass
+        time.sleep(0.5)
+
+    uploader = os.path.join(repo_root(), 'Tools', 'px4_uploader.py')
+    if not os.path.exists(uploader):
+        return None, 'uploader not found: {}'.format(uploader)
+
+    cmd = [sys.executable, uploader, px4_path,
+           '--port', connection, '--baud-flightstack', str(baud)]
+    print('[INFO] flashing: {}'.format(' '.join(cmd)), flush=True)
+
+    try:
+        proc = subprocess.Popen(cmd, stdout=subprocess.PIPE,
+                                stderr=subprocess.STDOUT, text=True)
+    except OSError as e:
+        return None, 'could not launch uploader: {}'.format(e)
+
+    state = {'reboot_attempts': 0, 'hinted': False}
+    reader = threading.Thread(target=_stream_uploader,
+                              args=(proc, state, interactive), daemon=True)
+    reader.start()
+    try:
+        rc = proc.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        try:
+            proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            pass
+        return None, 'uploader hung (killed after {:.0f}s, {} reboot attempts seen)'.format(
+            timeout, state['reboot_attempts'])
+    reader.join(timeout=5)
+
+    if rc != 0:
+        return None, 'uploader exited with code {}'.format(rc)
+
+    print('[INFO] upload done; waiting for the board to boot', flush=True)
+    try:
+        newmav, elapsed = wait_reconnect(connection, baud=baud, timeout=90)
+    except TimeoutError as e:
+        return None, 'board did not come back after flashing: {}'.format(e)
+    print('[INFO] board back after {:.1f}s'.format(elapsed), flush=True)
+    return newmav, None
+
+
+def build_firmware(target, timeout=BUILD_TIMEOUT_S, root=None):
+    """Run `make <target>` at the repo root; return (artifact_path, None).
+
+    Output is streamed. A cold NuttX build can exceed 10 minutes, hence the
+    generous default timeout. Returns (None, error_string) on failure.
+    """
+    root = root or repo_root()
+    print('[INFO] building: make {} (in {})'.format(target, root), flush=True)
+    try:
+        proc = subprocess.Popen(['make', target], cwd=root)
+    except OSError as e:
+        return None, 'could not launch make: {}'.format(e)
+    try:
+        rc = proc.wait(timeout=timeout)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        try:
+            proc.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            pass
+        return None, 'build timed out after {:.0f}s (see --build-timeout)'.format(timeout)
+    if rc != 0:
+        return None, 'make {} failed with exit code {}'.format(target, rc)
+
+    candidates = [os.path.join(root, 'build', target, target + '.px4'),
+                  os.path.join(root, 'build', target + '_default',
+                               target + '_default.px4')]
+    for path in candidates:
+        if os.path.exists(path):
+            return path, None
+    return None, 'build succeeded but no artifact at {}'.format(' or '.join(candidates))
+
+
+def source_tree_hash(root=None):
+    """git rev-parse HEAD of this checkout, or None."""
+    try:
+        out = subprocess.run(['git', 'rev-parse', 'HEAD'],
+                             cwd=root or repo_root(), capture_output=True,
+                             text=True, timeout=10)
+    except (OSError, subprocess.TimeoutExpired):
+        return None
+    return out.stdout.strip() if out.returncode == 0 else None
+
+
+def download_release(tag, target, dest_dir, timeout=DOWNLOAD_TIMEOUT_S):
+    """Download <target>_default.px4 from a GitHub release via gh.
+
+    tag is a release tag like v1.17.0, or 'latest'. Returns
+    (px4_path, None) or (None, error_string).
+    """
+    if shutil.which('gh') is None:
+        return None, "the 'gh' CLI is required for --release (brew install gh)"
+    asset = '{}_default.px4'.format(target)
+    cmd = ['gh', 'release', 'download', '--repo', GITHUB_REPO,
+           '--pattern', asset, '--dir', dest_dir, '--clobber']
+    if tag and tag != 'latest':
+        cmd.insert(3, tag)
+    print('[INFO] downloading: {}'.format(' '.join(cmd)), flush=True)
+    try:
+        proc = subprocess.run(cmd, timeout=timeout)
+    except subprocess.TimeoutExpired:
+        return None, 'gh release download timed out after {:.0f}s'.format(timeout)
+    except OSError as e:
+        return None, 'could not launch gh: {}'.format(e)
+    if proc.returncode != 0:
+        return None, 'gh release download failed (tag {!r}, asset {!r})'.format(tag, asset)
+    path = os.path.join(dest_dir, asset)
+    if not os.path.exists(path):
+        return None, 'gh reported success but {} is missing'.format(path)
+    return path, None

Tools/bench_test/run_bench_suite.py

@@ -12,14 +12,26 @@ second link is given) the dual-link nested-send lock stress.
 Each test enforces its own timeouts, and this orchestrator wraps each in a
 --per-test-timeout so even a fully wedged child is killed and recorded as a
 FAIL naming what hung. usb_replug is interactive and is never run here.
+
+Before any test starts, the firmware gate establishes exactly which build
+is on the board (and can flash one): qualification means knowing what you
+tested. The identity is printed, written to firmware.json in the suite
+report dir, and stamped into every test's report dir.
 """
 
 import argparse
+import json
 import os
 import signal
 import subprocess
 import sys
 import time
+from typing import NoReturn
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import px4bench
+from px4bench import firmware
 
 
 # Test order for the suite. link_forwarding is inserted after
@@ -105,6 +117,236 @@ def run_one(name, args):
     return 'FAIL', dur, 'exit code {}'.format(rc)
 
 
+def gate_abort(report, code=1) -> NoReturn:
+    """Print the preflight summary and abort the suite."""
+    report.finish()
+    print('firmware gate failed; no test was run.', flush=True)
+    sys.exit(code)
+
+
+def flash_and_verify(report, args, mav, ident, px4_path, source):
+    """Common tail for every firmware source: check board_id, flash, verify.
+
+    Returns (new_mav, new_identity); aborts the suite on any failure.
+    """
+    try:
+        meta = firmware.parse_px4_metadata(px4_path)
+    except ValueError as e:
+        report.fail('firmware_file', str(e))
+        gate_abort(report, 2)
+    report.ok('firmware_file', '{} ({}, board_id {})'.format(
+        px4_path, meta.get('git_identity') or 'no git identity', meta['board_id']))
+
+    # Early wrong-board check against boards/<vendor>/<model>/firmware.prototype;
+    # the uploader still enforces board_id against the bootloader at flash time.
+    target = args.target
+    if not target:
+        target, terr = firmware.infer_build_target(ident.get('hw_arch'))
+        if target is None:
+            report.info('board_id preflight check skipped: {}'.format(terr))
+    id_ok, id_detail = firmware.check_board_id(meta, target)
+    if id_ok is False:
+        report.fail('firmware_board_id', id_detail)
+        gate_abort(report)
+    elif id_ok is None:
+        report.info(id_detail)
+    else:
+        report.ok('firmware_board_id', id_detail)
+
+    newmav, err = firmware.flash_firmware(
+        px4_path, args.connection, baud=args.baudrate,
+        interactive=sys.stdin.isatty(), timeout=args.flash_timeout, mav=mav)
+    if newmav is None:
+        report.fail('firmware_flash', err)
+        gate_abort(report)
+    report.ok('firmware_flash', source)
+
+    expected = firmware.expected_hash_from_metadata(meta)
+    ok, new_ident = firmware.verify_identity(report, newmav, expected)
+    if not ok or new_ident is None:
+        gate_abort(report)
+    return newmav, new_ident
+
+
+def resolve_target(report, args, ident):
+    """--target, or inferred from the board's HW arch; aborts if ambiguous."""
+    if args.target:
+        return args.target
+    target, terr = firmware.infer_build_target(ident.get('hw_arch'))
+    if target is None:
+        report.fail('firmware_target',
+                    '{}; pass --target (e.g. --target px4_fmu-v6xrt)'.format(terr))
+        gate_abort(report, 2)
+    report.info('board target inferred from HW arch: {}'.format(target))
+    return target
+
+
+def preflight(args, suite_dir):
+    """Firmware gate: establish (and optionally control) the build under test.
+
+    Returns the final identity dict. Exits on any gate failure; no test
+    runs against an unknown or unintended build.
+    """
+    report = px4bench.Reporter('preflight')
+
+    gate_flag_given = any((args.any_firmware, args.expect_hash, args.firmware,
+                           args.build, args.release))
+    if not gate_flag_given and not sys.stdin.isatty():
+        report.fail('firmware_gate',
+                    'no firmware expectation given and stdin is not a TTY. '
+                    'Automation must state what it is testing: pass one of '
+                    '--firmware FILE, --build, --release TAG, --expect-hash PREFIX, '
+                    'or --any-firmware.')
+        gate_abort(report, 2)
+
+    try:
+        mav = px4bench.connect(args.connection, baud=args.baudrate)
+    except (TimeoutError, OSError) as e:
+        report.fail('connect', str(e))
+        gate_abort(report, 2)
+
+    ident, err = firmware.board_identity(mav)
+    if ident is None:
+        report.fail('board_identity', err)
+        gate_abort(report, 2)
+
+    print('=' * 70)
+    print('FIRMWARE ON BOARD ({})'.format(args.connection))
+    print(firmware.format_identity(ident))
+    print('=' * 70, flush=True)
+
+    source = 'kept firmware already on the board'
+
+    if args.any_firmware:
+        report.ok('firmware_identity',
+                  'any-firmware: testing {} as-is'.format(ident['git_hash']))
+
+    elif args.expect_hash:
+        ok = firmware.hashes_match(ident['git_hash'], args.expect_hash)
+        report.check('firmware_identity', ok,
+                     'board reports {} , expected {}'.format(
+                         ident['git_hash'], args.expect_hash))
+        if not ok:
+            mav.close()
+            gate_abort(report)
+
+    elif args.firmware:
+        mav, ident = flash_and_verify(report, args, mav, ident, args.firmware,
+                                      'flashed local file {}'.format(args.firmware))
+        source = 'flashed local file {}'.format(args.firmware)
+
+    elif args.build:
+        target = resolve_target(report, args, ident)
+        px4_path, err = firmware.build_firmware(target, timeout=args.build_timeout)
+        if px4_path is None:
+            report.fail('firmware_build', err)
+            gate_abort(report)
+        report.ok('firmware_build', px4_path)
+        head = firmware.source_tree_hash()
+        meta = firmware.parse_px4_metadata(px4_path)
+        if head and meta.get('git_hash') and not firmware.hashes_match(meta['git_hash'], head):
+            report.info('WARNING: artifact hash {} differs from source tree HEAD {}; '
+                        'stale build?'.format(meta['git_hash'], head))
+        mav, ident = flash_and_verify(report, args, mav, ident, px4_path,
+                                      'built and flashed {} from this tree'.format(target))
+        source = 'built {} from source tree (HEAD {})'.format(target, head or 'unknown')
+
+    elif args.release:
+        target = resolve_target(report, args, ident)
+        px4_path, err = firmware.download_release(args.release, target, suite_dir)
+        if px4_path is None:
+            report.fail('firmware_download', err)
+            gate_abort(report)
+        report.ok('firmware_download', px4_path)
+        mav, ident = flash_and_verify(report, args, mav, ident, px4_path,
+                                      'flashed release {} ({})'.format(args.release, target))
+        source = 'GitHub release {} ({})'.format(args.release, target)
+
+    else:
+        mav, ident, source = interactive_gate(report, args, mav, ident, suite_dir)
+
+    fw_info = {k: v for k, v in ident.items() if k != 'raw'}
+    fw_info['source'] = source
+    firmware.stamp_identity(suite_dir, ident, extra={'source': source})
+    os.environ[px4bench.FIRMWARE_INFO_ENV] = json.dumps(fw_info, sort_keys=True)
+
+    try:
+        mav.close()
+    except Exception:
+        pass
+
+    if report.finish() != 0:
+        print('firmware gate failed; no test was run.', flush=True)
+        sys.exit(1)
+    print()
+    return ident
+
+
+def interactive_gate(report, args, mav, ident, suite_dir):
+    """Operator menu when no firmware flag was given on a TTY."""
+    inferred, _ = firmware.infer_build_target(ident.get('hw_arch'))
+    build_label = inferred if inferred else '<target unknown, will prompt>'
+    while True:
+        print('\nNo firmware expectation given. Choose:')
+        print('  [1] continue on the current firmware ({})'.format(ident['git_hash'][:12]))
+        print('  [2] flash a local .px4 file')
+        print('  [3] build {} from this source tree and flash'.format(build_label))
+        print('  [4] download a GitHub release and flash')
+        print('  [q] abort')
+        choice = input('> ').strip().lower()
+
+        if choice == '1':
+            report.ok('firmware_identity',
+                      'operator kept current firmware {}'.format(ident['git_hash']))
+            return mav, ident, 'operator kept firmware already on the board'
+
+        if choice == '2':
+            path = input('.px4 path: ').strip()
+            if not path or not os.path.exists(path):
+                print('no such file: {!r}'.format(path))
+                continue
+            mav, ident = flash_and_verify(report, args, mav, ident, path,
+                                          'flashed local file {}'.format(path))
+            return mav, ident, 'flashed local file {}'.format(path)
+
+        if choice == '3':
+            target = args.target or inferred
+            if not target:
+                target = input('build target (e.g. px4_fmu-v6xrt): ').strip()
+                if not target:
+                    continue
+            px4_path, err = firmware.build_firmware(target, timeout=args.build_timeout)
+            if px4_path is None:
+                report.fail('firmware_build', err)
+                gate_abort(report)
+            report.ok('firmware_build', px4_path)
+            mav, ident = flash_and_verify(report, args, mav, ident, px4_path,
+                                          'built and flashed {}'.format(target))
+            return mav, ident, 'built {} from source tree'.format(target)
+
+        if choice == '4':
+            target = args.target or inferred
+            if not target:
+                target = input('board target (e.g. px4_fmu-v6xrt): ').strip()
+                if not target:
+                    continue
+            tag = input('release tag [latest]: ').strip() or 'latest'
+            px4_path, err = firmware.download_release(tag, target, suite_dir)
+            if px4_path is None:
+                report.fail('firmware_download', err)
+                gate_abort(report)
+            report.ok('firmware_download', px4_path)
+            mav, ident = flash_and_verify(report, args, mav, ident, px4_path,
+                                          'flashed release {} ({})'.format(tag, target))
+            return mav, ident, 'GitHub release {} ({})'.format(tag, target)
+
+        if choice == 'q':
+            report.fail('firmware_gate', 'aborted by operator')
+            gate_abort(report, 2)
+
+        print('unrecognized choice: {!r}'.format(choice))
+
+
 def main():
     parser = argparse.ArgumentParser(
         description='Run the non-interactive PX4 v1.18 bench-test suite.')
@@ -123,10 +365,33 @@ def main():
                         help='seconds before a test is killed as hung (default: %(default)s)')
     parser.add_argument('--stop-on-fail', action='store_true', default=False,
                         help='stop the suite at the first failing test')
+
+    gate = parser.add_mutually_exclusive_group()
+    gate.add_argument('--firmware', metavar='FILE',
+                      help='flash this .px4 before testing, then verify identity')
+    gate.add_argument('--build', action='store_true',
+                      help='build the board target from this source tree, flash, verify')
+    gate.add_argument('--release', metavar='TAG',
+                      help="flash a GitHub release artifact (a tag like v1.17.0, or 'latest')")
+    gate.add_argument('--expect-hash', metavar='PREFIX',
+                      help='verify the board already runs this git hash (prefix match), no flash')
+    gate.add_argument('--any-firmware', action='store_true',
+                      help='explicitly proceed with whatever firmware is on the board')
+    parser.add_argument('--target', default=None,
+                        help='board target for --build/--release (e.g. px4_fmu-v6xrt); '
+                             'inferred from the connected board when possible')
+    parser.add_argument('--build-timeout', type=float, default=1800,
+                        help='seconds before --build is killed (default: %(default)s)')
+    parser.add_argument('--flash-timeout', type=float, default=600,
+                        help='seconds before the uploader is killed (default: %(default)s)')
     args = parser.parse_args()
 
     skip = {s.strip() for s in args.skip.split(',') if s.strip()}
 
+    suite_dir = px4bench.make_report_dir(args.report_dir, 'suite')
+    print('suite report dir: {}'.format(suite_dir))
+    preflight(args, suite_dir)
+
     # build the run sequence
     sequence = []
     for name in BASE_SEQUENCE:

Tools/bench_test/sih/flight_mission.py

@@ -32,6 +32,7 @@ from px4bench.ftp import LOG_ROOT, ULOG_MAGIC7
 from px4bench.missions import BASE_LAT, BASE_LON, Item
 from px4bench.params import (drain_param_values, read_param, set_param_int32,
                              wait_param_echo)
+from px4bench import firmware as fw_gate
 
 from pymavlink import mavutil
 
@@ -359,6 +360,9 @@ def main():
     parser.add_argument('--viewer-port', type=int, default=19410)
     parser.add_argument('--keep-config', action='store_true',
                         help='stay on the SIH airframe when done')
+    parser.add_argument('--expect-hash', metavar='PREFIX', default=None,
+                        help='verify the board runs this git hash before flying '
+                             '(prefix match); mismatch aborts. No flashing here.')
     parser.add_argument('--report-dir', default='bench_reports')
     args = parser.parse_args()
 
@@ -373,6 +377,21 @@ def main():
         return report.finish()
     report.check('connect', True, 'heartbeat from {}'.format(args.connection))
 
+    ident, ident_err = fw_gate.board_identity(mav)
+    if ident is None:
+        report.fail('board_identity', ident_err)
+        return report.finish()
+    report.info('firmware on board:\n' + fw_gate.format_identity(ident))
+    fw_gate.stamp_identity(report_dir, ident)
+    if args.expect_hash:
+        ok = fw_gate.hashes_match(ident['git_hash'], args.expect_hash)
+        report.check('firmware_identity', ok,
+                     'board reports {} , expected {}'.format(
+                         ident['git_hash'], args.expect_hash))
+        if not ok:
+            mav.close()
+            return report.finish()
+
     original_autostart = original_hitl = None
     try:
         mav, original_autostart, original_hitl = enter_sih(