path: root/tools/rtt.py

#!/usr/bin/env python3
"""RTT console/capture over a debug probe — importable classes + CLI (the rtt
skill's SKILL.md is the manual).

Three routes (see the skill's transport matrix for which route a probe gets).
--backend is always explicit:

  J-Link route (console/capture, channel 0 only)
      rtt.py --backend jlink --probe <sn> --device <JLINK_DEVICE> [--seconds N] [-i]
  OpenOCD route (native probes: ST-Link/CMSIS-DAP; console/capture, any channel)
      rtt.py --backend openocd [--probe <sn>] [--vid-pid "0xVVVV 0xPPPP"] \\
             --cfg "-f interface/stlink.cfg -f target/stm32h7x.cfg" \\
             (--elf <flashed.elf> | --addr 0x2000xxxx) [--channel N] [--seconds N] [-i]
             [--reset-before-attach]   # capture from the target's boot (SystemView)
  Post-mortem ring dump (J-Link, no halt — debug-AP reads)
      rtt.py --backend jlink --dump <out.bin> --probe <sn> --device <JLINK_DEVICE> \\
             (--elf <flashed.elf> | --addr 0x...)

The probe is owned for the whole run: flash and reset BEFORE starting this, never
reset the target while it is attached. Pin the probe: rigs and benches run
several (jlink: --probe serial; openocd: --probe and/or --vid-pid).

The classes (JlinkRtt for J-Link, OpenocdRtt for openocd-driven probes) expose
the slice of pyserial the HIL harness uses — read/in_waiting/write/close/timeout,
reset_input_buffer, context-manager use, plus an `eof` latch — and are imported
by test/hil/helper/hil_util.py, so this file is HARNESS-CRITICAL: a change here
is classified like a test/hil/ harness change (tools/ci_select.py) and runs the
console unit tests (pre-commit hil-test hook, test/hil/test/test_hil_rtt.py).
Stdlib only — hil_util imports this file, never the other way around.
"""
import argparse
import contextlib
import os
import re
import select
import shlex
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time


class RttError(RuntimeError):
    """Every way a console can break: stall, closed, dead or reset server.

    A RuntimeError subclass so existing `except RuntimeError` callers keep working,
    but named so the harness can tell a console failure from an unrelated
    NotImplementedError / 'dictionary changed size during iteration' and stop
    reporting harness bugs as board failures."""


def _pos_float_env(name: str, default: float) -> float:
    # mirrors hil_util.pos_float_env, including its rejection of inf/nan: an infinite
    # write timeout is an unbounded write, the very thing this knob exists to bound
    raw = os.environ.get(name)
    if raw is None:
        return default
    try:
        v = float(raw)
    except ValueError:
        print(f'warning: {name} is not a number; using {default}', file=sys.stderr, flush=True)
        return default
    if not (v > 0 and v < float('inf')):
        print(f'warning: {name}={v} is not usable; using {default}', file=sys.stderr, flush=True)
        return default
    return v


# whole-call deadline for write() — same env knob as the harness's serial twin
RTT_WRITE_TIMEOUT = _pos_float_env('HIL_SERIAL_WRITE_TIMEOUT', 10)

# J-Link Commander's telnet greeting, sent at connect BEFORE (or without) the control
# block being found: never target output. Three lines; the middle one is the PROBE
# MODEL string, which in libjlinkarm carries no 'SEGGER ' prefix (J-Link OH3,
# J-Trace H9, ...) though some builds do prefix it — match both shapes. Consumers
# judging "did the target speak" must strip these lines first.
RTT_BANNER_RE = re.compile(r'^(SEGGER J-|J-Link[ 0-9]|J-Trace[ 0-9]|Process:\s)')


def strip_banner(data: bytes, complete_only: bool = False) -> bytes:
    """Target bytes only: drop the J-Link server banner lines and blanks.

    Both harness consumers (hil_test's device_info verdict, hil_pool_check's
    aliveness score) must judge "did the target speak" through this one filter,
    or the same byte stream scores differently per consumer. complete_only=True
    additionally drops a trailing unterminated line — for poll loops judging a
    growing buffer, where a banner FRAGMENT at a read boundary (b'SEGG', b'Proce')
    would defeat the prefix regex and count as target output; the final verdict
    after the window should pass complete_only=False to keep a genuine
    unterminated tail."""
    lines = data.splitlines(keepends=False)
    if complete_only and data and not data.endswith((b'\n', b'\r')) and lines:
        lines = lines[:-1]
    return b'\n'.join(l for l in lines
                      if l.strip() and not RTT_BANNER_RE.match(l.decode('utf-8', errors='ignore')))


