Merge pull request #278 from agessaman/feat/pre-1160-compatibility-sendfix

Feat/pre 1160 compatibility sendfix

repeater/engine.py

@@ -162,7 +162,7 @@ class RepeaterHandler(BaseHandler):
 
     async def __call__(
         self, packet: Packet, metadata: Optional[dict] = None, local_transmission: bool = False
-    ) -> None:
+    ) -> bool:
 
         if metadata is None:
             metadata = {}
@@ -259,19 +259,23 @@ class RepeaterHandler(BaseHandler):
                         f"Duty-cycle limit: deferring local TX by {wait_time:.1f}s "
                         f"(airtime={airtime_ms:.1f}ms)"
                     )
-                    self.forwarded_count += 1
-                    transmitted = True
                     tx_task = await self.schedule_retransmit(
                         fwd_pkt, deferred_delay, airtime_ms, local_transmission=True
                     )
                     try:
-                        await tx_task
+                        tx_success = await tx_task
                     except Exception as e:
-                        self.forwarded_count -= 1
                         transmitted = False
                         drop_reason = "TX failed (deferred)"
                         logger.warning(f"Deferred local TX failed: {e}")
                         raise
+                    if not tx_success:
+                        transmitted = False
+                        drop_reason = "TX failed (deferred)"
+                        self.dropped_count += 1
+                    else:
+                        self.forwarded_count += 1
+                        transmitted = True
                     tx_metadata = getattr(fwd_pkt, "_tx_metadata", None)
                     if tx_metadata:
                         lbt_attempts = tx_metadata.get("lbt_attempts", 0)
@@ -292,19 +296,23 @@ class RepeaterHandler(BaseHandler):
                     self.dropped_count += 1
                     drop_reason = "Duty cycle limit"
             else:
-                self.forwarded_count += 1
-                transmitted = True
                 tx_task = await self.schedule_retransmit(
                     fwd_pkt, delay, airtime_ms, local_transmission=local_transmission
                 )
                 try:
-                    await tx_task
+                    tx_success = await tx_task
                 except Exception as e:
-                    self.forwarded_count -= 1
                     transmitted = False
                     drop_reason = "TX failed"
                     logger.warning(f"Local TX failed: {e}")
                     raise
+                if not tx_success:
+                    transmitted = False
+                    drop_reason = "TX failed"
+                    self.dropped_count += 1
+                else:
+                    self.forwarded_count += 1
+                    transmitted = True
                 tx_metadata = getattr(fwd_pkt, "_tx_metadata", None)
                 if tx_metadata:
                     lbt_attempts = tx_metadata.get("lbt_attempts", 0)
@@ -415,6 +423,8 @@ class RepeaterHandler(BaseHandler):
             # Not a duplicate or first occurrence
             self._append_recent_packet(packet_record)
 
+        return transmitted
+
     def log_trace_record(self, packet_record: dict) -> None:
         """Manually log a packet trace record (used by external callers)"""
         self._append_recent_packet(packet_record)
@@ -1128,10 +1138,20 @@ class RepeaterHandler(BaseHandler):
                                 "Packet dropped at TX time: duty-cycle exceeded (airtime=%.1fms)",
                                 airtime_ms,
                             )
-                            return
+                            return False
 
                     try:
-                        await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False)
+                        sent = await self.dispatcher.send_packet(
+                            fwd_pkt, wait_for_ack=False
+                        )
+                        if not sent:
+                            logger.warning(
+                                "Retransmit failed (attempt %d): dispatcher returned false",
+                                attempt + 1,
+                            )
+                            if local_transmission and attempt == 0:
+                                continue
+                            return False
                         self._record_packet_sent(fwd_pkt)
                         if airtime_ms > 0:
                             self.airtime_mgr.record_tx(airtime_ms)
@@ -1140,13 +1160,14 @@ class RepeaterHandler(BaseHandler):
                             f"Retransmitted packet ({packet_size} bytes, "
                             f"{airtime_ms:.1f}ms airtime)"
                         )
