fix: resolve room-server push ACKs through the dispatcher

Room server pushes waited on dispatcher.wait_for_ack, but nothing in the
repeater could ever resolve it, so every push timed out and re-pushed on
the backoff schedule (issue #286's duplicate floods — worst for virtual
companions on the same instance, whose ACKs never even cross the air):

- no core AckHandler is registered (all RX lands in the router fallback),
  so received ACK CRCs were never fed to dispatcher ACK matching
- inject_packet waited on packet.get_crc(), a packet-hash CRC that no ACK
  sender produces; the crypto CRC only the room server knows was ignored
- PATH returns (how a flood-received DM is ACKed) were decrypted by
  PathHelper, but it read the encoded path_len wire byte as a raw count —
  an empty 3-byte-hash path (0x80) parsed as a 128-byte truncated
  payload — and the router's PATH branch never ran PathHelper for
  room/repeater destinations when any companion bridge existed

Fixes:
- router ACK branch feeds discrete ACK CRCs (RF and locally injected) to
  dispatcher._register_ack_received
- PATH packets addressed to a local server identity are processed by
  PathHelper regardless of companion bridges; PathHelper decodes the
  encoded path_len via PathUtils, keeps the encoded byte in out_path_len,
  and registers the embedded ACK via ack_received_fn
- inject_packet accepts expected_ack_crc/ack_timeout; the room server
  passes its crypto CRC and a hop-count-based timeout (the encoded byte
  would have produced a ~4-minute direct-push wait)

Verified live on a real mesh: push to a same-instance virtual companion
resolves via the PATH-embedded ACK (encoded path_len 0x80) and push to a
firmware client resolves via the discrete RF ACK, no re-push loops.

Fixes #286
Fixes #341

repeater/handler_helpers/path.py

@@ -5,10 +5,13 @@ logger = logging.getLogger("PathHelper")
 
 
 class PathHelper:
-    def __init__(self, acl_dict=None, log_fn=None):
+    def __init__(self, acl_dict=None, log_fn=None, ack_received_fn=None):
 
         self.acl_dict = acl_dict or {}
         self.log_fn = log_fn or logger.info
+        # Async callback fed with ACK CRCs found embedded in PATH payloads
+        # (dispatcher._register_ack_received) so local waiters resolve.
+        self.ack_received_fn = ack_received_fn
 
     async def process_path_packet(self, packet):
 
@@ -60,30 +63,61 @@ class PathHelper:
                 return False
 
             # Parse decrypted PATH data
-            # Format: path_len(1) + path[path_len] + extra_type(1) + extra[...]
+            # Format: path_len(1) + path[...] + extra_type(1) + extra[...]
+            # path_len is the ENCODED wire byte (bits 0-5 = hash count, bits
+            # 6-7 = hash size - 1), matching Packet.path_len — with 3-byte
+            # hashes an empty path is 0x80, not 0x00. Reading it as a raw byte
+            # count made every such path return look truncated.
             if len(decrypted) < 1:
                 logger.debug("Decrypted PATH data too short")
                 return False
 
-            path_len = decrypted[0]
-            if len(decrypted) < 1 + path_len:
+            from openhop_core.protocol.packet_utils import PathUtils
+
+            path_len_byte = decrypted[0]
+            if not PathUtils.is_valid_path_len(path_len_byte):
+                logger.debug(f"Invalid encoded path_len 0x{path_len_byte:02X} in PATH data")
+                return False
+            path_byte_len = PathUtils.get_path_byte_len(path_len_byte)
+            if len(decrypted) < 1 + path_byte_len:
                 logger.debug(
-                    f"PATH data truncated: need {1 + path_len} bytes, got {len(decrypted)}"
+                    f"PATH data truncated: need {1 + path_byte_len} bytes, got {len(decrypted)}"
                 )
                 return False
 
-            path_data = decrypted[1 : 1 + path_len]
+            path_data = decrypted[1 : 1 + path_byte_len]
 
-            # Update client's out_path (same as C++ memcpy)
+            # Update client's out_path (same as C++ memcpy); out_path_len keeps
+            # the encoded byte so direct sends put it on the wire as-is.
             client.out_path = bytearray(path_data)
-            client.out_path_len = path_len
+            client.out_path_len = path_len_byte
             client.last_activity = int(time.time())
 
             logger.info(
                 f"Updated out_path for client 0x{src_hash:02X} -> 0x{dest_hash:02X}: "
-                f"path_len={path_len}, path={[hex(b) for b in path_data]}"
+                f"path_len=0x{path_len_byte:02X}, path={[hex(b) for b in path_data]}"
             )
 