def free_ports(count: int) -> list:
    """Bind ephemeral ports and hand back the numbers. Boards run in parallel, so the
    RTT/GDB ports cannot be the SEGGER defaults or two boards collide.

    Known TOCTOU: the port is free when released here, but another process can claim
    it before the server binds it. Accepted — the server binds the port itself, so
    there is no fd to hand over. The post-connect re-poll catches the common outcome
    (our server lost the bind and died); a foreign listener that stays alive is not
    detectable here and would need the connected peer to be validated."""
    socks = []
    try:
        for _ in range(count):
            s = socket.socket()
            s.bind(('127.0.0.1', 0))
            socks.append(s)
        return [s.getsockname()[1] for s in socks]
    finally:
        for s in socks:
            s.close()


def nm_rtt_addr(elf: str, nm: str = None) -> int:
    """Control-block address from the FLASHED elf's symbol table. --addr is the way
    out when nm cannot read the file (another architecture, no toolchain)."""
    nm = nm or os.environ.get('RTT_NM', 'arm-none-eabi-nm')
    try:
        r = subprocess.run([nm, elf], capture_output=True, text=True, timeout=30)
    except FileNotFoundError:
        raise SystemExit(f'{nm} not on PATH — set RTT_NM=<your-nm>, or pass --addr')
    except subprocess.TimeoutExpired:
        raise SystemExit(f'{nm} did not finish reading {elf} in 30 s — pass --addr instead')
    if r.returncode != 0:
        raise SystemExit(f'{nm} could not read {elf}: {r.stderr.strip()[:200]}\n'
                         f'(wrong architecture? set RTT_NM=<your-nm>, or pass --addr)')
    for line in r.stdout.splitlines():
        # "<addr> <type> _SEGGER_RTT": a defined data symbol only — an undefined one
        # ("         U _SEGGER_RTT") has no address and would int('U', 16)
        m = re.match(r'^([0-9a-fA-F]+)\s+[bBdD]\s+_SEGGER_RTT$', line.strip())
        if m:
            return int(m.group(1), 16)
    raise SystemExit(f'no defined _SEGGER_RTT symbol in {elf} — was it built with LOGGER=rtt?')


