From 0344a0e0ae5b4feefcf6ecd718e8c222b3087377 Mon Sep 17 00:00:00 2001
From: DTTerastar <DT@Terastar.biz>
Date: Fri, 31 Jul 2026 01:08:43 -0400
Subject: [PATCH 1/3] hil: flood the bench with unique BLE addresses, and watch
 heap
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The HIL bench cannot currently reproduce ESPresense#2309, and could not detect
it if it did.

Load: ble_flood.py advertises from a fresh static random address ~40x/sec off a
USB Bluetooth adapter on the bench host. ESPresense fingerprints those by MAC
(ID_TYPE_RAND_STATIC_MAC), so every rotation costs a node a slot against a pool
of 100-200 — the churn of a busy room, compressed. Driven raw over
HCI_CHANNEL_USER so BlueZ is neither required nor able to fight for the adapter.

Detection: hil_monitor.py gains two checks it was missing.

  - Restart mid-run (exit 7). Nothing noticed a node going down and coming back,
    because every other signal here is satisfied by the fresh boot. An S3
    power-cycling every ~6h under the low-heap watchdog passes the current 8h
    window clean, which is exactly what happened on #2309.
  - Free heap decline (exit 8). /json already reports freeHeap/maxHeap/
    fingerprints, so sampling it needs no firmware change. Every sample is
    logged, so the run itself answers leak vs fragmentation vs churn — the
    question #2309 took months to settle by hand.

Gating: the flood only runs while a HIL step asks for it, via a request file in
the already-mounted /var/lock/woodpecker. A directory rather than one flag
because the four device steps finish at different times, and stale requests
expire so a killed container cannot leave the bench spraying junk MACs at every
ESPresense node in the house.

Checks: test_heap_trend.py runs the real #2309 shapes through the verdict (S3
slide fails, C3 hold passes, transient dip passes); ble_flood.py --selftest
covers HCI framing, static-random address rules, and both directions of the gate
without hardware.

Not yet run against a real adapter — needs the USB dongle passed through to the
bench VM first.
---
 bench/README.md            |  48 ++++++
 bench/ble-flood.service    |  24 +++
 scripts/ble_flood.py       | 290 +++++++++++++++++++++++++++++++++++++
 scripts/hil_monitor.py     |  98 ++++++++++++-
 scripts/test_heap_trend.py |  59 ++++++++
 5 files changed, 518 insertions(+), 1 deletion(-)
 create mode 100644 bench/README.md
 create mode 100644 bench/ble-flood.service
 create mode 100644 scripts/ble_flood.py
 create mode 100644 scripts/test_heap_trend.py

