fix(repeater): implement origin exclusion for injected packets

Enhance the PacketRouter to ensure that a companion does not receive its own transmitted packets. The `_companion_bridges_for_packet` method now tags packets with the originating companion's hash, preventing redelivery to the same bridge during fan-out. This change applies to various packet types, including advertisements and group messages. Additionally, tests have been added to verify that injected packets are correctly filtered, ensuring that the originating bridge does not process its own transmissions while still allowing other bridges to receive them.

repeater/packet_router.py

@@ -265,12 +265,30 @@ class PacketRouter:
         return False
 
     def _companion_bridges_for_packet(self, packet, metadata: dict) -> dict:
-        """Return companion bridges unless policy drop pre-check blocks delivery."""
+        """Return companion bridges unless policy drop or origin exclusion blocks delivery.
+
+        This is the single choke point every companion fan-out reads, so the
+        "a node never hears its own transmission" rule is applied here once and
+        covers ADVERT, ACK, PATH, RESPONSE, GRP_TXT, GRP_DATA and the dest-hash
+        candidate paths — the same rule ``inject_packet`` already applies to the
+        0x88 raw-RX echo.
+        """
         companion_bridges = getattr(self.daemon, "companion_bridges", {})
         if not companion_bridges:
             return {}
         if self._policy_blocks_companion(packet, metadata):
             return {}
+        # Both isinstance checks are load-bearing, not defensive noise: bridges are
+        # keyed by the int hash byte (main.py:880) while origin_hash is the "0xHH"
+        # text form the frame servers compare against (main.py:787, :1167), and a
+        # non-int key would raise inside the f-string. Do not simplify either away.
+        origin_hash = getattr(packet, "_injected_origin_hash", None)
+        if isinstance(origin_hash, str):
+            companion_bridges = {
+                key: bridge
+                for key, bridge in companion_bridges.items()
+                if not (isinstance(key, int) and f"0x{key:02x}" == origin_hash)
+            }
         return companion_bridges
 
     async def _fan_out_to_bridges(self, packet, bridges, *, context: str) -> tuple:
@@ -385,6 +403,14 @@ class PacketRouter:
                 "timestamp": getattr(packet, "timestamp", 0),
             }
 
+            # Tag the originating companion before anything reaches the air: enqueue()
+            # is a bare queue.put, so the packet itself is the only carrier that
+            # survives to _route_packet, where the fan-out withholds it from that
+            # bridge — the same "a node never hears its own transmission" rule the
+            # 0x88 echo below applies.  Set ahead of the TX so a core without the
+            # matching Packet slot fails before transmitting rather than after.
+            packet._injected_origin_hash = origin_hash
+
             # Serialize injects so one local TX completes before the next runs
             # (avoids duty-cycle or dispatcher races where a later packet goes out first)
             async with self._inject_lock:
@@ -421,7 +447,10 @@ class PacketRouter:
                 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)
+            # Enqueue so router can deliver to the *other* companion(s): TXT_MSG -> dest
+            # bridge, ACK/ADVERT/GRP_TXT/GRP_DATA -> every bridge except the origin.
+            # The originating companion never gets its own packet back: that exclusion
+            # is enforced once in _companion_bridges_for_packet via the tag set above.
             await self.enqueue(packet)
 
             if wait_for_ack:
@@ -572,8 +601,13 @@ class PacketRouter:
                 await deliver(snr, rssi, path_len, path_bytes, payload_bytes)
 
         elif payload_type == AdvertHandler.payload_type():
-            # Process advertisement packet for neighbor tracking
-            if self.daemon.advert_helper:
+            # Process advertisement packet for neighbor tracking. A locally injected
+            # advert is our own TX, never something heard off the air, so feeding it
+            # to the helper would record a phantom neighbour from the zeroed rssi/snr
+            # and burn a rate-limit token. AdvertHelper's own self-skip
+            # (handler_helpers/advert.py) only knows the repeater identity, not the
+            # co-hosted companions. Same _injected_for_tx idiom as the TRACE branches.
+            if self.daemon.advert_helper and not getattr(packet, "_injected_for_tx", False):
                 rssi = getattr(packet, "rssi", 0)
                 snr = getattr(packet, "snr", 0.0)
                 await self.daemon.advert_helper.process_advert_packet(packet, rssi, snr)

tests/test_companion_injector_wiring.py

