fix(router): try both companion and server on colliding dest hash

On-air dest hashes are one byte, so a companion and a room-server (or
repeater) identity can share one. The login and text dispatch used
`if dest_hash in companion_bridges: ... elif login_helper/text_helper`,
so a companion silently shadowed a same-hash room server: its login
never reached the room-server handler and failed with Invalid HMAC
(e.g. Lake Stevens pubkey f55d.. / hash 0xf5 colliding with a companion).

Offer the packet to both handlers when both are registered at the hash;
decryption disambiguates — the key owner replies, the other fails HMAC
and no-ops. Non-colliding paths are unchanged. Applied to both the
LoginServerHandler and TextMessageHandler branches.

Adds regression tests covering collision (both invoked), companion-only
(server helper skipped), and server-without-companion for each path.

repeater/packet_router.py

@@ -474,16 +474,30 @@ class PacketRouter:
                     logger.debug(f"Companion bridge advert error: {e}")
 
         elif payload_type == LoginServerHandler.payload_type():
-            # Route to companion if dest is a companion; else to login_helper (for logging into this repeater).
-            # When dest is remote (not handled), pass to engine so DIRECT/FLOOD ANON_REQ can be forwarded.
-            # Our own injected ANON_REQ is suppressed by the engine's duplicate (mark_seen) check.
+            # Route to the local identity that owns this login. The on-air dest
+            # hash is only one byte, so a companion and a room-server identity
+            # can share it; decryption is the only real disambiguator. When both
+            # are registered under the same hash, offer the packet to both — the
+            # owner whose key decrypts replies, the other fails HMAC and no-ops.
+            # When dest is remote (not handled), login_helper passes it to the
+            # engine so DIRECT/FLOOD ANON_REQ can be forwarded. Our own injected
+            # ANON_REQ is suppressed by the engine's duplicate (mark_seen) check.
             dest_hash = packet.payload[0] if packet.payload else None
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
-            if dest_hash is not None and dest_hash in companion_bridges:
+            login_helper = self.daemon.login_helper
+            login_handlers = getattr(login_helper, "handlers", {}) if login_helper else {}
+
+            has_companion = dest_hash is not None and dest_hash in companion_bridges
+            has_room_server = dest_hash is not None and dest_hash in login_handlers
+
+            if has_companion:
                 await companion_bridges[dest_hash].process_received_packet(packet)
                 processed_by_injection = True
-            elif self.daemon.login_helper:
-                handled = await self.daemon.login_helper.process_login_packet(packet)
+            # Offer to login_helper when a room-server identity shares this hash
+            # (collision) or when no local companion claims it at all (normal
+            # repeater/room-server login + remote-forward handling).
+            if login_helper and (has_room_server or not has_companion):
+                handled = await login_helper.process_login_packet(packet)
                 if handled:
                     processed_by_injection = True
             if processed_by_injection:
@@ -513,17 +527,27 @@ class PacketRouter:
                 await self._register_ack_with_dispatcher(ack_crc, "multi-ACK")
 
         elif payload_type == TextMessageHandler.payload_type():
+            # Same one-byte dest-hash collision handling as the login path above:
+            # a companion and a room-server text identity can share a hash, and
+            # only decryption tells them apart. Offer to both when both are
+            # registered so a companion never shadows a room-server message.
             dest_hash = packet.payload[0] if packet.payload else None
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
-            if dest_hash is not None and dest_hash in companion_bridges:
+            text_helper = self.daemon.text_helper
+            text_handlers = getattr(text_helper, "handlers", {}) if text_helper else {}
+
+            has_companion = dest_hash is not None and dest_hash in companion_bridges
+            has_text_identity = dest_hash is not None and dest_hash in text_handlers
+
+            if has_companion:
                 await companion_bridges[dest_hash].process_received_packet(packet)
                 processed_by_injection = True
-                self._record_for_ui(packet, metadata)
-            elif self.daemon.text_helper:
-                handled = await self.daemon.text_helper.process_text_packet(packet)
+            if text_helper and (has_text_identity or not has_companion):
+                handled = await text_helper.process_text_packet(packet)
                 if handled:
                     processed_by_injection = True
-                    self._record_for_ui(packet, metadata)
+            if processed_by_injection:
+                self._record_for_ui(packet, metadata)
 
         elif payload_type == PathHandler.payload_type():
             # Always let PathHelper inspect/decrypt PATH first so out_path and bundled ACK state

tests/test_packet_router.py

@@ -592,6 +592,64 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase):
         bridge.process_received_packet.assert_awaited_once()
         daemon.repeater_handler.assert_not_awaited()
 