diff --git a/bench/README.md b/bench/README.md
new file mode 100644
index 0000000..ede1f6e
--- /dev/null
+++ b/bench/README.md
@@ -0,0 +1,48 @@
+# Bench host setup
+
+Host-side pieces for the HIL bench (`tsi-ha`). Everything else in this repo runs inside
+the `firmware-tester` container; these do not, because they drive hardware the container
+has no access to.
+
+## BLE flood
+
+`scripts/ble_flood.py` advertises from a fresh random address ~40 times a second, so each
+rotation costs a node a new fingerprint slot. It exists to make the slow heap decline in
+[ESPresense#2309](https://github.com/ESPresense/ESPresense/issues/2309) show up in a HIL
+window instead of over days on someone's shelf.
+
+It talks to the adapter over `HCI_CHANNEL_USER`, so **BlueZ is not required and should not
+be installed** — the kernel hands over the controller exclusively and `bluetoothd` would
+only fight for it.
+
+Requires a USB Bluetooth adapter passed through to the VM. Verify the kernel sees it:
+
+```bash
+ls /sys/class/bluetooth        # expect hci0
+python3 ble_flood.py --selftest   # framing + address rules, no hardware needed
+sudo python3 ble_flood.py --index 0 --seconds 30   # 30s live burst
+```
+
+Install:
+
+```bash
+sudo install -m 755 scripts/ble_flood.py /usr/local/bin/ble_flood.py
+sudo install -m 644 bench/ble-flood.service /etc/systemd/system/
+sudo systemctl enable --now ble-flood
+```
+
+The service idles until a request file appears in `/var/lock/woodpecker/bleflood.d/`. Each
+HIL step creates one on entry and removes it on exit, so the flood only runs during tests —
+otherwise every ESPresense node within range spends the day logging junk MACs. It is a
+directory, not a single flag, because the four device steps run in parallel and the first
+one to finish must not cut the flood out from under the rest. Request files older than
+`--max-age` (9h) are ignored, so a hard-killed container cannot leave the bench advertising
+forever.
+
+To flood by hand (e.g. reproducing a report):
+
+```bash
+sudo mkdir -p /var/lock/woodpecker/bleflood.d
+sudo touch /var/lock/woodpecker/bleflood.d/manual   # journalctl -fu ble-flood to watch
+sudo rm /var/lock/woodpecker/bleflood.d/manual
+```
diff --git a/bench/ble-flood.service b/bench/ble-flood.service
new file mode 100644
index 0000000..52892b5
--- /dev/null
+++ b/bench/ble-flood.service
@@ -0,0 +1,24 @@
+[Unit]
+Description=BLE advertisement flood for the ESPresense HIL bench
+Documentation=https://github.com/ESPresense/firmware-tester
+After=network.target
+
+[Service]
+# Runs continuously but only advertises while a HIL step asks for it. Steps already
+# bind-mount /var/lock/woodpecker, so each one drops a request file there and removes it on
+# exit — no host access, no extra plumbing, and the bench is quiet between runs instead of
+# filling every ESPresense node in the house with junk fingerprints.
+ExecStart=/usr/local/bin/ble_flood.py --index 0 --rate 40 --flag-dir /var/lock/woodpecker/bleflood.d
+Restart=always
+RestartSec=5
+# HCI_CHANNEL_USER needs CAP_NET_ADMIN and an adapter no one else has powered up.
+AmbientCapabilities=CAP_NET_ADMIN
+CapabilityBoundingSet=CAP_NET_ADMIN
+NoNewPrivileges=yes
+ProtectSystem=strict
+ProtectHome=yes
+PrivateTmp=yes
+ReadWritePaths=/var/lock/woodpecker
+
+[Install]
+WantedBy=multi-user.target
diff --git a/scripts/ble_flood.py b/scripts/ble_flood.py
new file mode 100644
index 0000000..8e86125
--- /dev/null
+++ b/scripts/ble_flood.py
@@ -0,0 +1,290 @@
+#!/usr/bin/env python3
+"""Flood the bench with BLE advertisements from a fresh random address every time.
+
+Why: ESPresense fingerprints a *static random* address by MAC (BleFingerprint.cpp,
+ID_TYPE_RAND_STATIC_MAC when the top two bits of the MSB are set), so every rotation
+here costs the node a new fingerprint slot against a pool of 100-200. That is the
+churn a real room full of phones produces, compressed — the load under which
+ESPresense#2309's slow heap decline shows up in hours instead of days.
+
+The adapter is driven raw over HCI_CHANNEL_USER, which means BlueZ is not involved and
+does not need to be installed: the kernel hands us the controller exclusively and
+nothing else can fight us for it. Requires root (CAP_NET_ADMIN) and an adapter that no
+other process has powered up.
+
+  ble_flood.py --index 0 --rate 40
+  ble_flood.py --index 0 --flag /var/lock/woodpecker/bleflood.on   # only while flag exists
+  ble_flood.py --selftest                                          # no hardware needed
+"""
+
+import argparse
+import ctypes
+import fcntl
+import os
+import secrets
+import socket
+import struct
+import time
+
+AF_BLUETOOTH = 31
+BTPROTO_HCI = 1
+HCI_CHANNEL_USER = 1
+HCIDEVDOWN = 0x400448CA
+
+HCI_COMMAND_PKT = 0x01
+
+OCF_RESET = 0x0C03
+OCF_LE_SET_RANDOM_ADDRESS = 0x2005
+OCF_LE_SET_ADV_PARAMETERS = 0x2006
+OCF_LE_SET_ADV_DATA = 0x2008
+OCF_LE_SET_ADV_ENABLE = 0x200A
+
+ADV_NONCONN_IND = 0x03
+OWN_ADDR_TYPE_RANDOM = 0x01
+ADV_INTERVAL = 0x0020  # 20 ms — the BLE minimum, so a packet lands on all 3 channels
+
+
+def hci_command(opcode, params=b""):
+    """Frame one HCI command packet: type, opcode (LE), parameter length, parameters."""
+    if len(params) > 255:
+        raise ValueError(f"HCI parameters too long: {len(params)}")
+    return struct.pack("<BHB", HCI_COMMAND_PKT, opcode, len(params)) + params
+
+
+def random_static_address():
+    """A valid BLE static random address, in the little-endian order HCI wants.
+
+    Spec: the two most significant bits must both be 1, and the remaining 46 bits must
+    not be all-zeros or all-ones. The MSB is the *last* byte on the wire. Those top bits
+    are also exactly what makes ESPresense classify it as ID_TYPE_RAND_STATIC_MAC rather
+    than discarding it as a resolvable private address it cannot resolve.
+    """
+    while True:
+        addr = bytearray(secrets.token_bytes(6))
+        addr[5] |= 0xC0
+        rest = int.from_bytes(addr, "little") & ((1 << 46) - 1)
+        if rest not in (0, (1 << 46) - 1):
+            return bytes(addr)
+
+
+def advertising_payload(name=b"HIL-FLOOD"):
+    """A minimal but realistic non-connectable advert: flags + complete local name."""
+    fields = bytes([2, 0x01, 0x06]) + bytes([len(name) + 1, 0x09]) + name
+    if len(fields) > 31:
+        raise ValueError(f"advertising payload too long: {len(fields)}")
+    return bytes([len(fields)]) + fields.ljust(31, b"\x00")
+
+
+def adv_parameters():
+    return struct.pack(
+        "<HHBBB6sBB",
+        ADV_INTERVAL, ADV_INTERVAL,
+        ADV_NONCONN_IND,
+        OWN_ADDR_TYPE_RANDOM,
+        0x00,            # peer address type (unused for undirected)
+        b"\x00" * 6,     # peer address (unused)
+        0x07,            # all three advertising channels
+        0x00,            # no filtering — anyone may scan
+    )
+
+
+def open_adapter(index):
+    """Take exclusive raw control of hciN.
+
+    The kernel only allows HCI_CHANNEL_USER on a *down* adapter, which is also what
+    guarantees exclusivity: if this succeeds, nothing else is driving the controller.
+    """
+    ctl = socket.socket(AF_BLUETOOTH, socket.SOCK_RAW, BTPROTO_HCI)
+    try:
+        fcntl.ioctl(ctl.fileno(), HCIDEVDOWN, index)
+    except OSError:
+        # Already down is the normal case (no bluetoothd). A real problem — missing
+        # adapter, no privileges — surfaces with a clearer message on bind() below.
+        pass
+    finally:
+        ctl.close()
+
+    sock = socket.socket(AF_BLUETOOTH, socket.SOCK_RAW, BTPROTO_HCI)
+    # CPython's bind() for BTPROTO_HCI cannot set hci_channel, so build sockaddr_hci
+    # ourselves: { sa_family, hci_dev, hci_channel }, all u16.
+    addr = struct.pack("<HHH", AF_BLUETOOTH, index, HCI_CHANNEL_USER)
+    libc = ctypes.CDLL("libc.so.6", use_errno=True)
+    if libc.bind(sock.fileno(), ctypes.c_char_p(addr), len(addr)) != 0:
+        err = ctypes.get_errno()
+        sock.close()
+        raise OSError(err, f"bind(hci{index}, HCI_CHANNEL_USER) failed: {os.strerror(err)}. "
+                           f"Is bluetoothd holding the adapter, or is this not root?")
+    sock.setblocking(False)
+    return sock
+
+
+def send(sock, opcode, params=b""):
+    sock.sendall(hci_command(opcode, params))
+    drain(sock)
+
+
+def drain(sock):
+    """Consume controller events. Unread events eventually stall the socket."""
+    while True:
+        try:
+            sock.recv(1024)
+        except BlockingIOError:
+            return
+        except OSError:
+            return
+
+
+def flood_requested(flag_dir, max_age):
+    """True while any HIL step is asking for load.
+
+    A directory rather than a single on/off file because the four device steps run in
+    parallel and finish at different times — one step exiting must not cut the flood out
+    from under the other three. Each step owns one file and removes it on exit.
+
+    Entries older than max_age are ignored so a hard-killed container cannot leave the
+    bench advertising forever, which would fill every ESPresense node in range with junk.
+    """
+    if not flag_dir:
+        return True
+    now = time.time()
+    try:
+        with os.scandir(flag_dir) as entries:
+            for entry in entries:
+                try:
+                    if now - entry.stat().st_mtime < max_age:
+                        return True
+                except FileNotFoundError:
+                    continue  # step cleaned up mid-scan
+    except FileNotFoundError:
+        return False
+    return False
+
+
+def flood(sock, rate, flag_dir, max_age, stop_after):
+    """Rotate the advertised address forever (or until the flag clears / time runs out).
+
+    Order matters: the controller rejects LE Set Random Address while advertising is
+    enabled, so each rotation is disable -> re-address -> enable.
+    """
+    send(sock, OCF_RESET)
+    time.sleep(0.1)
+    send(sock, OCF_LE_SET_ADV_PARAMETERS, adv_parameters())
+    send(sock, OCF_LE_SET_ADV_DATA, advertising_payload())
+
+    interval = 1.0 / rate
+    started = time.monotonic()
+    rotations = 0
+    reported = started
+    advertising = False
+
+    while True:
+        if stop_after and time.monotonic() - started >= stop_after:
+            break
+        if not flood_requested(flag_dir, max_age):
+            if advertising:
+                send(sock, OCF_LE_SET_ADV_ENABLE, b"\x00")
+                advertising = False
+                print(f"[flood] paused — no requests in {flag_dir}", flush=True)
+            time.sleep(2)
+            continue
+        if not advertising and flag_dir:
+            print(f"[flood] resumed — request present in {flag_dir}", flush=True)
+
+        cycle = time.monotonic()
+        send(sock, OCF_LE_SET_ADV_ENABLE, b"\x00")
+        send(sock, OCF_LE_SET_RANDOM_ADDRESS, random_static_address())
+        send(sock, OCF_LE_SET_ADV_ENABLE, b"\x01")
+        advertising = True
+        rotations += 1
+
+        now = time.monotonic()
+        if now - reported >= 30:
+            print(f"[flood] {rotations} unique addresses in {now - started:.0f}s "
+                  f"({rotations / (now - started):.1f}/s)", flush=True)
+            reported = now
+        time.sleep(max(0.0, interval - (now - cycle)))
+
+    send(sock, OCF_LE_SET_ADV_ENABLE, b"\x00")
+    print(f"[flood] stopped after {rotations} unique addresses", flush=True)
+
+
+def selftest():
+    """Check the packet framing and address rules without touching an adapter."""
+    pkt = hci_command(OCF_LE_SET_ADV_ENABLE, b"\x01")
+    assert pkt == b"\x01\x0a\x20\x01\x01", pkt.hex()
+
+    assert hci_command(OCF_RESET) == b"\x01\x03\x0c\x00"
+
+    for _ in range(2000):
+        addr = random_static_address()
+        assert len(addr) == 6
+        # The MSB is the last byte on the wire; ESPresense keys ID_TYPE_RAND_STATIC_MAC
+        # off exactly this test, so if it ever fails the flood stops being fingerprinted.
+        assert addr[5] & 0xC0 == 0xC0, addr.hex()
+        rest = int.from_bytes(addr, "little") & ((1 << 46) - 1)
+        assert rest not in (0, (1 << 46) - 1)
+
+    assert len({random_static_address() for _ in range(5000)}) == 5000, "addresses repeat"
+
+    payload = advertising_payload()
+    assert len(payload) == 32 and payload[0] == 14, payload.hex()
+    assert len(adv_parameters()) == 15, len(adv_parameters())
+
+    try:
+        hci_command(OCF_LE_SET_ADV_DATA, b"\x00" * 256)
+        raise AssertionError("oversized parameters must be rejected")
+    except ValueError:
+        pass
+
+    # The gate decides whether the bench sprays junk MACs across the house, so both
+    # directions matter: it must come on for a live step and go off for a dead one.
+    import tempfile
+    with tempfile.TemporaryDirectory() as d:
+        assert not flood_requested(d, 3600), "empty dir must not request load"
+        assert not flood_requested(os.path.join(d, "gone"), 3600), "missing dir must not request"
+
+        live = os.path.join(d, "esp32s3")
+        open(live, "w").close()
+        assert flood_requested(d, 3600), "a step's request file must turn the flood on"
+
+        # A step killed hard leaves its file behind; staleness must not flood forever.
+        os.utime(live, (0, 0))
+        assert not flood_requested(d, 3600), "stale request must be ignored"
+
+        os.remove(live)
+        assert not flood_requested(d, 3600), "removed request must turn the flood off"
+    assert flood_requested(None, 3600), "ungated mode must always advertise"
+
+    print("selftest OK")
+
+
+def main():
+    p = argparse.ArgumentParser(description="BLE advertisement flood with unique addresses")
+    p.add_argument("--index", type=int, default=0, help="hciN adapter index")
+    p.add_argument("--rate", type=float, default=40.0, help="address rotations per second")
+    p.add_argument("--flag-dir", help="only advertise while this directory holds a request file")
+    p.add_argument("--max-age", type=float, default=9 * 3600,
+                   help="ignore request files older than this many seconds")
+    p.add_argument("--seconds", type=float, default=0, help="stop after N seconds (0 = forever)")
+    p.add_argument("--selftest", action="store_true", help="verify framing, no hardware")
+    args = p.parse_args()
+
+    if args.selftest:
+        selftest()
+        return
+    if args.rate <= 0:
+        p.error("--rate must be positive")
+
+    sock = open_adapter(args.index)
+    print(f"[flood] hci{args.index} claimed, rotating at {args.rate}/s"
+          + (f", gated on {args.flag_dir}" if args.flag_dir else ""), flush=True)
+    try:
+        flood(sock, args.rate, args.flag_dir, args.max_age, args.seconds)
+    except KeyboardInterrupt:
+        send(sock, OCF_LE_SET_ADV_ENABLE, b"\x00")
+    finally:
+        sock.close()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/hil_monitor.py b/scripts/hil_monitor.py
index 690b125..d64e975 100644
--- a/scripts/hil_monitor.py
+++ b/scripts/hil_monitor.py
@@ -10,12 +10,15 @@
   4 = no scan results seen
   5 = firmware MQTT reconnect limit reached
   6 = /json endpoint misbehaved under concurrent load
+  7 = node restarted mid-run
+  8 = free heap declined over the run
 """
 
 import argparse
 import http.client
 import json
 import re
+import statistics
 import sys
 import threading
 import time
@@ -35,11 +38,29 @@
 MQTT_RECONNECT_LIMIT_PATTERN = "Too many reconnect attempts; Restarting"
 DEFAULT_SCAN_RESULT_PATTERN = r"^\s*\d+\s+\w+\s+\|"
 
+# setup() prints this on every boot. Seeing it *after* boot was confirmed means the node
+# went down and came back — a silent restart loop otherwise passes the whole window,
+# because every other check here is satisfied by the fresh boot (ESPresense#2309).
+REBOOT_PATTERN = "Pre-Setup Free Mem:"
+# The firmware's own low-heap watchdog announcing a restart. Same failure, better message.
+OOM_RESTART_PATTERN = "Out of memory for"
+
 IP_PATTERN = re.compile(r"IP address:\s*(\d+\.\d+\.\d+\.\d+)")
 JSON_CHECK_DELAY_SECS = 15  # let the web server settle after boot
 JSON_CHECK_WORKERS = 3      # concurrent connections — enough to reliably collide on serveJson
 JSON_CHECK_REQUESTS = 12    # sequential requests per worker, on one kept-alive connection
 
+# Heap trend. /json already reports freeHeap/maxHeap/fingerprints, so sampling it needs no
+# firmware change. A leak shows as freeHeap sliding while fingerprints stays flat; a
+# fragmentation problem shows as maxHeap sliding while freeHeap holds; fingerprint churn
+# shows as both moving with the device count. All three are printed so the log answers
+# which one it is without a re-run.
+HEAP_SAMPLE_SECS = 60
+HEAP_SETTLE_SECS = 120        # ignore the post-boot allocation burst
+HEAP_TREND_MIN_SECS = 1800    # below this the window is too short for a slope to mean anything
+HEAP_TREND_EDGE = 5           # samples averaged at each end
+HEAP_DECLINE_FRAC = 0.25      # fail if the tail lost more than this much of the baseline
+
 
 def json_endpoint_check(ip, bug):
     """Detect the serveJson() double-send under concurrent load.
@@ -163,6 +184,60 @@ def worker(n, drops):
         print(f"[hil] /json check passed ({total} concurrent requests to {ip}){suffix}", flush=True)
 
 
+def heap_sampler(ip, samples, stop):
+    """Poll /json every HEAP_SAMPLE_SECS and record freeHeap/maxHeap/fingerprints.
+
+    Read-only and deliberately gentle — one request a minute is nothing next to the
+    concurrent burst json_endpoint_check already fires. Failures to sample are skipped
+    rather than recorded: a missed poll is a network hiccup, not a heap measurement, and
+    inventing a zero here would read as a catastrophic leak.
+    """
+    stop.wait(HEAP_SETTLE_SECS)
+    while not stop.is_set():
+        try:
+            conn = http.client.HTTPConnection(ip, timeout=10)
+            conn.request("GET", "/json")
+            resp = conn.getresponse()
+            body = resp.read()
+            conn.close()
+            if resp.status == 200:
+                doc = json.loads(body)
+                free, mx = doc.get("freeHeap"), doc.get("maxHeap")
+                if isinstance(free, int) and isinstance(mx, int):
+                    fp = doc.get("fingerprints")
+                    samples.append((time.monotonic(), free, mx, fp))
+                    print(f"[hil] heap freeHeap={free} maxHeap={mx} fingerprints={fp}", flush=True)
+        except (OSError, http.client.HTTPException, ValueError):
+            pass  # a missed sample is not a measurement — see docstring
+        stop.wait(HEAP_SAMPLE_SECS)
+
+
+def heap_verdict(samples, duration):
+    """Return a failure string if free heap trended down over the run, else None."""
+    if duration < HEAP_TREND_MIN_SECS:
+        return None  # a 3-minute PR run says nothing about a multi-hour slope
+    if len(samples) < HEAP_TREND_EDGE * 2:
+        print(f"[hil] heap trend SKIPPED — only {len(samples)} samples", flush=True)
+        return None
+
+    def edges(index):
+        head = statistics.median(s[index] for s in samples[:HEAP_TREND_EDGE])
+        tail = statistics.median(s[index] for s in samples[-HEAP_TREND_EDGE:])
+        return head, tail
+
+    free0, free1 = edges(1)
+    max0, max1 = edges(2)
+    fp0, fp1 = samples[0][3], samples[-1][3]
+    summary = (f"freeHeap {free0:.0f}->{free1:.0f}, maxHeap {max0:.0f}->{max1:.0f}, "
+               f"fingerprints {fp0}->{fp1}, over {format_duration(int(duration))}")
+
+    if free1 < free0 * (1 - HEAP_DECLINE_FRAC):
+        lost = (1 - free1 / free0) * 100
+        return f"Free heap fell {lost:.0f}% ({summary})"
+    print(f"[hil] heap trend OK ({summary})", flush=True)
+    return None
+
+
 class _Bug(Exception):
     """A /json contract violation that must fail the build."""
 
@@ -224,6 +299,8 @@ def main():
     booted = False
     saw_scan_result = False
     json_failure = []  # appended to by the /json check thread
+    heap_samples = []  # appended to by the heap sampler thread
+    stop_sampling = threading.Event()
 
     try:
         while True:
@@ -242,6 +319,11 @@ def main():
                         f"{format_duration(args.duration)}."
                     )
                     sys.exit(4)