@@ -0,0 +1,79 @@
+"""Every companion bridge must inject with its own hash as origin_hash.
+
+The origin exclusion in PacketRouter (the 0x88 raw-RX echo and the companion
+fan-out) is only as good as this binding: an injector built without
+``origin_hash``, or with the wrong companion's hash, silently sends a
+companion its own transmission back. Nothing else asserts the binding, so it
+is pinned here for both the config-load path and the hot-add path.
+"""
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from openhop_core import LocalIdentity
+from repeater.companion.frame_server import CompanionFrameServer
+from repeater.identity_manager import IdentityManager
+from repeater.main import RepeaterDaemon
+from repeater.packet_router import PacketRouter
+
+# Distinct seeds whose derived public keys do not collide on the first byte,
+# so the two companions occupy different companion_bridges slots.
+LOADED_KEY = "11" * 32  # public key starts 0xd0
+HOT_ADDED_KEY = "22" * 32  # public key starts 0xa0
+
+
+def _config(companions=()):
+    return {
+        "repeater": {"node_name": "n", "identity_key": b"\x10" * 32},
+        "logging": {},
+        "radio": {},
+        "identities": {"companions": list(companions), "room_servers": []},
+    }
+
+
+def _daemon(companions=()):
+    daemon = RepeaterDaemon(_config(companions), radio=object())
+    daemon.identity_manager = IdentityManager({})
+    daemon.router = PacketRouter(daemon)
+    return daemon
+
+
+def _expected_hash(identity_key_hex: str) -> str:
+    return f"0x{LocalIdentity(seed=bytes.fromhex(identity_key_hex)).get_public_key()[0]:02x}"
+
+
+def _assert_injector_bound(daemon, expected_hash: str) -> None:
+    """The bridge injects as expected_hash, and its frame server agrees."""
+    bridge = daemon.companion_bridges[int(expected_hash, 16)]
+    injector = bridge._packet_injector
+
+    assert injector.func == daemon.router.inject_packet
+    assert injector.keywords["origin_hash"] == expected_hash
+    # The frame server compares its own companion_hash against the exclude_hash
+    # inject_packet passes down, so the two must be the same string.
+    frame_server = next(fs for fs in daemon.companion_frame_servers if fs.bridge is bridge)
+    assert frame_server.companion_hash == expected_hash
+    # Same string again as the bridge's SQLite namespace key.
+    assert bridge._companion_hash == expected_hash
+
+
+@pytest.mark.asyncio
+async def test_companion_bridge_injector_is_bound_to_its_own_origin_hash():
+    daemon = _daemon(companions=({"name": "loaded", "identity_key": LOADED_KEY},))
+
+    with patch.object(CompanionFrameServer, "start", AsyncMock()):
+        await daemon._load_companion_identities()
+
+        assert len(daemon.companion_bridges) == 1
+        _assert_injector_bound(daemon, _expected_hash(LOADED_KEY))
+
+        # Hot-add path: a second companion added at runtime must be bound to its
+        # own hash, not to the one already loaded.
+        await daemon.add_companion_from_config(
+            {"name": "hot", "identity_key": HOT_ADDED_KEY, "settings": {"tcp_port": 5001}}
+        )
+
+    assert len(daemon.companion_bridges) == 2
+    _assert_injector_bound(daemon, _expected_hash(LOADED_KEY))
+    _assert_injector_bound(daemon, _expected_hash(HOT_ADDED_KEY))

tests/test_packet_router.py