class _SocketRtt:
    """Shared console core: a TCP socket onto an RTT server owned by self._proc.

    Subclasses build their server argv and call _spawn() + _connect() in __init__.
    One failure contract: RttError for every way the console can break (stall,
    closed, dead server) — callers are written for exactly it. A dead or resetting
    server LATCHES `eof` rather than raising from the read side, so read loops and
    the harness's `assert not ser.eof` triage see it without an exception racing
    them to a generic handler."""

    server = 'RTT server'   # for error messages

    def __init__(self, timeout: float = 0.1):
        self.timeout = timeout
        self._buf = b''
        self._eof = False
        self._sock = None
        self._proc = None
        self._log = None
        self._lock = threading.Lock()   # _buf is touched by the CLI pump thread too

    def _spawn(self, cmd: list, stdin=None) -> None:
        # server output spools to a temp file: a PIPE nobody drains blocks a
        # single-threaded server once 64 KiB of log accumulates (openocd at
        # polling_interval 1 against a resetting target fills that in minutes) and
        # the console goes silent with no error; the file also feeds _server_tail
        self._log = tempfile.NamedTemporaryFile(prefix='rtt-server-', suffix='.log')
        try:
            self._proc = subprocess.Popen(cmd, stdin=stdin, stdout=self._log,
                                          stderr=subprocess.STDOUT, start_new_session=True)
        except FileNotFoundError as e:
            self.close()
            raise RttError(f'RTT console: {e.filename or cmd[0]} not on PATH') from e
        except BaseException:
            # any other spawn failure (PermissionError...) must not leak the log fd
            self.close()
            raise

    def _connect(self, port: int) -> None:
        try:
            deadline = time.monotonic() + 15
            while time.monotonic() < deadline:
                try:
                    self._sock = socket.create_connection(('127.0.0.1', port), timeout=2)
                    break
                except OSError:
                    if self._proc.poll() is not None:
                        break
                    time.sleep(0.2)
            if self._sock is None:
                tail = self._server_tail()
                self.close()
                raise RttError(f'RTT console: {self.server} did not serve port {port}{tail}')
            if self._proc.poll() is not None:
                # the connect succeeded but our server is dead: a foreign process claimed
                # the port in the free_ports window — refuse a console wired to a stranger
                self.close()
                raise RttError(f'RTT console: {self.server} died after connect (port {port} hijacked?)')
            self._sock.setblocking(False)
        except (KeyboardInterrupt, SystemExit):
            # a signal mid-construction must not orphan the server we just spawned
            self.close()
            raise

    def _server_tail(self) -> str:
        log = getattr(self, '_log', None)
        if log:
            with contextlib.suppress(OSError, ValueError):
                with open(log.name, 'rb') as fh:
                    tail = fh.read()[-400:].decode(errors='replace')
                if tail:
                    return ' — ' + tail
        return ''

    def _drain(self) -> None:
        # LATCH, never raise: a peer reset or a socket closed under us ends the
        # stream exactly like an orderly EOF. Raising here raced the harness's
        # `assert not ser.eof` triage into a generic handler that re-flashes the
        # board, and leaked ConnectionResetError/ValueError to in_waiting callers.
        # the WHOLE body under the lock, not just the append: the CLI's -i pump thread
        # and the read loop drain the same socket concurrently, and recv->append being
        # non-atomic let chunks land out of order (measured: transposed 64-byte
        # segments in 3/6 stress trials)
        try:
            with self._lock:
                while self._sock and select.select([self._sock], [], [], 0)[0]:
                    try:
                        chunk = self._sock.recv(65536)
                    except (BlockingIOError, InterruptedError):
                        return
                    if not chunk:
                        self._eof = True
                        return
                    self._buf += chunk
        except (OSError, ValueError, TypeError, AttributeError):
            self._eof = True

    @property
    def eof(self) -> bool:
        """True once the server hung up AND everything it sent has been read out."""
        if self._sock is None:
            return True
        self._drain()
        return self._eof and not self._buf

    @property
    def in_waiting(self) -> int:
        if self._sock is None:
            # pyserial raises on a closed port; answering "N bytes waiting" from a
            # closed dead console would let a caller bug look like a healthy board
            raise RttError('RTT console is closed')
        self._drain()
        return len(self._buf)

    def read(self, size: int = 1) -> bytes:
        if size is None or size <= 0:
            # pyserial's read(0) returns b'' and consumes nothing; a negative size
            # must not silently hand over (or destroy) buffered bytes
            return b''
        if self._sock is None:
            raise RttError('RTT console is closed')
        self._drain()
        deadline = None if self.timeout is None else time.monotonic() + self.timeout
        while (len(self._buf) < size and not self._eof
               and (deadline is None or time.monotonic() < deadline)):
            time.sleep(0.005)
            self._drain()
        if self._eof and len(self._buf) < size:
            # dead server: pace the empty returns like a serial timeout would, so a
            # caller's read loop cannot busy-spin at 100% CPU (416k empty reads/s
            # measured unpaced). timeout=None deliberately diverges from pyserial's
            # block-forever: the eof latch makes "server is gone" knowable, and an
            # eternal block on it helps nobody -- paced empties + .eof is the contract.
            pace = self.timeout if self.timeout is not None else 0.1
            remaining = (deadline - time.monotonic()) if deadline is not None else pace
            time.sleep(max(0.0, min(remaining, pace)))
        with self._lock:
            out, self._buf = self._buf[:size], self._buf[size:]
        return out

    def reset_input_buffer(self) -> None:
        # pyserial surface: the host tests flush pre-reset backlog through this
        if self._sock is None:
            raise RttError('RTT console is closed')
        self._drain()
        with self._lock:
            self._buf = b''

    def write(self, data: bytes) -> int:
        # select+send, not sendall(): the socket is non-blocking for reads, and sendall()
        # on a non-blocking socket raises BlockingIOError as soon as the send buffer is
        # full, with no count of what already went out -- a caller cannot resume without
        # duplicating bytes. Same reason serial_write_all treats a short write as fatal.
        sock = self._sock   # snapshot: close() from another thread nulls the attribute
        if sock is None:
            raise RttError('RTT console is closed')
        self._drain()
        if self._eof:
            # TCP accepts exactly one send after peer death — without this the bytes
            # would "succeed" into the void and the read timeout gets blamed on the target
            raise RttError(f'RTT console write to a dead server ({self.server} gone)')
        sent = 0
        deadline = time.monotonic() + RTT_WRITE_TIMEOUT
        while sent < len(data):
            if time.monotonic() > deadline:
                raise RttError(f'RTT console write stalled after {sent}/{len(data)} bytes')
            try:
                if not select.select([], [sock], [], 0.1)[1]:
                    continue
                sent += sock.send(data[sent:])
            except (BlockingIOError, InterruptedError):
                continue
            except (OSError, ValueError, TypeError, AttributeError) as e:
                # peer death (BrokenPipe/ConnectionReset) or the socket closed under us
                # mid-call: keep the class's one failure contract
                raise RttError(f'RTT console write failed after {sent}/{len(data)} bytes: {e}') from e
        return sent

    def _gentle_stop(self, proc) -> None:
        """Subclass hook: ask the server to exit before the group takedown."""

    def close(self) -> None:
        self._eof = True   # latch: post-close eof reads True, like a hung-up server
        if getattr(self, '_sock', None):
            self._sock.close()
            self._sock = None
        with self._lock:
            self._buf = b''   # pyserial contract: nothing is readable after close
        proc = getattr(self, '_proc', None)
        if proc:
            if proc.poll() is None:
                self._gentle_stop(proc)
                with contextlib.suppress(subprocess.TimeoutExpired):
                    proc.wait(timeout=5)
        if proc and proc.poll() is None:
            # own session (start_new_session), so the group takedown gets the server and
            # anything it spawned; leaving one alive would hold the probe for the next test
            try:
                os.killpg(proc.pid, signal.SIGTERM)
                proc.wait(timeout=5)
            except (ProcessLookupError, PermissionError):
                pass
            except subprocess.TimeoutExpired:
                with contextlib.suppress(ProcessLookupError, PermissionError):
                    os.killpg(proc.pid, signal.SIGKILL)
                # reap, or the server stays a zombie for the caller's lifetime
                with contextlib.suppress(subprocess.TimeoutExpired):
                    proc.wait(timeout=2)
        if proc:
            for pipe in (proc.stdin, proc.stdout):
                if pipe:
                    with contextlib.suppress(OSError, ValueError):
                        pipe.close()
        # the server spool file: one fd plus a /tmp file per console, and the server
        # grows it while alive -- GC is not a release policy on a rig
        log = getattr(self, '_log', None)
        if log:
            with contextlib.suppress(OSError, ValueError):
                log.close()
            self._log = None

    # a console dropped without close() must not hold the probe for the process's life
    def __enter__(self):
        return self

    def __exit__(self, *exc):
        self.close()

    def __del__(self):
        with contextlib.suppress(Exception):
            self.close()


