ci(e2e): on-demand Meshtasticator e2e job for the docker-enabled runner

Adds .gitea/workflows/meshtasticator-e2e.yaml (workflow_dispatch until
validated green, pull_request trigger commented out): builds the pinned
Hermes contract env like release.yaml, installs the plugin with deps,
checks out Meshtasticator at a pinned commit, installs the simulator's
own (conflicting-version) requirements in an isolated venv, then runs
scripts/meshtasticator_e2e/run_e2e.sh --mode docker.

Supporting changes:
- run_e2e.sh: SIM_VENV support; automatic localhost->daemon TCP forwards
  (scripts/meshtasticator_e2e/tcp_forward.py) when DOCKER_HOST is a tcp://
  daemon, because Meshtasticator hardcodes localhost for its node control
  connections; node-boot settle wait; fixed docker cleanup (container
  'Meshtastic', volume 'Meshtasticator').
- e2e tests: connect via a module-long event loop in a background thread
  (adapter requires a running loop for watchdog + inbound dispatch, so
  per-call asyncio.run was wrong) with retry while the node TCP API boots.
- README: sim-venv guidance, remote-daemon forwarding, CI section.

Docker access follows the deployment examples in
eric/actions-docker (docker is available to runner jobs).

.gitea/workflows/meshtasticator-e2e.yaml

@@ -0,0 +1,90 @@
+name: meshtasticator-e2e
+
+# Radio-level end-to-end validation against Meshtasticator's interactive
+# simulator (real MeshtasticD per node + simulated LoRa PHY). Runs on demand
+# until it is validated green on the runner; afterwards switch `on` to also
+# trigger on pull_request touching plugin/ and this workflow.
+#
+# Docker access follows the deployment examples in
+# https://git.ericxliu.me/eric/actions-docker (docker is available to jobs);
+# Meshtasticator's docker mode publishes node TCP API ports (4404+n) from the
+# meshtastic/meshtasticd container and the simulator connects to them on
+# localhost. scripts/meshtasticator_e2e/run_e2e.sh starts localhost->daemon
+# TCP forwards automatically when DOCKER_HOST is a tcp:// daemon, which is how
+# this job is configured.
+
+on:
+  workflow_dispatch:
+  # pull_request:
+  #   paths:
+  #     - 'plugin/**'
+  #     - 'scripts/meshtasticator_e2e/**'
+  #     - '.gitea/workflows/meshtasticator-e2e.yaml'
+
+concurrency:
+  # One e2e run at a time: Meshtasticator hardcodes the container/volume name
+  # ("Meshtastic"/"Meshtasticator") on the shared docker daemon.
+  group: meshtasticator-e2e
+  cancel-in-progress: true
+
+jobs:
+  e2e:
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    env:
+      DOCKER_HOST: tcp://docker.local:2375
+      MESHTASTICATOR_COMMIT: 17ceb8231079d87b070abc6132181e4c6b20202d
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python 3.13 (matches Hermes image)
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.13'
+
+      - name: Install uv
+        run: pip install uv
+
+      # --- Hermes gateway contract env (mirrors .gitea/workflows/release.yaml) ---
+      - name: Clone pinned Hermes gateway source (deployed image tag)
+        run: |
+          git clone --depth 1 --branch v2026.8.31 \
+            https://github.com/NousResearch/hermes-agent.git plugin/.hermes-src
+          test "$(git -C plugin/.hermes-src rev-parse HEAD)" = \
+            "29112bef099274229cadff79cdff7bf7b99c4b77"
+
+      - name: Editable install hermes-agent + pytest + plugin (meshtastic client lib)
+        run: |
+          uv venv --python 3.13 plugin/.venv-contract
+          uv pip install --python plugin/.venv-contract/bin/python -e plugin/.hermes-src
+          uv pip install --python plugin/.venv-contract/bin/python pytest
+          uv pip install --python plugin/.venv-contract/bin/python -e plugin
+
+      # --- Simulator env (isolated: meshtasticator pins meshtastic~=2.6.1) ---
+      - name: Clone Meshtasticator at pinned commit
+        run: |
+          git clone https://github.com/meshtastic/meshtasticator.git "$HOME/meshtasticator"
+          git -C "$HOME/meshtasticator" checkout "$MESHTASTICATOR_COMMIT"
+          test "$(git -C "$HOME/meshtasticator" rev-parse HEAD)" = "$MESHTASTICATOR_COMMIT"
+
+      - name: Simulator venv (its own requirements + docker SDK)
+        run: |
+          uv venv --python 3.13 "$HOME/sim-venv"
+          uv pip install --python "$HOME/sim-venv/bin/python" \
+            -r "$HOME/meshtasticator/requirements.txt" docker
+
+      - name: Run Meshtasticator e2e suite
+        run: |
+          export MESHTASTICATOR_DIR="$HOME/meshtasticator"
+          export SIM_VENV="$HOME/sim-venv"
+          export E2E_VENV="plugin/.venv-contract"
+          scripts/meshtasticator_e2e/run_e2e.sh --mode docker
+
+      - name: Print simulator log (on failure)
+        if: failure()
+        run: |
+          echo "===== meshtasticator-sim.log ====="
+          tail -n 200 meshtasticator-sim.log 2>/dev/null || true
+          echo "===== docker ps ====="
+          docker ps -a 2>/dev/null || true

.gitea/workflows/release.yaml

@@ -8,8 +8,49 @@ on:
       - 'v*'
 
 jobs:
+  # Gateway contract gate: the plugin must bind and wire against the REAL
+  # Hermes gateway API (pinned to the deployed image source) before any
+  # release can be built or published. Without this, adapter/API drift used
+  # to surface only as a gateway crash after deploy (the set_message_handler
+  # AttributeError). HERMES_CONTRACT_REQUIRED=1 turns a missing hermes-agent
+  # into a hard failure so this gate can never silently skip.
+  contract-test:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python 3.13 (matches Hermes image)
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.13'
+
+      - name: Install uv
+        run: pip install uv
+
+      - name: Clone pinned Hermes gateway source (deployed image tag)
+        run: |
+          git clone --depth 1 --branch v2026.8.31 \
+            https://github.com/NousResearch/hermes-agent.git plugin/.hermes-src
+          test "$(git -C plugin/.hermes-src rev-parse HEAD)" = \
+            "29112bef099274229cadff79cdff7bf7b99c4b77"
+
+      - name: Editable install hermes-agent + pytest
+        run: |
+          uv venv --python 3.13 plugin/.venv-contract
+          uv pip install --python plugin/.venv-contract/bin/python -e plugin/.hermes-src
+          uv pip install --python plugin/.venv-contract/bin/python pytest
+
+      - name: Run gateway contract tests
+        env:
+          HERMES_CONTRACT_REQUIRED: '1'
+        run: |
+          cd plugin
+          .venv-contract/bin/python -m pytest tests/contract -q
+
   test-and-release:
     runs-on: ubuntu-latest
