Add RTT console/capture tooling (tools/rtt.py), rtt skill, and HIL harness support (#3853)

Promote SEGGER RTT from an inline debugging technique to a standalone skill
backed by one stdlib-only implementation in tools/rtt.py: a CLI and importable
module for console/capture over J-Link (RTTTelnetPort) and OpenOCD (rtt server)
probes, with probe selection by serial or VID:PID, control-block address via
--elf or --addr, bidirectional console, post-mortem ring dump, and
--reset-before-attach for boot-time capture. The HIL harness reads a board's
console over RTT when its probe has no VCOM ("logger": "rtt" plus a LOGGER=rtt
variant define), covering device_info, pool-check aliveness, and CI wiring.
Validated on 22 boards across both backends; 26 unit tests run in pre-commit.

7 files changed, 711 insertions, 29 deletions

diff --git a/test/hil/helper/hil_pool_check.py b/test/hil/helper/hil_pool_check.py
index b92f0aee0..4623ce45f 100644
--- a/test/hil/helper/hil_pool_check.py
+++ b/test/hil/helper/hil_pool_check.py
@@ -360,7 +360,47 @@ def check_host_serial(board: dict, do_reset: bool = True, want_hello: bool = Fal
 
     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."""
+    the same probe can fail transiently and leave the target halted.
+
+    "logger": "rtt" boards have no VCOM: the same check runs over the probe's RTT
+    console instead. The reset happens BEFORE the console opens (it owns the probe),
+    which also zeroes the .bss ring — so pre-reset backlog cannot count as life, and
+    without a reset Commander delivers the boot burst the preceding flash left."""
+    if board.get('logger') == 'rtt':
+        if do_reset:
+            # a failed reset leaves the previous run's ring intact: attaching anyway would
+            # score stale output as life, so bail to host_alive's board_test reflash ladder
+            rc, err = call_flasher(getattr(hil_flash, f'reset_{board["flasher"]["name"].lower()}'), board)
+            if rc:
+                say(f'{board["name"]:26} reset failed: {err}')
+                return None
+        try:
+            ser = hil_util.JlinkRtt(board, timeout=0.3)
+        except hil_util.RttError as e:
+            say(f'{board["name"]:26} no RTT console: {e}')
+            return None
+        try:
+            data = b''
+            deadline = time.monotonic() + SERIAL_WAIT
+            while time.monotonic() < deadline:
+                ser.write(b'U')
+                data += ser.read(256)
+                # JLinkExe's banner arrives whether or not the target is alive --
+                # judged unfiltered it scores a dead board 'alive'. Same shared filter
+                # as test_host_device_info; complete_only drops a trailing partial
+                # line, so a banner FRAGMENT split by this read boundary cannot count
+                # as target output either.
+                td = hil_util.strip_banner(data, complete_only=True)
+                if want_hello:
+                    if b'Hello from TinyUSB' in td:
+                        return td
+                elif td and not boardtest_output(td):
+                    return td
+            return hil_util.strip_banner(data)
+        except hil_util.RttError:
+            return None  # console died mid-poll (server exited, probe dropped)
+        finally:
+            ser.close()
     import serial
     try:
         port = hil_util.get_serial_dev(board['flasher']['uid'], None, None, 0)
diff --git a/test/hil/helper/hil_util.py b/test/hil/helper/hil_util.py
index 03d01270f..f279cfa77 100644
--- a/test/hil/helper/hil_util.py
+++ b/test/hil/helper/hil_util.py
@@ -1,7 +1,8 @@
 #!/usr/bin/env python3
 # SPDX-License-Identifier: MIT
 # Bottom layer of the HIL harness: the bounded command runner plus the shared helpers and
-# data every other module needs. Stays stdlib-only and imports nothing local -- everything
+# data every other module needs. Stays stdlib-only; its one local dependency is
+# tools/rtt.py (the RTT console, loaded by path below) -- everything
 # else imports this, including the unit tests on GitHub's bare runner; never import them
 # from here. Callers set the module global `verbose`.
 
@@ -499,6 +500,29 @@ def run_alongside(argv: list, work, timeout: int) -> subprocess.CompletedProcess
     return _reap()
 
 
+# The RTT console implementation lives in tools/rtt.py (importable classes + CLI,
+# stdlib-only, harness-critical — see its module docstring). Loaded by file path so
+# no sys.path entry for tools/ can shadow other imports; re-exported here so the
+# harness keeps addressing hil_util.JlinkRtt.
+import importlib.util as _ilu
+
+_rtt_path = TINYUSB_ROOT / 'tools' / 'rtt.py'
+if not _rtt_path.exists():
+    # name the real cause: a bare FileNotFoundError out of an exec_module here reads
+    # as a harness bug, when the actual problem is an incompletely staged tree
+    raise ImportError(f'{_rtt_path} is missing — the RTT console lives there and the '
+                      f'harness depends on it; stage it alongside test/hil (hil_ci.sh does)')
+_rtt_spec = _ilu.spec_from_file_location('tinyusb_tools_rtt', _rtt_path)
+_rtt = _ilu.module_from_spec(_rtt_spec)
+sys.modules[_rtt_spec.name] = _rtt   # registered: RttError must be picklable across the fork Pool
+_rtt_spec.loader.exec_module(_rtt)
+JlinkRtt = _rtt.JlinkRtt
+OpenocdRtt = _rtt.OpenocdRtt
+RttError = _rtt.RttError
+RTT_BANNER_RE = _rtt.RTT_BANNER_RE
+strip_banner = _rtt.strip_banner
+
+
 def _cmd_label(cmd) -> str:
     """A one-line name for a banner. An argv whose payload is a `python3 -c` program would
     otherwise dump the whole body into the CI log, where run_cmd's banners are already the
diff --git a/test/hil/hil_ci.sh b/test/hil/hil_ci.sh
index daa787242..43ede5795 100644
--- a/test/hil/hil_ci.sh
+++ b/test/hil/hil_ci.sh
@@ -288,6 +288,9 @@ scp -q "$ROOT_DIR/test/hil/helper/__init__.py" \
        "$ROOT_DIR/test/hil/helper/hil_lock.py" \
        "$ROOT_DIR/test/hil/helper/hil_report.py" \
        "$REMOTE:$REMOTE_DIR/test/hil/helper/"
+# the rtt console/capture tool (rtt skill), harness-critical: hil_util imports it
+ssh "$REMOTE" mkdir -p "$REMOTE_DIR/tools"
+scp -q "$ROOT_DIR/tools/rtt.py" "$REMOTE:$REMOTE_DIR/tools/"
 
 # 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_test.py b/test/hil/hil_test.py
index e32998420..233627ec7 100755
--- a/test/hil/hil_test.py
+++ b/test/hil/hil_test.py
@@ -206,6 +206,7 @@ class Board(TypedDict):
     # needs one carries a single variant named after itself (metro_m4_express /
     # MAX3421_HOST=1), which is exactly what the `or [...]` default below synthesises
     variant: NotRequired[list[VariantCfg]]
+    logger: NotRequired[str]  # "rtt": console = the debug probe's RTT channel 0, not a VCOM (rtt skill)
     toolchain: NotRequired[str]  # CI build bucket override, e.g. "riscv-gcc" (consumed by hil_ci_set_matrix.py)
 
 
@@ -292,6 +293,25 @@ def open_serial_dev(port: str):
     return ser
 
 
+def open_board_console(board: Board):
+    """The board's log console: its probe's VCOM, or RTT when the probe has none.
+
+    Both ends expose the same read/in_waiting/write/close surface, so the tests read one
+    the same way they read the other."""
+    if board.get('logger') == 'rtt':
+        # JlinkRtt speaks JLinkExe only; an openocd/stlink flasher would yield
+        # `-device ''` and fail 15 s later with a misleading port error. The OpenOCD
+        # RTT route is validated manually on native probes but has no harness backend
+        # yet (rtt skill; followup doc) — and never point it at ea4088's LPC-Link2
+        # (measured: knocks that probe off USB; other J-Link-OB probes untested)
+        assert board['flasher']['name'].lower() == 'jlink', \
+            f'{board["name"]}: "logger": "rtt" needs a jlink flasher, not {board["flasher"]["name"]}'
+        return hil_util.JlinkRtt(board)
+    ser = open_serial_dev(hil_util.get_serial_dev(board['flasher']["uid"], None, None, 0))
+    ser.timeout = 0.1
+    return ser
+
+
 def serial_write_all(ser: serial.Serial, data: bytes):
     # write_timeout is a deadline for the whole call. A timeout means the device stopped
     # draining, and it is fatal: pyserial loses the partial-write count on raise, so
@@ -300,8 +320,19 @@ def serial_write_all(ser: serial.Serial, data: bytes):
         ser.write(data)
     except serial.SerialTimeoutException:
         raise AssertionError(f'Serial write timeout after {SERIAL_WRITE_TIMEOUT:.1f}s')
+    except hil_util.RttError as e:
+        # the RTT console's failure contract (stall/closed/peer death): same
+        # drain-stopped meaning as the serial timeout -- a test failure, not a harness
+        # crash. Deliberately NOT bare RuntimeError: NotImplementedError and CPython's
+        # own 'dictionary changed size during iteration' are RuntimeErrors too, and a
+        # harness bug must not be reported as this board misbehaving.
+        raise AssertionError(f'Console write failed: {e}')
 
 
+# J-Link Commander's telnet greeting: never target output (defined with the console
+# in tools/rtt.py; hil_pool_check strips it through the same object)
+RTT_BANNER_RE = hil_util.RTT_BANNER_RE
+
 LP_OPEN_TIMEOUT = 5   # bound on opening the printer lp node; see test_device_printer_to_cdc
 # Runs under hil_util.run_alongside as `python3 -c`. Inline rather than a file so hil_ci.sh's
 # staging list does not need another entry to keep the rig working.
@@ -508,34 +539,53 @@ def test_host_device_info(board):
     flasher = board['flasher']
     declared_devs = [f'{d["vid_pid"]}_{d["serial"]}' for d in board['tests']['dev_attached']]
 
-    port = hil_util.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 = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
-    assert ret.returncode == 0, 'Failed to reset device'
+    if board.get('logger') == 'rtt':
+        # The RTT console owns the probe, so reset BEFORE opening it (Commander then
+        # delivers the buffered boot burst). Unconditional, not only under --skip-flash:
+        # a previous run's console drained the ring, and the enumeration lines print
+        # only once — without this a re-run on unchanged firmware reads an empty ring.
+        ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
+        assert ret.returncode == 0, 'Failed to reset device'
+    ser = open_board_console(board)
+    try:
+        if board.get('logger') != 'rtt':
+            # reset device since we can miss the first line; on the VCOM the console
+            # survives the reset, so resetting after open catches the boot banner.
+            ret = getattr(hil_flash, f'reset_{flasher["name"].lower()}')(board)
+            assert ret.returncode == 0, 'Failed to reset device'
 
-    data = b''
-    timeout = enum_timeout()
-    while timeout > 0:
-        new_data = ser.read(ser.in_waiting or 1)
-        if new_data:
-            data += new_data
-        enum_dev_sn = []
-        for l in data.decode('utf-8', errors='ignore').splitlines():
-            vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
-            if vid_pid_sn:
-                enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')
-        if set(declared_devs).issubset(set(enum_dev_sn)):
-            break
-        time.sleep(0.1)
-        timeout -= 0.1
-    ser.close()
+        data = b''
+        timeout = enum_timeout()
+        while timeout > 0:
+            # infra death is not a board failure: without this a dead JLinkExe/probe
+            # would burn the whole timeout and report as 'No data from device'
+            assert not getattr(ser, 'eof', False), \
+                'RTT console died (its server exited or the probe dropped off USB)'
+            new_data = ser.read(ser.in_waiting or 1)
+            if new_data:
+                data += new_data
+            enum_dev_sn = []
+            for l in data.decode('utf-8', errors='ignore').splitlines():
+                vid_pid_sn = re.search(r'ID ([0-9a-fA-F]+):([0-9a-fA-F]+) SN (\w+)', l)
+                if vid_pid_sn:
+                    enum_dev_sn.append(f'{vid_pid_sn.group(1)}_{vid_pid_sn.group(2)}_{vid_pid_sn.group(3)}')
+            if set(declared_devs).issubset(set(enum_dev_sn)):
+                break
+            time.sleep(0.1)
+            timeout -= 0.1
+    finally:
+        ser.close()
 
-    if len(data) == 0:
-        assert False, 'No data from device'
     lines = data.decode('utf-8', errors='ignore').splitlines()
+    if board.get('logger') == 'rtt':
+        # JLinkExe's telnet banner is delivered at connect, whether or not it ever
+        # finds the control block, so len(data) alone cannot tell "board said nothing"
+        # from "console never attached to the ring" -- drop the banner first
+        target_lines = hil_util.strip_banner(data).splitlines()
+        assert target_lines, ('No data from device: the RTT console attached but the target '
+                              'produced nothing -- firmware built without LOGGER=rtt, or SWD lost')
+    elif len(data) == 0:
+        assert False, 'No data from device'
 
     enum_dev_sn = []
     for l in lines:
@@ -1729,7 +1779,7 @@ def test_example(board: Board, variant: str, example: str) -> tuple[int, str, st
 
 def build_board(board: Board) -> tuple[str, int]:
     """Build firmware for this board via tools/build.py.
-    Honors board config's variant list (name, defines, flags).
+    Honors board config's variant list.
     Output goes to cmake-build/cmake-build-<variant>/ (tools/build.py layout).
 
     Unbounded on purpose: --build is a local convenience (no CI workflow passes it), so
@@ -2337,6 +2387,56 @@ def main() -> None:
         config_boards = [e for e in config['boards'] if e['name'] in boards]
     config_boards = [e for e in config_boards if e['flasher']['name'] not in args.exclude_flasher
                      and (not args.flasher or e['flasher']['name'] in args.flasher)]
+
+    # fail rtt misconfigurations before the first flash cycle -- but only for boards
+    # this run actually touches: one bad roster entry must not abort other runs' subsets
+    def _rtt_config_abort(msg: str):
+        # loud AND leaving evidence, like the no-boards branch below: exiting with no
+        # report at all lets the PR comment keep the previous push's stale table
+        print(f'ERROR: {msg}', flush=True)
+        rd = Path(os.environ.get('HIL_REPORT_DIR', '.'))
+        hil_report.mark_report_no_boards(rd, f'config error: {msg}', fresh=not args.accumulate)
+        sys.exit(1)
+
+    bad_logger = [e['name'] for e in config_boards if e.get('logger') not in (None, 'rtt')]
+    if bad_logger:
+        # only the exact string activates RTT handling; anything else would silently
+        # mean VCOM and reproduce the misleading 'No serial device found' failure
+        _rtt_config_abort(f'unknown "logger" value (only "rtt" is supported): {", ".join(bad_logger)}')
+    bad_rtt = [e['name'] for e in config_boards
+               if e.get('logger') == 'rtt' and e['flasher']['name'].lower() != 'jlink']
+    if bad_rtt:
+        # JlinkRtt speaks JLinkExe only (the OpenOCD RTT route is manual — rtt skill)
+        _rtt_config_abort(f'"logger": "rtt" needs a jlink flasher: {", ".join(bad_rtt)}')
+    rtt_no_logger_def = [e['name'] for e in config_boards
+                         if e.get('logger') == 'rtt'
+                         and any('LOGGER=rtt' not in (v.get('defines') or [])
+                                 for v in (e.get('variant') or [{}]))]
+    if rtt_no_logger_def:
+        # a prebuilt cmake-build-<board> configured with -DLOGGER=rtt is a legitimate
+        # build path the roster need not describe, so warn there -- but when this run is
+        # responsible for the firmware (--build, or CI where the hil-build job compiled
+        # the artifact from these same defines) the flashed image is UART-logger and every
+        # test times out as 'the target produced nothing'. An always-on define is
+        # expressed as a single self-named variant (see the Board comment).
+        msg = (f'"logger": "rtt" board has a variant without LOGGER=rtt in its defines '
+               f'({", ".join(rtt_no_logger_def)})')
+        if args.build or os.environ.get('GITHUB_ACTIONS'):
+            _rtt_config_abort(f'{msg} -- the firmware built for this run cannot serve the '
+                              f'configured RTT console')
+        print(f'warning: {msg} -- fine for prebuilt example sets, wrong for --build/CI '
+              f'builds', flush=True)
+    rtt_fixture = [e['name'] for e in config_boards
+                   if e.get('logger') == 'rtt'
+                   and any(d.get('is_cdc') or d.get('is_msc')
+                           for d in e.get('tests', {}).get('dev_attached', []))]
+    if rtt_fixture:
+        # interim guard, removed when the followup lands: cdc_msc_hid/msc_file_explorer
+        # still open the flasher VCOM directly and would die mid-run on an rtt board
+        _rtt_config_abort(f'"logger": "rtt" boards cannot carry is_cdc/is_msc fixtures yet '
+                          f'(host cdc/msc tests bypass the RTT console — see '
+                          f'the rtt harness-adoption doc in docs/superpowers/followup/): {", ".join(rtt_fixture)}')
+
     if not config_boards:
         # same reason the unknown -b board exits 1: 'No tests were run.' with rc 0 reads as
         # a green HIL leg, so a roster edit emptying a leg's filter stops testing silently
diff --git a/test/hil/test/test_ci_select.py b/test/hil/test/test_ci_select.py
index ace230246..22fbde17b 100644
--- a/test/hil/test/test_ci_select.py
+++ b/test/hil/test/test_ci_select.py
@@ -499,6 +499,8 @@ class TestPortAndCoreRoleUseExtras(unittest.TestCase):
         self.assertFalse(s['full'])
         for board in boards:
             tests = s['boards'][board]
+            if tests == 'all':
+                continue  # a board whose whole allowed set is selected collapses to 'all'
             self.assertIn('device/hid_composite_freertos', tests)
             self.assertIn('device/cdc_msc_freertos', tests)
             self.assertIn('device/audio_test_freertos', tests)
@@ -510,6 +512,8 @@ class TestPortAndCoreRoleUseExtras(unittest.TestCase):
         self.assertFalse(s['full'])
         for board in boards:
             tests = s['boards'][board]
+            if tests == 'all':
+                continue  # a board whose whole allowed set is selected collapses to 'all'
             self.assertIn('device/hid_composite_freertos', tests)
             self.assertIn('device/cdc_msc_freertos', tests)
             self.assertIn('device/audio_test_freertos', tests)
@@ -987,6 +991,7 @@ class TestTheHarnessTestsAreNotTheHarness(unittest.TestCase):
             'test/hil/test/test_hil_bounded.py',
             'test/hil/test/test_hil_health.py',
             'test/hil/test/test_hil_report.py',
+            'test/hil/test/test_hil_rtt.py',
             'test/hil/test/test_hil_util.py',
         ], 'test/hil/test/ gained or lost a file; it is carved out of rule 2, so confirm '
            'the rig still does not read anything in there before updating this list')