+            # Extra section after the path: extra_type(1) + extra[...]. Firmware
+            # answers a flood-received DM with a path return that embeds the
+            # delivery ACK here (createPathReturn, extra_type=PAYLOAD_TYPE_ACK);
+            # register it so local waiters (e.g. room server pushes) resolve.
+            from openhop_core.protocol.constants import PAYLOAD_TYPE_ACK
+
+            extra_start = 1 + path_byte_len
+            if (
+                self.ack_received_fn is not None
+                and len(decrypted) >= extra_start + 5
+                and decrypted[extra_start] == PAYLOAD_TYPE_ACK
+            ):
+                ack_crc = int.from_bytes(
+                    bytes(decrypted[extra_start + 1 : extra_start + 5]), "little"
+                )
+                await self.ack_received_fn(ack_crc)
+                logger.info(
+                    f"PATH from 0x{src_hash:02X} carried embedded ACK CRC={ack_crc:08X}"
+                )
+
             # Don't mark as do_not_retransmit - let it forward normally
             return False
 

repeater/handler_helpers/room_server.py

@@ -6,6 +6,7 @@ from typing import Dict
 
 from openhop_core.protocol import CryptoUtils, PacketBuilder
 from openhop_core.protocol.constants import PAYLOAD_TYPE_TXT_MSG
+from openhop_core.protocol.packet_utils import PathUtils
 
 logger = logging.getLogger("RoomServer")
 
@@ -377,18 +378,25 @@ class RoomServer:
                 route_type=route_type,
             )
 
-            # Add stored path for direct routing
+            # Add stored path for direct routing. out_path_len is the encoded
+            # wire byte (bits 0-5 = hash count, bits 6-7 = hash size - 1);
+            # out_path already holds exactly the path bytes.
             if route_type == "direct" and len(client_info.out_path) > 0:
-                packet.path = bytearray(client_info.out_path[: client_info.out_path_len])
+                packet.path = bytearray(client_info.out_path)
                 packet.path_len = client_info.out_path_len
 
-            # Calculate ACK timeout
+            # Calculate ACK timeout from the HOP count, not the encoded byte
+            # (0x80 = empty 3-byte-hash path would otherwise give a ~4min wait)
             if route_type == "flood":
                 ack_timeout = PUSH_ACK_TIMEOUT_FLOOD_MS / 1000.0
             else:
-                path_len = client_info.out_path_len if client_info.out_path_len >= 0 else 0
+                hops = (
+                    PathUtils.get_path_hash_count(client_info.out_path_len)
+                    if client_info.out_path_len >= 0
+                    else 0
+                )
                 ack_timeout = (
-                    PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (path_len + 1)
+                    PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (hops + 1)
                 ) / 1000.0
 
             # Update client sync state with pending ACK
@@ -399,9 +407,16 @@ class RoomServer:
                 push_post_timestamp=post["post_timestamp"],
                 ack_timeout_time=time.time() + ack_timeout,
             )
-            # Send packet (dispatcher will track ACK automatically)
+            # Send and wait for the client's delivery ACK. The injector must be
+            # told the crypto ACK CRC we computed above — its default
+            # (packet.get_crc()) is a packet-hash CRC no client ever sends.
             # This blocks for the entire transmission duration (0.5-9 seconds)
-            success = await self.packet_injector(packet, wait_for_ack=True)
+            success = await self.packet_injector(
+                packet,
+                wait_for_ack=True,
+                expected_ack_crc=expected_ack_crc,
+                ack_timeout=ack_timeout,
+            )
 
             # SAFETY: Release transmission lock AFTER send completes
             self.global_limiter.release()

repeater/main.py

@@ -412,6 +412,9 @@ class RepeaterDaemon:
             self.path_helper = PathHelper(
                 acl_dict=self.login_helper.get_acl_dict(),  # Per-identity ACLs
                 log_fn=logger.info,
+                # Embedded PATH ACKs must reach the dispatcher's ACK matching or
+                # wait_for_ack() (e.g. room server pushes) never resolves.
+                ack_received_fn=self.dispatcher._register_ack_received,
             )
             logger.info("PATH packet processing helper initialized")
 

repeater/packet_router.py

@@ -271,7 +271,14 @@ class PacketRouter:
                 pass
         await self.queue.put(packet)
 