+    needs: contract-test
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
@@ -40,21 +81,37 @@ jobs:
           echo "Publishing release for $tag_name"
           wheel_file=$(ls plugin/dist/*.whl | head -n 1)
           tar_file=$(ls plugin/dist/*.tar.gz | head -n 1)
-          
-          # Create release
-          response=$(curl -s -k -X POST "${{ github.server_url }}/api/v1/repos/${{ github.repository }}/releases" \
-            -H "Authorization: token $GITEA_TOKEN" \
-            -H "Content-Type: application/json" \
-            -d "{\"tag_name\": \"$tag_name\", \"name\": \"$tag_name\", \"body\": \"Release $tag_name for hermes-meshtastic plugin\"}")
-          
-          release_id=$(echo "$response" | grep -o '"id":[0-9]*' | head -n 1 | cut -d: -f2)
+          api_url="${{ github.server_url }}/api/v1/repos/${{ github.repository }}"
+
+          # server_url is http and Gitea answers with a 308 redirect to https.
+          # curl needs -L to follow, AND --location-trusted: without it curl
+          # drops the Authorization header on the scheme-changing redirect and
+          # Gitea replies "token is required". Release may already exist
+          # (retried run) — reuse it instead of failing.
+          fetch_id() { python3 -c "import json,sys;print(json.load(sys.stdin).get('id') or '')" 2>/dev/null; }
+
+          response=$(curl -s -k -L --location-trusted -H "Authorization: token $GITEA_TOKEN" \
+            "$api_url/releases/tags/$tag_name")
+          release_id=$(printf '%s' "$response" | fetch_id)
+
+          if [ -z "$release_id" ]; then
+            response=$(curl -s -k -L --location-trusted -X POST "$api_url/releases" \
+              -H "Authorization: token $GITEA_TOKEN" \
+              -H "Content-Type: application/json" \
+              -d "{\"tag_name\": \"$tag_name\", \"name\": \"$tag_name\", \"body\": \"Release $tag_name for hermes-meshtastic plugin\"}")
+            release_id=$(printf '%s' "$response" | fetch_id)
+          fi
+
           if [ -n "$release_id" ]; then
-            curl -s -k -X POST "${{ github.server_url }}/api/v1/repos/${{ github.repository }}/releases/$release_id/assets?name=$(basename $wheel_file)" \
+            curl -s -k -L --location-trusted -X POST "$api_url/releases/$release_id/assets?name=$(basename "$wheel_file")" \
               -H "Authorization: token $GITEA_TOKEN" \
               -H "Content-Type: application/octet-stream" \
               --data-binary "@$wheel_file"
-            curl -s -k -X POST "${{ github.server_url }}/api/v1/repos/${{ github.repository }}/releases/$release_id/assets?name=$(basename $tar_file)" \
+            curl -s -k -L --location-trusted -X POST "$api_url/releases/$release_id/assets?name=$(basename "$tar_file")" \
               -H "Authorization: token $GITEA_TOKEN" \
               -H "Content-Type: application/octet-stream" \
               --data-binary "@$tar_file"
+          else
+            echo "ERROR: could not create or fetch release $tag_name" >&2
+            exit 1
           fi

.gitignore

@@ -4,3 +4,7 @@
 .pio/
 *.log
 plugin/dist/
+plugin/.venv-contract/
+plugin/.hermes-src/
+__pycache__/
+*.pyc

plugin/.gitignore

@@ -0,0 +1,6 @@
+dist/
+__pycache__/
+*.pyc
+.venv-contract/
+.hermes-src/
+.pytest_cache/

plugin/hermes_meshtastic/__init__.py

@@ -1,16 +1,23 @@
-"""Meshtastic Platform Adapter plugin for Hermes Agent."""
+"""Meshtastic Platform Adapter plugin for Hermes Agent.
 
-from .adapter import MeshtasticAdapter, check_requirements, MAX_LORA_MESSAGE_LENGTH
+The ``register(ctx)`` entry point resolves the adapter class lazily so the real
+Hermes gateway API (``gateway.platforms.base``) is imported only at registration
+time — never at plugin-discovery import time (see ``adapter.get_adapter_class``).
+"""
+
+from .formatting import MAX_LORA_MESSAGE_LENGTH
 
 
 def register(ctx):
     """Hermes plugin entrypoint."""
-    from gateway.platform_registry import PlatformEntry
+    from .adapter import check_requirements, get_adapter_class
+
+    adapter_cls = get_adapter_class()
 
     ctx.register_platform(
         name="meshtastic",
         label="Meshtastic LoRa",
-        adapter_factory=lambda config: MeshtasticAdapter(config),
+        adapter_factory=lambda config: adapter_cls(config),
         check_fn=check_requirements,
         max_message_length=MAX_LORA_MESSAGE_LENGTH,
         emoji="📻",

plugin/hermes_meshtastic/adapter.py

@@ -3,306 +3,345 @@
 Connects to a Meshtastic node via TCP (e.g. meshtastic.local:4403), listening for
 LoRa packets, stripping formatting, and dispatching to Hermes agent sessions. Outbound
 replies are chunked into concise LoRa-friendly packets.
+
+Contract note
+-------------
+The Hermes gateway API (``gateway.platforms.base.BasePlatformAdapter`` and friends)
+is resolved LAZILY, inside ``get_adapter_class()``, never at module import time.
+Hermes' own guidance (``hermes_cli/plugins.py``) puts gateway imports inside factory
+bodies because third-party plugin modules are imported during plugin discovery, which
+runs before/independently of the gateway runtime being fully importable. Importing
+``gateway.platforms.base`` at module scope silently failed in the deployed runtime
+(an ``ImportError`` swallowed by an earlier fallback bound the adapter to a local stub
+class lacking ``set_message_handler`` — see git history). Deferred resolution keeps
+this module importable in isolation (unit tests / CI without Hermes) while always
+binding the adapter to the REAL gateway base at registration/creation time.
+
+The contract tests in ``tests/contract/`` assert this invariant against the real
+``hermes-agent`` package (pinned to the deployed image tag).
 """
 
 from __future__ import annotations
 
 import asyncio
 import logging
-import re
 import time
 from typing import Any, Dict, List, Optional
 
-try:
-    from gateway.platforms.base import (
-        BasePlatformAdapter,
-        MessageEvent,
-        MessageType,
-        SendResult,
-    )
-except ImportError:
-    # Allow importing and testing utility functions without gateway in environment
-    class BasePlatformAdapter:  # type: ignore[no-redef]
-        def __init__(self, config: Any, platform: Any = "meshtastic"):
-            self.config = config
-            self.platform = platform
-
-        def build_source(self, **kwargs):
-            return kwargs
-
-        async def handle_message(self, event: Any) -> None:
-            pass
-
-    class SendResult:  # type: ignore[no-redef]
-        def __init__(self, success: bool, message_id: Optional[str] = None, error: Optional[str] = None):
-            self.success = success
-            self.message_id = message_id
-            self.error = error
-
-    class MessageType:  # type: ignore[no-redef]
-        TEXT = "text"
-
-    class MessageEvent:  # type: ignore[no-redef]
-        def __init__(self, **kwargs):
-            for k, v in kwargs.items():
-                setattr(self, k, v)
+from .formatting import MAX_LORA_MESSAGE_LENGTH, chunk_text, strip_markdown
 
 logger = logging.getLogger(__name__)
 
-MAX_LORA_MESSAGE_LENGTH = 180  # Characters per packet to prevent LoRa airtime congestion
-
-_MARKDOWN_RULES = (
-    (r"\*\*(.+?)\*\*", r"\1"),  # bold
-    (r"__(.+?)__", r"\1"),
-    (r"\*(.+?)\*", r"\1"),      # italic
-    (r"(?<!\w)_(.+?)_(?!\w)", r"\1"),
-    (r"`(.+?)`", r"\1"),        # inline code
-    (r"```[\w]*\n?", ""),       # code fences
-    (r"!\[([^\]]*)\]\(([^)]+)\)", r"\1"),
-    (r"\[([^\]]+)\]\(([^)]+)\)", r"\1 (\2)"),
-    (r"^#+\s*", ""),            # headings
-    (r"^\s*[-*+]\s+", "- "),    # normalize lists
-)
+__all__ = [
+    "MAX_LORA_MESSAGE_LENGTH",
+    "check_requirements",
+    "chunk_text",
+    "get_adapter_class",
+    "strip_markdown",
+]
 
 
-def strip_markdown(text: str) -> str:
-    """Convert markdown formatting into plain readable text suitable for radio."""
-    if not text:
-        return ""
-    result = text
-    for pattern, replacement in _MARKDOWN_RULES:
-        result = re.sub(pattern, replacement, result, flags=re.MULTILINE)
-    # Collapse multiple blank lines
-    result = re.sub(r"\n{3,}", "\n\n", result)
-    return result.strip()
+# ---------------------------------------------------------------------------
+# Hermes gateway API — resolved lazily and cached (see module docstring).
+# ---------------------------------------------------------------------------
+
+_gateway_api: Optional[Dict[str, Any]] = None
 
 
-def chunk_text(text: str, limit: int = MAX_LORA_MESSAGE_LENGTH) -> List[str]:
-    """Split text into smaller chunks for LoRa transmission, breaking at whitespace/newlines."""
-    if not text:
-        return []
-    if len(text) <= limit:
-        return [text]
+def _load_gateway_api() -> Dict[str, Any]:
+    """Import the real Hermes gateway types and bind them as module globals.
 
-    chunks: List[str] = []
-    remaining = text
-    while remaining:
-        if len(remaining) <= limit:
-            chunks.append(remaining.strip())
-            break
-        split_at = remaining.rfind(" ", 0, limit)
-        if split_at <= 0 or split_at < (limit // 3):
-            newline_at = remaining.rfind("\n", 0, limit)
-            if newline_at > 0:
-                split_at = newline_at
-            else:
-                split_at = limit
-        chunk = remaining[:split_at].strip()
-        if chunk:
-            chunks.append(chunk)
-        remaining = remaining[split_at:].lstrip()
-    return chunks
+    Called from ``get_adapter_class()`` at adapter-creation time (the gateway
+    runtime is fully loaded by then). Method bodies reference these names as
+    module globals, so they are injected into this module's namespace.
+    """
+    global _gateway_api
+    if _gateway_api is None:
+        from gateway.config import Platform
+        from gateway.platforms.base import (  # type: ignore[import-not-found]
+            BasePlatformAdapter,
+            MessageEvent,
+            MessageType,
+            SendResult,
+        )
+
+        _gateway_api = {
+            "Platform": Platform,
+            "BasePlatformAdapter": BasePlatformAdapter,
+            "MessageEvent": MessageEvent,
+            "MessageType": MessageType,
+            "SendResult": SendResult,
+        }
+        globals().update(_gateway_api)
+    return _gateway_api
+
+
+# ---------------------------------------------------------------------------
+# Dependency probe (passive — never installs anything).
+# ---------------------------------------------------------------------------
 
 
 def check_requirements() -> bool:
     """Check if meshtastic is installed."""
     try:
-        import meshtastic
-        import meshtastic.tcp_interface
+        import meshtastic  # noqa: F401
+        import meshtastic.tcp_interface  # noqa: F401
         return True
     except ImportError:
         return False
 
 
-class MeshtasticAdapter(BasePlatformAdapter):
-    """Hermes gateway adapter for Meshtastic LoRa radios over TCP."""
+# ---------------------------------------------------------------------------
+# Adapter — class body defined once the real gateway API is available.
+# ---------------------------------------------------------------------------
 
-    def __init__(self, config: Any):
-        super().__init__(config, platform="meshtastic")
-        self.extra = getattr(config, "extra", {}) or {}
-        self.host = self.extra.get("host", "meshtastic.local")
-        self.port = int(self.extra.get("port", 4403))
-        self.channel_index = int(self.extra.get("channel_index", 1))
-        self.allowed_nodes: List[str] = [
-            n.strip() for n in self.extra.get("allowed_nodes", []) if n.strip()
-        ]
-        self._iface = None
-        self._loop: Optional[asyncio.AbstractEventLoop] = None
-        self._connected = False
-        self._running = False
-        self._task: Optional[asyncio.Task] = None
+_adapter_class = None
 
-    async def connect(self) -> None:
-        """Connect to Meshtastic radio via TCP and start subscriber."""
-        self._loop = asyncio.get_running_loop()
-        self._running = True
-        self._task = asyncio.create_task(self._connection_supervisor())
 
-    async def _connection_supervisor(self) -> None:
-        backoff = 2.0
-        while self._running:
-            try:
-                import meshtastic.tcp_interface
-                from pubsub import pub
+def get_adapter_class():
+    """Return the ``MeshtasticAdapter`` class bound to the real gateway base.
 
-                logger.info("[Meshtastic] Connecting to radio at %s:%s...", self.host, self.port)
-                # TCPInterface handles initial protobuf handshake synchronously
-                self._iface = await asyncio.to_thread(
-                    meshtastic.tcp_interface.TCPInterface,
-                    hostname=self.host,
-                    portNumber=self.port,
-                    noProto=False,
-                )
-                self._connected = True
-                backoff = 2.0
-                my_id = getattr(self._iface.myInfo, "my_node_num", "unknown")
-                logger.info("[Meshtastic] Connected to radio (my_node_num=%s, listening on ch=%d)", my_id, self.channel_index)
+    The class is defined (and cached) on first call so that ``BasePlatformAdapter``
+    is the REAL Hermes class — a fallback stub at this point would reproduce the
+    ``set_message_handler`` AttributeError this design exists to prevent.
+    """
+    global _adapter_class
+    if _adapter_class is None:
+        _load_gateway_api()
 
-                pub.subscribe(self._on_meshtastic_receive, "meshtastic.receive.text")
+        class MeshtasticAdapter(BasePlatformAdapter):
+            """Hermes gateway adapter for Meshtastic LoRa radios over TCP."""
 
-                # Keep-alive loop
-                while self._running and self._connected:
-                    await asyncio.sleep(5)
-                    if not self._iface or not self._iface.isConnected:
-                        break
-
-            except asyncio.CancelledError:
-                break
-            except Exception as exc:
-                logger.warning("[Meshtastic] Radio connection failed (%s). Retrying in %.1fs...", exc, backoff)
+            def __init__(self, config: Any):
+                super().__init__(config=config, platform=Platform("meshtastic"))
+                self.extra = getattr(config, "extra", {}) or {}
+                self.host = self.extra.get("host", "meshtastic.local")
+                self.port = int(self.extra.get("port", 4403))
+                self.channel_index = int(self.extra.get("channel_index", 1))
+                self.allowed_nodes: List[str] = [
+                    n.strip() for n in self.extra.get("allowed_nodes", []) if n.strip()
+                ]
+                self._iface = None
+                self._loop: Optional[asyncio.AbstractEventLoop] = None
                 self._connected = False
-                await asyncio.sleep(backoff)
-                backoff = min(backoff * 1.5, 60.0)
+                self._running = False
+                self._task: Optional[asyncio.Task] = None
+                self._connecting = False
 
-    def _on_meshtastic_receive(self, packet: Dict[str, Any], interface: Any) -> None:
-        """Handle incoming text packet from Meshtastic pubsub (runs in worker thread)."""
-        try:
-            ch = packet.get("channel", 0)
-            if self.channel_index is not None and ch != self.channel_index:
-                # Ignore packets from different channels (e.g. public channel 0 when listening on 1)
-                return
+            async def connect(self, *, is_reconnect: bool = False) -> bool:
+                """Establish the Meshtastic TCP session; return success.
 
-            from_id = packet.get("fromId") or f"!{packet.get('from', 0):08x}"
-            my_info = getattr(interface, "myInfo", None)
-            my_node_num = getattr(my_info, "my_node_num", 0)
-            if packet.get("from") == my_node_num:
-                # Prevent self-message loops
-                return
+                Mirrors the gateway contract ``async connect(*, is_reconnect=False)
+                -> bool`` (see tests/contract/). On initial failure the watchdog
+                keeps retrying with exponential backoff in the background.
+                """
+                self._loop = asyncio.get_running_loop()
+                self._running = True
+                try:
+                    ok = await self._try_connect()
+                except Exception as exc:
+                    logger.warning(
+                        "[Meshtastic] Initial connect failed (%s); watchdog will retry.", exc
+                    )
+                    ok = False
+                self._connected = ok
+                self._ensure_watchdog()
+                return ok
 
-            if self.allowed_nodes and from_id not in self.allowed_nodes:
-                logger.debug("[Meshtastic] Ignoring message from unauthorized node: %s", from_id)
-                return
+            async def _try_connect(self) -> bool:
+                if self._connecting:
+                    return False  # single-flight: gateway + watchdog must not double-connect
+                self._connecting = True
+                try:
+                    import meshtastic.tcp_interface
+                    from pubsub import pub
 
-            decoded = packet.get("decoded", {})
-            text = decoded.get("text", "").strip()
-            if not text:
-                return
+                    logger.info(
+                        "[Meshtastic] Connecting to radio at %s:%s (is_reconnect=%s)...",
+                        self.host, self.port, getattr(self, "_last_reconnect", False),
+                    )
+                    # TCPInterface performs the protobuf handshake synchronously.
+                    self._iface = await asyncio.to_thread(
+                        meshtastic.tcp_interface.TCPInterface,
+                        hostname=self.host,
+                        portNumber=self.port,
+                        noProto=False,
+                    )
+                    my_id = getattr(self._iface.myInfo, "my_node_num", "unknown")
+                    logger.info(
+                        "[Meshtastic] Connected to radio (my_node_num=%s, listening on ch=%d)",
+                        my_id, self.channel_index,
+                    )
+                    pub.subscribe(self._on_meshtastic_receive, "meshtastic.receive.text")
+                    return True
+                finally:
+                    self._connecting = False
 
-            # Determine sender and target
-            to_id = packet.get("toId")
-            is_dm = to_id != "^all" and packet.get("to") == my_node_num
-            chat_id = from_id if is_dm else f"channel_{ch}"
-            chat_type = "dm" if is_dm else "channel"
+            def _ensure_watchdog(self) -> None:
+                if self._task is None or self._task.done():
+                    self._task = asyncio.create_task(self._connection_watchdog())
 
-            rx_snr = packet.get("rxSnr", 0.0)
-            logger.info(
-                "[Meshtastic] Inbound message from %s on ch=%s (is_dm=%s, SNR=%.1fdB): %s",
-                from_id, ch, is_dm, rx_snr, text
-            )
+            async def _connection_watchdog(self) -> None:
+                """Keep the radio session alive; reconnect with backoff on drop."""
+                backoff = 2.0
+                while self._running:
+                    await asyncio.sleep(5)
+                    try:
+                        iface = self._iface
+                        if iface is not None and not iface.isConnected:
+                            self._connected = False
+                    except Exception:
+                        self._connected = False
+                    if self._connected:
+                        backoff = 2.0
+                        continue
+                    # Radio went away (or never came up) — reconnect.
+                    self._last_reconnect = True
+                    try:
+                        ok = await self._try_connect()
+                        self._connected = ok
+                        backoff = 2.0
+                    except Exception as exc:
+                        logger.warning(
+                            "[Meshtastic] Radio reconnect failed (%s). Retrying in %.1fs...",
+                            exc, backoff,
+                        )
+                        self._connected = False
+                        await asyncio.sleep(backoff)
+                        backoff = min(backoff * 1.5, 60.0)
 
-            # Build Hermes SessionSource and MessageEvent
-            source = self.build_source(
-                chat_id=chat_id,
-                chat_name=f"Meshtastic {chat_id}",
-                chat_type=chat_type,
-                user_id=from_id,
-                user_name=from_id,
-                message_id=str(packet.get("id", int(time.time()))),
-            )
+            def _on_meshtastic_receive(self, packet: Dict[str, Any], interface: Any) -> None:
+                """Handle incoming text packet from Meshtastic pubsub (worker thread)."""
+                try:
+                    ch = packet.get("channel", 0)
+                    if self.channel_index is not None and ch != self.channel_index:
+                        # Ignore packets from different channels (e.g. public channel 0).
+                        return
 
-            event = MessageEvent(
-                text=text,
-                message_type=MessageType.TEXT,
-                user_id=from_id,
-                user_name=from_id,
-                source=source,
-                raw_message=packet,
-                message_id=str(packet.get("id", int(time.time()))),
-            )
+                    from_id = packet.get("fromId") or f"!{packet.get('from', 0):08x}"
+                    my_info = getattr(interface, "myInfo", None)
+                    my_node_num = getattr(my_info, "my_node_num", 0)
+                    if packet.get("from") == my_node_num:
+                        # Prevent self-message loops.
+                        return
 
-            if self._loop and self._loop.is_running():
-                asyncio.run_coroutine_threadsafe(self.handle_message(event), self._loop)
+                    if self.allowed_nodes and from_id not in self.allowed_nodes:
+                        logger.debug(
+                            "[Meshtastic] Ignoring message from unauthorized node: %s", from_id
+                        )
+                        return
 
-        except Exception as exc:
-            logger.exception("[Meshtastic] Error processing received packet: %s", exc)
+                    decoded = packet.get("decoded", {})
+                    text = decoded.get("text", "").strip()
+                    if not text:
+                        return
 
-    async def send(
-        self,
-        chat_id: str,
-        content: str,
-        reply_to: Optional[str] = None,
-        metadata: Optional[Dict[str, Any]] = None,
-    ) -> SendResult:
-        """Send a message out over LoRa (implements BasePlatformAdapter.send)."""
-        if not self._connected or not self._iface:
-            return SendResult(success=False, error="Meshtastic radio not connected")
+                    to_id = packet.get("toId")
+                    is_dm = to_id != "^all" and packet.get("to") == my_node_num
+                    chat_id = from_id if is_dm else f"channel_{ch}"
+                    chat_type = "dm" if is_dm else "channel"
 
-        clean_text = strip_markdown(content)
-        chunks = chunk_text(clean_text, MAX_LORA_MESSAGE_LENGTH)
+                    rx_snr = packet.get("rxSnr", 0.0)
+                    logger.info(
+                        "[Meshtastic] Inbound message from %s on ch=%s (is_dm=%s, SNR=%.1fdB): %s",
+                        from_id, ch, is_dm, rx_snr, text,
+                    )
 
-        # Destination: if chat_id starts with '!', it is a direct node ID
-        destination = chat_id if chat_id.startswith("!") else "^all"
-        channel_index = self.channel_index if destination == "^all" else 0
+                    source = self.build_source(
+                        chat_id=chat_id,
+                        chat_name=f"Meshtastic {chat_id}",
+                        chat_type=chat_type,
+                        user_id=from_id,
+                        user_name=from_id,
+                        message_id=str(packet.get("id", int(time.time()))),
+                    )
 
-        last_id = None
-        try:
-            for idx, chunk in enumerate(chunks):
-                if len(chunks) > 1:
-                    formatted_chunk = f"({idx+1}/{len(chunks)}) {chunk}"
-                else:
-                    formatted_chunk = chunk
+                    event = MessageEvent(
+                        text=text,
+                        message_type=MessageType.TEXT,
+                        user_id=from_id,
+                        user_name=from_id,
+                        source=source,
+                        raw_message=packet,
+                        message_id=str(packet.get("id", int(time.time()))),
+                    )
 
-                logger.info(
-                    "[Meshtastic] Sending chunk to %s on ch=%s: %s",
-                    destination, channel_index, formatted_chunk
-                )
+                    if self._loop and self._loop.is_running():
+                        asyncio.run_coroutine_threadsafe(
+                            self.handle_message(event), self._loop
+                        )
+                except Exception as exc:
+                    logger.exception("[Meshtastic] Error processing received packet: %s", exc)
 
-                # Send via mesh interface thread
-                await asyncio.to_thread(
-                    self._iface.sendText,
-                    text=formatted_chunk,
-                    destinationId=destination,
-                    channelIndex=channel_index,
-                )
-                last_id = str(int(time.time() * 1000))
-                # Inter-packet delay to comply with LoRa duty cycle
-                if len(chunks) > 1:
-                    await asyncio.sleep(1.2)
+            async def send(
+                self,
+                chat_id: str,
+                content: str,
+                reply_to: Optional[str] = None,
+                metadata: Optional[Dict[str, Any]] = None,
+            ) -> SendResult:
+                """Send a message out over LoRa (implements BasePlatformAdapter.send)."""
+                if not self._connected or not self._iface:
+                    return SendResult(success=False, error="Meshtastic radio not connected")
 
-            return SendResult(success=True, message_id=last_id or str(int(time.time())))
-        except Exception as exc:
-            logger.exception("[Meshtastic] Failed to send message: %s", exc)
-            return SendResult(success=False, error=str(exc))
+                clean_text = strip_markdown(content)
+                chunks = chunk_text(clean_text, MAX_LORA_MESSAGE_LENGTH)
 
-    async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
-        """Get chat/channel info (implements BasePlatformAdapter.get_chat_info)."""
-        is_dm = chat_id.startswith("!")
-        return {
-            "name": f"Node {chat_id}" if is_dm else f"Channel {chat_id}",
-            "type": "dm" if is_dm else "channel",
-        }
+                # Destination: if chat_id starts with '!', it is a direct node ID.
+                destination = chat_id if chat_id.startswith("!") else "^all"
+                channel_index = self.channel_index if destination == "^all" else 0
 
-    async def disconnect(self) -> None:
-        """Disconnect and clean up resources."""
-        self._running = False
-        if self._task:
-            self._task.cancel()
-        if self._iface:
-            try:
-                from pubsub import pub
-                pub.unsubscribe(self._on_meshtastic_receive, "meshtastic.receive.text")
-            except Exception:
-                pass
-            await asyncio.to_thread(self._iface.close)
-            self._connected = False
-            logger.info("[Meshtastic] Radio interface closed")
+                last_id = None
+                try:
+                    for idx, chunk in enumerate(chunks):
+                        if len(chunks) > 1:
+                            formatted_chunk = f"({idx+1}/{len(chunks)}) {chunk}"
+                        else:
+                            formatted_chunk = chunk
+
+                        logger.info(
+                            "[Meshtastic] Sending chunk to %s on ch=%s: %s",
+                            destination, channel_index, formatted_chunk,
+                        )
+                        await asyncio.to_thread(
+                            self._iface.sendText,
+                            text=formatted_chunk,
+                            destinationId=destination,
+                            channelIndex=channel_index,
+                        )
+                        last_id = str(int(time.time() * 1000))
+                        # Inter-packet delay to comply with LoRa duty cycle.
+                        if len(chunks) > 1:
+                            await asyncio.sleep(1.2)
+                    return SendResult(success=True, message_id=last_id or str(int(time.time())))
+                except Exception as exc:
+                    logger.exception("[Meshtastic] Failed to send message: %s", exc)
+                    return SendResult(success=False, error=str(exc))
+
+            async def get_chat_info(self, chat_id: str) -> Dict[str, Any]:
+                """Get chat/channel info (implements BasePlatformAdapter.get_chat_info)."""
+                is_dm = chat_id.startswith("!")
+                return {
+                    "name": f"Node {chat_id}" if is_dm else f"Channel {chat_id}",
+                    "type": "dm" if is_dm else "channel",
+                }
+
+            async def disconnect(self) -> None:
+                """Disconnect and clean up resources."""
+                self._running = False
+                if self._task:
+                    self._task.cancel()
+                if self._iface:
+                    try:
+                        from pubsub import pub
+                        pub.unsubscribe(
+                            self._on_meshtastic_receive, "meshtastic.receive.text"
+                        )
+                    except Exception:
+                        pass
+                    await asyncio.to_thread(self._iface.close)
+                    self._connected = False
+                    logger.info("[Meshtastic] Radio interface closed")
+
+        _adapter_class = MeshtasticAdapter
+    return _adapter_class

plugin/hermes_meshtastic/formatting.py

@@ -0,0 +1,61 @@
+"""Pure formatting utilities for LoRa-friendly text.
+
+These helpers have NO dependency on the Hermes gateway package so they can be
+imported and unit-tested in isolation (CI runs them without hermes-agent).
+"""
+
+import re
+
+MAX_LORA_MESSAGE_LENGTH = 180  # Characters per packet to prevent LoRa airtime congestion
+
+_MARKDOWN_RULES = (
+    (r"\*\*(.+?)\*\*", r"\1"),  # bold
+    (r"__(.+?)__", r"\1"),
+    (r"\*(.+?)\*", r"\1"),      # italic
+    (r"(?<!\w)_(.+?)_(?!\w)", r"\1"),
+    (r"`(.+?)`", r"\1"),        # inline code
+    (r"```[\w]*\n?", ""),       # code fences
+    (r"!\[([^\]]*)\]\(([^)]+)\)", r"\1"),
+    (r"\[([^\]]+)\]\(([^)]+)\)", r"\1 (\2)"),
+    (r"^#+\s*", ""),            # headings
+    (r"^\s*[-*+]\s+", "- "),    # normalize lists
+)
+
+
+def strip_markdown(text: str) -> str:
+    """Convert markdown formatting into plain readable text suitable for radio."""
+    if not text:
+        return ""
+    result = text
+    for pattern, replacement in _MARKDOWN_RULES:
+        result = re.sub(pattern, replacement, result, flags=re.MULTILINE)
+    # Collapse multiple blank lines
+    result = re.sub(r"\n{3,}", "\n\n", result)
+    return result.strip()
+
+
+def chunk_text(text: str, limit: int = MAX_LORA_MESSAGE_LENGTH) -> list:
+    """Split text into smaller chunks for LoRa transmission, breaking at whitespace/newlines."""
+    if not text:
+        return []
+    if len(text) <= limit:
+        return [text]
+
+    chunks = []
+    remaining = text
+    while remaining:
+        if len(remaining) <= limit:
+            chunks.append(remaining.strip())
+            break
+        split_at = remaining.rfind(" ", 0, limit)
+        if split_at <= 0 or split_at < (limit // 3):
+            newline_at = remaining.rfind("\n", 0, limit)
+            if newline_at > 0:
+                split_at = newline_at
+            else:
+                split_at = limit
+        chunk = remaining[:split_at].strip()
+        if chunk:
+            chunks.append(chunk)
+        remaining = remaining[split_at:].lstrip()
+    return chunks

plugin/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "hermes-meshtastic"
-version = "0.1.2"
+version = "0.1.3"
 description = "Meshtastic LoRa mesh gateway adapter plugin for Hermes Agent"
 readme = "README.md"
 requires-python = ">=3.10"
@@ -23,3 +23,10 @@ meshtastic-platform = "hermes_meshtastic"
 
 [tool.hatch.build.targets.wheel]
 packages = ["hermes_meshtastic"]
+
+[tool.hatch.build.targets.sdist]
+exclude = [
+    ".venv-contract",
+    ".hermes-src",
+    "tests",
+]

plugin/tests/contract/test_gateway_contract.py

@@ -0,0 +1,230 @@
+"""Contract tests: hermes-meshtastic vs the REAL Hermes gateway API.
+
+These tests exist so adapter development converges WITHOUT deploying: they
+replay the gateway's adapter bootstrap against the actual ``hermes-agent``
+package (pinned to the deployed image tag, see ``scripts/contract-test.sh``)
+and fail loudly on any API drift that used to surface only as a gateway
+crash after release.
+
+Runtime ordering modelled here (verified against ``gateway/run.py`` and
+``gateway/config.py``):
+  1. plugin discovery runs the plugin's ``register(ctx)`` (registers the
+     platform in ``gateway.platform_registry``),
+  2. the gateway later instantiates the adapter via the registry entry and
+     wires it: ``set_message_handler`` -> ... -> ``connect()``,
+  3. ``Platform("meshtastic")`` (auto-created enum member) is only valid
+     AFTER registration — the enum's ``_missing_`` queries the registry.
+
+Layers covered
+--------------
+L0  The adapter subclasses the REAL ``gateway.platforms.base.BasePlatformAdapter``
+    (regression for the fallback-stub bug that caused the
+    ``AttributeError: ... 'set_message_handler'`` crash at ``gateway/run.py``).
+L0b ABC instantiation gate — construction with a real ``PlatformConfig`` fails
+    if any abstract member is missing or the platform enum is misused.
+L1  Wiring replay — every ``set_*`` call ``gateway/run.py`` makes at startup
+    must exist and accept a handler.
+L2  Signature conformance — ``connect``/``send`` match the gateway contract.
+L3  Event shape — the inbound ``MessageEvent``/``SessionSource`` construction
+    must be valid against the real gateway types.
+
+Skip policy: skipped when ``hermes-agent`` is not installed so the standalone
+unit suite (CI without Hermes) stays green. Set ``HERMES_CONTRACT_REQUIRED=1``
+to turn absence into a hard failure — the dedicated contract CI job sets this
+so the gate can never silently skip.
+"""
+
+import inspect
+import os
+
+import pytest
+
+try:
+    from gateway.config import Platform, PlatformConfig
+    from gateway.platforms.base import BasePlatformAdapter, MessageEvent, MessageType
+except ImportError:  # pragma: no cover - exercised when hermes-agent is absent
+    if os.environ.get("HERMES_CONTRACT_REQUIRED"):
+        raise AssertionError(
+            "hermes-agent is required for contract tests (HERMES_CONTRACT_REQUIRED=1). "
+            "Run `scripts/contract-test.sh` or the CI contract job."
+        )
+    pytest.skip(
+        "hermes-agent not installed — contract tests skipped (scripts/contract-test.sh)",
+        allow_module_level=True,
+    )
+
+import hermes_meshtastic  # noqa: F401,E402  (package must import without gateway side effects)
+from hermes_meshtastic import adapter as plugin_adapter  # noqa: E402
+
+MINIMAL_CONFIG = PlatformConfig(
+    enabled=True,
+    extra={"host": "127.0.0.1", "port": 1, "channel_index": 1},
+)
+
+# Exactly the set_* wiring gateway/run.py performs right before connect()
+# (observed at the crash site: adapter.set_message_handler(...) etc.).
+GATEWAY_WIRING_SETTERS = [
+    "set_message_handler",
+    "set_fatal_error_handler",
+    "set_session_store",
+    "set_busy_session_handler",
+    "set_topic_recovery_fn",
+    "set_authorization_check",
+    "set_platform_event_handler",
+    "set_reaction_handler",
+]
+
+_REGISTER_KWARGS = {}
+
+
+class _Ctx:
+    """Minimal ctx mirroring how discovery calls the plugin's register()."""
+
+    def register_platform(self, **kwargs):
+        _REGISTER_KWARGS.update(kwargs)
+
+
+def _noop(*_args, **_kwargs):
+    return None
+
+
+@pytest.fixture(scope="module", autouse=True)
+def registered_entry():
+    """Replay discovery: run register(), then register the PlatformEntry.
+
+    Mirrors how gateway/run.py later instantiates the adapter from the
+    registry — and why Platform("meshtastic") is valid only after this.
+    """
+    from gateway.platform_registry import PlatformEntry, platform_registry
+
+    hermes_meshtastic.register(_Ctx())
+    kw = dict(_REGISTER_KWARGS)
+    assert kw.get("name") == "meshtastic"
+
+    if not platform_registry.is_registered("meshtastic"):
+        entry = PlatformEntry(
+            name=kw["name"],
+            label=kw["label"],
+            adapter_factory=kw["adapter_factory"],
+            check_fn=kw["check_fn"],
+            required_env=[],
+            max_message_length=kw.get("max_message_length", 0),
+            emoji=kw.get("emoji", "🔌"),
+            platform_hint=kw.get("platform_hint", ""),
+        )
+        platform_registry.register(entry)
+    return platform_registry.get("meshtastic")
+
+
+def _make_adapter(entry=None):
+    from gateway.platform_registry import platform_registry
+
+    if entry is None:
+        entry = platform_registry.get("meshtastic")
+    assert entry is not None, "meshtastic platform not registered (fixture order broken)"
+    return entry.adapter_factory(MINIMAL_CONFIG)
+
+
+# --- L0: real base binding -------------------------------------------------
+
+
+def test_adapter_subclasses_real_gateway_base():
+    """Regression: adapter must bind the real base, never a local stub."""
+    cls = plugin_adapter.get_adapter_class()
+    assert inspect.isclass(cls)
+    assert issubclass(cls, BasePlatformAdapter)
+    assert cls.__mro__[1].__module__ == "gateway.platforms.base"
+
+
+def test_abstract_members_are_implemented():
+    """ABC gate: BasePlatformAdapter requires connect/send/disconnect/get_chat_info."""
+    cls = plugin_adapter.get_adapter_class()
+    for member in BasePlatformAdapter.__abstractmethods__:
+        impl = getattr(cls, member, None)
+        assert callable(impl), f"adapter missing abstract member: {member}"
+
+
+# --- L0b: instantiation against the real contract --------------------------
+
+
+def test_adapter_instantiates_with_real_platform_config(registered_entry):
+    adapter = _make_adapter(registered_entry)
+    assert isinstance(adapter, BasePlatformAdapter)
+    # super().__init__ must receive the Platform enum (auto-created member),
+    # not a bare string — the gateway calls Platform methods on adapter.platform.
+    assert isinstance(adapter.platform, Platform)
+    assert adapter.platform.value == "meshtastic"
+
+
+def test_registered_entry_carries_radio_limits(registered_entry):
+    assert registered_entry.max_message_length == plugin_adapter.MAX_LORA_MESSAGE_LENGTH
+    assert registered_entry.emoji == "📻"
+
+
+# --- L1: gateway wiring replay ---------------------------------------------
+
+
+def test_adapter_survives_gateway_wiring(registered_entry):
+    """Replay run.py's startup wiring; any missing setter crashes the gateway."""
+    adapter = _make_adapter(registered_entry)
+    for setter in GATEWAY_WIRING_SETTERS:
+        fn = getattr(adapter, setter, None)
+        assert callable(fn), (
+            f"adapter missing callable '{setter}' — gateway startup crashes here "
+            "(the original set_message_handler AttributeError)"
+        )
+        fn(_noop)
+    # run.py also pokes these attributes directly after wiring.
+    adapter._busy_text_mode = False
+
+
+# --- L2: signature conformance ----------------------------------------------
+
+
+def test_connect_signature_matches_gateway_contract():
+    cls = plugin_adapter.get_adapter_class()
+    sig = inspect.signature(cls.connect)
+    assert inspect.iscoroutinefunction(cls.connect)
+    assert "is_reconnect" in sig.parameters, "connect() must accept is_reconnect kwarg"
+    param = sig.parameters["is_reconnect"]
+    assert param.kind == inspect.Parameter.KEYWORD_ONLY
+    assert param.default is False
+
+
+def test_send_signature_matches_gateway_contract():
+    base_sig = inspect.signature(BasePlatformAdapter.send)
+    cls = plugin_adapter.get_adapter_class()
+    sig = inspect.signature(cls.send)
+    base_params = [p for p in base_sig.parameters if p != "self"]
+    own_params = [p for p in sig.parameters if p != "self"]
+    assert own_params == base_params, (
+        f"send() params {own_params} drift from gateway contract {base_params}"
+    )
+    assert inspect.iscoroutinefunction(cls.send)
+
+
+# --- L3: inbound event shape against real types -----------------------------
+
+
+def test_inbound_event_construction_is_valid(registered_entry):
+    """The exact source/event the adapter builds on receive must type-check."""
+    adapter = _make_adapter(registered_entry)
+    source = adapter.build_source(
+        chat_id="!abcd1234",
+        chat_name="Meshtastic !abcd1234",
+        chat_type="dm",
+        user_id="!abcd1234",
+        user_name="!abcd1234",
+        message_id="42",
+    )
+    event = MessageEvent(
+        text="hello over LoRa",
+        message_type=MessageType.TEXT,
+        user_id="!abcd1234",
+        user_name="!abcd1234",
+        source=source,
+        raw_message={"id": 42},
+        message_id="42",
+    )
+    assert event.source.platform.value == "meshtastic"
+    assert event.text == "hello over LoRa"

plugin/tests/e2e/test_meshtasticator_e2e.py

@@ -0,0 +1,266 @@
+"""End-to-end tests: hermes-meshtastic adapter against live Meshtasticator nodes.
+
+What these tests validate (the "radio-level" layer no contract test can):
+  * the adapter's real TCP/protobuf handshake against MeshtasticD (the actual
+    device software, not a mock) as run by the Meshtasticator interactive
+    simulator (https://meshtastic.org/docs/software/meshtasticator/),
+  * inbound text reception across the simulated LoRa mesh,
+  * outbound chunked sends arriving at another simulated node,
+  * the whole path: pubsub callback -> MessageEvent construction (real gateway
+    types) -> handle_message dispatch.
+
+What they do NOT validate (out of scope even for Meshtasticator): true RF
+(silicon-level LoRa, real propagation/interference, regulatory duty cycle).
+
+Running
+-------
+  1. On a Linux host (or a machine with Docker): clone meshtastic/meshtasticator
+     and boot the interactive simulator (see scripts/meshtasticator_e2e/).
+  2. Point the plugin's own contract venv at the booted nodes and run:
+       cd plugin && MESHTASTICATOR_E2E=1 E2E_PORTS=4404,4405 \
+         E2E_CHANNEL_INDEX=0 .venv-contract/bin/python -m pytest tests/e2e -q
+
+Environment
+-----------
+  MESHTASTICATOR_E2E  = "1" enables these tests (required; without it the
+                        module skips so ordinary unit/CI runs stay fast).
+  E2E_PORTS           = comma-separated TCP API ports of the simulated nodes
+                        (first = the node the adapter connects to, default 4404).
+  E2E_HOST            = host where the simulator exposes the ports (default 127.0.0.1).
+  E2E_CHANNEL_INDEX   = Meshtastic channel index the adapter listens/sends on
+                        (default 0 = primary channel, matching a stock sim node).
+
+NOTE: this module is an initial scaffold. Structural validation (imports,
+fixtures, flow) passes, but a full green run requires a Linux/Docker host with
+a MeshtasticD build — tracked as the remaining verification step.
+"""
+
+import asyncio
+import os
+
+import pytest
+
+if os.environ.get("MESHTASTICATOR_E2E") != "1":
+    pytest.skip(
+        "set MESHTASTICATOR_E2E=1 to run against Meshtasticator nodes "
+        "(see scripts/meshtasticator_e2e/README.md)",
+        allow_module_level=True,
+    )
+
+# Real Hermes gateway + real Meshtastic client library are required.
+pytest.importorskip("gateway.platforms.base")
+pytest.importorskip("meshtastic.tcp_interface")
+
+from gateway.config import PlatformConfig  # noqa: E402
+from gateway.platform_registry import (  # noqa: E402
+    PlatformEntry,
+    platform_registry,
+)
+
+import hermes_meshtastic  # noqa: E402
+from hermes_meshtastic import adapter as plugin_adapter  # noqa: E402
+
+E2E_HOST = os.environ.get("E2E_HOST", "127.0.0.1")
+E2E_PORTS = [int(p) for p in os.environ.get("E2E_PORTS", "4404,4405").split(",") if p]
+E2E_CHANNEL_INDEX = int(os.environ.get("E2E_CHANNEL_INDEX", "0"))
+assert len(E2E_PORTS) >= 2, "E2E_PORTS needs at least two node ports (adapter + peer)"
+
+NODE0_PORT = E2E_PORTS[0]  # the node our adapter connects to
+NODE1_PORT = E2E_PORTS[1]  # the peer node that sends / observes
+CONNECT_TIMEOUT_S = 30
+RADIO_SETTLE_S = 2
+
+
+def _register_meshtastic_platform():
+    """Register the plugin (mirrors discovery) so Platform('meshtastic') resolves."""
+    if platform_registry.is_registered("meshtastic"):
+        return
+    kwargs = {}
+
+    class _Ctx:
+        def register_platform(self, **kw):
+            kwargs.update(kw)
+
+    hermes_meshtastic.register(_Ctx())
+    platform_registry.register(
+        PlatformEntry(
+            name=kwargs["name"],
+            label=kwargs["label"],
+            adapter_factory=kwargs["adapter_factory"],
+            check_fn=kwargs["check_fn"],
+            required_env=[],
+            max_message_length=kwargs.get("max_message_length", 0),
+            emoji=kwargs.get("emoji", "🔌"),
+            platform_hint=kwargs.get("platform_hint", ""),
+        )
+    )
+
+
+class _ProbeAdapter(plugin_adapter.get_adapter_class()):
+    """MeshtasticAdapter that records inbound MessageEvents instead of
+    dispatching to a Hermes agent handler (no agent runs in these tests)."""
+
+    def __init__(self, config):
+        super().__init__(config)
+        self.received_events = []
+
+    async def handle_message(self, event):
+        self.received_events.append(event)
+        # Intentionally do not call super(): the real base would forward to a
+        # message handler that only exists inside a running Hermes gateway.
+        return None
+
+
+@pytest.fixture(scope="module")
+def adapter():
+    _register_meshtastic_platform()
+    cfg = PlatformConfig(
+        enabled=True,
+        extra={
+            "host": E2E_HOST,
+            "port": NODE0_PORT,
+            "channel_index": E2E_CHANNEL_INDEX,
+        },
+    )
+    return _ProbeAdapter(cfg)
+
+
+@pytest.fixture(scope="module")
+def loop():
+    """Long-lived event loop in a background thread.
+
+    The adapter mirrors the Hermes gateway contract: connect()/watchdog/
+    handle_message() assume a *running* loop (the gateway's). A per-call
+    asyncio.run() would close the loop right after connect() and strand both
+    the watchdog and inbound dispatch, so the e2e module keeps one loop alive
+    and drives coroutines into it with run_coroutine_threadsafe.
+    """
+    import threading
+
+    _loop = asyncio.new_event_loop()
+    thread = threading.Thread(target=_loop.run_forever, daemon=True)
+    thread.start()
+    try:
+        yield _loop
+    finally:
+        _loop.call_soon_threadsafe(_loop.stop)
+        thread.join(timeout=5)
+
+
+def _run_in_loop(loop, coro, timeout_s=CONNECT_TIMEOUT_S):
+    import concurrent.futures
+
+    try:
+        return asyncio.run_coroutine_threadsafe(coro, loop).result(timeout=timeout_s)
+    except concurrent.futures.TimeoutError:
+        pytest.fail(f"coroutine did not finish within {timeout_s}s")
+
+
+@pytest.fixture(scope="module")
+def node0(adapter, loop):
+    """Connect the adapter to simulated node 0 (retrying while the node's TCP
+    API comes up); disconnect afterwards."""
+    import time
+
+    deadline = time.monotonic() + CONNECT_TIMEOUT_S
+    last_ok = False
+    while time.monotonic() < deadline:
+        last_ok = bool(_run_in_loop(loop, adapter.connect()))
+        if last_ok:
+            break
+        time.sleep(3)
+    assert last_ok, (
+        f"adapter failed to connect to simulated node at {E2E_HOST}:{NODE0_PORT} "
+        "(is the simulator running and the node reachable?)"
+    )
+    try:
+        yield adapter
+    finally:
+        try:
+            _run_in_loop(loop, adapter.disconnect(), timeout_s=10)
+        except Exception:  # loop teardown must not mask test failures
+            pass
+
+
+def _wait_for(predicate, timeout_s=15.0, interval_s=0.5, what="condition"):
+    """Poll until *predicate* (callable -> bool) is true, else pytest.fail."""
+    import time
+
+    deadline = time.monotonic() + timeout_s
+    while time.monotonic() < deadline:
+        if predicate():
+            return
+        time.sleep(interval_s)
+    pytest.fail(f"timed out waiting for {what}")
+
+
+# ---------------------------------------------------------------------------
+# Radio-level flows
+# ---------------------------------------------------------------------------
+
+
+def test_adapter_receives_broadcast_from_peer(node0):
+    """Peer node 1 broadcasts text over the simulated mesh; the adapter must
+    receive it as a real gateway MessageEvent."""
+    import time
+
+    from meshtastic import tcp_interface
+
+    sent = f"e2e-inbound-{int(time.time())}"
+
+    peer = tcp_interface.TCPInterface(hostname=E2E_HOST, portNumber=NODE1_PORT)
+    try:
+        time.sleep(RADIO_SETTLE_S)
+
+        def _got():
+            return any(getattr(e, "text", None) == sent for e in node0.received_events)
+
+        # Radio delivery is not guaranteed per attempt; retry a few times.
+        for _ in range(3):
+            if _got():
+                break
+            peer.localNode.sendText(
+                text=sent, destinationId="^all", channelIndex=E2E_CHANNEL_INDEX
+            )
+            time.sleep(1)
+        _wait_for(_got, timeout_s=25, what="inbound broadcast from peer node")
+    finally:
+        peer.close()
+
+
+def test_adapter_chunked_send_reaches_peer(node0, loop):
+    """A long outbound message must be chunked and arrive at the peer node."""
+    import time
+
+    from meshtastic import tcp_interface
+    from pubsub import pub
+
+    content = " ".join(f"word{n}" for n in range(60))  # well over 180 chars -> chunks
+    received = []
+
+    def _on_receive(packet, interface):
+        decoded = packet.get("decoded", {}) or {}
+        text = decoded.get("text", "")
+        if text:
+            received.append(text)
+
+    peer = tcp_interface.TCPInterface(hostname=E2E_HOST, portNumber=NODE1_PORT)
+    try:
+        time.sleep(RADIO_SETTLE_S)
+        pub.subscribe(_on_receive, "meshtastic.receive.text")
+
+        async def _send():
+            result = await node0.send("channel_e2e", content)
+            assert result.success, f"adapter.send failed: {result.error}"
+
+        _run_in_loop(loop, _send(), timeout_s=60)
+
+        def _all_chunks_arrived():
+            # Adapter frames multi-part sends as "(i/n) <chunk>"; the final
+            # chunk carries the message tail.
+            return any(str(r).rstrip().endswith("word59") for r in received)
+
+        _wait_for(_all_chunks_arrived, timeout_s=45, what="chunked outbound send at peer")
+    finally:
+        pub.unsubscribe(_on_receive, "meshtastic.receive.text")
+        peer.close()

scripts/contract-test.sh

@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+# Gateway contract tests for hermes-meshtastic.
+#
+# Runs the plugin's contract suite (tests/contract/) against the REAL Hermes
+# gateway API, pinned to the same upstream commit the deployed container image
+# is built from (NousResearch/hermes-agent @ v2026.8.31 — the image tag used in
+# k8s-flux apps/base/coder/hermes-agent).
+#
+# Why: the adapter used to be developed by guessing the gateway API and only
+# failing at gateway startup after a release (AttributeError: ... no attribute
+# 'set_message_handler'). These tests replay the gateway's adapter bootstrap
+# (registry registration -> factory -> set_* wiring) so API drift fails here,
+# in seconds, without deploying.
+#
+# Requires: uv (https://docs.astral.sh/uv/) and network on first run.
+#
+# Usage:  scripts/contract-test.sh
+set -euo pipefail
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+PLUGIN_DIR="$REPO_ROOT/plugin"
+VENV="$PLUGIN_DIR/.venv-contract"
+HERMES_SRC="$PLUGIN_DIR/.hermes-src"
+HERMES_TAG="${HERMES_TAG:-v2026.8.31}"   # must match k8s-flux hermes image tag
+HERMES_COMMIT="${HERMES_COMMIT:-29112bef099274229cadff79cdff7bf7b99c4b77}"
+
+if ! command -v uv >/dev/null 2>&1; then
+  echo "error: uv is required (https://docs.astral.sh/uv/)" >&2
+  exit 1
+fi
+
+if [ ! -d "$HERMES_SRC/.git" ]; then
+  echo "== cloning hermes-agent @ $HERMES_TAG (deployed image source) =="
+  git clone --depth 1 --branch "$HERMES_TAG" \
+    https://github.com/NousResearch/hermes-agent.git "$HERMES_SRC"
+fi
+actual="$(git -C "$HERMES_SRC" rev-parse HEAD)"
+if [ "$actual" != "$HERMES_COMMIT" ]; then
+  echo "error: $HERMES_SRC is at $actual, expected $HERMES_COMMIT ($HERMES_TAG)." >&2
+  echo "       Refresh deliberately: HERMES_TAG=<new tag> scripts/contract-test.sh" >&2
+  exit 1
+fi
+
+if [ ! -x "$VENV/bin/python" ]; then
+  echo "== creating contract venv (python 3.13, matching the Hermes image) =="
+  uv venv --python 3.13 "$VENV"
+fi
+uv pip install --python "$VENV/bin/python" -q -e "$HERMES_SRC"
+uv pip install --python "$VENV/bin/python" -q pytest
+
+echo "== running contract tests (HERMES_CONTRACT_REQUIRED=1) =="
+(
+  cd "$PLUGIN_DIR"
+  HERMES_CONTRACT_REQUIRED=1 "$VENV/bin/python" -m pytest tests/contract -q "$@"
+)
+
+echo "== running standalone unit tests (no gateway needed) =="
+(
+  cd "$PLUGIN_DIR"
+  "$VENV/bin/python" -m pytest tests/test_adapter.py -q
+)

scripts/meshtasticator_e2e/README.md

@@ -0,0 +1,133 @@
+# Meshtasticator e2e — radio-level validation for hermes-meshtastic
+
+This directory is the **radio-level counterpart** to the contract suite
+(`plugin/tests/contract`). The contract suite guarantees the adapter matches
+the Hermes gateway API; it cannot prove the adapter actually talks to a
+Meshtastic node. These e2e tests boot the
+[Meshtasticator](https://meshtastic.org/docs/software/meshtasticator/)
+interactive simulator — which runs the **real MeshtasticD device software**
+per node with a simulated LoRa PHY — and drive the adapter against a live
+simulated node.
+
+| Layer | Validated by |
+|---|---|
+| Adapter ↔ Hermes gateway API | `plugin/tests/contract` (no deployment) |
+| Adapter ↔ real MeshtasticD TCP/protobuf session | **this suite** |
+| Inbound text across the (simulated) LoRa mesh | **this suite** |
+| Outbound chunked sends arriving at another node | **this suite** |
+| True RF (silicon LoRa, real propagation, duty cycle) | physical radio only |
+
+## Layout
+
+```
+scripts/meshtasticator_e2e/
+  run_e2e.sh                  orchestrator: boots N nodes, runs pytest, tears down
+  tcp_forward.py              localhost->daemon TCP forward (remote docker daemons)
+  README.md                   this file
+plugin/tests/e2e/
+  test_meshtasticator_e2e.py  the e2e assertions (skip-gated)
+.gitea/workflows/
+  meshtasticator-e2e.yaml     on-demand CI job (docker-enabled runner)
+```
+
+## Prerequisites
+
+* **uv** + Python 3.13.
+* A contract venv with the real gateway **and** the plugin (with deps) installed:
+  ```bash
+  scripts/contract-test.sh                 # creates plugin/.venv-contract (hermes-agent pinned)
+  uv pip install --python plugin/.venv-contract/bin/python -e plugin
+  ```
+  (`-e plugin` installs the meshtastic client library the tests also use.)
+* A clone of Meshtasticator: `git clone https://github.com/meshtastic/meshtasticator`
+* Runtime for the simulator — pick one:
+  * **Docker** (simplest, works on macOS/Windows): the simulator pulls the
+    `meshtastic/meshtasticd` image on first run (slow once). Node TCP API ports
+    are published on your docker host's localhost, which is what the simulator
+    itself expects.
+  * **Linux native**: build MeshtasticD's `native` PlatformIO target and pass
+    the build dir with `--program`.
+* Optional but recommended: an isolated simulator venv, because Meshtasticator
+  pins old client libs (`meshtastic~=2.6.1` in its requirements.txt):
+  ```bash
+  uv venv --python 3.13 /tmp/sim-venv
+  uv pip install --python /tmp/sim-venv/bin/python \
+    -r <meshtasticator>/requirements.txt docker
+  export SIM_VENV=/tmp/sim-venv     # else the script uses plain `python3`
+  ```
+
+## Run
+
+```bash
+# Docker mode (default)
+MESHTASTICATOR_DIR=/path/to/meshtasticator \
+  scripts/meshtasticator_e2e/run_e2e.sh
+
+# Linux native mode
+MESHTASTICATOR_DIR=/path/to/meshtasticator \
+  scripts/meshtasticator_e2e/run_e2e.sh \
+    --mode native --program /path/to/firmware/.pio/build/native
+
+# Tune the scenario
+  ... --nodes 3 --channel-index 0 --host 127.0.0.1
+```
+
+The orchestrator boots the simulator, waits until `N` node TCP API ports accept
+connections (Meshtasticator assigns node *n* → port `4404 + n`), then runs
+`pytest plugin/tests/e2e` with `MESHTASTICATOR_E2E=1`. Simulator output lands in
+`meshtasticator-sim.log` at the repo root.
+
+**Remote docker daemon** (`DOCKER_HOST=tcp://host:2375`): the simulator's own
+control connections assume nodes listen on localhost, so the script starts a
+`tcp_forward.py` per node port (localhost → daemon host) before booting the
+simulator. Nothing is forwarded when the daemon is local (unset/unix
+`DOCKER_HOST`).
+
+Without `MESHTASTICATOR_E2E=1` the e2e module skips, so ordinary unit/CI runs
+stay fast and gateway-free.
+
+## CI
+
+`.gitea/workflows/meshtasticator-e2e.yaml` runs the whole flow on a
+docker-enabled runner (on demand via `workflow_dispatch` until validated, then
+flip on the commented `pull_request` trigger): it builds the Hermes contract
+env (pinned like `release.yaml`), installs the plugin with deps, clones
+Meshtasticator at a pinned commit, installs the simulator's own requirements in
+an isolated venv, and calls `run_e2e.sh --mode docker`. Simulator logs print on
+failure. Only one e2e run at a time (the sim hardcodes the `Meshtastic`
+container/volume name on the shared daemon).
+
+## What the tests do
+
+1. **Inbound**: peer node 1 broadcasts text; the adapter (connected to node 0)
+   must receive it as a real gateway `MessageEvent` (`host/port/channel_index`
+   come from `E2E_PORTS`/`E2E_HOST`/`E2E_CHANNEL_INDEX`).
+2. **Outbound**: the adapter sends a >180-char message; it must be chunked and
+   the final chunk (`(i/n) … word59`) must arrive at peer node 1.
+
+A `_ProbeAdapter` subclass records inbound events instead of dispatching to a
+Hermes agent handler (no agent runs in these tests; dispatch is already covered
+by the contract suite).
+
+## Status / open items (v1 scaffold)
+
+This is an initial, structurally validated scaffold. A full green run still
+needs to happen on a Linux/Docker host with a MeshtasticD build, and these
+items are expected to need iteration there:
+
+* **Node adjacency**: the simulator's default random placement may put nodes out
+  of range. For deterministic tests, seed close-together coordinates (the sim
+  supports `--from-file` with an `out/nodeConfig.yaml`).
+* **Channel semantics**: tests default to channel index 0 (stock primary
+  channel). Mirroring production (channel 1 / private channel) requires
+  configuring matching channels on the simulated nodes.
+* **Reconnect flow**: killing/restarting a single simulated node needs per-node
+  lifecycle control (native mode subprocesses; docker mode currently runs all
+  nodes in one container), so the adapter's watchdog reconnect is **not** yet
+  exercised here. See the watchdog unit coverage in the adapter for now.
+* **CI validation run**: `.gitea/workflows/meshtasticator-e2e.yaml` is drafted
+  and on-demand only; it still needs a first green run on the docker-enabled
+  runner (topology assumptions: docker reachable from the job, node ports on
+  the job's localhost via the automatic tcp forwards). Trigger it via
+  `workflow_dispatch`, watch the sim log step, and only then enable the
+  `pull_request` trigger.

scripts/meshtasticator_e2e/run_e2e.sh

@@ -0,0 +1,175 @@
+#!/usr/bin/env bash
+# Boot N Meshtasticator nodes and run the plugin's radio-level e2e tests
+# (plugin/tests/e2e) against node 0.
+#
+# Meshtasticator's interactive simulator (https://meshtastic.org/docs/software/meshtasticator/)
+# runs the real MeshtasticD device software per node and simulates the LoRa
+# PHY, so these tests exercise the adapter's real TCP/protobuf session,
+# inbound/outbound text flow across the simulated mesh, and chunked sends —
+# without physical radios. It cannot validate silicon-level RF.
+#
+# Prerequisites
+#   * Linux host, OR Docker (macOS/Windows): the simulator auto-falls back to
+#     Docker on non-Linux and can build MeshtasticD itself (`-d`).
+#   * Native Linux mode: build MeshtasticD's 'native' target (PlatformIO,
+#     select 'native') and pass --program <dir containing binary 'program'>.
+#   * uv + python 3.13 for the contract venv (plugin/.venv-contract).
+#
+# Docker on a remote daemon: if DOCKER_HOST points at a tcp:// daemon, the
+# simulator's own control connections assume nodes listen on localhost, so
+# this script starts lightweight localhost->daemon TCP forwarders for the node
+# ports first (scripts/meshtasticator_e2e/tcp_forward.py). Nothing is
+# forwarded when the daemon is local (unset/unix DOCKER_HOST).
+#
+# Usage
+#   MESHTASTICATOR_DIR=/path/to/meshtasticator scripts/meshtasticator_e2e/run_e2e.sh
+#   # docker mode (default):     ... run_e2e.sh --mode docker
+#   # native mode:               ... run_e2e.sh --mode native --program /path/to/firmware/.pio/build/native
+#   # node count / channel:      ... run_e2e.sh --nodes 3 --channel-index 0
+#
+# Environment
+#   MESHTASTICATOR_DIR  path to a clone of github.com/meshtastic/meshtasticator
+#   E2E_VENV            python env with hermes-agent + plugin deps installed
+#                       (default: plugin/.venv-contract — see scripts/contract-test.sh,
+#                       then `uv pip install --python plugin/.venv-contract/bin/python -e plugin`)
+#   SIM_VENV            optional python env holding the simulator's deps
+#                       (meshtasticator/requirements.txt + `pip install docker`);
+#                       defaults to plain `python3`.
+set -euo pipefail
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+SIM_DIR="${MESHTASTICATOR_DIR:?set MESHTASTICATOR_DIR to a meshtasticator clone}"
+NODES=3
+MODE=docker
+PROGRAM=""
+CHANNEL_INDEX=0
+HOST=127.0.0.1
+VENV="${E2E_VENV:-$REPO_ROOT/plugin/.venv-contract}"
+if [ -n "${SIM_VENV:-}" ]; then
+  SIM_PYTHON="$SIM_VENV/bin/python"
+else
+  SIM_PYTHON="${SIM_PYTHON:-python3}"
+fi
+
+while [ $# -gt 0 ]; do
+  case "$1" in
+    --nodes) NODES="$2"; shift 2 ;;
+    --mode) MODE="$2"; shift 2 ;;
+    --program) PROGRAM="$2"; shift 2 ;;
+    --channel-index) CHANNEL_INDEX="$2"; shift 2 ;;
+    --host) HOST="$2"; shift 2 ;;
+    *) echo "unknown arg: $1" >&2; exit 2 ;;
+  esac
+done
+
+command -v uv >/dev/null 2>&1 || { echo "uv required" >&2; exit 1; }
+[ -d "$SIM_DIR" ] || { echo "meshtasticator not found at $SIM_DIR" >&2; exit 1; }
+[ -x "$VENV/bin/python" ] || { echo "e2e venv missing at $VENV (see scripts/contract-test.sh)" >&2; exit 1; }
+
+export MPLBACKEND=Agg   # headless matplotlib (no display needed)
+
+# ---------------------------------------------------------------------------
+# Remote-daemon port forwarding: Meshtasticator talks to nodes on localhost.
+# ---------------------------------------------------------------------------
+DOCKER_TARGET=""
+if [[ "${DOCKER_HOST:-}" == tcp://* ]]; then
+  DOCKER_TARGET="${DOCKER_HOST#tcp://}"
+  DOCKER_TARGET="${DOCKER_TARGET%%:*}"
+  if [ "$DOCKER_TARGET" = "127.0.0.1" ] || [ "$DOCKER_TARGET" = "localhost" ]; then
+    DOCKER_TARGET=""
+  fi
+fi
+FWD_PIDS=()
+PORT_BASE=4404  # matches meshtasticator lib/interactive.py TCP_PORT_OFFSET
+
+start_forwards() {
+  for ((i = 0; i < NODES; i++)); do
+    port=$((PORT_BASE + i))
+    "$VENV/bin/python" "$REPO_ROOT/scripts/meshtasticator_e2e/tcp_forward.py" \
+      --listen "127.0.0.1:$port" --target "$DOCKER_TARGET:$port" >/dev/null 2>&1 &
+    FWD_PIDS+=("$!")
+  done
+  sleep 1
+}
+
+SIM_LOG="$REPO_ROOT/meshtasticator-sim.log"
+SIM_ARGS=("$NODES")
+if [ "$MODE" = native ]; then
+  [ -n "$PROGRAM" ] && [ -d "$PROGRAM" ] || { echo "--mode native needs --program <dir>" >&2; exit 2; }
+  SIM_ARGS+=(-p "$PROGRAM")
+else
+  SIM_ARGS+=(-d)
+fi
+
+if [ -n "$DOCKER_TARGET" ]; then
+  echo "== remote docker daemon detected ($DOCKER_TARGET): starting localhost port forwards =="
+  start_forwards
+fi
+
+echo "== booting Meshtasticator ($MODE) with $NODES node(s) from $SIM_DIR =="
+(
+  cd "$SIM_DIR"
+  "$SIM_PYTHON" interactiveSim.py "${SIM_ARGS[@]}" >"$SIM_LOG" 2>&1
+) &
+SIM_PID=$!
+
+cleanup() {
+  echo "== tearing down simulator (pid $SIM_PID) =="
+  kill "$SIM_PID" 2>/dev/null || true
+  wait "$SIM_PID" 2>/dev/null || true
+  for pid in "${FWD_PIDS[@]:-}"; do
+    kill "$pid" 2>/dev/null || true
+  done
+  if [ "$MODE" = docker ]; then
+    # The sim names its container/volume (auto_remove only fires on stop).
+    docker rm -f Meshtastic >/dev/null 2>&1 || true
+    docker volume rm Meshtasticator >/dev/null 2>&1 || true
+  fi
+}
+trap cleanup EXIT
+
+echo "== waiting for $NODES node TCP API port(s) to accept connections =="
+READY=""
+for _ in $(seq 1 180); do   # up to ~5 min: docker pulls meshtastic/meshtasticd on first run
+  READY="$("$VENV/bin/python" - "$HOST" "$NODES" <<'PY' || true
+import socket, sys
+host, count = sys.argv[1], int(sys.argv[2])
+found = []
+# Meshtasticator assigns node n -> TCP port starting at 4404 (see
+# lib/interactive.py TCP_PORT_OFFSET); probe a small window to be safe.
+for port in range(4403, 4403 + 16):
+    try:
+        with socket.create_connection((host, port), timeout=0.5):
+            found.append(port)
+    except OSError:
+        pass
+    if len(found) >= count:
+        break
+print(",".join(map(str, found[:count])))
+PY
+)"
+  if [ -n "$READY" ]; then
+    break
+  fi
+  sleep 2
+done
+if [ -z "$READY" ]; then
+  echo "error: no Meshtasticator node ports became reachable on $HOST (see $SIM_LOG)" >&2
+  exit 1
+fi
+echo "   node ports: $READY"
+
+# Ports accept before the node firmware finished booting inside the container;
+# give meshtasticd a moment to open its TCP API for real.
+echo "== settling (node firmware boot) =="
+sleep 15
+
+echo "== running radio-level e2e tests =="
+(
+  cd "$REPO_ROOT"
+  MESHTASTICATOR_E2E=1 \
+  E2E_HOST="$HOST" \
+  E2E_PORTS="$READY" \
+  E2E_CHANNEL_INDEX="$CHANNEL_INDEX" \
+    "$VENV/bin/python" -m pytest plugin/tests/e2e -q "$@"
+)

scripts/meshtasticator_e2e/tcp_forward.py

@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+"""Minimal TCP forwarder used by run_e2e.sh when the Docker daemon is remote.
+
+Meshtasticator's interactive simulator always connects to its simulated nodes
+at localhost:<port> (lib/interactive.py init_communication). On a Gitea
+Actions runner whose Docker daemon is remote (e.g. DOCKER_HOST=tcp://docker.local:2375)
+the container-published node ports live on the *daemon* host, not the job's
+localhost — so we bind 127.0.0.1:<port> in the job and forward to the daemon
+host, restoring the localhost view the simulator (and our e2e tests) expect.
+
+Usage: tcp_forward.py --listen 127.0.0.1:4404 --target docker.local:4404
+"""
+
+import argparse
+import socket
+import threading
+
+
+def _forward(src: socket.socket, dst: socket.socket):
+    try:
+        while True:
+            data = src.recv(65536)
+            if not data:
+                break
+            dst.sendall(data)
+    except OSError:
+        pass
+    finally:
+        try:
+            dst.shutdown(socket.SHUT_WR)
+        except OSError:
+            pass
+
+
+def _handle(client: socket.socket, target_addr):
+    try:
+        upstream = socket.create_connection(target_addr, timeout=10)
+    except OSError:
+        client.close()
+        return
+    t = threading.Thread(target=_forward, args=(client, upstream), daemon=True)
+    t.start()
+    _forward(upstream, client)
+    client.close()
+    upstream.close()
+
+
+def main():
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("--listen", required=True, help="local bind, e.g. 127.0.0.1:4404")
+    ap.add_argument("--target", required=True, help="forward target, e.g. docker.local:4404")
+    args = ap.parse_args()
+
+    lhost, lport = args.listen.rsplit(":", 1)
+    thost, tport = args.target.rsplit(":", 1)
+    target_addr = (thost, int(tport))
+
+    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+    listener.bind((lhost, int(lport)))
+    listener.listen(16)
+    print(f"forwarding {args.listen} -> {args.target}", flush=True)
+
+    while True:
+        client, _ = listener.accept()
+        threading.Thread(target=_handle, args=(client, target_addr), daemon=True).start()
+
+
+if __name__ == "__main__":
+    main()