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).

plugin/tests/e2e/test_meshtasticator_e2e.py

@@ -126,19 +126,60 @@ def adapter():
 
 
 @pytest.fixture(scope="module")
-def node0(adapter):
-    """Connect the adapter to simulated node 0; disconnect afterwards."""
-    async def _connect():
-        return await adapter.connect()
+def loop():
+    """Long-lived event loop in a background thread.
 
-    assert asyncio.run(_connect()) is True, (
+    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:
-        asyncio.run(adapter.disconnect())
+        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"):
@@ -187,7 +228,7 @@ def test_adapter_receives_broadcast_from_peer(node0):
         peer.close()
 
 
-def test_adapter_chunked_send_reaches_peer(node0):
+def test_adapter_chunked_send_reaches_peer(node0, loop):
     """A long outbound message must be chunked and arrive at the peer node."""
     import time
 
@@ -200,9 +241,7 @@ def test_adapter_chunked_send_reaches_peer(node0):
     def _on_receive(packet, interface):
         decoded = packet.get("decoded", {}) or {}
         text = decoded.get("text", "")
-        if text and text.startswith("(1/"):  # our chunked framing
-            received.append(text)
-        elif text:
+        if text:
             received.append(text)
 
     peer = tcp_interface.TCPInterface(hostname=E2E_HOST, portNumber=NODE1_PORT)
@@ -214,7 +253,7 @@ def test_adapter_chunked_send_reaches_peer(node0):
             result = await node0.send("channel_e2e", content)
             assert result.success, f"adapter.send failed: {result.error}"
 
-        asyncio.run(_send())
+        _run_in_loop(loop, _send(), timeout_s=60)
 
         def _all_chunks_arrived():
             # Adapter frames multi-part sends as "(i/n) <chunk>"; the final