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..820a2a8
--- /dev/null
+++ b/scripts/ble_flood.py
@@ -0,0 +1,361 @@
+#!/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 select
+import socket
+import struct
+import time
+
+AF_BLUETOOTH = 31
+BTPROTO_HCI = 1
+HCI_CHANNEL_USER = 1
+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
+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(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)}")
+    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
+
+
+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.
+
+    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:
+            pkt = sock.recv(258)
+        except BlockingIOError:
+            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):
+    """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())
+
+    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()
+        address = random_static_address()
+        send(sock, OCF_LE_SET_ADV_ENABLE, b"\x00")
+        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
+
+        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"
+
+    # 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:
+        hci_command(OCF_LE_SET_ADV_DATA, b"\x00" * 256)
+        raise AssertionError("oversized parameters must be rejected")
+    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
+    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..2308655 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,34 @@
 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, 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
+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
+
 
 def json_endpoint_check(ip, bug):
     """Detect the serveJson() double-send under concurrent load.
@@ -163,6 +189,78 @@ def worker(n, drops):
         print(f"[hil] /json check passed ({total} concurrent requests to {ip}){suffix}", flush=True)
 
 
+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
+    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", 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)
+                # 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)
+
+
+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
+
+    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]
+    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(span))}")
+
+    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 +322,9 @@ 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_problems = []  # ditto, for "the check could not run" conditions
+    stop_sampling = threading.Event()
 
     try:
         while True:
@@ -242,6 +343,11 @@ def main():
                         f"{format_duration(args.duration)}."
                     )
                     sys.exit(4)
+                stop_sampling.set()
+                decline = heap_verdict(heap_samples, elapsed, heap_problems)
+                if decline:
+                    print(f"FAIL: {decline}")
+                    sys.exit(8)
                 print(f"PASS: {format_duration(args.duration)} elapsed cleanly.")
                 sys.exit(0)
 
@@ -269,15 +375,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, heap_problems),
+                        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..23c620d
--- /dev/null
+++ b/scripts/test_heap_trend.py
@@ -0,0 +1,70 @@
+#!/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")
+
+# 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")