class JlinkRtt(_SocketRtt):
    """Bidirectional console over SEGGER RTT channel 0, for J-Link probes (the only
    console on boards whose probe has no VCOM or whose BSP has no UART).

    J-Link Commander (JLinkExe) owns the probe and serves RTT channel 0 on
    -RTTTelnetPort -- what JLinkRTTClient talks to, minus its banner. It keeps
    hunting for the control block and streams whatever the buffer already holds,
    where JLinkRTTLogger searches once when it attaches and gives up. It also
    carries input, which the host tests that drive a menu need.

    The probe is held for as long as this is open, so flashing and resetting the
    board must happen before it is created or after close(). Select the probe by
    serial: rigs run more than one."""

    server = 'JLinkExe'

    def __init__(self, board: dict, timeout: float = 0.1):
        super().__init__(timeout)
        flasher = board['flasher']
        args = shlex.split(flasher.get('args', ''))
        if '-device' not in args:
            # fail with the real cause now: JLinkExe without a device blocks prompting
            # and would surface 15 s later as a misleading port error
            raise RttError(f'RTT console: no -device in flasher args: {flasher.get("args")!r}')
        port = free_ports(1)[0]
        # defaults first, the roster's args after so they can override (-if jtag,
        # -JLinkScriptFile, an explicit -speed). NOTE: hil_flash orders it the other
        # way (roster args first, its own -if/-speed last, so ITS defaults win) --
        # a roster override honored here is ignored by flash/reset; align them if a
        # roster ever carries such args. -ExitOnError makes a failed target connect
        # EXIT Commander
        # (a clean error with the log tail) instead of leaving a banner-only console
        cmd = ['JLinkExe', '-USB', str(flasher['uid']), '-if', 'swd',
               '-JTAGConf', '-1,-1', '-speed', 'auto', '-NoGui', '1',
               '-ExitOnError', '1', '-AutoConnect', '1',
               *args, '-RTTTelnetPort', str(port)]
        # stdin stays open: Commander exits when it runs out of input; close() writes
        # 'exit' there.
        self._spawn(cmd, stdin=subprocess.PIPE)
        self._connect(port)

    def _gentle_stop(self, proc) -> None:
        with contextlib.suppress(OSError, ValueError):
            proc.stdin.write(b'exit\n')
            proc.stdin.flush()
        # close our pipe end in its own suppress: a BrokenPipe on the write above must
        # not skip it (the base close also closes it for the server-already-dead path)
        with contextlib.suppress(OSError, ValueError):
            proc.stdin.close()


