feat(companion): echo injected TX to companion clients as raw RX (0x88)

Push locally-injected TX packets to connected companion frame server
clients as PUSH_CODE_LOG_RX_DATA (0x88) with snr=0/rssi=0, so apps that
decrypt locally from raw RX (e.g. RemoteTerm) see companion-originated
channel traffic. The originating companion is excluded so a node never
hears its own transmission, matching physical firmware behavior.
inject_packet now takes an origin_hash (threaded per-companion via the
packet_injector partial); _on_raw_rx_for_companions gains exclude_hash
to skip that companion's frame server. OTA RX is unaffected.

repeater/main.py

@@ -572,7 +572,12 @@ class RepeaterDaemon:
 
                 bridge = RepeaterCompanionBridge(
                     identity=identity,
-                    packet_injector=self.router.inject_packet,
+                    # Tag the injector with this companion's hash so inject_packet can
+                    # skip its own frame server when echoing TX as raw RX (a node never
+                    # hears its own transmission).
+                    packet_injector=functools.partial(
+                        self.router.inject_packet, origin_hash=companion_hash_str
+                    ),
                     node_name=node_name,
                     radio_config=radio_config,
                     sqlite_handler=sqlite_handler,
@@ -819,12 +824,21 @@ class RepeaterDaemon:
             f"port={tcp_port}, bind={bind_address}, client_idle_timeout_sec={client_idle_timeout_sec}"
         )
 
-    async def _on_raw_rx_for_companions(self, data: bytes, rssi: int, snr: float) -> None:
-        """Raw RX subscriber: push PUSH_CODE_LOG_RX_DATA (0x88) to connected companion clients."""
+    async def _on_raw_rx_for_companions(
+        self, data: bytes, rssi: int, snr: float, exclude_hash: str | None = None
+    ) -> None:
+        """Raw RX subscriber: push PUSH_CODE_LOG_RX_DATA (0x88) to connected companion clients.
+
+        ``exclude_hash`` skips the frame server for that companion hash; used when
+        echoing a companion's own injected TX so it never hears its own transmission.
+        OTA RX subscribers leave it unset, so received packets reach every companion.
+        """
         servers = getattr(self, "companion_frame_servers", [])
         if not servers:
             return
         for fs in servers:
+            if exclude_hash is not None and getattr(fs, "companion_hash", None) == exclude_hash:
+                continue
             try:
                 fs.push_rx_raw(snr, rssi, data)
             except Exception as e:

repeater/packet_router.py

@@ -160,7 +160,7 @@ class PacketRouter:
                 pass
         await self.queue.put(packet)
 