-    async def inject_packet(self, packet, wait_for_ack: bool = False, origin_hash=None):
+    async def inject_packet(
+        self,
+        packet,
+        wait_for_ack: bool = False,
+        origin_hash=None,
+        expected_ack_crc=None,
+        ack_timeout=None,
+    ):
         try:
             metadata = {
                 "rssi": getattr(packet, "rssi", 0),
@@ -327,8 +334,16 @@ class PacketRouter:
                     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)
+                            # Callers that know the crypto ACK CRC (e.g. room
+                            # server pushes) must pass it: packet.get_crc() is a
+                            # packet-hash CRC that no ACK sender ever produces.
+                            expected_crc = (
+                                expected_ack_crc
+                                if expected_ack_crc is not None
+                                else packet.get_crc()
+                            )
+                            timeout = ack_timeout if ack_timeout is not None else 5.0
+                            ack_ok = await dispatcher.wait_for_ack(expected_crc, timeout=timeout)
                             if not ack_ok:
                                 logger.warning(
                                     "Injected packet ACK timeout (crc=%08X)", expected_crc
@@ -463,6 +478,20 @@ class PacketRouter:
                 self._record_for_ui(packet, metadata)
 
         elif payload_type == AckHandler.payload_type():
+            # Feed the dispatcher's ACK matching: the repeater registers no core
+            # AckHandler (all RX lands in this router via the fallback), so
+            # without this no dispatcher.wait_for_ack() — e.g. a room server
+            # push — can ever resolve. Covers RF ACKs and locally injected ones
+            # (companions hosted on this same instance never go over the air).
+            if packet.payload is not None and len(packet.payload) >= 4:
+                register_ack = getattr(
+                    getattr(self.daemon, "dispatcher", None), "_register_ack_received", None
+                )
+                if register_ack is not None:
+                    try:
+                        await register_ack(int.from_bytes(bytes(packet.payload[:4]), "little"))
+                    except Exception as e:
+                        logger.debug("ACK registration failed: %s", e)
             # ACK has no dest in payload (4-byte CRC only); deliver to all bridges so sender sees send_confirmed.
             # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop.
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
@@ -488,6 +517,19 @@ class PacketRouter:
         elif payload_type == PathHandler.payload_type():
             dest_hash = packet.payload[0] if packet.payload else None
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
+            # PATH addressed to a local server identity (room server/repeater):
+            # process for client out_path updates and any embedded ACK before
+            # bridge delivery. Previously the all-bridges branch below swallowed
+            # these whenever any companion was configured, so room servers never
+            # saw path returns — or the delivery ACKs firmware embeds in them.
+            path_helper = getattr(self.daemon, "path_helper", None)
+            is_local_identity_dest = (
+                path_helper is not None
+                and dest_hash is not None
+                and dest_hash in getattr(path_helper, "acl_dict", {})
+            )
+            if is_local_identity_dest:
+                await path_helper.process_path_packet(packet)
             if dest_hash is not None and dest_hash in companion_bridges:
                 if self._should_deliver_path_to_companions(packet):
                     await companion_bridges[dest_hash].process_received_packet(packet)
@@ -506,8 +548,8 @@ class PacketRouter:
                     len(companion_bridges),
                 )
                 # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop.
-            elif self.daemon.path_helper:
-                await self.daemon.path_helper.process_path_packet(packet)
+            elif path_helper and not is_local_identity_dest:
+                await path_helper.process_path_packet(packet)
 
         elif payload_type == LoginResponseHandler.payload_type():
             # PAYLOAD_TYPE_RESPONSE (0x01): payload is dest_hash(1)+src_hash(1)+encrypted.

tests/test_handler_helpers_path_protocol_text.py

@@ -67,6 +67,74 @@ async def test_path_helper_updates_client_out_path_on_valid_decrypt():
     assert isinstance(client.last_activity, int)
 
 