diff --git a/test/hil/test/test_hil_rtt.py b/test/hil/test/test_hil_rtt.py
new file mode 100644
index 000000000..3a07f13ec
--- /dev/null
+++ b/test/hil/test/test_hil_rtt.py
@@ -0,0 +1,506 @@
+#!/usr/bin/env python3
+# SPDX-License-Identifier: MIT
+# Unit tests for hil_util.JlinkRtt and the rtt.py CLI against a fake JLinkExe
+# on PATH — real subprocesses and sockets, no hardware, stdlib only, so the pre-commit
+# hil-test hook can run this on GitHub's bare runner. Run directly:
+#   python3 test/hil/test/test_hil_rtt.py
+import os
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from contextlib import suppress as contextlib_suppress
+from pathlib import Path
+
+# the module under test lives in the parent dir's helper/ package
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from helper import hil_util
+
+CLI = Path(__file__).resolve().parents[3] / 'tools' / 'rtt.py'
+
+# Serves -RTTTelnetPort like J-Link Commander: greets, echoes input uppercased, exits on
+# stdin 'exit' (JlinkRtt.close()'s contract). FAKE_JLINK_MODE=die_after_greet sends the
+# greeting then drops the connection and exits — the probe-unplug/crash case;
+# FAKE_JLINK_MODE=tick also streams a line every 50 ms — the continuous-capture case.
+FAKE_JLINK = '''#!/usr/bin/env python3
+import os, socket, sys, threading, time
+port = int(sys.argv[sys.argv.index('-RTTTelnetPort') + 1])
+srv = socket.socket(); srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+srv.bind(('127.0.0.1', port)); srv.listen(1)
+mode = os.environ.get('FAKE_JLINK_MODE', '')
+def serve():
+    conn, _ = srv.accept()
+    # the real server sends its banner AT CONNECT, before the control block is
+    # found — target data only flows later; the CLI's -i gate must not release
+    # on the banner
+    conn.sendall(b'SEGGER J-Link fake - Real time terminal output\\r\\n'
+                 b'J-Link FakeProbe V1.0, SN=000\\r\\nProcess: JLinkExe\\r\\n')
+    if mode == 'banner_only':
+        while True:
+            if not conn.recv(4096): os._exit(0)
+    if mode == 'rst':
+        import struct
+        conn.recv(4096)   # wait for the client to speak, then reset the connection
+        conn.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, struct.pack('ii', 1, 0))
+        conn.close(); os._exit(0)
+    if mode == 'late_cb':
+        # models JLinkExe before it finds the control block: client bytes sent in
+        # this window are silently dropped, output starts only after the "attach"
+        end = time.time() + 1.0
+        conn.setblocking(False)
+        while time.time() < end:
+            try:
+                conn.recv(4096)   # discard early input like the real server
+            except OSError:
+                pass
+            time.sleep(0.05)
+        conn.setblocking(True)
+    conn.sendall(b'hello from target\\r\\n')
+    if mode == 'die_after_greet':
+        conn.close(); os._exit(0)
+    if mode == 'tick':
+        def tick():
+            try:
+                while True:
+                    time.sleep(0.05); conn.sendall(b'tick\\r\\n')
+            except OSError:
+                pass
+        threading.Thread(target=tick, daemon=True).start()
+    while True:
+        d = conn.recv(4096)
+        if not d: return
+        conn.sendall(d.upper())
+threading.Thread(target=serve, daemon=True).start()
+for line in sys.stdin:
+    if line.strip() == 'exit': break
+'''
+
+BOARD = {'flasher': {'uid': '000', 'args': '-device FAKE'}}
+
+
[email protected](os.name == 'nt', 'POSIX PATH/exec semantics')
+class JlinkRttFakeProbe(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls._dir = tempfile.TemporaryDirectory()
+        fake = Path(cls._dir.name) / 'JLinkExe'
+        fake.write_text(FAKE_JLINK)
+        fake.chmod(0o755)
+        cls._path = f'{cls._dir.name}{os.pathsep}{os.environ["PATH"]}'
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._dir.cleanup()
+
+    def _fake_path(self):
+        # register the restore BEFORE mutating, then prepend the fake tool dir
+        self.addCleanup(os.environ.__setitem__, 'PATH', os.environ['PATH'])
+        os.environ['PATH'] = self._path
+
+    def _console(self, mode=''):
+        self._fake_path()
+        if mode:
+            os.environ['FAKE_JLINK_MODE'] = mode
+        self.addCleanup(os.environ.pop, 'FAKE_JLINK_MODE', None)
+        con = hil_util.JlinkRtt(BOARD, timeout=0.1)
+        self.addCleanup(con.close)
+        return con
+
+    def _read_until(self, con, want, timeout=3):
+        out = b''
+        end = time.monotonic() + timeout
+        while want not in out and time.monotonic() < end:
+            out += con.read(con.in_waiting or 1)
+        return out
+
+    def test_read_and_echo_write(self):
+        con = self._console()
+        self.assertIn(b'hello from target', self._read_until(con, b'hello from target'))
+        self.assertEqual(con.write(b'ping'), 4)
+        self.assertIn(b'PING', self._read_until(con, b'PING'))
+
+    def test_eof_latched_when_server_dies(self):
+        con = self._console(mode='die_after_greet')
+        self._read_until(con, b'hello from target')
+        end = time.monotonic() + 3
+        while not con.eof and time.monotonic() < end:
+            time.sleep(0.05)
+        self.assertTrue(con.eof)               # dead server is detected, not spun on
+        t0 = time.monotonic()
+        self.assertEqual(con.read(64), b'')    # empty, paced like a serial timeout
+        elapsed = time.monotonic() - t0
+        self.assertLess(elapsed, 0.5)          # bounded by the 0.1 s timeout, not hung
+        self.assertGreater(elapsed, 0.02)      # ...but not a busy-spin fast return
+        con.timeout = None                     # pyserial's block-forever mode must
+        t0 = time.monotonic()                  # ALSO pace (0.1 s default), not spin
+        self.assertEqual(con.read(64), b'')
+        elapsed = time.monotonic() - t0
+        self.assertLess(elapsed, 0.5)
+        self.assertGreater(elapsed, 0.02)
+        con.timeout = 0.1
+
+    def test_reset_input_buffer(self):
+        con = self._console()
+        self._read_until(con, b'hello from target')
+        con.write(b'x')
+        time.sleep(0.3)
+        con.reset_input_buffer()
+        self.assertEqual(con.in_waiting, 0)
+
+    def test_write_after_close_raises_runtimeerror(self):
+        con = self._console()
+        con.close()
+        with self.assertRaises(RuntimeError):
+            con.write(b'x')
+
+    def test_write_after_server_death_raises(self):
+        # TCP accepts one send after peer death — write() must refuse instead of
+        # "succeeding" into the void
+        con = self._console(mode='die_after_greet')
+        self._read_until(con, b'hello from target')
+        end = time.monotonic() + 3
+        while not con.eof and time.monotonic() < end:
+            time.sleep(0.05)
+        with self.assertRaises(RuntimeError):
+            con.write(b'ping')
+
+    def test_read_after_close_raises_runtimeerror(self):
+        con = self._console()
+        self._read_until(con, b'hello from target')
+        con.close()
+        with self.assertRaises(RuntimeError):
+            con.read(1)
+
+    def test_missing_jlinkexe_raises_runtimeerror(self):
+        self._fake_path()
+        os.environ['PATH'] = self._dir.name  # no python3 either, but JLinkExe fails first
+        os.rename(f'{self._dir.name}/JLinkExe', f'{self._dir.name}/JLinkExe.off')
+        self.addCleanup(os.rename, f'{self._dir.name}/JLinkExe.off', f'{self._dir.name}/JLinkExe')
+        with self.assertRaises(RuntimeError):
+            hil_util.JlinkRtt(BOARD, timeout=0.1)
+
+    def test_close_reaps_the_server(self):
+        con = self._console()
+        proc = con._proc
+        con.close()
+        self.assertIsNotNone(proc.poll())      # no zombie, no probe held
+
+    def test_cli_exits_when_server_dies(self):
+        # --seconds 0 must end on server EOF (rc 1), not hang forever
+        env = dict(os.environ, PATH=self._path, FAKE_JLINK_MODE='die_after_greet')
+        r = subprocess.run([sys.executable, str(CLI),
+                            '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '0'],
+                           env=env, capture_output=True, timeout=20)
+        self.assertEqual(r.returncode, 1)
+        self.assertIn(b'hello from target', r.stdout)
+        self.assertIn(b'server closed', r.stderr)
+
+    def test_peer_reset_latches_eof(self):
+        # a killed server closes with RST when bytes are unread; the read side must
+        # LATCH eof (so the harness's `assert not ser.eof` triage fires) and never
+        # leak ConnectionResetError/ValueError to in_waiting/eof callers
+        con = self._console(mode='rst')
+        # rst mode sends only the banner (it RSTs on first input) -- wait for the
+        # banner tail, not target output that never comes
+        self._read_until(con, b'Process: JLinkExe')
+        con.write(b'x')                       # fake resets the connection on input
+        end = time.monotonic() + 3
+        try:
+            while not con.eof and time.monotonic() < end:
+                con.in_waiting                # must not raise across the RST
+                time.sleep(0.05)
+        except Exception as e:                # noqa: BLE001 - the regression this guards
+            self.fail(f'{type(e).__name__} escaped the latch-only contract: {e}')
+        self.assertTrue(con.eof)
+        with self.assertRaises(hil_util.RttError):
+            con.write(b'y')                   # dead server refuses writes
+
+    def test_write_timeout_env_rejects_inf(self):
+        # hil_util's twin rejects inf for the same reason: an unbounded write is what
+        # this knob exists to bound
+        import importlib.util as ilu
+        from pathlib import Path as _P
+        spec = ilu.spec_from_file_location('rtt_env_probe', _P(CLI))
+        mod = ilu.module_from_spec(spec)
+        old = os.environ.get('HIL_SERIAL_WRITE_TIMEOUT')
+        os.environ['HIL_SERIAL_WRITE_TIMEOUT'] = 'inf'
+        self.addCleanup(lambda: os.environ.__setitem__('HIL_SERIAL_WRITE_TIMEOUT', old)
+                        if old is not None else os.environ.pop('HIL_SERIAL_WRITE_TIMEOUT', None))
+        spec.loader.exec_module(mod)
+        self.assertEqual(mod.RTT_WRITE_TIMEOUT, 10)
+
+    def test_cli_rejects_bad_seconds_and_jlink_channel(self):
+        def run(*a):
+            return subprocess.run([sys.executable, str(CLI), *a], capture_output=True, timeout=15)
+        for bad in ('-5', 'nan'):
+            r = run('--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', bad)
+            self.assertEqual(r.returncode, 2, f'--seconds {bad} was accepted')
+        # the jlink telnet route serves channel 0 only; asking for another is an error,
+        # not silence (--dump can read any ring, so it stays allowed there)
+        r = run('--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--channel', '1')
+        self.assertEqual(r.returncode, 2)
+        self.assertIn(b'channel 0 only', r.stderr)
+        # a negative index would walk backwards off aUp[] (dump route included)
+        r = run('--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--channel', '-1')
+        self.assertEqual(r.returncode, 2)
+        self.assertIn(b'>= 0', r.stderr)
+
+    def test_pyserial_surface_contracts(self):
+        con = self._console()
+        self._read_until(con, b'hello from target')
+        con.write(b'abcdef')
+        self._read_until(con, b'ABC')          # echo queued
+        before = con.in_waiting
+        self.assertEqual(con.read(0), b'')     # pyserial: consumes nothing
+        self.assertEqual(con.read(-1), b'')    # never hand over/destroy bytes
+        self.assertEqual(con.in_waiting, before)
+        con.timeout = None                     # pyserial: block until satisfied
+        con.write(b'xy')                       # fresh echo guarantees the read returns
+        self.assertEqual(len(con.read(2)), 2)
+        con.timeout = 0.1
+        con.close()
+        with self.assertRaises(hil_util.RttError):
+            con.in_waiting                     # closed console reports closed, not healthy
+        self.assertTrue(con.eof)
+
+    def test_context_manager_closes(self):
+        self._fake_path()
+        with hil_util.JlinkRtt(BOARD, timeout=0.1) as con:
+            proc = con._proc
+        self.assertIsNotNone(proc.poll())      # __exit__ released the probe
+
+    def test_staging_and_banner_coupling(self):
+        # tripwires for couplings no import-walk can see:
+        # (a) hil_ci.sh must stage tools/rtt.py -- hil_util exec_module's it, so an
+        #     unstaged rig tree kills every harness import
+        hil_ci = (Path(__file__).resolve().parents[1] / 'hil_ci.sh').read_text()
+        self.assertIn('tools/rtt.py', hil_ci)
+        # (b) the shared RTT banner filter must drop ALL THREE J-Link banner lines,
+        #     including the middle one, which is the PROBE MODEL string and in
+        #     libjlinkarm carries no 'SEGGER ' prefix (J-Link OH3, J-Trace H9...)
+        banner_re = hil_util.RTT_BANNER_RE
+        for line in ('SEGGER J-Link V9.66 - Real time terminal output',
+                     'SEGGER J-Link LPC-Link 2 V1.0, SN=611000000',
+                     'J-Link OH3 V1.0, SN=123456789',
+                     'J-Trace H9 V2.0, SN=123456789002',
+                     'Process: JLinkExe'):
+            self.assertTrue(banner_re.match(line), f'banner line not filtered: {line!r}')
+        for line in ('Hello from TinyUSB', 'USBD init on controller 0',
+                     'ID 1a86:8010 SN 7FD88F0604B5', 'echo:p'):
+            self.assertFalse(banner_re.match(line), f'target line wrongly filtered: {line!r}')
+
+    def test_pool_check_dead_rtt_board_is_not_alive(self):
+        # JLinkExe's banner alone must not score a dead board 'alive': pool_check's
+        # rtt aliveness judges only target bytes (the bug: unfiltered, the banner
+        # made `not boardtest_output(data)` true on the first poll)
+        sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+        from helper import hil_pool_check
+        # a dead board burns the whole poll window; the verdict is the same at 0.5 s
+        self.addCleanup(setattr, hil_pool_check, 'SERIAL_WAIT', hil_pool_check.SERIAL_WAIT)
+        hil_pool_check.SERIAL_WAIT = 0.5
+        self._fake_path()
+        os.environ['FAKE_JLINK_MODE'] = 'banner_only'
+        self.addCleanup(os.environ.pop, 'FAKE_JLINK_MODE', None)
+        board = dict(BOARD, name='deadboard', logger='rtt')
+        got = hil_pool_check.check_host_serial(board, do_reset=False, want_hello=True)
+        self.assertEqual(got, b'')            # dead, not "alive on banner"
+
+    def test_cli_arg_contract(self):
+        # --backend is explicit (no default); vid-pid is openocd-only; the openocd
+        # backend accepts --addr instead of --elf and --vid-pid instead of --probe
+        def run(*a, inp=b''):
+            return subprocess.run([sys.executable, str(CLI), *a],
+                                  input=inp, capture_output=True, timeout=15)
+        r = run('--probe', '000', '--device', 'FAKE')          # no --backend
+        self.assertEqual(r.returncode, 2)
+        self.assertIn(b'--backend', r.stderr)
+        r = run('--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--vid-pid', '0x1 0x2')
+        self.assertEqual(r.returncode, 2)                       # vid-pid is openocd-only
+        r = run('--backend', 'openocd', '--cfg', '-f x.cfg', '--addr', '0x20000000')
+        self.assertEqual(r.returncode, 2)                       # needs --probe or --vid-pid
+        self.assertIn(b'vid-pid', r.stderr)
+        r = run('--backend', 'openocd', '--probe', '000', '--cfg', '-f x.cfg', '--addr', 'nothex')
+        self.assertEqual(r.returncode, 2)
+        self.assertIn(b'hex', r.stderr)
+
+    def test_cli_interactive_echo(self):
+        env = dict(os.environ, PATH=self._path)
+        r = subprocess.run([sys.executable, str(CLI),
+                            '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '2', '-i'],
+                           env=env, input=b'hi', capture_output=True, timeout=20)
+        self.assertEqual(r.returncode, 0)
+        self.assertIn(b'HI', r.stdout)         # bytes forwarded without needing a newline
+        self.assertNotIn(b'never forwarded', r.stderr)   # forwarding happened: no false alarm
+
+    def test_cli_interactive_input_held_until_output(self):
+        # input piped at process start must survive the server's control-block hunt
+        # (the real JLinkExe drops client bytes until the block is found — measured
+        # on the rig: instant 'ping' lost, delayed 'ping' echoed)
+        env = dict(os.environ, PATH=self._path, FAKE_JLINK_MODE='late_cb')
+        r = subprocess.run([sys.executable, str(CLI),
+                            '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '3', '-i'],
+                           env=env, input=b'hi', capture_output=True, timeout=25)
+        self.assertEqual(r.returncode, 0)
+        self.assertIn(b'HI', r.stdout)
+
+    def test_cli_interactive_no_input_diagnostic(self):
+        # -i with stdin closed immediately: the diagnostic must say stdin was never
+        # forwarded (true), keyed on actual forwarding -- not on the attach gate,
+        # which releases after 5 s and forwards anyway on longer runs
+        env = dict(os.environ, PATH=self._path, FAKE_JLINK_MODE='banner_only')
+        r = subprocess.run([sys.executable, str(CLI),
+                            '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '1', '-i'],
+                           env=env, input=b'', capture_output=True, timeout=20)
+        self.assertEqual(r.returncode, 0)
+        self.assertIn(b'never forwarded', r.stderr)
+        self.assertIn(b'no target output', r.stderr)
+
+    def test_cli_downstream_pipe_close(self):
+        # a real `rtt.py | head`-style consumer: close the read end mid-stream
+        # and the CLI must exit 0 via its BrokenPipe path, not traceback (this test
+        # fails if the handler is removed — subprocess.run capture can't cover it)
+        env = dict(os.environ, PATH=self._path, FAKE_JLINK_MODE='tick')
+        p = subprocess.Popen([sys.executable, str(CLI),
+                              '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '8'],
+                             env=env, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        p.stdout.read(10)          # let it stream a little
+        p.stdout.close()           # downstream hangs up
+        rc = p.wait(timeout=20)
+        err = p.stderr.read()
+        p.stderr.close()
+        self.assertEqual(rc, 0, err)
+        self.assertNotIn(b'Traceback', err)
+
+    def test_cli_feeder_races_shutdown(self):
+        # a feeder still writing when --seconds expires must not crash the CLI
+        # (pump thread vs close() race: historically tracebacks and SIGABRT rc 134)
+        env = dict(os.environ, PATH=self._path)
+        for _ in range(3):
+            p = subprocess.Popen([sys.executable, str(CLI),
+                                  '--backend', 'jlink', '--probe', '000', '--device', 'FAKE', '--seconds', '1', '-i'],
+                                 env=env, stdin=subprocess.PIPE, stdout=subprocess.DEVNULL,
+                                 stderr=subprocess.PIPE)
+            try:
+                while True:
+                    p.stdin.write(b'hi\n')
+                    p.stdin.flush()
+                    time.sleep(0.01)
+            except (BrokenPipeError, OSError):
+                pass
+            rc = p.wait(timeout=20)
+            err = p.stderr.read()
+            p.stderr.close()
+            with contextlib_suppress(OSError, ValueError):
+                p.stdin.close()
+            self.assertEqual(rc, 0, err)
+            self.assertNotIn(b'Exception in thread', err)
+
+
+
+class StripBanner(unittest.TestCase):
+    # both harness consumers (device_info verdict, pool_check aliveness) judge
+    # target-aliveness through this ONE filter -- pin its shape here
+    def test_drops_banner_keeps_target(self):
+        raw = (b'SEGGER J-Link V9.66 - Real time terminal output\r\n'
+               b'J-Link OH3 V1.0, SN=123456789\r\nProcess: JLinkExe\r\n'
+               b'Hello from TinyUSB\r\n')
+        self.assertEqual(hil_util.strip_banner(raw), b'Hello from TinyUSB')
+
+    def test_complete_only_drops_split_banner_fragment(self):
+        # a poll loop can catch the banner mid-line at a read boundary; the
+        # fragment must not defeat the prefix regex and score as target output
+        frag = b'SEGGER J-Link V9.66 - Real time terminal output\r\nProce'
+        self.assertEqual(hil_util.strip_banner(frag, complete_only=True), b'')
+        # the final verdict keeps a genuine unterminated target tail
+        self.assertEqual(hil_util.strip_banner(b'tud_task\r\nrunn'), b'tud_task\nrunn')
+        self.assertEqual(hil_util.strip_banner(b'', complete_only=True), b'')
+
+
+# Serves like `openocd ... -c "rtt server start PORT CH"`: parses the port from its
+# single shell-quoted command line, greets, echoes uppercased. No banner (matches the
+# real openocd rtt server, which sends target data only).
+FAKE_OPENOCD = '''#!/usr/bin/env python3
+import os, re, socket, sys, threading, time
+if os.environ.get('FAKE_OPENOCD_ARGV'):
+    open(os.environ['FAKE_OPENOCD_ARGV'], 'w').write(' '.join(sys.argv))
+port = int(re.search(r'rtt server start (\\d+)', ' '.join(sys.argv)).group(1))
+srv = socket.socket(); srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+srv.bind(('127.0.0.1', port)); srv.listen(1)
+conn, _ = srv.accept()
+conn.sendall(b'hello from target\\r\\n')
+while True:
+    d = conn.recv(4096)
+    if not d: break
+    conn.sendall(d.upper())
+'''
+
+
[email protected](os.name == 'nt', 'POSIX PATH/exec semantics')
+class OpenocdRttFakeProbe(unittest.TestCase):
+    """The openocd-backend class shares its whole read/write/eof contract with
+    JlinkRtt via the base class (covered above); this exercises the parts it owns:
+    spawn/connect, echo round-trip, teardown."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls._dir = tempfile.TemporaryDirectory()
+        fake = Path(cls._dir.name) / 'openocd'
+        fake.write_text(FAKE_OPENOCD)
+        fake.chmod(0o755)
+        cls._path = f'{cls._dir.name}{os.pathsep}{os.environ["PATH"]}'
+
+    @classmethod
+    def tearDownClass(cls):
+        cls._dir.cleanup()
+
+    def _fake_path(self):
+        self.addCleanup(os.environ.__setitem__, 'PATH', os.environ['PATH'])
+        os.environ['PATH'] = self._path
+
+    def test_reset_before_attach_shapes_the_command(self):
+        # SystemView-style consumers need the server draining WHEN the target boots
+        # (its Init record is emitted once); the opt-in flag must put `reset run`
+        # between init and rtt setup, and must not appear otherwise
+        self._fake_path()
+        argv_file = os.path.join(self._dir.name, 'argv.txt')
+        os.environ['FAKE_OPENOCD_ARGV'] = argv_file
+        self.addCleanup(os.environ.pop, 'FAKE_OPENOCD_ARGV', None)
+        for flag, want in ((True, True), (False, False)):
+            con = hil_util.OpenocdRtt('-f fake.cfg', 0x20000000, 1, serial_no='000',
+                                      reset_before_attach=flag)
+            try:
+                argv = Path(argv_file).read_text()
+            finally:
+                con.close()
+            self.assertEqual('reset run' in argv, want, argv)
+            if want:   # ordering is the whole point: reset, settle, THEN attach
+                self.assertLess(argv.index('reset run'), argv.index('rtt setup'), argv)
+                self.assertIn('sleep 2000', argv)
+            self.assertIn('rtt server start', argv)
+            self.assertTrue(argv.rstrip().endswith('1'), argv)   # channel threaded through
+
+    def test_openocd_route_echo_and_teardown(self):
+        self._fake_path()
+        con = hil_util.OpenocdRtt('-f fake.cfg', 0x20000000, 0,
+                                  serial_no='000', vid_pid='0x1234 0x5678')
+        self.addCleanup(con.close)
+        out = b''
+        end = time.monotonic() + 3
+        while b'hello from target' not in out and time.monotonic() < end:
+            out += con.read(con.in_waiting or 1)
+        self.assertIn(b'hello from target', out)
+        con.write(b'ping')
+        end = time.monotonic() + 3
+        while b'PING' not in out and time.monotonic() < end:
+            out += con.read(con.in_waiting or 1)
+        self.assertIn(b'PING', out)
+        proc = con._proc
+        con.close()
+        self.assertIsNotNone(proc.poll())      # no zombie, no probe held
+        with self.assertRaises(RuntimeError):
+            con.write(b'x')                    # same post-close contract as JlinkRtt
+
+
+if __name__ == '__main__':
+    unittest.main()
diff --git a/test/hil/test/test_hil_util.py b/test/hil/test/test_hil_util.py
index e06d2ba8b..1a283bda7 100644
--- a/test/hil/test/test_hil_util.py
+++ b/test/hil/test/test_hil_util.py
@@ -145,9 +145,13 @@ class BottomLayer(unittest.TestCase):
         # hil_pool_check included: test_hil_util_is_a_single_module_instance imports it
         # on the bare runner, and its `import serial` is function-local for exactly
         # this reason -- hoisting it must fail HERE, not on every PR's pre-commit CI
+        # ../../tools/rtt: hil_util exec_module's it at import (helper/hil_util.py's
+        # loader block), so a non-stdlib import THERE kills ci_select on the bare
+        # runner just as surely -- and the spec_from_file_location call is invisible to
+        # the ast.Import walk below, which is why it must be listed explicitly
         for mod in ('helper/hil_util', 'hil_flash', '../../tools/ci_select',
                     'helper/hil_health', 'helper/hil_lock', 'helper/hil_pool_check',
-                    '../../tools/build', '../../tools/build_utils'):
+                    '../../tools/build', '../../tools/build_utils', '../../tools/rtt'):
             tree = ast.parse((hil_dir / f'{mod}.py').read_text())
             # module level only: a deferred import inside a function cannot break
             # importability (hil_pool_check keeps `import serial` function-local