+                stop_sampling.set()
+                decline = heap_verdict(heap_samples, elapsed)
+                if decline:
+                    print(f"FAIL: {decline}")
+                    sys.exit(8)
                 print(f"PASS: {format_duration(args.duration)} elapsed cleanly.")
                 sys.exit(0)
 
@@ -269,15 +351,29 @@ def main():
                 print(f"[hil] Boot confirmed at {elapsed:.1f}s")
                 match = IP_PATTERN.search(line)
                 if match:
-                    # Background thread so the serial buffer keeps draining while we probe.
+                    # Background threads so the serial buffer keeps draining while we probe.
                     threading.Thread(
                         target=json_endpoint_check,
                         args=(match.group(1), json_failure),
                         daemon=True,
                     ).start()
+                    threading.Thread(
+                        target=heap_sampler,
+                        args=(match.group(1), heap_samples, stop_sampling),
+                        daemon=True,
+                    ).start()
                 else:
                     print(f"[hil] /json check SKIPPED — no IP in {line!r}")
 
+            # A restart after boot resets every other signal here — fresh boot, fresh heap,
+            # fresh scan results — so without this the window passes while the node is
+            # actually power-cycling every few hours (ESPresense#2309 on the S3).
+            elif booted and (REBOOT_PATTERN in line or OOM_RESTART_PATTERN in line):
+                why = ("firmware low-heap watchdog fired" if OOM_RESTART_PATTERN in line
+                       else "unexpected restart")
+                print(f"FAIL: Node restarted {elapsed:.0f}s into the run — {why}: {line.strip()!r}")
+                sys.exit(7)
+
             if MQTT_RECONNECT_LIMIT_PATTERN in line:
                 print(f"FAIL: Firmware MQTT reconnect limit reached: '{MQTT_RECONNECT_LIMIT_PATTERN}'")
                 sys.exit(5)