+@pytest.mark.asyncio
+async def test_path_helper_registers_embedded_ack():
+    """Firmware path returns embed the delivery ACK after the path
+    (extra_type=PAYLOAD_TYPE_ACK + 4-byte CRC); it must reach ack_received_fn
+    so local waiters (e.g. room server pushes) resolve."""
+    client = _FakeClient(pubkey=bytes([0x22]) + b"x" * 31, shared_secret=b"k" * 32)
+    acl = _FakeACL([client])
+    ack_fn = AsyncMock()
+    helper = PathHelper(acl_dict={0x11: acl}, ack_received_fn=ack_fn)
+
+    packet = _PathPacket(payload=b"\x11\x22\xaa\xbb\xcc")
+    # path_len(2) + path + extra_type(PAYLOAD_TYPE_ACK=3) + crc(4, LE)
+    decrypted = b"\x02\x99\x88" + bytes([0x03]) + bytes.fromhex("4dabaf95")
+    with patch(
+        "openhop_core.protocol.crypto.CryptoUtils.mac_then_decrypt",
+        return_value=decrypted,
+    ):
+        await helper.process_path_packet(packet)
+
+    ack_fn.assert_awaited_once_with(0x95AFAB4D)
+    assert client.out_path_len == 2  # path update still applied
+
+
+@pytest.mark.asyncio
+async def test_path_helper_handles_encoded_path_len_with_embedded_ack():
+    """path_len in a path return is the ENCODED wire byte: with 3-byte hashes an
+    empty path is 0x80, not 0x00. Reading it as a raw count (128) made the
+    helper bail as 'truncated' before the embedded ACK was registered, so
+    room-server pushes to same-instance companions timed out forever.
+    Bytes below are a decrypted path return captured from a live mesh."""
+    client = _FakeClient(pubkey=bytes([0x22]) + b"x" * 31, shared_secret=b"k" * 32)
+    acl = _FakeACL([client])
+    ack_fn = AsyncMock()
+    helper = PathHelper(acl_dict={0x11: acl}, ack_received_fn=ack_fn)
+
+    packet = _PathPacket(payload=b"\x11\x22\xaa\xbb\xcc")
+    # path_len 0x80 (3-byte hashes, 0 hops) + extra_type ACK + crc + AES padding
+    decrypted = bytes.fromhex("80038d48208500000000000000000000")
+    with patch(
+        "openhop_core.protocol.crypto.CryptoUtils.mac_then_decrypt",
+        return_value=decrypted,
+    ):
+        await helper.process_path_packet(packet)
+
+    ack_fn.assert_awaited_once_with(0x8520488D)
+    assert client.out_path_len == 0x80  # encoded byte preserved
+    assert bytes(client.out_path) == b""
+
+
+@pytest.mark.asyncio
+async def test_path_helper_ignores_non_ack_extra():
+    client = _FakeClient(pubkey=bytes([0x22]) + b"x" * 31, shared_secret=b"k" * 32)
+    acl = _FakeACL([client])
+    ack_fn = AsyncMock()
+    helper = PathHelper(acl_dict={0x11: acl}, ack_received_fn=ack_fn)
+
+    packet = _PathPacket(payload=b"\x11\x22\xaa\xbb\xcc")
+    # extra_type 0x08 (PATH) instead of ACK: nothing to register
+    decrypted = b"\x02\x99\x88" + bytes([0x08]) + b"\x01\x02\x03\x04"
+    with patch(
+        "openhop_core.protocol.crypto.CryptoUtils.mac_then_decrypt",
+        return_value=decrypted,
+    ):
+        await helper.process_path_packet(packet)
+
+    ack_fn.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_path_helper_returns_false_for_non_matching_or_invalid_inputs():
     client = _FakeClient(pubkey=bytes([0x22]) + b"x" * 31, shared_secret=b"k" * 32)

tests/test_handler_helpers_room_server.py

@@ -146,7 +146,12 @@ async def test_room_server_push_post_to_client_success_direct_route_sets_path_an
     assert ok is True
     assert bytes(packet.path) == b"\xaa\xbb"
     assert packet.path_len == 2
-    injector.assert_awaited_once_with(packet, wait_for_ack=True)
+    injector.assert_awaited_once_with(
+        packet,
+        wait_for_ack=True,
+        expected_ack_crc=int.from_bytes(b"\x01\x02\x03\x04", "little"),
+        ack_timeout=10.0,  # PUSH_TIMEOUT_BASE + FACTOR * (path_len 2 + 1)
+    )
     rs._handle_ack_received.assert_awaited_once_with(
         client.id.get_public_key(), post["post_timestamp"]
     )
@@ -174,9 +179,11 @@ async def test_room_server_push_expected_ack_matches_firmware_signed_ack():
 
     db = _FakeDB()
     sent = []
+    injector_kwargs = []
 
-    async def injector(packet, wait_for_ack=False):
+    async def injector(packet, wait_for_ack=False, **kwargs):
         sent.append(packet)