-                        return
+                        return True
                     except Exception as e:
                         logger.error(f"Retransmit failed (attempt {attempt + 1}): {e}")
                         if local_transmission and attempt == 0:
                             pass  # release lock, outer loop sleeps, then retries
                         else:
                             raise
+            return False
 
         return asyncio.create_task(delayed_send())
 

repeater/handler_helpers/discovery.py

@@ -7,11 +7,21 @@ allowing other nodes to discover repeaters on the mesh network.
 
 import asyncio
 import logging
+import secrets
 
 from pymc_core.node.handlers.control import ControlHandler
 
 logger = logging.getLogger("DiscoveryHelper")
 
+# Default upper bound (ms) for the randomized pre-send jitter applied to node
+# discovery responses. A node-discover request is a broadcast that every
+# in-range repeater answers at once, so without jitter they all transmit at the
+# same engine-scheduled instant and collide. Mirrors the firmware, which spreads
+# these replies deliberately (MyMesh.cpp:797, sendZeroHop with
+# getRetransmitDelay*4). Safe to be generous: the requester's discovery window is
+# 60s (firmware pending_discover_until = futureMillis(60000)).
+DEFAULT_DISCOVERY_RESPONSE_JITTER_MS = 2000
+
 
 class DiscoveryHelper:
     """Helper class for processing discovery requests in the repeater."""
@@ -23,6 +33,7 @@ class DiscoveryHelper:
         node_type: int = 2,
         log_fn=None,
         debug_log_fn=None,
+        response_jitter_ms: int = DEFAULT_DISCOVERY_RESPONSE_JITTER_MS,
     ):
         """
         Initialize the discovery helper.
@@ -34,10 +45,14 @@ class DiscoveryHelper:
             log_fn: Optional logging function for ControlHandler
             debug_log_fn: Optional logging for verbose ControlHandler messages (e.g. callback
                 presence). Pass logger.debug to avoid INFO noise when forwarding to companions.
+            response_jitter_ms: Upper bound (ms) for the randomized delay added before
+                transmitting a discovery response, to avoid multiple repeaters colliding
+                when answering the same broadcast. Set to 0 to disable (e.g. in tests).
         """
         self.local_identity = local_identity
         self.packet_injector = packet_injector  # Function to inject packets into router
         self.node_type = node_type