diff --git a/scripts/test_heap_trend.py b/scripts/test_heap_trend.py
new file mode 100644
index 0000000..071cd91
--- /dev/null
+++ b/scripts/test_heap_trend.py
@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+"""Prove heap_verdict fails on a real decline and stays quiet on healthy noise.
+
+The shapes come from ESPresense#2309: the S3 slid 85KB -> 33KB over ~6h with the
+fingerprint count flat, while the C3 sat at ~110KB for the same window. One of those
+must fail the build and the other must not.
+"""
+import os
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from hil_monitor import HEAP_TREND_MIN_SECS, heap_verdict  # noqa: E402
+
+HOURS_6 = 6 * 3600
+
+
+def samples(values, fingerprints=29):
+    return [(i * 60.0, v, v // 3, fingerprints) for i, v in enumerate(values)]
+
+
+def ramp(start, end, n):
+    step = (end - start) / (n - 1)
+    return [int(start + step * i) for i in range(n)]
+
+
+def wobble(level, n):
+    """Healthy heap is noisy, not flat — alternate around the level."""
+    return [level + (2000 if i % 2 else -2000) for i in range(n)]
+
+
+# The S3: a steady slide with the fingerprint count flat. Must fail.
+verdict = heap_verdict(samples(ramp(85000, 33000, 360)), HOURS_6)
+assert verdict and "fell" in verdict, f"leak not caught: {verdict!r}"
+assert "fingerprints 29->29" in verdict, verdict
+print(f"leak      -> {verdict}")
+
+# The C3: noisy but level. Must pass.
+verdict = heap_verdict(samples(wobble(110000, 360)), HOURS_6)
+assert verdict is None, f"healthy node failed: {verdict!r}"
+print("healthy   -> pass")
+
+# A dip that recovers is not a leak — the tail is what matters, and the median at each
+# edge keeps a single ugly sample from deciding the build.
+dip = wobble(110000, 150) + ramp(110000, 60000, 60) + ramp(60000, 108000, 150)
+verdict = heap_verdict(samples(dip), HOURS_6)
+assert verdict is None, f"transient dip failed the build: {verdict!r}"
+print("dip       -> pass")
+
+# A 25% floor means a shallow slide is tolerated; anything past it is not.
+assert heap_verdict(samples(ramp(100000, 80000, 360)), HOURS_6) is None, "20% should pass"
+assert heap_verdict(samples(ramp(100000, 70000, 360)), HOURS_6), "30% should fail"
+print("threshold -> 20% passes, 30% fails")
+
+# A 3-minute PR run cannot say anything about a multi-hour slope, so it must not try.
+assert heap_verdict(samples(ramp(85000, 33000, 360)), 180) is None, "short run must not judge"
+assert heap_verdict(samples(ramp(85000, 33000, 8)), HOURS_6) is None, "too few samples to judge"
+print("short run -> skipped")
+
+print("all heap trend checks passed")

From d55f609a489ddc8eb18f52b23c818c42bffadf01 Mon Sep 17 00:00:00 2001
From: DTTerastar <DT@Terastar.biz>
Date: Fri, 31 Jul 2026 18:16:31 -0400
Subject: [PATCH 2/3] hil: sample heap from /json/tele, and fail if it is
 missing
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two corrections found by running this against real hardware.

Unique identities: the flood advertised a constant name, and a live node
reported two different MACs both as id "name:hil-flood" — ESPresense ranks a
name (ID_TYPE_NAME, 35) above a static random address (5), so every advert
collapsed onto one logical id. The slot pool still churned, but the id space
this exists to exercise did not. The name now carries the full MAC; a short
suffix would collide tens of thousands of times over an 8h soak and quietly
merge ids again.

Sampling: /json only ever carried room/ver/firm, so the heap check would have
collected nothing and skipped silently forever. It now reads /json/tele
(ESPresense#2430), which is allocation-free and answers under low heap —
unlike /json, which refuses with 429 when it cannot afford a 12KB document and
would therefore hide the numbers exactly as a node degraded.

A missing /json/tele now fails the run rather than skipping. A heap check that
samples nothing and still reports PASS is how #2309 survived an 8h soak.
Unreachable nodes stay a skip: that is not a firmware fault.

Also: send() now waits for each command's completion event and checks status,
so a wedged adapter can no longer report thousands of addresses while
radiating nothing.
---
 scripts/ble_flood.py       | 101 +++++++++++++++++++++++++++++++------
 scripts/hil_monitor.py     |  44 +++++++++++-----
 scripts/test_heap_trend.py |  11 ++++
 3 files changed, 129 insertions(+), 27 deletions(-)

diff --git a/scripts/ble_flood.py b/scripts/ble_flood.py
index 8e86125..820a2a8 100644
--- a/scripts/ble_flood.py
+++ b/scripts/ble_flood.py
@@ -22,6 +22,7 @@
 import fcntl
 import os
 import secrets
+import select
 import socket
 import struct
 import time
@@ -32,6 +33,9 @@
 HCIDEVDOWN = 0x400448CA
 
 HCI_COMMAND_PKT = 0x01
+HCI_EVENT_PKT = 0x04
+EVT_CMD_COMPLETE = 0x0E
+EVT_CMD_STATUS = 0x0F
 
 OCF_RESET = 0x0C03
 OCF_LE_SET_RANDOM_ADDRESS = 0x2005
@@ -67,8 +71,19 @@ def random_static_address():
             return bytes(addr)
 
 
-def advertising_payload(name=b"HIL-FLOOD"):
-    """A minimal but realistic non-connectable advert: flags + complete local name."""
+def advertising_payload(address):
+    """Flags + a complete local name that is unique to this address.
+
+    The name must vary per rotation. ESPresense ranks a name (ID_TYPE_NAME, 35) above a
+    static random address (ID_TYPE_RAND_STATIC_MAC, 5), so a constant name would collapse
+    every advert in the flood onto one logical id — the slot pool would still churn, but
+    the id space this is meant to exercise would not. Confirmed on a live node, which
+    reported two different MACs both as id "name:hil-flood".
+
+    The full MAC goes in the name rather than a short suffix: at 40/s a 3-byte suffix
+    collides tens of thousands of times over an 8h soak, quietly merging ids again.
+    """
+    name = b"HIL-" + address[::-1].hex().encode()  # MSB-first, matching how nodes show it
     fields = bytes([2, 0x01, 0x06]) + bytes([len(name) + 1, 0x09]) + name
     if len(fields) > 31:
         raise ValueError(f"advertising payload too long: {len(fields)}")
@@ -118,20 +133,50 @@ def open_adapter(index):
     return sock
 
 
-def send(sock, opcode, params=b""):
-    sock.sendall(hci_command(opcode, params))
-    drain(sock)
+class HciError(Exception):
+    """The controller rejected a command, or never answered one."""
+
+
+def command_status(pkt, opcode):
+    """Status byte from a Command Complete/Status event for opcode, else None."""
+    if len(pkt) < 6 or pkt[0] != HCI_EVENT_PKT:
+        return None
+    if pkt[1] == EVT_CMD_COMPLETE:  # type, 0x0e, plen, ncmd, opcode(2), status
+        if struct.unpack_from("<H", pkt, 4)[0] != opcode:
+            return None
+        return pkt[6] if len(pkt) > 6 else 0x00
+    if pkt[1] == EVT_CMD_STATUS:    # type, 0x0f, plen, status, ncmd, opcode(2)
+        if len(pkt) < 7 or struct.unpack_from("<H", pkt, 5)[0] != opcode:
+            return None
+        return pkt[3]
+    return None
+
 
+def send(sock, opcode, params=b"", timeout=2.0):
+    """Send one command and confirm the controller accepted it.
 
-def drain(sock):
-    """Consume controller events. Unread events eventually stall the socket."""
+    Waiting for the completion event is what makes a run mean something: without it a
+    wedged or unplugged adapter silently swallows every command and the flood reports
+    thousands of addresses while radiating nothing at all.
+    """
+    sock.sendall(hci_command(opcode, params))
+    deadline = time.monotonic() + timeout
     while True:
+        remaining = deadline - time.monotonic()
+        if remaining <= 0:
+            raise HciError(f"opcode 0x{opcode:04x}: no completion event within {timeout}s")
+        if not select.select([sock], [], [], remaining)[0]:
+            continue
         try:
-            sock.recv(1024)
+            pkt = sock.recv(258)
         except BlockingIOError:
-            return
-        except OSError:
-            return
+            continue
+        status = command_status(pkt, opcode)
+        if status is None:
+            continue  # an unrelated event (advertising reports etc.) — keep draining
+        if status != 0x00:
+            raise HciError(f"opcode 0x{opcode:04x} rejected with status 0x{status:02x}")
+        return
 
 
 def flood_requested(flag_dir, max_age):
@@ -169,7 +214,6 @@ def flood(sock, rate, flag_dir, max_age, stop_after):
     send(sock, OCF_RESET)
     time.sleep(0.1)
     send(sock, OCF_LE_SET_ADV_PARAMETERS, adv_parameters())
-    send(sock, OCF_LE_SET_ADV_DATA, advertising_payload())
 
     interval = 1.0 / rate
     started = time.monotonic()
@@ -191,8 +235,10 @@ def flood(sock, rate, flag_dir, max_age, stop_after):
             print(f"[flood] resumed — request present in {flag_dir}", flush=True)
 
         cycle = time.monotonic()
+        address = random_static_address()
         send(sock, OCF_LE_SET_ADV_ENABLE, b"\x00")
-        send(sock, OCF_LE_SET_RANDOM_ADDRESS, random_static_address())
+        send(sock, OCF_LE_SET_RANDOM_ADDRESS, address)
+        send(sock, OCF_LE_SET_ADV_DATA, advertising_payload(address))
         send(sock, OCF_LE_SET_ADV_ENABLE, b"\x01")
         advertising = True
         rotations += 1
@@ -226,8 +272,15 @@ def selftest():
 
     assert len({random_static_address() for _ in range(5000)}) == 5000, "addresses repeat"
 
-    payload = advertising_payload()
-    assert len(payload) == 32 and payload[0] == 14, payload.hex()
+    # Every advert must carry an identity unique to its address, or ESPresense merges them.
+    addr_a, addr_b = random_static_address(), random_static_address()
+    payload = advertising_payload(addr_a)
+    assert len(payload) == 32, len(payload)
+    assert payload[0] == 21 and payload[1:4] == b"\x02\x01\x06", payload.hex()
+    assert payload[4:6] == b"\x11\x09", payload.hex()  # 16-byte name, "complete local name"
+    assert payload[6:22] == b"HIL-" + addr_a[::-1].hex().encode(), payload.hex()
+    assert advertising_payload(addr_b) != payload, "payload must vary with the address"
+    assert len({advertising_payload(random_static_address()) for _ in range(2000)}) == 2000
     assert len(adv_parameters()) == 15, len(adv_parameters())
 
     try:
@@ -236,6 +289,24 @@ def selftest():
     except ValueError:
         pass
 
+    # Completion parsing decides whether a rejected command is noticed at all. If this
+    # goes wrong the flood happily reports thousands of addresses while radiating none.
+    op = OCF_LE_SET_ADV_ENABLE
+    complete_ok = bytes([HCI_EVENT_PKT, EVT_CMD_COMPLETE, 4, 1]) + struct.pack("<H", op) + b"\x00"
+    complete_bad = bytes([HCI_EVENT_PKT, EVT_CMD_COMPLETE, 4, 1]) + struct.pack("<H", op) + b"\x12"
+    status_ok = bytes([HCI_EVENT_PKT, EVT_CMD_STATUS, 4, 0x00, 1]) + struct.pack("<H", op)
+    status_bad = bytes([HCI_EVENT_PKT, EVT_CMD_STATUS, 4, 0x0C, 1]) + struct.pack("<H", op)
+    other_op = bytes([HCI_EVENT_PKT, EVT_CMD_COMPLETE, 4, 1]) + struct.pack("<H", OCF_RESET) + b"\x00"
+    adv_report = bytes([HCI_EVENT_PKT, 0x3E, 12]) + b"\x02" * 12
+
+    assert command_status(complete_ok, op) == 0x00
+    assert command_status(complete_bad, op) == 0x12
+    assert command_status(status_ok, op) == 0x00
+    assert command_status(status_bad, op) == 0x0C
+    assert command_status(other_op, op) is None, "another command's reply must not be claimed"
+    assert command_status(adv_report, op) is None, "an advertising report is not a completion"
+    assert command_status(b"", op) is None and command_status(b"\x04\x0e", op) is None
+
     # The gate decides whether the bench sprays junk MACs across the house, so both
     # directions matter: it must come on for a live step and go off for a dead one.
     import tempfile
diff --git a/scripts/hil_monitor.py b/scripts/hil_monitor.py
index d64e975..81526be 100644
--- a/scripts/hil_monitor.py
+++ b/scripts/hil_monitor.py
@@ -50,11 +50,16 @@
 JSON_CHECK_WORKERS = 3      # concurrent connections — enough to reliably collide on serveJson
 JSON_CHECK_REQUESTS = 12    # sequential requests per worker, on one kept-alive connection
 
-# Heap trend. /json already reports freeHeap/maxHeap/fingerprints, so sampling it needs no
-# firmware change. A leak shows as freeHeap sliding while fingerprints stays flat; a
-# fragmentation problem shows as maxHeap sliding while freeHeap holds; fingerprint churn
-# shows as both moving with the device count. All three are printed so the log answers
-# which one it is without a re-run.
+# Heap trend, sampled from /json/tele. Deliberately not /json: that endpoint refuses with
+# 429 when it cannot afford a 12KB document, so heap numbers hung off it would vanish
+# exactly as a node degraded — the samples would stop before the decline they are meant to
+# measure and the verdict below would compare a healthier window. /json/tele is
+# allocation-free and answers at any heap level.
+#
+# A leak shows as freeHeap sliding while fingerprints stays flat; a fragmentation problem
+# shows as maxHeap sliding while freeHeap holds; fingerprint churn shows as both moving
+# with the device count. All three are printed so the log answers which without a re-run.
+HEAP_TELE_PATH = "/json/tele"
 HEAP_SAMPLE_SECS = 60
 HEAP_SETTLE_SECS = 120        # ignore the post-boot allocation burst
 HEAP_TREND_MIN_SECS = 1800    # below this the window is too short for a slope to mean anything
@@ -184,8 +189,8 @@ def worker(n, drops):
         print(f"[hil] /json check passed ({total} concurrent requests to {ip}){suffix}", flush=True)
 
 
-def heap_sampler(ip, samples, stop):
-    """Poll /json every HEAP_SAMPLE_SECS and record freeHeap/maxHeap/fingerprints.
+def heap_sampler(ip, samples, stop, problems):
+    """Poll /json/tele every HEAP_SAMPLE_SECS and record freeHeap/maxHeap/fingerprints.
 
     Read-only and deliberately gentle — one request a minute is nothing next to the
     concurrent burst json_endpoint_check already fires. Failures to sample are skipped
@@ -196,10 +201,18 @@ def heap_sampler(ip, samples, stop):
     while not stop.is_set():
         try:
             conn = http.client.HTTPConnection(ip, timeout=10)
-            conn.request("GET", "/json")
+            conn.request("GET", HEAP_TELE_PATH)
             resp = conn.getresponse()
             body = resp.read()
             conn.close()
+            if resp.status == 404:
+                # Firmware predates the endpoint. Recorded rather than ignored: a heap check
+                # that quietly samples nothing is worse than no check, because the run still
+                # reports PASS.
+                if "missing" not in problems:
+                    problems.append("missing")
+                    print(f"[hil] {HEAP_TELE_PATH} returned 404 — firmware too old", flush=True)
+                return
             if resp.status == 200:
                 doc = json.loads(body)
                 free, mx = doc.get("freeHeap"), doc.get("maxHeap")
@@ -212,11 +225,17 @@ def heap_sampler(ip, samples, stop):
         stop.wait(HEAP_SAMPLE_SECS)
 
 
-def heap_verdict(samples, duration):
+def heap_verdict(samples, duration, problems=()):
     """Return a failure string if free heap trended down over the run, else None."""
     if duration < HEAP_TREND_MIN_SECS:
         return None  # a 3-minute PR run says nothing about a multi-hour slope
+    if "missing" in problems:
+        # Not a heap failure, but the run cannot claim to have checked heap either.
+        return (f"{HEAP_TELE_PATH} is missing from this firmware, so heap was never "
+                f"sampled over {format_duration(int(duration))}")
     if len(samples) < HEAP_TREND_EDGE * 2:
+        # Unreachable-from-runner looks the same as a quiet network; neither is a firmware
+        # fault, so this stays a skip. A 404 is handled above precisely because it is one.
         print(f"[hil] heap trend SKIPPED — only {len(samples)} samples", flush=True)
         return None
 
@@ -299,7 +318,8 @@ def main():
     booted = False
     saw_scan_result = False
     json_failure = []  # appended to by the /json check thread
-    heap_samples = []  # appended to by the heap sampler thread
+    heap_samples = []   # appended to by the heap sampler thread
+    heap_problems = []  # ditto, for "the check could not run" conditions
     stop_sampling = threading.Event()
 
     try:
@@ -320,7 +340,7 @@ def main():
                     )
                     sys.exit(4)
                 stop_sampling.set()
-                decline = heap_verdict(heap_samples, elapsed)
+                decline = heap_verdict(heap_samples, elapsed, heap_problems)
                 if decline:
                     print(f"FAIL: {decline}")
                     sys.exit(8)
@@ -359,7 +379,7 @@ def main():
                     ).start()
                     threading.Thread(
                         target=heap_sampler,
-                        args=(match.group(1), heap_samples, stop_sampling),
+                        args=(match.group(1), heap_samples, stop_sampling, heap_problems),
                         daemon=True,
                     ).start()
                 else:
diff --git a/scripts/test_heap_trend.py b/scripts/test_heap_trend.py
index 071cd91..23c620d 100644
--- a/scripts/test_heap_trend.py
+++ b/scripts/test_heap_trend.py
@@ -56,4 +56,15 @@ def wobble(level, n):
 assert heap_verdict(samples(ramp(85000, 33000, 8)), HOURS_6) is None, "too few samples to judge"
 print("short run -> skipped")
 
+# Firmware without /json/tele must fail loudly. A heap check that silently samples nothing
+# and still reports PASS is how #2309 survived an 8h soak in the first place.
+verdict = heap_verdict([], HOURS_6, ["missing"])
+assert verdict and "missing" in verdict, f"old firmware not flagged: {verdict!r}"
+print(f"no endpoint -> {verdict}")
+
+# But a node the runner simply cannot reach is not a firmware fault — still a skip.
+assert heap_verdict([], HOURS_6) is None, "unreachable node must not fail the build"
+assert heap_verdict([], 180, ["missing"]) is None, "short run still must not judge"
+print("unreachable -> skipped")
+
 print("all heap trend checks passed")

From b9498dee2219ae35070376867dde2273fbacfa7a Mon Sep 17 00:00:00 2001
From: DTTerastar <DT@Terastar.biz>
Date: Sat, 1 Aug 2026 13:46:49 -0400
Subject: [PATCH 3/3] hil: address review on heap sampler/verdict
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

CodeRabbit on #9:
- heap_sampler: guard that the /json/tele body is a dict before doc.get(). A
  null/non-object body (the low-heap path) made None.get() raise AttributeError,
  which isn't in the caught tuple — so the sampler thread died silently exactly
  when heap was declining, the one thing this check exists to catch.
- heap_verdict: report the actual sampled span (from sample timestamps) rather
  than the full monitor duration, which overstates the window near the 30-min floor.
- Fix the HEAP_TREND_EDGE comment: median at each end, not average.
---
 scripts/hil_monitor.py | 18 +++++++++++-------
 1 file changed, 11 insertions(+), 7 deletions(-)

diff --git a/scripts/hil_monitor.py b/scripts/hil_monitor.py
index 81526be..2308655 100644
--- a/scripts/hil_monitor.py
+++ b/scripts/hil_monitor.py
@@ -63,7 +63,7 @@
 HEAP_SAMPLE_SECS = 60
 HEAP_SETTLE_SECS = 120        # ignore the post-boot allocation burst
 HEAP_TREND_MIN_SECS = 1800    # below this the window is too short for a slope to mean anything
-HEAP_TREND_EDGE = 5           # samples averaged at each end
+HEAP_TREND_EDGE = 5           # samples used for the median at each end
 HEAP_DECLINE_FRAC = 0.25      # fail if the tail lost more than this much of the baseline
 
 
@@ -215,11 +215,14 @@ def heap_sampler(ip, samples, stop, problems):
                 return
             if resp.status == 200:
                 doc = json.loads(body)
-                free, mx = doc.get("freeHeap"), doc.get("maxHeap")
-                if isinstance(free, int) and isinstance(mx, int):
-                    fp = doc.get("fingerprints")
-                    samples.append((time.monotonic(), free, mx, fp))
-                    print(f"[hil] heap freeHeap={free} maxHeap={mx} fingerprints={fp}", flush=True)
+                # A null/non-object body is the low-heap path (see json_endpoint_check) — the
+                # exact moment this sampler must not die on `None.get(...)`. Skip the sample.
+                if isinstance(doc, dict):
+                    free, mx = doc.get("freeHeap"), doc.get("maxHeap")
+                    if isinstance(free, int) and isinstance(mx, int):
+                        fp = doc.get("fingerprints")
+                        samples.append((time.monotonic(), free, mx, fp))
+                        print(f"[hil] heap freeHeap={free} maxHeap={mx} fingerprints={fp}", flush=True)
         except (OSError, http.client.HTTPException, ValueError):
             pass  # a missed sample is not a measurement — see docstring
         stop.wait(HEAP_SAMPLE_SECS)
@@ -247,8 +250,9 @@ def edges(index):
     free0, free1 = edges(1)
     max0, max1 = edges(2)
     fp0, fp1 = samples[0][3], samples[-1][3]
+    span = samples[-1][0] - samples[0][0]  # actual sampled window, not the full monitor duration
     summary = (f"freeHeap {free0:.0f}->{free1:.0f}, maxHeap {max0:.0f}->{max1:.0f}, "
-               f"fingerprints {fp0}->{fp1}, over {format_duration(int(duration))}")
+               f"fingerprints {fp0}->{fp1}, over {format_duration(int(span))}")
 
     if free1 < free0 * (1 - HEAP_DECLINE_FRAC):
         lost = (1 - free1 / free0) * 100