diff --git a/tests/hardware/lxmf_echo_bot.py b/tests/hardware/lxmf_echo_bot.py
new file mode 100644
index 00000000..3754b033
--- /dev/null
+++ b/tests/hardware/lxmf_echo_bot.py
@@ -0,0 +1,331 @@
+#!/usr/bin/env python3
+"""
+Mac-side LXMF echo bot for diagnosing pyxis T-Deck delivery issues.
+
+Joins the same RNS network as the T-Deck via AutoInterface (link-local
+IPv6 multicast). Announces an identity called "Mac Echo Bot" and echoes
+any received DIRECT message back to the sender. Every announce send,
+announce receive, message receive, and message send is logged with a
+timestamp.
+
+Usage:
+    pip install rns lxmf
+    python3 /tmp/lxmf_echobot.py
+
+Expected output when working:
+    [12:00:00.123] Reticulum up. Identity hash=<hex>. Delivery dest hash=<hex>.
+    [12:00:00.234] Sending announce
+    [12:00:01.456] Received announce from <hex8>: Some Peer
+    [12:00:05.789] Received DIRECT message from <hex8>: 'hello'
+    [12:00:05.890] Echoing back: 'echo: hello'
+    [12:00:06.012] Echo SENT (state=delivered)
+"""
+import os, sys, time, threading, datetime
+
+# Pin to a local checkout of RNS + LXMF if one exists, so the bot
+# uses the same library pyxis interops with rather than whatever
+# pip points at. Override with RNS_REPO / LXMF_REPO env vars.
+for env_var, default_subdir in (("RNS_REPO", "repos/Reticulum"),
+                                ("LXMF_REPO", "repos/LXMF")):
+    p = os.environ.get(env_var) \
+        or os.path.expanduser(os.path.join("~", default_subdir))
+    if os.path.isdir(p):
+        sys.path.insert(0, p)
+
+import RNS
+import LXMF
+
+
+def ts():
+    return datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
+
+
+def log(msg):
+    print(f"[{ts()}] {msg}", flush=True)
+
+
+CONFIG_DIR = "/tmp/echobot-rnsconfig"
+STORAGE_DIR = "/tmp/echobot-storage"
+DISPLAY_NAME = "Mac Echo Bot"
+
+# Optional: when the harness launches the bot, it sets ECHOBOT_PEER_HEX
+# to pyxis's delivery destination hash. We use that to proactively
+# request a path / cached-announce so we have pyxis's identity in
+# `Identity.recall` cache before the harness drives messages — otherwise
+# we depend on rnsd's announce rebroadcast and miss it if rnsd's
+# per-announce rebroadcast limit was already reached.
+PEER_HEX = os.environ.get("ECHOBOT_PEER_HEX", "").strip()
+
+# lxmd propagation node the bot routes PROPAGATED messages through.
+# Deployment-specific, so it comes from PYXIS_PROPAGATION_NODE_HEX (no
+# hardcoded default — keeps environment-specific hashes out of source).
+# When unset, the bot skips all propagation setup and the harness skips
+# its PROPAGATED rounds; DIRECT + OPPORTUNISTIC + the bz2 probe still run.
+PROPAGATION_NODE_HEX = os.environ.get("PYXIS_PROPAGATION_NODE_HEX", "").strip()
+PROPAGATION_STAMP_COST = 16  # lxmd default is 16; 13 is the floor.
+
+# How often to pull queued messages from the PN. The bot's not running
+# any UI, so it relies on this loop to actually receive PROPAGATED
+# messages.
+PROP_SYNC_INTERVAL_SEC = 8
+
+os.makedirs(CONFIG_DIR, exist_ok=True)
+os.makedirs(STORAGE_DIR, exist_ok=True)
+
+# Minimal RNS config: TCPClient to the local rnsd at 127.0.0.1:4242.
+# Pyxis on the T-Deck connects as a TCP CLIENT to the same rnsd's
+# TCPServerInterface (built into rnsd's default config) so this bot
+# and pyxis end up on the same Reticulum network with the rnsd acting
+# as a hub.
+#
+# We use an isolated config dir / non-default control ports so this
+# bot does NOT accidentally join the user's primary rnsd shared
+# instance (which would mix the bot's identity into their personal
+# Reticulum state). The bot is its own RNS process.
+config_path = os.path.join(CONFIG_DIR, "config")
+with open(config_path, "w") as f:
+    f.write("""\
+[reticulum]
+enable_transport = Yes
+share_instance = No
+shared_instance_port = 47428
+instance_control_port = 47429
+
+[logging]
+loglevel = 4
+
+[interfaces]
+  [[TCP to local rnsd]]
+    type = TCPClientInterface
+    enabled = yes
+    target_host = 127.0.0.1
+    target_port = 4242
+""")
+
+reticulum = RNS.Reticulum(configdir=CONFIG_DIR, loglevel=4)
+
+# Persistent identity stored in the config dir so reruns reuse the
+# same delivery destination hash (peers don't have to re-learn the
+# path on every restart).
+ident_path = os.path.join(CONFIG_DIR, "identity")
+if os.path.exists(ident_path):
+    identity = RNS.Identity.from_file(ident_path)
+    log(f"Loaded identity from {ident_path}")
+else:
+    identity = RNS.Identity()
+    identity.to_file(ident_path)
+    log(f"Created new identity, saved to {ident_path}")
+
+router = LXMF.LXMRouter(identity=identity, storagepath=STORAGE_DIR)
+delivery_destination = router.register_delivery_identity(
+    identity, display_name=DISPLAY_NAME
+)
+delivery_destination.set_default_app_data(
+    lambda: router.get_announce_app_data(delivery_destination.hash)
+)
+
+log(f"Reticulum up. Identity hash={identity.hash.hex()}.")
+log(f"Delivery dest hash={delivery_destination.hash.hex()}.")
+log(f"Display name: {DISPLAY_NAME}")
+
+# Configure propagation node so the bot can SEND propagated and SYNC
+# its inbox. We have to wait for the PN to be path-known before
+# `set_outbound_propagation_node` is useful; the path arrives via
+# the rnsd<->lxmd shared instance. Set it right away anyway —
+# request_messages_from_propagation_node tolerates "no path yet"
+# and retries the path request internally.
+prop_node_hash = bytes.fromhex(PROPAGATION_NODE_HEX) if PROPAGATION_NODE_HEX else None
+if prop_node_hash is not None:
+    router.set_outbound_propagation_node(prop_node_hash)
+    router.outbound_propagation_node = prop_node_hash
+    log(f"Configured outbound propagation node: {PROPAGATION_NODE_HEX}")
+else:
+    log("No PYXIS_PROPAGATION_NODE_HEX set — propagation disabled (DIRECT/OPP/bz2 only)")
+
+
+def on_announce(destination_hash, announced_identity, app_data):
+    name = LXMF.display_name_from_app_data(app_data) or "(no name)"
+    log(f"Announce RX: dest={destination_hash.hex()} name={name!r}")
+
+
+announce_handler = type(
+    "Handler", (), {
+        "aspect_filter": "lxmf.delivery",
+        # Without `receive_path_responses = True`, RNS only fires this
+        # handler for LIVE announces (received via broadcast/forward).
+        # Path responses (the cached announce sent back by a next-hop
+        # in reply to RNS.Transport.request_path) are skipped. The
+        # harness's eager-path-acquirer relies on path responses to
+        # learn pyxis's identity quickly, so we opt in here.
+        "receive_path_responses": True,
+        "received_announce": staticmethod(on_announce),
+    }
+)
+RNS.Transport.register_announce_handler(announce_handler)
+
+
+def on_delivery(message):
+    src_hex = message.source_hash.hex() if message.source_hash else "?"
+    method = {
+        LXMF.LXMessage.OPPORTUNISTIC: "OPPORTUNISTIC",
+        LXMF.LXMessage.DIRECT: "DIRECT",
+        LXMF.LXMessage.PROPAGATED: "PROPAGATED",
+    }.get(message.method, str(message.method))
+    try:
+        content = message.content.decode("utf-8")
+    except Exception:
+        content = repr(message.content)
+    title = ""
+    try:
+        if message.title:
+            title = message.title.decode("utf-8")
+    except Exception:
+        title = repr(message.title)
+    log(
+        f"Message RX: from={src_hex} method={method} "
+        f"title={title!r} content={content!r}"
+    )
+
+    # Echo back. Mirror the sender's method so PROPAGATED messages
+    # echo via the PN (round-trip: sender uploads → PN persists → bot
+    # syncs down → bot replies via PN → sender syncs down) and DIRECT
+    # messages echo over a fresh link.
+    try:
+        # Recall the sender's identity so we can construct a destination.
+        sender_identity = RNS.Identity.recall(message.source_hash)
+        if sender_identity is None:
+            log(f"  Cannot echo: sender identity not yet known for {src_hex}")
+            return
+        dest = RNS.Destination(
+            sender_identity, RNS.Destination.OUT, RNS.Destination.SINGLE,
+            "lxmf", "delivery"
+        )
+        echo_method = (
+            LXMF.LXMessage.PROPAGATED
+            if message.method == LXMF.LXMessage.PROPAGATED
+            else LXMF.LXMessage.DIRECT
+        )
+        # bz2-on-receive probe: when the harness sends "BZ2PROBE", reply with
+        # a large, highly-compressible payload. It exceeds microLXMF's Resource
+        # threshold AND compresses ~99%, so python LXMF sends it as a bz2-
+        # compressed Resource (auto_compress, since microLXMF announces
+        # compression-capable). That forces pyxis's Resource::assemble() to
+        # bz2_decompress on receive — the path that, unpatched upstream, marks
+        # the resource CORRUPT and tears down the link. If pyxis surfaces the
+        # full payload via T:RX, decompress-on-receive works on hardware.
+        if content.startswith("BZ2PROBE"):
+            echo_body = "BZ2OK:" + ("ABCDABCDABCDABCD" * 96)  # ~1542 bytes, repeating
+        else:
+            echo_body = f"echo: {content}"
+        echo_kwargs = dict(
+            destination=dest,
+            source=delivery_destination,
+            content=echo_body.encode("utf-8"),
+            title=b"echo",
+            desired_method=echo_method,
+        )
+        if echo_method == LXMF.LXMessage.PROPAGATED:
+            # python LXMF requires `include_ticket=False` here so the
+            # outbound propagation flow doesn't trip on missing ticket
+            # state. The default is fine but spelled out for clarity.
+            echo_kwargs["include_ticket"] = False
+        echo = LXMF.LXMessage(**echo_kwargs)
+
+        def on_delivered(msg):
+            log(f"  Echo DELIVERED to {src_hex}")
+        def on_failed(msg):
+            log(f"  Echo FAILED to {src_hex} state={msg.state}")
+        def on_sent(msg):
+            log(f"  Echo SENT (PRF received) to {src_hex}")
+
+        echo.register_delivery_callback(on_delivered)
+        echo.register_failed_callback(on_failed)
+        # `sent` only fires on PROPAGATED in upstream LXMF; we still set it
+        # for parity with the `delivery` callback nomenclature.
+        try:
+            echo.register_sent_callback(on_sent)
+        except Exception:
+            pass
+
+        router.handle_outbound(echo)
+        log(f"  Echo queued for delivery to {src_hex}")
+    except Exception as e:
+        log(f"  Echo construction failed: {e}")
+
+
+router.register_delivery_callback(on_delivery)
+
+
+# Announce loop — every 30s so the T-Deck hears us multiple times.
+def announcer():
+    while True:
+        delivery_destination.announce()
+        log("Announce TX")
+        time.sleep(30)
+
+
+# Propagation sync loop — periodically pull queued messages from the PN.
+# Without this the bot would never receive PROPAGATED messages (no UI
+# tap to drive a manual sync; the LXMRouter only auto-syncs when its
+# `delivery_destination` has someone tapping `sync_inbound`).
+#
+# lxmd's default propagation announce_interval is 5 MINUTES (the value
+# in the config is multiplied by 60), so waiting passively for an
+# announce can stall the first sync several minutes. We proactively
+# `RNS.Transport.request_path()` until we get a path.
+def prop_syncer():
+    while not RNS.Transport.has_path(prop_node_hash):
+        log("  Requesting path to PN...")
+        RNS.Transport.request_path(prop_node_hash)
+        for _ in range(10):
+            if RNS.Transport.has_path(prop_node_hash):
+                break
+            time.sleep(0.5)
+    log(f"Path to PN known; starting periodic sync every {PROP_SYNC_INTERVAL_SEC}s")
+    while True:
+        try:
+            router.request_messages_from_propagation_node(identity)
+        except Exception as e:
+            log(f"  prop sync error: {e}")
+        time.sleep(PROP_SYNC_INTERVAL_SEC)
+
+
+threading.Thread(target=announcer, daemon=True).start()
+if prop_node_hash is not None:
+    threading.Thread(target=prop_syncer, daemon=True).start()
+
+
+# Eager peer-path acquisition. When the harness tells us pyxis's hash
+# via env, request a path immediately. RNS forwards the cached announce
+# from the next-hop, which fires our announce-handler and populates
+# Identity.recall_app_data. Without this, if the bot connects AFTER
+# rnsd hit its rebroadcast limit for the peer's announce, we never
+# learn pyxis's identity and can't echo PROPAGATED messages.
+if PEER_HEX:
+    def peer_path_acquirer():
+        try:
+            peer_hash = bytes.fromhex(PEER_HEX)
+        except Exception as e:
+            log(f"  Bad ECHOBOT_PEER_HEX={PEER_HEX!r}: {e}")
+            return
+        for _ in range(10):
+            if RNS.Transport.has_path(peer_hash):
+                log(f"Path to peer {PEER_HEX} known via cached announce")
+                return
+            log(f"  Requesting path to peer {PEER_HEX}...")
+            RNS.Transport.request_path(peer_hash)
+            for _ in range(20):
+                if RNS.Transport.has_path(peer_hash):
+                    log(f"Path to peer {PEER_HEX} known via cached announce")
+                    return
+                time.sleep(0.5)
+        log(f"  WARN: never got path to peer {PEER_HEX}")
+
+    threading.Thread(target=peer_path_acquirer, daemon=True).start()
+log(f"Echo bot ready. Press Ctrl-C to stop. Storage: {STORAGE_DIR}")
+
+try:
+    while True:
+        time.sleep(60)
+except KeyboardInterrupt:
+    log("Shutting down.")
diff --git a/tests/hardware/run_e2e.sh b/tests/hardware/run_e2e.sh
new file mode 100755
index 00000000..fcbac7c2
--- /dev/null
+++ b/tests/hardware/run_e2e.sh
@@ -0,0 +1,68 @@
+#!/usr/bin/env bash
+#
+# Full end-to-end on-device test for pyxis: build + flash a T-Deck, bring up a
+# Mac-side LXMF echo bot over the local rnsd, and drive LXMF round-trips
+# (DIRECT / OPPORTUNISTIC / optional PROPAGATED) plus a bz2-on-receive probe
+# via the firmware's `T:` serial command surface (-DPYXIS_TEST_HOOKS).
+#
+# All environment-specific values are read from the environment (nothing
+# deployment-specific is committed):
+#   PYXIS_TEST_TCP_HOST   Mac IP the T-Deck dials for rnsd (default: en0 IPv4)
+#   PYXIS_TEST_TCP_PORT   rnsd TCPServerInterface port      (default: 4242)
+#   PYXIS_SERIAL_PORT     T-Deck USB serial                 (default: first /dev/cu.usbmodem*)
+#   PYXIS_ENV             platformio env                    (default: tdeck)
+#   PYXIS_PROPAGATION_NODE_HEX  lxmd PN hash (optional; enables PROPAGATED rounds)
+#
+# Usage:
+#   tests/hardware/run_e2e.sh                 # build, flash, one pass + bz2 probe
+#   tests/hardware/run_e2e.sh --soak-hours 1  # ... then soak round-trips for 1h
+#   PYXIS_SKIP_FLASH=1 tests/hardware/run_e2e.sh   # reuse already-flashed fw
+#
+# Exits 0 only if every round (incl. the bz2 probe) passed and no crash fired.
+set -euo pipefail
+
+HERE="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPO="$(cd "$HERE/../.." && pwd)"
+SOAK_HOURS="0"
+[ "${1:-}" = "--soak-hours" ] && SOAK_HOURS="${2:-0}"
+
+PYXIS_ENV="${PYXIS_ENV:-tdeck}"
+PYXIS_TEST_TCP_PORT="${PYXIS_TEST_TCP_PORT:-4242}"
+PYXIS_TEST_TCP_HOST="${PYXIS_TEST_TCP_HOST:-$(ipconfig getifaddr en0 2>/dev/null || true)}"
+PYXIS_SERIAL_PORT="${PYXIS_SERIAL_PORT:-$(ls /dev/cu.usbmodem* 2>/dev/null | head -1 || true)}"
+PIO="$(command -v pio || echo /opt/homebrew/bin/pio)"
+# The harness needs pyserial; PlatformIO's bundled python has it. Prefer an
+# explicit PYXIS_HARNESS_PY, else the penv python, else any python with serial.
+PIO_PY="${PYXIS_HARNESS_PY:-}"
+[ -z "$PIO_PY" ] && PIO_PY="$(ls "$HOME"/.platformio/penv/bin/python3 2>/dev/null | head -1 || true)"
+[ -z "$PIO_PY" ] && for c in /opt/homebrew/Cellar/platformio/*/libexec/bin/python3; do [ -x "$c" ] && PIO_PY="$c" && break; done
+
+echo "== pyxis on-device e2e =="
+echo "  repo:        $REPO"
+echo "  env:         $PYXIS_ENV"
+echo "  serial:      ${PYXIS_SERIAL_PORT:-<none found>}"
+echo "  rnsd target: ${PYXIS_TEST_TCP_HOST:-<unset>}:$PYXIS_TEST_TCP_PORT"
+echo "  prop node:   ${PYXIS_PROPAGATION_NODE_HEX:-<unset — PROPAGATED skipped>}"
+echo "  soak hours:  $SOAK_HOURS"
+echo "  harness py:  ${PIO_PY:-<not found>}"
+
+[ -z "${PYXIS_SERIAL_PORT:-}" ] && { echo "ERROR: no T-Deck serial port found (set PYXIS_SERIAL_PORT)"; exit 2; }
+[ -z "${PYXIS_TEST_TCP_HOST:-}" ] && { echo "ERROR: could not determine Mac IP (set PYXIS_TEST_TCP_HOST)"; exit 2; }
+
+# rnsd must be listening so both the T-Deck (LAN) and the bot (127.0.0.1) can
+# attach to the same Reticulum hub.
+if ! lsof -nP -iTCP:"$PYXIS_TEST_TCP_PORT" -sTCP:LISTEN >/dev/null 2>&1; then
+    echo "ERROR: nothing listening on TCP:$PYXIS_TEST_TCP_PORT — start rnsd with a"
+    echo "       TCPServerInterface on that port first (e.g. \`rnsd\`)."
+    exit 2
+fi
+
+export PYXIS_TEST_TCP_HOST PYXIS_TEST_TCP_PORT PYXIS_SERIAL_PORT PYXIS_PROPAGATION_NODE_HEX
+
+if [ "${PYXIS_SKIP_FLASH:-0}" != "1" ]; then
+    echo "== build + flash ($PYXIS_ENV) with TCP target baked in =="
+    "$PIO" run -e "$PYXIS_ENV" -t upload --upload-port "$PYXIS_SERIAL_PORT"
+fi
+
+echo "== running harness (resets device, drives T: commands) =="
+exec "$PIO_PY" "$HERE/tdeck_harness.py" --soak-hours "$SOAK_HOURS"
diff --git a/tests/hardware/tdeck_harness.py b/tests/hardware/tdeck_harness.py
new file mode 100644
index 00000000..ba78db5a
--- /dev/null
+++ b/tests/hardware/tdeck_harness.py
@@ -0,0 +1,645 @@
+#!/usr/bin/env python3
+"""
+T-Deck ↔ Mac echo-bot integration harness.
+
+This script drives a pyxis-built T-Deck firmware (with -DPYXIS_TEST_HOOKS)
+and a Mac-side LXMF echo bot together, exercising LXMF DIRECT delivery
+in both directions over the local rnsd's TCPServerInterface
+(host:port configured at firmware-build-time via PYXIS_TEST_TCP_HOST /
+PYXIS_TEST_TCP_PORT in platformio.ini).
+
+What it does:
+
+1. Open the T-Deck serial port (/dev/cu.usbmodem1101) and reset the device.
+   Stream every line into a captured log; parse `T:OK` / `T:ERR` /
+   `T:RXMSG` / `T:PATH` lines for the harness, pass everything else
+   through to the log.
+
+2. Wait for pyxis boot: `BOOT END: ui_manager` line, then a TCP
+   connect-success log to confirm the rnsd link is up.
+
+3. Start the echo bot AFTER pyxis is fully up (subprocess) so its
+   announce arrives while pyxis is listening. Bot uses the same rnsd
+   the T-Deck talks to (shared instance via shared_instance_port).
+
+4. Wait for pyxis to learn the bot's path (`T:HASPATH <bot_dest>`
+   returns `T:OK 1`), with bot manually re-announcing if needed.
+
+5. Test loop, repeating until either a failure or the soak deadline:
+   - Send short message from pyxis to bot via `T:SEND`.
+   - Wait for bot to log RX + ECHO SENT.
+   - Wait for pyxis to log RX (poll `T:RX`).
+   - Verify content matches `echo: <orig>`.
+   - Send long (~1KB) message and repeat.
+   - Sleep `--cadence` seconds between rounds.
+
+6. Soak: run for at least `--soak-hours` hours (default 1.1). Crash =
+   fail. Mid-test failures are logged but don't abort.
+
+Usage:
+    # Run with a python that has pyserial available. PlatformIO's
+    # bundled python works:
+    "$(pio system info | awk -F: '/Python Executable/{print $2}' | xargs)" \\
+        tests/soak/tdeck_soak_harness.py [--soak-hours 1.1] [--cadence 30]
+
+Outputs:
+    /tmp/tdeck-harness.log         — combined timestamped event log
+    /tmp/tdeck-harness-tdeck.log   — verbatim serial output from T-Deck
+    /tmp/echobot.log               — echo bot's own log (subprocess stdout)
+
+Exits 0 on full success, 1 on failure.
+"""
+import os, sys, time, threading, subprocess, argparse, queue, signal
+import datetime
+
+# pyserial: try the active python first, fall back to PlatformIO's
+# bundled site-packages if the user runs us through system python.
+try:
+    import serial  # noqa: F401
+except ImportError:
+    pio_site = os.environ.get("PIO_SITE_PACKAGES")
+    if pio_site and os.path.isdir(pio_site):
+        sys.path.insert(0, pio_site)
+    import serial  # second try; let it raise if still missing
+
+# Default serial port for a USB-attached T-Deck Plus on macOS. Override
+# with PYXIS_SERIAL_PORT (eg "/dev/ttyUSB0" on Linux).
+PORT = os.environ.get("PYXIS_SERIAL_PORT", "/dev/cu.usbmodem1101")
+BAUD = 115200
+HARNESS_LOG = "/tmp/tdeck-harness.log"
+TDECK_LOG = "/tmp/tdeck-harness-tdeck.log"
+ECHOBOT_LOG = "/tmp/echobot.log"
+ECHOBOT_PY = os.path.join(os.path.dirname(os.path.abspath(__file__)),
+                          "lxmf_echo_bot.py")
+
+
+def ts():
+    return datetime.datetime.now().strftime("%H:%M:%S.%f")[:-3]
+
+
+_log_lock = threading.Lock()
+_log_fh = None
+
+
+def log(category, msg):
+    line = f"[{ts()}] [{category}] {msg}\n"
+    with _log_lock:
+        sys.stdout.write(line)
+        sys.stdout.flush()
+        if _log_fh is not None:
+            _log_fh.write(line)
+            _log_fh.flush()
+
+
+class TDeck:
+    """Serial driver: reset, command, parse response."""
+
+    def __init__(self, port=PORT, baud=BAUD):
+        self.ser = serial.Serial(port, baud, timeout=0.1)
+        self._line_q = queue.Queue()           # raw text lines
+        self._tdeck_log = open(TDECK_LOG, "wb")
+        self._stop = threading.Event()
+        self._reader = threading.Thread(target=self._read_loop, daemon=True)
+        self._reader.start()
+
+    def reset(self):
+        """Pulse DTR/RTS like esptool to reboot the ESP32 cleanly."""
+        log("HARNESS", "Resetting T-Deck via DTR/RTS pulse")
+        self.ser.dtr = False
+        self.ser.rts = True
+        time.sleep(0.1)
+        self.ser.dtr = True
+        self.ser.rts = False
+        time.sleep(0.1)
+        self.ser.dtr = False
+        self.ser.rts = False
+
+    def _read_loop(self):
+        buf = b""
+        while not self._stop.is_set():
+            try:
+                d = self.ser.read(4096)
+            except Exception as e:
+                log("HARNESS", f"Serial read error: {e}")
+                return
+            if not d:
+                continue
+            self._tdeck_log.write(d)
+            self._tdeck_log.flush()
+            buf += d
+            while b"\n" in buf:
+                line, buf = buf.split(b"\n", 1)
+                try:
+                    text = line.rstrip(b"\r").decode("utf-8", errors="replace")
+                except Exception:
+                    text = repr(line)
+                self._line_q.put(text)
+
+    def drain_lines(self):
+        """Pop all currently-buffered lines (non-blocking)."""
+        out = []
+        while True:
+            try:
+                out.append(self._line_q.get_nowait())
+            except queue.Empty:
+                break
+        return out
+
+    def wait_for_line(self, predicate, timeout):
+        """Block until a line matches predicate(text) or timeout."""
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            try:
+                line = self._line_q.get(timeout=min(1.0, deadline - time.time()))
+            except queue.Empty:
+                continue
+            if predicate(line):
+                return line
+        return None
+
+    def send_command(self, cmd, response_timeout=10.0):
+        """Send a `T:` command, wait for the next `T:OK` or `T:ERR` line.
+
+        Returns the response line (without the `T:OK` / `T:ERR` prefix
+        portion left intact for inspection), or None on timeout.
+        """
+        log("HARNESS-TX", cmd)
+        # Drain any stale T: responses queued up from earlier
+        deadline = time.time() + response_timeout
+        # Send
+        self.ser.write((cmd + "\n").encode("utf-8"))
+        self.ser.flush()
+        # Read until we see a T:OK or T:ERR
+        while time.time() < deadline:
+            try:
+                line = self._line_q.get(timeout=min(1.0, deadline - time.time()))
+            except queue.Empty:
+                continue
+            if line.startswith("T:OK") or line.startswith("T:ERR"):
+                log("HARNESS-RX", line)
+                return line
+        log("HARNESS", f"send_command({cmd!r}) timed out")
+        return None
+
+    def close(self):
+        self._stop.set()
+        self.ser.close()
+        self._tdeck_log.close()
+
+
+def wait_for_pyxis_boot(t, timeout=90):
+    log("HARNESS", "Waiting for pyxis boot to complete...")
+    line = t.wait_for_line(
+        lambda L: "BOOT" in L and "ui_manager" in L and "END" in L,
+        timeout,
+    )
+    if not line:
+        return False
+    log("HARNESS", f"Boot complete: {line}")
+    return True
+
+
+def wait_for_tcp_link(t, timeout=60):
+    log("HARNESS", "Waiting for TCP interface to connect to rnsd...")
+    line = t.wait_for_line(
+        lambda L: "TCPClientInterface" in L and (
+            "connected" in L.lower() or "Connected" in L or "started" in L.lower()
+        ),
+        timeout,
+    )
+    if line:
+        log("HARNESS", f"TCP link up: {line}")
+    else:
+        log("HARNESS", "(TCP-link line not seen — continuing anyway, "
+                       "pyxis may have already attached.)")
+    return line is not None
+
+
+def start_echobot(peer_hex=""):
+    log("HARNESS", f"Launching echo bot (peer_hex={peer_hex})...")
+    fh = open(ECHOBOT_LOG, "w")
+    env = dict(os.environ)
+    if peer_hex:
+        # The bot uses this to proactively request_path() so it picks up
+        # pyxis's cached announce from rnsd, even if rnsd's announce
+        # rebroadcast limit was already reached when the bot connected.
+        env["ECHOBOT_PEER_HEX"] = peer_hex
+    proc = subprocess.Popen(
+        ["/usr/bin/python3", ECHOBOT_PY],
+        stdout=fh, stderr=subprocess.STDOUT,
+        env=env,
+    )
+    return proc, fh
+
+
+def echobot_announce_dest(echobot_log_path, timeout=20):
+    """Read the echobot's own log to extract its delivery destination hash."""
+    deadline = time.time() + timeout
+    last = ""
+    while time.time() < deadline:
+        try:
+            with open(echobot_log_path) as f:
+                last = f.read()
+        except FileNotFoundError:
+            time.sleep(0.5)
+            continue
+        for line in last.splitlines():
+            # `Delivery dest hash=<hex>.`
+            if "Delivery dest hash=" in line:
+                hex_part = line.split("Delivery dest hash=", 1)[1].rstrip(".").strip()
+                # may have trailing punctuation
+                hex_part = hex_part.split()[0].rstrip(".")
+                return hex_part
+        time.sleep(0.5)
+    return None
+
+
+def wait_for_path(t, dest_hex, timeout=120):
+    log("HARNESS", f"Polling pyxis for path to bot {dest_hex}...")
+    deadline = time.time() + timeout
+    last_paths_dump = 0
+    while time.time() < deadline:
+        resp = t.send_command(f"T:HASPATH {dest_hex}", response_timeout=5)
+        if resp and resp.startswith("T:OK 1"):
+            log("HARNESS", "Pyxis has path to bot.")
+            return True
+        # Every 20s, dump T:PATHS to see what pyxis DOES know about
+        if time.time() - last_paths_dump > 20:
+            last_paths_dump = time.time()
+            t.ser.write(b"T:PATHS\n")
+            t.ser.flush()
+            ok = t.wait_for_line(lambda L: L.startswith("T:OK count="), timeout=5)
+            if ok:
+                log("HARNESS", f"T:PATHS {ok}")
+                # Drain the T:PATH lines that follow
+                try:
+                    count = int(ok.split("=", 1)[1])
+                except Exception:
+                    count = 0
+                for _ in range(count):
+                    p = t.wait_for_line(lambda L: L.startswith("T:PATH "), timeout=2)
+                    if p:
+                        log("HARNESS", f"   {p}")
+        time.sleep(2.0)
+    return False
+
+
+def echobot_log_after(marker, predicate, timeout=30):
+    """Wait until the echobot log has a line matching predicate appearing
+    after `marker` (a substring known to be present already, used to
+    avoid matching old log entries from a previous round)."""
+    deadline = time.time() + timeout
+    last_size = 0
+    while time.time() < deadline:
+        try:
+            with open(ECHOBOT_LOG) as f:
+                txt = f.read()
+        except FileNotFoundError:
+            time.sleep(0.5)
+            continue
+        if marker:
+            mi = txt.find(marker)
+            if mi < 0:
+                time.sleep(0.5)
+                continue
+            txt = txt[mi + len(marker):]
+        for line in txt.splitlines():
+            if predicate(line):
+                return line
+        time.sleep(0.5)
+    return None
+
+
+def poll_tdeck_rx(t, expected_substring, timeout=60):
+    """Poll T:RX until a received message contains expected_substring."""
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        # T:RX returns multiple lines; we have to read more than one
+        log("HARNESS-TX", "T:RX")
+        t.ser.write(b"T:RX\n")
+        t.ser.flush()
+        # Collect: T:OK count=N then N T:RXMSG lines
+        ok_line = t.wait_for_line(
+            lambda L: L.startswith("T:OK count=") or L.startswith("T:ERR"),
+            timeout=5,
+        )
+        if not ok_line:
+            time.sleep(2.0)
+            continue
+        try:
+            count = int(ok_line.split("=", 1)[1])
+        except Exception:
+            count = 0
+        rx_msgs = []
+        for _ in range(count):
+            line = t.wait_for_line(lambda L: L.startswith("T:RXMSG"), timeout=2)
+            if line:
+                rx_msgs.append(line)
+        for line in rx_msgs:
+            if expected_substring in line:
+                log("HARNESS", f"Pyxis RX confirmed: {line}")
+                return line
+        time.sleep(2.0)
+    return None
+
+
+def main():
+    global _log_fh
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--soak-hours", type=float, default=0.0,
+                        help="0 = one pass of each method + the bz2 probe (default); "
+                             ">0 loops the round-trip tests for a soak")
+    parser.add_argument("--cadence", type=float, default=30,
+                        help="seconds between message rounds")
+    parser.add_argument("--no-reset", action="store_true",
+                        help="don't pulse DTR (use if pyxis is already booted)")
+    args = parser.parse_args()
+
+    _log_fh = open(HARNESS_LOG, "w")
+    log("HARNESS", f"Harness starting; soak target = {args.soak_hours} hours")
+
+    # 1. Open T-Deck serial, reset, wait for boot
+    t = TDeck()
+    if not args.no_reset:
+        t.reset()
+
+    if not wait_for_pyxis_boot(t, timeout=90):
+        log("HARNESS", "FAILED: pyxis boot did not complete in 90s")
+        return 1
+
+    wait_for_tcp_link(t, timeout=30)
+    time.sleep(2.0)
+
+    pyxis_dest_resp = t.send_command("T:DEST", response_timeout=5)
+    if not pyxis_dest_resp or not pyxis_dest_resp.startswith("T:OK"):
+        log("HARNESS", f"FAILED: T:DEST returned {pyxis_dest_resp}")
+        return 1
+    pyxis_dest = pyxis_dest_resp.split(" ", 1)[1].strip()
+    log("HARNESS", f"Pyxis delivery dest = {pyxis_dest}")
+
+    # 2. Start echo bot AFTER pyxis is up. Pass pyxis's destination
+    # hash so the bot can proactively request_path() and pick up
+    # the cached announce from rnsd.
+    bot, bot_fh = start_echobot(peer_hex=pyxis_dest)
+    bot_dest = echobot_announce_dest(ECHOBOT_LOG, timeout=20)
+    if not bot_dest:
+        log("HARNESS", "FAILED: could not extract bot destination hash")
+        bot.terminate()
+        return 1
+    log("HARNESS", f"Echo bot delivery dest = {bot_dest}")
+
+    # 3. Wait for pyxis to learn the bot's path
+    if not wait_for_path(t, bot_dest, timeout=120):
+        log("HARNESS", "FAILED: pyxis never learned bot's path")
+        bot.terminate()
+        return 1
+
+    # 3b. Drive pyxis to announce so the bot learns pyxis's identity
+    # for echo construction. Two ways the bot can end up with the
+    # identity: (a) live announce delivered through rnsd while the bot
+    # is connected — fires our announce_handler and we log "Announce
+    # RX" — or (b) a path-response from rnsd's cache when the bot
+    # calls RNS.Transport.request_path (which the bot does eagerly
+    # for ECHOBOT_PEER_HEX). Path (b) populates Identity.recall but
+    # doesn't always fire the announce_handler (path-response
+    # delivery is gated on `receive_path_responses` and hits a thread
+    # boundary). So treat "bot has pyxis path" OR "bot logged Announce
+    # RX" as either-or success. Drive a few T:ANN cycles to give path
+    # (a) a chance and not just rely on the eager request_path.
+    log("HARNESS", "Driving pyxis to announce, "
+                   "waiting for bot to learn pyxis's identity...")
+    bot_saw_pyxis = False
+    for attempt in range(6):
+        t.send_command("T:ANN", response_timeout=5)
+        # Either: bot's announce_handler fired (live announce) OR
+        # bot's eager request_path landed (path-response → cached
+        # announce → Identity.remember). The latter logs "Path to
+        # peer <hex> known via cached announce" via the
+        # peer_path_acquirer thread.
+        deadline = time.time() + 10
+        while time.time() < deadline:
+            try:
+                with open(ECHOBOT_LOG) as f:
+                    txt = f.read()
+            except FileNotFoundError:
+                txt = ""
+            if (("Announce RX:" in txt and pyxis_dest in txt)
+                    or f"Path to peer {pyxis_dest} known" in txt):
+                bot_saw_pyxis = True
+                break
+            time.sleep(0.5)
+        if bot_saw_pyxis:
+            log("HARNESS", "Bot has pyxis identity (via announce or path-response)")
+            break
+        log("HARNESS", f"  attempt {attempt+1}: bot not yet — re-announce")
+    if not bot_saw_pyxis:
+        log("HARNESS", "FAILED: bot never learned pyxis's identity — "
+                       "echoes will fail")
+        bot.terminate()
+        return 1
+
+    fails = 0
+    successes = 0
+
+    # 4. bz2-on-receive probe — the graft's riskiest new code. Send a short
+    #    "BZ2PROBE" trigger (fits the USB-CDC single-line limit); the echo bot
+    #    replies with a ~1.5KB highly-compressible payload. python LXMF sends
+    #    that as a bz2-COMPRESSED Resource (auto_compress — microLXMF announces
+    #    compression-capable). pyxis must bz2_decompress it in
+    #    Resource::assemble() and surface the full body via T:RX. Unpatched
+    #    upstream marks a compressed inbound resource CORRUPT and tears down
+    #    the link, so a clean decompressed receive here proves the graft's
+    #    decompress-on-receive port works on real hardware.
+    log("HARNESS", "=== bz2-on-receive probe ===")
+    with open(ECHOBOT_LOG) as f:
+        probe_marker = f.read()[-256:]
+    probe_resp = t.send_command(f"T:SEND {bot_dest} BZ2PROBE", response_timeout=10)
+    if probe_resp and probe_resp.startswith("T:OK"):
+        bot_rx = echobot_log_after(
+            probe_marker,
+            lambda L: "Message RX:" in L and "BZ2PROBE" in L,
+            timeout=60,
+        )
+        if not bot_rx:
+            log("HARNESS", "FAIL [bz2-probe]: bot never received BZ2PROBE trigger")
+            fails += 1
+        else:
+            rx = poll_tdeck_rx(t, "BZ2OK:", timeout=90)
+            if rx and len(rx) > 800:
+                log("HARNESS", f"PASS [bz2-probe]: pyxis received + decompressed "
+                               f"compressed Resource ({len(rx)}-char T:RXMSG line)")
+                successes += 1
+            else:
+                log("HARNESS", f"FAIL [bz2-probe]: no decompressed payload surfaced "
+                               f"(got {rx!r}) — check for link teardown / CORRUPT")
+                fails += 1
+        t.send_command("T:RXCLR", response_timeout=5)
+    else:
+        log("HARNESS", f"FAIL [bz2-probe]: T:SEND BZ2PROBE returned {probe_resp!r}")
+        fails += 1
+
+    # 5. Round-trip tests. DIRECT + OPPORTUNISTIC always run; PROPAGATED runs
+    #    only when PYXIS_PROPAGATION_NODE_HEX is set (deployment-specific hash
+    #    kept out of source). Runs one pass minimum, then loops for --soak-hours.
+    deadline = time.time() + args.soak_hours * 3600
+    round_n = 0
+    payloads = [
+        ("direct-short",         "T:SEND",    "hi t-deck"),
+        ("direct-medium",        "T:SEND",    "this is a 100-char-ish payload to verify pyxis can frame a 1-packet LXMF message cleanly y"),
+        ("opportunistic-short",  "T:SENDOPP", "opp t-deck"),
+        ("opportunistic-medium", "T:SENDOPP", "opp 100-char payload should fit a single LXMF packet without any link establishment xx"),
+    ]
+
+    propagation_node_hex = os.environ.get("PYXIS_PROPAGATION_NODE_HEX", "").strip()
+    if propagation_node_hex:
+        log("HARNESS", f"Configuring propagation node on pyxis: {propagation_node_hex}")
+        set_resp = t.send_command(f"T:SETPROP {propagation_node_hex} 16", response_timeout=5)
+        if not set_resp or not set_resp.startswith("T:OK"):
+            log("HARNESS", f"WARN: T:SETPROP failed: {set_resp!r}")
+        if not wait_for_path(t, propagation_node_hex, timeout=60):
+            log("HARNESS", "WARN: pyxis never learned PN path; propagation rounds will fail")
+        else:
+            log("HARNESS", "Waiting for PN identity to land in recall cache...")
+            identity_deadline = time.time() + 60
+            while time.time() < identity_deadline:
+                r = t.send_command(f"T:RECALL {propagation_node_hex}", response_timeout=5)
+                if r and "size=" in r and int(r.split("size=", 1)[1].split()[0]) > 0:
+                    log("HARNESS", f"PN identity known: {r}")
+                    break
+                time.sleep(2.0)
+        payloads.append(("propagation-short", "T:SENDPROP", "prop t-deck"))
+    else:
+        log("HARNESS", "No PYXIS_PROPAGATION_NODE_HEX set — skipping PROPAGATED rounds")
+
+    first_pass = True
+    while first_pass or time.time() < deadline:
+        first_pass = False
+        round_n += 1
+        for label, send_cmd, content in payloads:
+            log("HARNESS", f"=== Round {round_n} / {label} ===")
+            # Mark current echobot log so we only look at lines after this point
+            with open(ECHOBOT_LOG) as f:
+                marker_text = f.read()
+            marker = marker_text[-256:] if len(marker_text) > 256 else marker_text
+
+            # 4a. T-Deck → Bot
+            send_resp = t.send_command(f"{send_cmd} {bot_dest} {content}", response_timeout=10)
+            if not send_resp or not send_resp.startswith("T:OK"):
+                log("HARNESS", f"FAIL [{label}]: T:SEND returned {send_resp}")
+                fails += 1
+                continue
+            # Parse the message hash so we can poll state
+            try:
+                msg_hash = send_resp.split("hash=", 1)[1].split()[0]
+            except Exception:
+                msg_hash = None
+                log("HARNESS", f"WARN [{label}]: could not parse msg_hash from {send_resp}")
+
+            # Wait for bot to log RX
+            rx_line = echobot_log_after(
+                marker,
+                lambda L: "Message RX:" in L and "from=" + pyxis_dest in L,
+                timeout=60,
+            )
+            if not rx_line:
+                log("HARNESS", f"FAIL [{label}]: bot never received from pyxis")
+                fails += 1
+                continue
+            log("HARNESS", f"OK  [{label}]: bot RX: {rx_line.strip()}")
+
+            # Wait for echo back
+            echo_line = echobot_log_after(
+                marker,
+                lambda L: "Echo queued for delivery" in L
+                          or "Echo SENT" in L or "Echo DELIVERED" in L,
+                timeout=20,
+            )
+            if not echo_line:
+                log("HARNESS", f"WARN [{label}]: bot did not log echo send "
+                               f"(may still arrive)")
+
+            # For PROPAGATED rounds the bot's echo doesn't go directly
+            # to pyxis — it gets uploaded to the PN first, then pyxis
+            # has to sync down. Wait for the bot's "Echo DELIVERED" (or
+            # "Echo SENT") log line before kicking a sync, then retry
+            # sync a few times because the first sync after bot upload
+            # can race the PN's index update.
+            if send_cmd == "T:SENDPROP":
+                upload_line = echobot_log_after(
+                    marker,
+                    lambda L: ("Echo DELIVERED" in L
+                               or "Echo SENT" in L
+                               or "Echo queued" in L),
+                    timeout=20,
+                )
+                # Echo "DELIVERED" for PROPAGATED in python LXMF means
+                # uploaded to PN. Even after that the PN index update
+                # can lag a few seconds, so we give it a small grace
+                # period.
+                if upload_line and ("DELIVERED" in upload_line
+                                    or "SENT" in upload_line):
+                    log("HARNESS", "  Bot uploaded echo to PN; "
+                                   "waiting 3s for PN to index it")
+                    time.sleep(3)
+                else:
+                    log("HARNESS", "  WARN: didn't see echo upload; "
+                                   "syncing anyway and hoping for the best")
+
+                # Try up to 4 sync rounds. Each call to T:SYNCPROP
+                # restarts the FSM if it's idle/complete/failed; we
+                # poll T:SYNCSTATE until PR_COMPLETE (=6) or FAILED (=7).
+                for sync_attempt in range(4):
+                    log("HARNESS",
+                        f"  T:SYNCPROP attempt {sync_attempt + 1}/4")
+                    t.send_command("T:SYNCPROP", response_timeout=5)
+                    sync_deadline = time.time() + 30
+                    while time.time() < sync_deadline:
+                        sresp = t.send_command("T:SYNCSTATE",
+                                               response_timeout=5)
+                        if sresp and ("state=6" in sresp
+                                      or "state=7" in sresp):
+                            break
+                        time.sleep(2)
+                    # Quick peek for the echo before kicking another sync
+                    quick_match = poll_tdeck_rx(t, "echo:", timeout=5)
+                    if quick_match:
+                        break
+
+            # Wait for pyxis to RX the echo
+            rx_timeout = 60 if send_cmd != "T:SENDPROP" else 30
+            rx_match = poll_tdeck_rx(t, "echo:", timeout=rx_timeout)
+            if not rx_match:
+                log("HARNESS", f"FAIL [{label}]: pyxis never received echo")
+                fails += 1
+                continue
+            log("HARNESS", f"OK  [{label}]: pyxis RX echo: {rx_match}")
+
+            # Clear the pyxis RX ring so the next round starts clean
+            t.send_command("T:RXCLR", response_timeout=5)
+
+            successes += 1
+            log("HARNESS", f"=== Round {round_n} / {label}: PASS "
+                           f"(total: {successes} pass, {fails} fail) ===")
+
+        # Stability check: peek for any panic/abort/assert in the T-Deck log
+        # (drain new lines but don't block on them).
+        for line in t.drain_lines():
+            if any(k in line for k in (
+                "Guru Meditation", "PANIC", "abort", "assertion", "rst:0x"
+            )):
+                log("HARNESS", f"CRASH detected: {line}")
+                fails += 1
+                deadline = 0
+                break
+
+        time.sleep(args.cadence)
+
+    log("HARNESS", f"Soak complete. successes={successes} fails={fails}")
+    bot.terminate()
+    bot_fh.close()
+    t.close()
+    return 0 if fails == 0 and successes > 0 else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())