-    async def inject_packet(self, packet, wait_for_ack: bool = False):
+    async def inject_packet(self, packet, wait_for_ack: bool = False, origin_hash=None):
         try:
             metadata = {
                 "rssi": getattr(packet, "rssi", 0),
@@ -182,6 +182,32 @@ class PacketRouter:
             # Mark so when this packet is dequeued we don't pass to engine again (avoid double-send / double-count)
             packet._injected_for_tx = True
 
+            # Echo this local TX to companion frame server clients as raw RX
+            # (PUSH_CODE_LOG_RX_DATA 0x88, snr=0/rssi=0 = local origin) so apps that
+            # decrypt locally from raw RX (e.g. RemoteTerm) see companion-originated
+            # traffic, matching what other mesh nodes would hear off the air. The
+            # originating companion (origin_hash) is excluded so it never hears its own TX.
+            push_rx = getattr(self.daemon, "_on_raw_rx_for_companions", None)
+            if push_rx is not None:
+                try:
+                    raw = packet.write_to()
+                    await push_rx(raw, 0, 0.0, exclude_hash=origin_hash)
+                    servers = getattr(self.daemon, "companion_frame_servers", [])
+                    pushed = sum(
+                        1
+                        for fs in servers
+                        if getattr(fs, "companion_hash", None) != origin_hash
+                    )
+                    logger.debug(
+                        "Echoed injected TX as raw RX (0x88) to %d companion client(s) "
+                        "(%d bytes, origin=%s excluded)",
+                        pushed,
+                        len(raw),
+                        origin_hash,
+                    )
+                except Exception as e:
+                    logger.debug("Failed to echo injected TX to companions: %s", e)
+
             # Enqueue so router can deliver to companion(s): TXT_MSG -> dest bridge, ACK -> all bridges (sender sees ACK)
             await self.enqueue(packet)
 

tests/test_main_py_coverage.py

@@ -143,6 +143,15 @@ async def test_raw_rx_and_duplicate_logging_hooks():
     await daemon._on_raw_rx_for_companions(b"abc", rssi=-90, snr=2.0)
     fs_ok.push_rx_raw.assert_called_once()
 
+    # exclude_hash skips the matching companion's own frame server (no self-echo)
+    fs_self = SimpleNamespace(companion_hash="0x1a", push_rx_raw=MagicMock())
+    fs_other = SimpleNamespace(companion_hash="0x2b", push_rx_raw=MagicMock())
+    daemon.companion_frame_servers = [fs_self, fs_other]
+    await daemon._on_raw_rx_for_companions(b"xyz", rssi=0, snr=0.0, exclude_hash="0x1a")
+    fs_self.push_rx_raw.assert_not_called()
+    fs_other.push_rx_raw.assert_called_once()
+    daemon.companion_frame_servers = [fs_ok, fs_fail]
+
     engine = SimpleNamespace(
         is_duplicate=MagicMock(side_effect=[False, True]),
         record_duplicate=MagicMock(),

tests/test_packet_router.py

@@ -561,3 +561,70 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase):
         await router._route_packet(pkt)
         b1.process_received_packet.assert_awaited_once()
         daemon.repeater_handler.assert_awaited_once()
+
+
+class TestInjectedTxRawEcho(unittest.IsolatedAsyncioTestCase):
+    """inject_packet echoes local TX to companion clients as raw RX (0x88)."""
+
+    async def test_inject_packet_echoes_raw_tx_to_companions(self):
+        """Successful local TX is pushed via _on_raw_rx_for_companions with snr=0/rssi=0."""
+        daemon = _make_daemon()
+        daemon._on_raw_rx_for_companions = AsyncMock()
+        router = PacketRouter(daemon)
+        pkt = _make_packet()
+        pkt.write_to.return_value = b"\x10\x20\x30"
+
+        ok = await router.inject_packet(pkt)
+
+        self.assertTrue(ok)
+        daemon._on_raw_rx_for_companions.assert_awaited_once_with(
+            b"\x10\x20\x30", 0, 0.0, exclude_hash=None
+        )
+
+    async def test_inject_packet_excludes_originating_companion(self):
+        """A companion's own TX is echoed with its hash excluded (no self-echo)."""
+        daemon = _make_daemon()
+        daemon._on_raw_rx_for_companions = AsyncMock()
+        router = PacketRouter(daemon)
+        pkt = _make_packet()
+        pkt.write_to.return_value = b"\xaa\xbb"
+
+        ok = await router.inject_packet(pkt, origin_hash="0x1a")
+
+        self.assertTrue(ok)
+        daemon._on_raw_rx_for_companions.assert_awaited_once_with(
+            b"\xaa\xbb", 0, 0.0, exclude_hash="0x1a"
+        )
+
+    async def test_inject_packet_no_echo_when_tx_fails(self):
+        """A failed local transmission must not echo a raw RX frame."""
+        daemon = _make_daemon()
+        daemon.repeater_handler = AsyncMock(return_value=False)
+        daemon._on_raw_rx_for_companions = AsyncMock()
+        router = PacketRouter(daemon)
+
+        ok = await router.inject_packet(_make_packet())
+
+        self.assertFalse(ok)
+        daemon._on_raw_rx_for_companions.assert_not_awaited()
+
+    async def test_inject_packet_survives_echo_failure(self):
+        """An error while echoing must not fail the injection."""
+        daemon = _make_daemon()
+        daemon._on_raw_rx_for_companions = AsyncMock(side_effect=RuntimeError("boom"))
+        router = PacketRouter(daemon)
+
+        ok = await router.inject_packet(_make_packet())
+
+        self.assertTrue(ok)
+        daemon._on_raw_rx_for_companions.assert_awaited_once()
+
+    async def test_inject_packet_without_echo_hook(self):
+        """Injection succeeds even if the daemon has no raw-RX companion hook."""
+        daemon = _make_daemon()
+        daemon._on_raw_rx_for_companions = None
+        router = PacketRouter(daemon)
+
+        ok = await router.inject_packet(_make_packet())
+
+        self.assertTrue(ok)