+        injector_kwargs.append(kwargs)
         return False  # no ACK: leaves the pending upsert as the only db write
 
     rs = RoomServer(
@@ -205,6 +212,11 @@ async def test_room_server_push_expected_ack_matches_firmware_signed_ack():
     assert len(upserts) == 1
     expected_ack_crc = upserts[0]["pending_ack_crc"]
 
+    # The injector must be told the crypto CRC (and the computed timeout) so
+    # dispatcher.wait_for_ack matches the client's actual ACK.
+    assert injector_kwargs[0]["expected_ack_crc"] == expected_ack_crc
+    assert injector_kwargs[0]["ack_timeout"] > 0
+
     # Decrypt the pushed datagram and verify the signed-plain layout.
     pkt = sent[0]
     encrypted = bytes(pkt.payload[2 : pkt.payload_len])

tests/test_packet_router.py

@@ -56,6 +56,9 @@ def _make_daemon():
     daemon.text_helper = None
     daemon.path_helper = None
     daemon.protocol_request_helper = None
+    daemon.dispatcher = MagicMock()
+    daemon.dispatcher._register_ack_received = AsyncMock()
+    daemon.dispatcher.wait_for_ack = AsyncMock(return_value=True)
     return daemon
 
 
@@ -614,6 +617,66 @@ class TestPacketRouterRoutingBranches(unittest.IsolatedAsyncioTestCase):
         b2.process_received_packet.assert_awaited_once()
         daemon.repeater_handler.assert_awaited_once()
 
+    async def test_route_ack_registers_crc_with_dispatcher(self):
+        """Discrete ACKs must feed dispatcher ACK matching or wait_for_ack never resolves."""
+        daemon = _make_daemon()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(AckHandler.payload_type())
+        # 6-byte firmware ACK: 4-byte CRC (LE) + ext-attempt + random byte
+        pkt.payload = bytes.fromhex("4dabaf95") + b"\x00\x7f"
+        await router._route_packet(pkt)
+        daemon.dispatcher._register_ack_received.assert_awaited_once_with(0x95AFAB4D)
+
+    async def test_route_ack_short_payload_not_registered(self):
+        daemon = _make_daemon()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(AckHandler.payload_type())
+        pkt.payload = b"\x01\x02"
+        await router._route_packet(pkt)
+        daemon.dispatcher._register_ack_received.assert_not_awaited()
+
+    async def test_route_locally_injected_ack_registers_crc(self):
+        """ACKs from same-instance companions never cross the radio: the injected
+        packet re-entering the router is the only chance to resolve local waiters
+        (e.g. a room server pushing to a virtual companion on this repeater)."""
+        daemon = _make_daemon()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(AckHandler.payload_type())
+        pkt.payload = b"\xaa\xbb\xcc\xdd"
+        pkt._injected_for_tx = True
+        await router._route_packet(pkt)
+        daemon.dispatcher._register_ack_received.assert_awaited_once_with(0xDDCCBBAA)
+        daemon.repeater_handler.assert_not_awaited()  # already transmitted
+
+    async def test_inject_packet_waits_on_expected_ack_crc(self):
+        """Callers that know the crypto ACK CRC pass it; packet.get_crc() is a
+        packet-hash CRC no ACK sender ever produces."""
+        daemon = _make_daemon()
+        router = PacketRouter(daemon)
+        pkt = _make_packet(TextMessageHandler.payload_type())
+        ok = await router.inject_packet(
+            pkt, wait_for_ack=True, expected_ack_crc=0x1234ABCD, ack_timeout=12.0
+        )
+        self.assertTrue(ok)
+        daemon.dispatcher.wait_for_ack.assert_awaited_once_with(0x1234ABCD, timeout=12.0)
+        pkt.get_crc.assert_not_called()
+
+    async def test_route_path_to_local_identity_runs_path_helper_despite_bridges(self):
+        """PATH addressed to a room server/repeater identity must reach the path
+        helper (out_path update + embedded ACK) even when companion bridges exist;
+        previously the all-bridges branch swallowed it."""
+        daemon = _make_daemon()
+        bridge = _make_bridge()
+        daemon.companion_bridges = {0x01: bridge}
+        daemon.path_helper = MagicMock()
+        daemon.path_helper.acl_dict = {0x48: MagicMock()}
+        daemon.path_helper.process_path_packet = AsyncMock(return_value=False)
+        router = PacketRouter(daemon)
+        pkt = _make_packet(PathHandler.payload_type())
+        pkt.payload = bytes([0x48, 0x77, 0xAA, 0xBB])
+        await router._route_packet(pkt)
+        daemon.path_helper.process_path_packet.assert_awaited_once_with(pkt)
+
     async def test_route_path_dedupes_companion_delivery(self):
         daemon = _make_daemon()
         bridge = _make_bridge()