+    async def test_route_login_server_hash_collision_offers_to_both(self):
+        """A one-byte dest hash shared by a companion and a room-server identity
+        must be offered to BOTH handlers; only the one whose key decrypts replies.
+        Regression: previously the companion shadowed the room server, so the
+        room-server login never ran and failed with Invalid HMAC."""
+        daemon = _make_daemon()
+        bridge = _make_bridge()
+        daemon.companion_bridges = {0xF5: bridge}
+        daemon.login_helper = MagicMock()
+        daemon.login_helper.handlers = {0xF5: MagicMock()}
+        daemon.login_helper.process_login_packet = AsyncMock(return_value=True)
+        daemon.repeater_handler = AsyncMock()
+        daemon.repeater_handler.storage = MagicMock()
+        daemon.repeater_handler.record_packet_only = MagicMock()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(LoginServerHandler.payload_type())
+        pkt.payload = bytes([0xF5, 0x99])
+        await router._route_packet(pkt)
+        bridge.process_received_packet.assert_awaited_once()
+        daemon.login_helper.process_login_packet.assert_awaited_once()
+        daemon.repeater_handler.assert_not_awaited()
+
+    async def test_route_login_server_companion_only_skips_login_helper(self):
+        """No collision: a companion-owned hash with no room server registered
+        there must not also invoke login_helper."""
+        daemon = _make_daemon()
+        bridge = _make_bridge()
+        daemon.companion_bridges = {0x7A: bridge}
+        daemon.login_helper = MagicMock()
+        daemon.login_helper.handlers = {}  # no room-server identity at this hash
+        daemon.login_helper.process_login_packet = AsyncMock(return_value=False)
+        daemon.repeater_handler = AsyncMock()
+        daemon.repeater_handler.storage = MagicMock()
+        daemon.repeater_handler.record_packet_only = MagicMock()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(LoginServerHandler.payload_type())
+        pkt.payload = bytes([0x7A, 0x99])
+        await router._route_packet(pkt)
+        bridge.process_received_packet.assert_awaited_once()
+        daemon.login_helper.process_login_packet.assert_not_awaited()
+
+    async def test_route_login_server_room_server_without_companion(self):
+        """No local companion claims the hash: login_helper handles the
+        room-server (or forwards a remote) login as before."""
+        daemon = _make_daemon()
+        daemon.companion_bridges = {}
+        daemon.login_helper = MagicMock()
+        daemon.login_helper.handlers = {0xF5: MagicMock()}
+        daemon.login_helper.process_login_packet = AsyncMock(return_value=True)
+        daemon.repeater_handler = AsyncMock()
+        daemon.repeater_handler.storage = MagicMock()
+        daemon.repeater_handler.record_packet_only = MagicMock()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(LoginServerHandler.payload_type())
+        pkt.payload = bytes([0xF5, 0x99])
+        await router._route_packet(pkt)
+        daemon.login_helper.process_login_packet.assert_awaited_once()
+
     async def test_route_text_to_helper_marks_processed(self):
         daemon = _make_daemon()
         daemon.text_helper = MagicMock()
@@ -605,6 +663,44 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase):
         daemon.text_helper.process_text_packet.assert_awaited_once()
         daemon.repeater_handler.assert_not_awaited()
 
+    async def test_route_text_hash_collision_offers_to_both(self):
+        """A dest hash shared by a companion and a room-server text identity must
+        reach both handlers so a companion never shadows a room-server message."""
+        daemon = _make_daemon()
+        bridge = _make_bridge()
+        daemon.companion_bridges = {0xF5: bridge}
+        daemon.text_helper = MagicMock()
+        daemon.text_helper.handlers = {0xF5: MagicMock()}
+        daemon.text_helper.process_text_packet = AsyncMock(return_value=True)
+        daemon.repeater_handler = AsyncMock()
+        daemon.repeater_handler.storage = MagicMock()
+        daemon.repeater_handler.record_packet_only = MagicMock()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(TextMessageHandler.payload_type())
+        pkt.payload = bytes([0xF5, 0x01])
+        await router._route_packet(pkt)
+        bridge.process_received_packet.assert_awaited_once()
+        daemon.text_helper.process_text_packet.assert_awaited_once()
+        daemon.repeater_handler.assert_not_awaited()
+
+    async def test_route_text_companion_only_skips_text_helper(self):
+        """No collision: a companion-owned text hash must not also hit text_helper."""
+        daemon = _make_daemon()
+        bridge = _make_bridge()
+        daemon.companion_bridges = {0xEE: bridge}
+        daemon.text_helper = MagicMock()
+        daemon.text_helper.handlers = {}  # no room-server text identity here
+        daemon.text_helper.process_text_packet = AsyncMock(return_value=False)
+        daemon.repeater_handler = AsyncMock()
+        daemon.repeater_handler.storage = MagicMock()
+        daemon.repeater_handler.record_packet_only = MagicMock()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(TextMessageHandler.payload_type())
+        pkt.payload = bytes([0xEE, 0x01])
+        await router._route_packet(pkt)
+        bridge.process_received_packet.assert_awaited_once()
+        daemon.text_helper.process_text_packet.assert_not_awaited()
+
     async def test_route_ack_delivers_to_all_bridges_and_engine(self):
         daemon = _make_daemon()
         b1 = _make_bridge()