class OpenocdRtt(_SocketRtt):
    """The console surface over an openocd `rtt server` (native probes:
    ST-Link/CMSIS-DAP — never point openocd at ea4088's LPC-Link2, measured to
    knock that probe off USB; other J-Link-OB probes untested).

    Exact control-block address (never a full-RAM scan), polling_interval 1
    (default 100 ms polling loses most of a busy stream), attach WITHOUT reset —
    flash and reset before starting; `rtt start` needs the block to exist.
    reset_before_attach opts into an in-session reset for streams that only
    decode from byte 0 (SystemView)."""

    server = 'openocd'

    def __init__(self, cfg: str, addr: int, channel: int, serial_no: str = None,
                 vid_pid: str = None, timeout: float = 0.1, reset_before_attach: bool = False):
        super().__init__(timeout)
        port = free_ports(1)[0]
        # argv, never a shell string: cfg/serial/vid_pid come from roster JSON and the
        # command line, and a '$', backtick or quote in any of them would otherwise be
        # substituted by the shell or break out of it
        cmd = ['openocd', '-c', 'tcl_port disabled', '-c', 'gdb_port disabled',
               '-c', 'telnet_port disabled']
        # probe pin: vid_pid keeps discovery from opening foreign usbfs nodes (a
        # wedged one hangs the open), serial disambiguates same-model probes —
        # both before the -f scripts, like hil_flash does
        if vid_pid:
            if not re.fullmatch(r'0x[0-9a-fA-F]{1,4} 0x[0-9a-fA-F]{1,4}', vid_pid.strip()):
                # openocd only WARNS and exits 0 on a malformed value, so the pin
                # silently does not apply and discovery reopens every usbfs node --
                # the convoy hil_flash.valid_vid_pid exists to stop
                raise RttError(f'--vid-pid must be "0xVVVV 0xPPPP", got {vid_pid!r}')
            cmd += ['-c', f'adapter usb vid_pid {vid_pid.strip()}']
        if serial_no:
            cmd += ['-c', f'adapter serial {serial_no}']
        cmd += shlex.split(cfg)
        cmd += ['-c', 'init']
        # opt-in: reset the target INSIDE this session, give it 2 s to boot, THEN
        # attach and drain. The order is forced: `rtt start` needs the control block
        # to already exist in RAM (the firmware creates it at init), and attaching
        # ahead of the reset would latch the PREVIOUS run's stale block. Byte 0 still
        # reaches the consumer because NO_BLOCK_SKIP retains the ring's HEAD: a boot
        # burst bigger than the ring loses its tail until the drain catches up, never
        # its first bytes -- which is the part a boot-anchored decoder needs
        # (SystemView's Init record, carrying the timestamp frequency, is emitted once
        # at boot; a mid-flight attach yields a stream no decoder can lock onto; size
        # BUFFER_SIZE_UP to the boot burst if the tail matters too). Costs the tool's
        # usual no-reset invariant, and is unsafe on parts where an in-session reset
        # leaves the core held (SAMD5x DSU) or perturbs the target (WCH SDI).
        if reset_before_attach:
            cmd += ['-c', 'reset run', '-c', 'sleep 2000']
        cmd += ['-c', f'rtt setup 0x{addr:x} 0x800 "SEGGER RTT"',
                '-c', 'rtt polling_interval 1', '-c', 'rtt start',
                '-c', f'rtt server start {port} {channel}']
        self._spawn(cmd)
        self._connect(port)

    def _gentle_stop(self, proc) -> None:
        # no stdin channel to ask openocd to exit, and it keeps its listener up after
        # the client disconnects: go straight to the group takedown instead of blocking
        # the base class's 5 s wait on a process that has no reason to leave
        with contextlib.suppress(ProcessLookupError, PermissionError):
            os.killpg(proc.pid, signal.SIGTERM)