@@ -15,6 +15,7 @@ or:
 """
 
 import asyncio
+import functools
 import hashlib
 import time
 import unittest
@@ -36,12 +37,14 @@ from openhop_core.node.handlers.text import TextMessageHandler
 from openhop_core.node.handlers.trace import TraceHandler
 from openhop_core.protocol.constants import (
     PAYLOAD_TYPE_GRP_DATA,
+    PAYLOAD_TYPE_GRP_TXT,
     ROUTE_TYPE_DIRECT,
     ROUTE_TYPE_FLOOD,
     ROUTE_TYPE_TRANSPORT_DIRECT,
 )
 from openhop_core.protocol import LocalIdentity, Packet, PacketBuilder
 
+from repeater.companion.bridge import RepeaterCompanionBridge
 from repeater.packet_router import (
     PacketRouter,
     _companion_dedup_key,
@@ -83,6 +86,7 @@ def _make_packet(payload_type: int = 0xFF):
     pkt.snr = 5.0
     pkt.timestamp = time.time()
     pkt._injected_for_tx = False
+    pkt._injected_origin_hash = None
     pkt.path = bytearray()
     pkt.calculate_packet_hash.return_value = b"\x01" * 32
     pkt.mark_do_not_retransmit = MagicMock()
@@ -1323,6 +1327,119 @@ class TestInjectedTxRawEcho(unittest.IsolatedAsyncioTestCase):
         self.assertTrue(ok)
 
 
+class TestInjectedOriginExclusion(unittest.IsolatedAsyncioTestCase):
+    """A companion never hears its own transmission back through the router.
+
+    inject_packet tags the packet with the originating companion's "0xHH" hash
+    and _companion_bridges_for_packet withholds that bridge from every fan-out,
+    the same rule the 0x88 raw-RX echo already applies to the frame servers.
+    """
+
+    @staticmethod
+    def _daemon_with_two_bridges():
+        daemon = _make_daemon()
+        daemon._on_raw_rx_for_companions = None
+        origin = _make_bridge()
+        other = _make_bridge()
+        daemon.companion_bridges = {0x1A: origin, 0x2B: other}
+        return daemon, origin, other
+
+    @staticmethod
+    async def _inject_and_drain(router, pkt, origin_hash):
+        await router.start()
+        try:
+            injected = await router.inject_packet(pkt, origin_hash=origin_hash)
+            await asyncio.sleep(0.05)  # let queue drain into _route_packet
+        finally:
+            await router.stop()
+        return injected
+
+    async def test_injected_advert_is_not_redelivered_to_the_originating_bridge(self):
+        """A companion's own advert reaches the other bridge only."""
+        daemon, origin, other = self._daemon_with_two_bridges()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(AdvertHandler.payload_type())
+
+        self.assertTrue(await self._inject_and_drain(router, pkt, "0x1a"))
+
+        origin.process_received_packet.assert_not_awaited()
+        other.process_received_packet.assert_awaited_once()
+
+    async def test_inject_without_origin_hash_still_fans_out_to_all_bridges(self):
+        """Repeater-originated injects (trace, discovery, login, text) have no
+        origin companion, so every bridge must still be offered the packet."""
+        daemon, origin, other = self._daemon_with_two_bridges()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(AdvertHandler.payload_type())
+
+        self.assertTrue(await self._inject_and_drain(router, pkt, None))
+
+        origin.process_received_packet.assert_awaited_once()
+        other.process_received_packet.assert_awaited_once()
+
+    async def test_ota_advert_still_reaches_every_bridge(self):
+        """Anti-over-blocking control: an advert heard off the air is untagged,
+        so the origin filter must leave both bridges in place."""
+        daemon, origin, other = self._daemon_with_two_bridges()
+        router = PacketRouter(daemon)
+
+        await router._route_packet(_make_packet(AdvertHandler.payload_type()))
+
+        origin.process_received_packet.assert_awaited_once()
+        other.process_received_packet.assert_awaited_once()
+
+    async def test_injected_group_message_is_not_returned_to_its_origin(self):
+        """GRP_TXT fans out to every companion, so it needs the same exclusion."""
+        daemon, origin, other = self._daemon_with_two_bridges()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(PAYLOAD_TYPE_GRP_TXT)
+
+        self.assertTrue(await self._inject_and_drain(router, pkt, "0x1a"))
+
+        origin.process_received_packet.assert_not_awaited()
+        other.process_received_packet.assert_awaited_once()
+
+    async def test_injected_advert_skips_advert_helper_neighbour_tracking(self):
+        """Our own TX is not a neighbour observation: feeding it to the helper
+        would record a phantom neighbour from the zeroed rssi/snr and burn a
+        rate-limit token. An advert heard off the air must still reach it."""
+        daemon, _origin, _other = self._daemon_with_two_bridges()
+        daemon.advert_helper = MagicMock()
+        daemon.advert_helper.process_advert_packet = AsyncMock()
+        router = PacketRouter(daemon)
+
+        injected = _make_packet(AdvertHandler.payload_type())
+        self.assertTrue(await self._inject_and_drain(router, injected, "0x1a"))
+        daemon.advert_helper.process_advert_packet.assert_not_awaited()
+
+        await router._route_packet(_make_packet(AdvertHandler.payload_type()))
+        daemon.advert_helper.process_advert_packet.assert_awaited_once()
+
+    async def test_self_advert_never_returns_to_its_own_companion_end_to_end(self):
+        """A real bridge advertising through the real injector must not end up
+        holding itself as a contact once the router queue has drained."""
+        identity = LocalIdentity()
+        companion_hash = identity.get_public_key()[0]
+        daemon = _make_daemon()
+        daemon._on_raw_rx_for_companions = None
+        router = PacketRouter(daemon)
+        bridge = RepeaterCompanionBridge(
+            identity,
+            functools.partial(router.inject_packet, origin_hash=f"0x{companion_hash:02x}"),
+            companion_hash=f"0x{companion_hash:02x}",
+        )
+        daemon.companion_bridges = {companion_hash: bridge}
+
+        await router.start()
+        try:
+            self.assertTrue(await bridge.advertise())
+            await asyncio.sleep(0.05)  # let queue drain into _route_packet
+        finally:
+            await router.stop()
+
+        self.assertEqual(bridge.get_contact_count(), 0)
+
+
 class TestCompanionDeliveryFailureHandling(unittest.IsolatedAsyncioTestCase):
     """Dedupe marking and candidate-loop behaviour when companion bridges raise."""