+        self.response_jitter_ms = max(0, int(response_jitter_ms))
 
         # Create ControlHandler internally as a parsing utility
         self.control_handler = ControlHandler(
@@ -147,6 +162,18 @@ class DiscoveryHelper:
             tag: The tag for logging purposes
         """
         try:
+            # Randomized pre-send jitter so multiple repeaters answering the same
+            # zero-hop discovery broadcast don't transmit at the same engine-scheduled
+            # instant and collide (the engine's DIRECT delay is fixed, not random).
+            # Mirrors firmware MyMesh.cpp:797. Uses secrets like the engine's TX jitter.
+            if self.response_jitter_ms > 0:
+                jitter_s = secrets.randbelow(self.response_jitter_ms + 1) / 1000.0
+                if jitter_s > 0:
+                    logger.debug(
+                        f"Discovery response jitter {jitter_s * 1000:.0f}ms for tag 0x{tag:08X}"
+                    )
+                    await asyncio.sleep(jitter_s)
+
             success = await self.packet_injector(packet, wait_for_ack=False)
             if success:
                 logger.info(f"Response sent for tag 0x{tag:08X}")

repeater/handler_helpers/login.py

@@ -6,7 +6,9 @@ This module processes login requests and manages authentication for all identiti
 
 import asyncio
 import logging
+import time
 
+from pymc_core.node.handlers.anon_request import AnonRateLimiter, AnonRequestHandler
 from pymc_core.node.handlers.login_server import LoginServerHandler
 from pymc_core.protocol.constants import PAYLOAD_TYPE_ANON_REQ
 
@@ -14,15 +16,27 @@ logger = logging.getLogger("LoginHelper")
 
 
 class LoginHelper:
-    def __init__(self, identity_manager, packet_injector=None, log_fn=None):
+    def __init__(
+        self,
+        identity_manager,
+        packet_injector=None,
+        log_fn=None,
+        sqlite_handler=None,
+        config=None,
+    ):
 
         self.identity_manager = identity_manager
         self.packet_injector = packet_injector
         self.log_fn = log_fn or logger.info
+        self.sqlite_handler = sqlite_handler
+        self.config = config or {}
 
         self.handlers = {}
         self.acls = {}  # Per-identity ACLs keyed by hash_byte
         self._pending_tasks = set()
+        # Shared across all identities so the node's total anon-reply rate is
+        # bounded (mirrors firmware anon_limiter: ~4 requests / 2 min).
+        self.anon_limiter = AnonRateLimiter()
 
     def _track_task(self, task: asyncio.Task) -> None:
         self._pending_tasks.add(task)
@@ -126,12 +140,88 @@ class LoginHelper:
             is_room_server=(identity_type == "room_server"),
         )
 
-        handler.set_send_packet_callback(self._send_packet_with_delay)
+        # Wrap the login handler in an anon-request dispatcher so anonymous
+        # regions/owner/basic discovery queries are answered instead of being
+        # mis-parsed as failed logins (MeshCore 1.16.0 discovery feature).
+        anon_handler = AnonRequestHandler(
+            local_identity=identity,
+            log_fn=self.log_fn,
+            login_handler=handler,
+            anon_limiter=self.anon_limiter,
+            region_names_fn=self._format_region_names,
+            owner_info_fn=self._make_owner_info_fn(name, config),
+            features_fn=self._make_features_fn(config),
+            clock_fn=lambda: int(time.time()),
+        )
+        # Wires the send callback through to both the wrapper and login handler.
+        anon_handler.set_send_packet_callback(self._send_packet_with_delay)
 
-        self.handlers[hash_byte] = handler
+        self.handlers[hash_byte] = anon_handler
 
         logger.info(f"Registered {identity_type} '{name}' login handler: hash=0x{hash_byte:02X}")
 
+    def _format_region_names(self) -> str:
+        """Build the comma-separated region-names string for an anon regions reply.
+
+        Mirrors firmware ``RegionMap::exportNamesTo`` with ``REGION_DENY_FLOOD``:
+        emit the ``*`` wildcard region first (unless unscoped flood is denied),
+        then each allow-flood named region with a leading ``#`` stripped, with no
+        trailing comma. The firmware wildcard is the always-present default flood
+        scope; pyMC_repeater models that via ``mesh.unscoped_flood_allow``
+        (falling back to ``mesh.global_flood_allow``, default allow).
+        """
+        parts = []
+
+        mesh_cfg = self.config.get("mesh", {}) if isinstance(self.config, dict) else {}
+        unscoped_allow = mesh_cfg.get(
+            "unscoped_flood_allow", mesh_cfg.get("global_flood_allow", True)
+        )
+        if unscoped_allow:
+            parts.append("*")
+
+        if self.sqlite_handler:
+            try:
+                keys = self.sqlite_handler.get_transport_keys()
+            except Exception as e:
+                logger.warning(f"Failed to read transport keys for regions reply: {e}")
+                keys = []
+            for rec in keys or []:
+                if rec.get("flood_policy", "deny") != "allow":
+                    continue
+                name = (rec.get("name") or "").strip()
+                if not name or name == "*":
+                    continue  # wildcard handled above
+                parts.append(name[1:] if name.startswith("#") else name)
+
+        return ",".join(parts)
+
+    @staticmethod
+    def _make_owner_info_fn(name: str, config: dict):
+        """Build an owner-info callback returning ``(node_name, owner_info)``."""
+
+        def owner_info_fn():
+            cfg = config or {}
+            repeater_cfg = cfg.get("repeater", {})
+            node_name = repeater_cfg.get("node_name") or name or "pyMC"
+            owner = repeater_cfg.get("owner_info", "") or ""
+            return (node_name, owner)
+
+        return owner_info_fn
+
+    @staticmethod
+    def _make_features_fn(config: dict):
+        """Build a feature-flags callback (bit0 = bridge, bit7 = forwarding disabled)."""
+
+        def features_fn():
+            cfg = config or {}
+            mode = cfg.get("repeater", {}).get("mode", "forward")
+            features = 0
+            if mode != "forward":  # monitor / no_tx => not forwarding
+                features |= 0x80
+            return features
+
+        return features_fn
+
     async def process_login_packet(self, packet):
 
         try:
@@ -147,9 +237,16 @@ class LoginHelper:
                 packet.mark_do_not_retransmit()
                 return True
             else:
-                # ANON_REQ to other nodes (e.g. owner-info to firmware) is normal; skip log to avoid spam
+                # ANON_REQ to other nodes (e.g. another repeater's regions/owner
+                # query overheard on-air) is normal; log at DEBUG so the dest is
+                # visible when diagnosing "why didn't my repeater answer".
                 ptype = getattr(packet, "get_payload_type", lambda: None)()
-                if ptype != PAYLOAD_TYPE_ANON_REQ:
+                if ptype == PAYLOAD_TYPE_ANON_REQ:
+                    logger.debug(
+                        f"ANON_REQ for hash 0x{dest_hash:02X} not addressed to a local "
+                        f"identity ({sorted(f'0x{h:02X}' for h in self.handlers)}); ignoring"
+                    )
+                else:
                     logger.debug(
                         f"No login handler registered for hash 0x{dest_hash:02X}, allowing forward"
                     )

repeater/main.py

@@ -268,6 +268,12 @@ class RepeaterDaemon:
                 identity_manager=self.identity_manager,
                 packet_injector=self.router.inject_packet,
                 log_fn=logger.info,
+                sqlite_handler=(
+                    self.repeater_handler.storage.sqlite_handler
+                    if self.repeater_handler and self.repeater_handler.storage
+                    else None
+                ),  # For anon regions-discovery replies
+                config=self.config,  # For owner-info / feature-flags replies
             )
 
             # Register default repeater identity
@@ -566,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,
@@ -813,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:
@@ -1093,7 +1113,7 @@ class RepeaterDaemon:
                 logger.debug("Marked own advert as seen in duplicate cache")
 
             logger.info(
-                "Sent flood advert '%s' at (% .6f, % .6f) source=%s",
+                "Sent flood advert '%s' at (%.6f, %.6f) source=%s",
                 node_name,
                 latitude,
                 longitude,

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),
@@ -172,14 +172,67 @@ class PacketRouter:
             # (avoids duty-cycle or dispatcher races where a later packet goes out first)
             async with self._inject_lock:
                 # Use local_transmission=True to bypass forwarding logic
-                await self.daemon.repeater_handler(packet, metadata, local_transmission=True)
+                sent = await self.daemon.repeater_handler(
+                    packet, metadata, local_transmission=True
+                )
+            if not sent:
+                logger.warning("Injected packet failed local transmission")
+                return False
 
             # 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)
 
+            if wait_for_ack:
+                ptype = getattr(packet, "get_payload_type", lambda: None)()
+                if ptype not in {
+                    AckHandler.payload_type(),
+                    AdvertHandler.payload_type(),
+                }:
+                    dispatcher = getattr(self.daemon, "dispatcher", None)
+                    if dispatcher and hasattr(dispatcher, "wait_for_ack"):
+                        try:
+                            expected_crc = packet.get_crc()
+                            ack_ok = await dispatcher.wait_for_ack(
+                                expected_crc, timeout=5.0
+                            )
+                            if not ack_ok:
+                                logger.warning(
+                                    "Injected packet ACK timeout (crc=%08X)", expected_crc
+                                )
+                                return False
+                        except Exception as e:
+                            logger.warning("Injected packet ACK wait failed: %s", e)
+                            return False
+
             packet_len = len(packet.payload) if packet.payload else 0
             logger.debug(
                 f"Injected packet processed by engine as local transmission ({packet_len} bytes)"
@@ -461,4 +514,11 @@ class PacketRouter:
                 "snr": getattr(packet, "snr", 0.0),
                 "timestamp": getattr(packet, "timestamp", 0),
             }
-            await self.daemon.repeater_handler(packet, metadata)
+            sent = await self.daemon.repeater_handler(packet, metadata)
+            if sent is False:
+                logger.warning(
+                    "Inbound packet not transmitted by repeater handler "
+                    "(type=%s, header=0x%02x)",
+                    payload_type,
+                    getattr(packet, "header", 0),
+                )

tests/test_engine.py

@@ -1072,6 +1072,46 @@ class TestTxMode:
             await asyncio.sleep(0)
         handler.dispatcher.send_packet.assert_called_once()
 
+    async def test_local_tx_exception_does_not_underflow_forwarded_count(self, handler):
+        """TX exceptions must not decrement forwarded_count before any increment happened."""
+        handler.config["repeater"]["mode"] = "forward"
+        pkt = _make_flood_packet()
+        handler.forwarded_count = 0
+
+        async def _boom():
+            raise RuntimeError("simulated tx failure")
+
+        with patch("repeater.engine.asyncio.sleep", new_callable=AsyncMock):
+            with patch.object(
+                handler, "schedule_retransmit", new=AsyncMock(return_value=asyncio.create_task(_boom()))
+            ):
+                with pytest.raises(RuntimeError, match="simulated tx failure"):
+                    await handler(pkt, {"snr": 0.0, "rssi": -80}, local_transmission=True)
+
+        assert handler.forwarded_count == 0
+
+    async def test_rx_tx_failure_increments_dropped_count_and_preserves_accounting(self, handler):
+        """A graceful TX failure should count as dropped and keep counters balanced."""
+        handler.config["repeater"]["mode"] = "forward"
+        pkt = _make_flood_packet()
+        handler.rx_count = 0
+        handler.forwarded_count = 0
+        handler.dropped_count = 0
+
+        async def _tx_false():
+            return False
+
+        with patch.object(
+            handler, "schedule_retransmit", new=AsyncMock(return_value=asyncio.create_task(_tx_false()))
+        ):
+            transmitted = await handler(pkt, {"snr": 0.0, "rssi": -80}, local_transmission=False)
+
+        assert transmitted is False
+        assert handler.rx_count == 1
+        assert handler.forwarded_count == 0
+        assert handler.dropped_count == 1
+        assert handler.rx_count == handler.forwarded_count + handler.dropped_count
+
 
 # ===================================================================
 # 16. Airtime calculation correctness
@@ -1793,7 +1833,7 @@ class TestEngineTransmissionAndBackgroundLifecycle:
         with patch.object(handler, "_calculate_tx_delay", return_value=0.5):
             loop = asyncio.get_running_loop()
             completed = loop.create_future()
-            completed.set_result(None)
+            completed.set_result(True)
 
             async def _fake_schedule(packet, delay, airtime_ms, local_transmission=False):
                 packet._tx_metadata = {

tests/test_handler_helpers_trace_discovery_login.py

@@ -207,7 +207,10 @@ def test_discovery_request_without_identity_does_not_send():
 @pytest.mark.asyncio
 async def test_discovery_send_packet_async_success_failure_and_exception():
     injector = AsyncMock(side_effect=[True, False, RuntimeError("send fail")])
-    helper = DiscoveryHelper(local_identity=FakeIdentity(0x42), packet_injector=injector)
+    # jitter disabled so the test doesn't sleep
+    helper = DiscoveryHelper(
+        local_identity=FakeIdentity(0x42), packet_injector=injector, response_jitter_ms=0
+    )
 
     await helper._send_packet_async(packet=object(), tag=0x11)
     await helper._send_packet_async(packet=object(), tag=0x12)
@@ -216,6 +219,41 @@ async def test_discovery_send_packet_async_success_failure_and_exception():
     assert injector.await_count == 3
 
 
+@pytest.mark.asyncio
+async def test_discovery_response_applies_bounded_jitter_before_send():
+    injector = AsyncMock(return_value=True)
+    helper = DiscoveryHelper(
+        local_identity=FakeIdentity(0x42), packet_injector=injector, response_jitter_ms=2000
+    )
+
+    slept = []
+
+    async def fake_sleep(secs):
+        slept.append(secs)
+
+    with patch("repeater.handler_helpers.discovery.asyncio.sleep", side_effect=fake_sleep):
+        await helper._send_packet_async(packet=object(), tag=0x55)
+
+    # Jitter applied exactly once, bounded to [0, 2.0]s, before the injection.
+    assert len(slept) == 1
+    assert 0.0 <= slept[0] <= 2.0
+    injector.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_discovery_response_jitter_disabled_does_not_sleep():
+    injector = AsyncMock(return_value=True)
+    helper = DiscoveryHelper(
+        local_identity=FakeIdentity(0x42), packet_injector=injector, response_jitter_ms=0
+    )
+
+    with patch("repeater.handler_helpers.discovery.asyncio.sleep") as sleep_mock:
+        await helper._send_packet_async(packet=object(), tag=0x56)
+
+    sleep_mock.assert_not_called()
+    injector.assert_awaited_once()
+
+
 def test_discovery_send_response_without_injector_is_safe():
     helper = DiscoveryHelper(local_identity=FakeIdentity(0x42), packet_injector=None)
 
@@ -251,12 +289,16 @@ def test_login_register_identity_repeater_creates_acl_and_handler():
     identity = FakeIdentity(0x52)
     acl_obj = MagicMock()
     handler_obj = MagicMock()
+    anon_obj = MagicMock()
 
     with (
         patch("repeater.handler_helpers.acl.ACL", return_value=acl_obj) as acl_cls,
         patch(
             "repeater.handler_helpers.login.LoginServerHandler", return_value=handler_obj
         ) as handler_cls,
+        patch(
+            "repeater.handler_helpers.login.AnonRequestHandler", return_value=anon_obj
+        ) as anon_cls,
     ):
         helper.register_identity(
             name="repeater-main",
@@ -271,11 +313,65 @@ def test_login_register_identity_repeater_creates_acl_and_handler():
 
     acl_cls.assert_called_once()
     handler_cls.assert_called_once()
-    handler_obj.set_send_packet_callback.assert_called_once()
-    assert helper.handlers[0x52] is handler_obj
+    # The login handler is wrapped in an AnonRequestHandler, and that wrapper is
+    # what gets stored + wired with the send callback.
+    anon_cls.assert_called_once()
+    assert anon_cls.call_args.kwargs["login_handler"] is handler_obj
+    anon_obj.set_send_packet_callback.assert_called_once()
+    assert helper.handlers[0x52] is anon_obj
     assert helper.acls[0x52] is acl_obj
 
 
+class _FakeSqlite:
+    def __init__(self, keys):
+        self._keys = keys
+
+    def get_transport_keys(self):
+        return self._keys
+
+
+def test_format_region_names_filters_and_strips():
+    keys = [
+        {"name": "#VHF", "flood_policy": "allow"},
+        {"name": "USA", "flood_policy": "allow"},
+        {"name": "secret", "flood_policy": "deny"},
+        {"name": "*", "flood_policy": "allow"},  # duplicate wildcard ignored
+        {"name": "", "flood_policy": "allow"},
+    ]
+    # Default config => unscoped flood allowed => wildcard '*' present.
+    helper = LoginHelper(identity_manager=MagicMock(), sqlite_handler=_FakeSqlite(keys))
+    # Wildcard first (from policy), '#' stripped, deny + empty + literal '*' excluded.
+    assert helper._format_region_names() == "*,VHF,USA"
+
+
+def test_format_region_names_wildcard_suppressed_when_unscoped_denied():
+    keys = [{"name": "USA", "flood_policy": "allow"}]
+    helper = LoginHelper(
+        identity_manager=MagicMock(),
+        sqlite_handler=_FakeSqlite(keys),
+        config={"mesh": {"unscoped_flood_allow": False}},
+    )
+    # No wildcard when unscoped flood is denied (firmware: wildcard deny-flood).
+    assert helper._format_region_names() == "USA"
+
+
+def test_format_region_names_without_storage_is_just_wildcard():
+    # No named regions, but unscoped flood allowed by default => bare wildcard.
+    helper = LoginHelper(identity_manager=MagicMock(), sqlite_handler=None)
+    assert helper._format_region_names() == "*"
+
+
+def test_owner_and_features_callbacks_from_config():
+    config = {"repeater": {"node_name": "node-x", "owner_info": "me", "mode": "monitor"}}
+    helper = LoginHelper(identity_manager=MagicMock(), config=config)
+
+    assert helper._make_owner_info_fn("fallback", config)() == ("node-x", "me")
+    # Non-forward mode sets the forwarding-disabled bit (0x80).
+    assert helper._make_features_fn(config)() == 0x80
+    # Forwarding mode clears it.
+    assert helper._make_features_fn({"repeater": {"mode": "forward"}})() == 0x00
+
+
 @pytest.mark.asyncio
 async def test_login_process_packet_routes_to_registered_handler_and_marks_no_retransmit():
     helper = LoginHelper(identity_manager=MagicMock(), packet_injector=AsyncMock())

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

@@ -46,7 +46,7 @@ from repeater.packet_router import (
 def _make_daemon():
     """Minimal daemon that satisfies PacketRouter without touching hardware."""
     daemon = MagicMock()
-    daemon.repeater_handler = AsyncMock(return_value=None)
+    daemon.repeater_handler = AsyncMock(return_value=True)
     daemon.trace_helper = None
     daemon.discovery_helper = None
     daemon.advert_helper = None
@@ -201,6 +201,19 @@ class TestInFlightCap(unittest.IsolatedAsyncioTestCase):
         finally:
             await router.stop()
 
+    async def test_non_injected_handler_false_is_logged(self):
+        """Inbound packets should log when repeater_handler reports TX failure."""
+        daemon = _make_daemon()
+        daemon.repeater_handler = AsyncMock(return_value=False)
+        router = PacketRouter(daemon)
+        pkt = _make_packet(payload_type=0xFF)
+
+        with patch("repeater.packet_router.logger.warning") as mock_warn:
+            await router._route_packet(pkt)
+
+        daemon.repeater_handler.assert_awaited_once()
+        mock_warn.assert_called()
+
     # ── 3. Shutdown: in-flight tasks drained ────────────────────────────────
 
     async def test_stop_waits_for_in_flight_tasks(self):
@@ -548,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)

tests/test_tx_lock.py

@@ -255,7 +255,8 @@ class TestTxLockSerialisation(unittest.IsolatedAsyncioTestCase):
         task = await h.schedule_retransmit(
             pkt, delay=0.0, airtime_ms=100.0, local_transmission=True
         )
-        await task  # should complete without error (gate returns silently)
+        result = await task
+        self.assertFalse(result, "Duty-cycle drop should report TX failure")
 
         self.assertEqual(
             send_calls[0], 1, "send_packet called on retry despite duty-cycle rejection"