def dump_ring(probe: str, device: str, addr: int, out_path: str, channel: int = 0) -> int:
    """Post-mortem: read aUp[channel]'s ring over the debug AP (no halt) via JLinkExe.
    NO_BLOCK_SKIP means an undrained ring holds the FIRST KB after boot, not the
    tail — interpretation rules in the target-debug skill."""
    if re.search(r'[\s"\']', out_path):
        raise SystemExit(f'--dump path must not contain whitespace or quotes: {out_path!r} '
                         f'(it is spliced into a JLinkExe script line)')
    # a stale file from an earlier run must not satisfy the success check below
    with contextlib.suppress(OSError):
        os.remove(out_path)
    # SEGGER_RTT_CB: acID[16], MaxNumUpBuffers, MaxNumDownBuffers, then aUp[] at 0x18,
    # each ring 6 words {sName, pBuffer, SizeOfBuffer, WrOff, RdOff, Flags}. Read the
    # counts with the descriptor so an out-of-range channel is rejected instead of
    # reading whatever RAM follows the array.
    jlink = ['JLinkExe', '-USB', probe, '-device', device, '-if', 'swd',
             '-speed', '4000', '-NoGui', '1', '-AutoConnect', '1']

    def _jlink_run(script: str):
        # same clean-exit contract as nm_rtt_addr/_spawn: a missing binary or a wedged
        # probe must not reach the CLI as a traceback
        try:
            return subprocess.run(jlink, input=script, capture_output=True, text=True, timeout=60)
        except FileNotFoundError:
            raise SystemExit('JLinkExe not on PATH — the --dump route needs J-Link Commander')
        except subprocess.TimeoutExpired:
            raise SystemExit('JLinkExe did not finish in 60 s — probe wedged or target unreachable?')

    script = f'mem32 {addr + 0x10:#x}, 2\nmem32 {addr + 0x18 + channel * 24:#x}, 6\nexit\n'
    r = _jlink_run(script)
    words = []
    for line in r.stdout.splitlines():
        # UNANCHORED: when the script arrives on stdin, some JLinkExe versions glue
        # the 'J-Link>' prompt onto the result line with no newline between
        m = re.search(r'([0-9A-Fa-f]{8}) = ((?:[0-9A-Fa-f]{8} ?)+)$', line.strip())
        if m:
            words += [int(w, 16) for w in m.group(2).split()]
    if len(words) < 8:
        print(r.stdout[-500:], file=sys.stderr)
        raise SystemExit(f'could not read the aUp[{channel}] descriptor — wrong control block address?')
    max_up = words[0]
    if not 0 < max_up <= 32:
        raise SystemExit(f'control block at {addr:#x} looks uninitialized '
                         f'(MaxNumUpBuffers={max_up}) — the target has not written to RTT yet, '
                         f'or the address is wrong')
    if channel >= max_up:
        raise SystemExit(f'--channel {channel}: this firmware has {max_up} up-buffer(s) (0..{max_up - 1})')
    _, pbuf, size, wroff, rdoff, _ = words[2:8]
    if not pbuf or not size:
        raise SystemExit(f'up-buffer {channel} is not initialized (pBuffer={pbuf:#x} size={size}) — '
                         f'the target has not written to it yet')
    script = f'savebin {out_path}, {pbuf:#x}, {size:#x}\nexit\n'
    _jlink_run(script)
    # JLinkExe exits 0 even when a command inside its script fails, so the only proof
    # savebin worked is the file itself: it must hold the WHOLE ring, since a read that
    # dies partway (probe disconnect, unreadable address) still leaves a short file that
    # would otherwise be reported as a complete dump. Removing it also keeps the
    # invariant above -- no stale file can satisfy a later run's check.
    got = os.path.getsize(out_path) if os.path.exists(out_path) else 0
    if got < size:
        with contextlib.suppress(OSError):
            os.remove(out_path)
        if got == 0:
            raise SystemExit(f'savebin produced no data at {out_path} — probe or address problem')
        raise SystemExit(f'savebin wrote {got}/{size} B to {out_path} (truncated dump removed) '
                         f'— probe or address problem')
    print(f'ring: {size} B at {pbuf:#x}, WrOff={wroff:#x} RdOff={rdoff:#x} -> {out_path}\n'
          f'valid bytes wrap at WrOff; default NO_BLOCK_SKIP holds the FIRST data after '
          f'boot, not the tail', file=sys.stderr)
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument('--backend', choices=['jlink', 'openocd'], required=True,
                    help='transport route — explicit, no default (skill transport matrix)')
    ap.add_argument('--probe', help='probe serial (JLinkExe -USB / openocd "adapter serial")')
    ap.add_argument('--vid-pid', help='openocd probe pin by USB IDs, e.g. "0x2e8a 0x000c" '
                                      '(with or instead of --probe)')
    ap.add_argument('--device', help='JLINK_DEVICE from board.cmake/family.cmake (jlink backend)')
    ap.add_argument('--cfg', help='openocd -f/-c args, e.g. "-f interface/stlink.cfg -f target/stm32h7x.cfg"')
    ap.add_argument('--elf', help='the FLASHED elf: exact _SEGGER_RTT address via nm (openocd/--dump)')
    ap.add_argument('--addr', help='SEGGER RTT control block address (hex), instead of --elf')
    ap.add_argument('--channel', type=int, default=0, help='up-buffer index (0 console, 1 SysView)')
    ap.add_argument('--seconds', type=float, default=0, help='capture duration; 0 = until Ctrl-C/EOF')
    ap.add_argument('-i', '--interactive', action='store_true', help='forward stdin to the target')
    ap.add_argument('--reset-before-attach', action='store_true',
                    help='openocd: reset the target inside the capture session so the '
                         'server is draining when it boots (needed for streams that must '
                         'include the boot preamble, e.g. SystemView); unsafe on SAMD5x/WCH')
    ap.add_argument('--dump', metavar='OUT.bin',
                    help='post-mortem ring dump (jlink backend; needs --elf or --addr)')
    args = ap.parse_args()

    if args.seconds < 0 or args.seconds != args.seconds:   # negative or nan
        ap.error(f'--seconds must be >= 0 (0 = until Ctrl-C/EOF), got {args.seconds}')
    if args.channel < 0:
        # a negative index would walk backwards off aUp[] into the control-block
        # header and read garbage as a descriptor
        ap.error(f'--channel must be >= 0, got {args.channel}')

    def rtt_addr():
        if args.addr:
            try:
                return int(args.addr, 16)
            except ValueError:
                ap.error(f'--addr must be hex, got {args.addr!r}')
        if args.elf:
            return nm_rtt_addr(args.elf)
        ap.error('need --elf (flashed elf, address via nm) or --addr')

    if args.backend == 'jlink':
        if args.reset_before_attach:
            ap.error('--reset-before-attach is openocd-only (the J-Link route attaches '
                     'to a running target; flash and reset before starting it)')
        if args.channel and not args.dump:
            # -RTTTelnetPort serves the Terminal buffer only; --dump can read any ring
            ap.error('the jlink backend streams channel 0 only (use --backend openocd '
                     'for another channel, or --dump to read one)')
        if args.vid_pid:
            ap.error('--vid-pid is openocd-only; J-Link probes are selected by serial (--probe)')
        if not (args.probe and args.device):
            ap.error('the jlink backend needs --probe and --device')
    elif not (args.probe or args.vid_pid):
        ap.error('the openocd backend needs --probe and/or --vid-pid')

    if args.dump:
        if args.backend != 'jlink':
            ap.error('--dump uses the jlink backend (debug-AP reads via JLinkExe)')
        return dump_ring(args.probe, args.device, rtt_addr(), args.dump, args.channel)

    # install BEFORE the console exists: an external `timeout`/kill during the
    # up-to-15 s connect window must still reach the cleanup below, or the openocd
    # route leaves a server holding the probe and the port (JLinkExe would exit on
    # stdin EOF; openocd has no such channel and its own session shields it)
    def _terminate(signum, _frame):
        raise KeyboardInterrupt
    for _sig in (signal.SIGTERM, signal.SIGHUP):
        with contextlib.suppress(ValueError, OSError):
            signal.signal(_sig, _terminate)

    try:
        if args.backend == 'openocd':
            if not args.cfg:
                ap.error('--backend openocd needs --cfg')
            con = OpenocdRtt(args.cfg, rtt_addr(), args.channel,
                             serial_no=args.probe, vid_pid=args.vid_pid,
                             reset_before_attach=args.reset_before_attach)
        else:
            con = JlinkRtt({'flasher': {'uid': args.probe, 'args': f'-device {args.device}'}},
                           timeout=0.1)
    except RttError as e:
        print(e, file=sys.stderr)
        return 1
    except KeyboardInterrupt:
        return 130   # constructors clean up after themselves on the way out

    saw_output = threading.Event()
    forwarded = threading.Event()
    if args.interactive:
        def pump_stdin():
            # Hold input until the capture side has seen TARGET output (or 5 s for a
            # quiet firmware): the J-Link telnet route silently DROPS client bytes
            # until Commander locates the control block, so input forwarded at attach
            # vanishes (measured on the rig: instant 'ping' lost, delayed 'ping'
            # echoed). The gate must ignore the server's own banner — it arrives at
            # connect, BEFORE the block is found. Raw os.read, not sys.stdin.buffer:
            # bytes with no newline wait, and no BufferedReader lock — a daemon
            # thread blocked holding that lock at interpreter shutdown aborts
            # CPython (_enter_buffered_busy).
            saw_output.wait(5)
            try:
                while True:
                    data = os.read(0, 4096)
                    if not data:
                        return
                    con.write(data)
                    forwarded.set()
            except (RttError, OSError, ValueError):
                return   # console closed/stalled/dead; capture side reports the state
        threading.Thread(target=pump_stdin, daemon=True).start()

    deadline = time.monotonic() + args.seconds if args.seconds else None
    rc = 0
    seen = b''   # pre-release accumulator for the banner check only
    try:
        while deadline is None or time.monotonic() < deadline:
            try:
                chunk = con.read(con.in_waiting or 1)
            except RttError as e:
                print(f'rtt: {e}', file=sys.stderr)
                rc = 1
                break
            if chunk:
                if args.interactive and not saw_output.is_set():
                    # target data = anything past the J-Link banner's final line
                    # ('Process: <name>'); the openocd server has no banner
                    seen = (seen + chunk)[-65536:]
                    if args.backend != 'jlink':
                        saw_output.set()
                    else:
                        i = seen.find(b'Process: ')
                        j = seen.find(b'\n', i) if i >= 0 else -1
                        if j >= 0 and len(seen) > j + 1:
                            saw_output.set()
                sys.stdout.buffer.write(chunk)
                sys.stdout.buffer.flush()
            elif con.eof:
                print('rtt: server closed the connection', file=sys.stderr)
                rc = 1
                break
    except KeyboardInterrupt:
        pass
    except BrokenPipeError:
        # downstream consumer (head/grep -m) closed the pipe: a normal way to end a
        # capture, not an error. Point stdout at devnull so interpreter shutdown does
        # not raise on the final implicit flush.
        os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno())
    finally:
        if args.interactive and not forwarded.is_set():
            # only claim what is true: the gate releases after 5 s and forwards anyway,
            # so "never forwarded" must come from the forwarded flag, not the gate
            print('rtt: -i stdin was never forwarded to the target (no input arrived, '
                  'or the console closed first)', file=sys.stderr)
        if args.interactive and not saw_output.is_set():
            print('rtt: no target output within the window', file=sys.stderr)
        # a late TERM landing during the up-to-12 s teardown must not skip the kill
        # escalation and orphan the server -- cleanup is committed at this point
        for _sig in (signal.SIGTERM, signal.SIGHUP):
            with contextlib.suppress(ValueError, OSError):
                signal.signal(_sig, signal.SIG_IGN)
        con.close()
    return rc


if __name__ == '__main__':
    sys.exit(main())