fix(repeater): relay multipart ACKs as regenerated direct ACKs

Match MeshCore's forwardMultipartDirect: a multipart packet is never
relayed as an ordinary directed packet. At an intermediate direct hop a
multipart ACK is re-emitted as a plain DIRECT ACK with the wrapper byte
stripped and this node removed from the path, deduped on the regenerated
ACK form and spaced by (remaining + 1) * 300ms. Flood routing, final
hops, and non-ACK embedded types are dropped rather than forwarded.

Coverage includes literal firmware-format wire vectors for one- and
two-byte hash widths and transport-direct input, end-to-end receive-to-
transmit relaying, and the drop cases.

repeater/engine.py

@@ -10,10 +10,13 @@ from openhop_core.node.handlers.base import BaseHandler
 from openhop_core.protocol import Packet
 from openhop_core.protocol.constants import (
     MAX_PATH_SIZE,
+    PAYLOAD_TYPE_ACK,
     PAYLOAD_TYPE_ADVERT,
     PAYLOAD_TYPE_ANON_REQ,
+    PAYLOAD_TYPE_MULTIPART,
     PAYLOAD_TYPE_TRACE,
     PH_ROUTE_MASK,
+    PH_TYPE_SHIFT,
     ROUTE_TYPE_DIRECT,
     ROUTE_TYPE_FLOOD,
     ROUTE_TYPE_TRANSPORT_DIRECT,
@@ -1036,6 +1039,84 @@ class RepeaterHandler(BaseHandler):
 
         return packet
 
+    # Multipart ACKs are spaced ~300ms apart per remaining fragment, matching
+    # MeshCore's forwardMultipartDirect (`(remaining + 1) * 300`).
+    MULTIPART_ACK_SPACING_MS = 300
+
+    def forward_multipart_direct(
+        self, packet: Packet, packet_hash: Optional[str] = None
+    ) -> Optional[Tuple[Packet, float]]:
+        """Forward a MULTIPART packet the way MeshCore's forwardMultipartDirect does.
+
+        MeshCore never relays the multipart wrapper as an ordinary directed
+        packet.  Only a multipart-ACK at an intermediate direct hop is relayed,
+        and it is relayed by regenerating the embedded ACK as a plain DIRECT ACK
+        (the wrapper byte stripped) rather than repeating the wrapper.  Every
+        other multipart case — flood routing, a final hop, or a non-ACK embedded
+        type — is dropped.  Returns ``(ack_packet, delay_seconds)`` when a
+        regenerated ACK should be sent, otherwise ``None``.
+
+        INVARIANT: purely synchronous, mirroring flood_forward / direct_forward —
+        the is_duplicate + mark_seen pair must stay atomic within the event loop.
+        """
+        # MeshCore only forwards multipart on the direct-route branch; the flood
+        # switch case updates ACK state locally but never re-routes it.
+        if not packet.is_route_direct():
+            packet.drop_reason = "Multipart: not direct-routed"
+            return None
+
+        valid, reason = self.validate_packet(packet)
+        if not valid:
+            packet.drop_reason = reason
+            return None
+
+        if packet.is_marked_do_not_retransmit():
+            packet.drop_reason = packet.drop_reason or "Marked do not retransmit"
+            return None
+
+        payload = packet.payload or b""
+        remaining = payload[0] >> 4
+        embedded_type = payload[0] & 0x0F
+        # Only multipart-ACKs are relayed; MeshCore leaves other types for future
+        # use and does not forward them.
+        if embedded_type != PAYLOAD_TYPE_ACK or len(payload) < 5:
+            packet.drop_reason = "Multipart: unsupported embedded type"
+            return None
+
+        hash_size = packet.get_path_hash_size()
+        hop_count = packet.get_path_hash_count()
+
+        # We must be the next hop (a final-hop multipart has no remaining path and
+        # is handled locally, not forwarded).
+        if not packet.path or len(packet.path) < hash_size:
+            packet.drop_reason = "Direct: no path"
+            return None
+        if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]:
+            packet.drop_reason = "Direct: not for us"
+            return None
+
+        # Rebuild the received wrapper as the plain DIRECT ACK MeshCore would emit:
+        # drop the multipart header byte and force the ACK payload/route type. The
+        # path is left intact for the seen-check, then this node is removed from it,
+        # matching forwardMultipartDirect's hasSeen()/removeSelfFromPath() order.
+        packet.payload = bytearray(payload[1:])
+        packet.payload_len = len(packet.payload)
+        packet.header = (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT
+
+        # Dedupe on the regenerated ACK form (full path) before removing ourselves,
+        # so a repeated multipart ACK — or an equivalent plain ACK — is not relayed
+        # twice.  The pre-computed hash belongs to the original wrapper, so recompute.
+        if self.is_duplicate(packet):
+            packet.drop_reason = "Duplicate"
+            return None
+        self.mark_seen(packet)
+
+        packet.path = bytearray(packet.path[hash_size:])
+        packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1)
+
+        delay_s = ((remaining + 1) * self.MULTIPART_ACK_SPACING_MS) / 1000.0
+        return packet, delay_s
+
     @staticmethod
     def calculate_packet_score(snr: float, packet_len: int, spreading_factor: int = 8) -> float:
 
@@ -1120,6 +1201,13 @@ class RepeaterHandler(BaseHandler):
         """
         route_type = packet.header & PH_ROUTE_MASK
 
+        # MeshCore routes multipart traffic through its own branch, never the
+        # generic flood/direct path — it regenerates embedded ACKs and drops
+        # everything else.  Its delay is fixed by the fragment count, so return
+        # the (packet, delay) pair directly instead of _calculate_tx_delay.
+        if packet.get_payload_type() == PAYLOAD_TYPE_MULTIPART:
+            return self.forward_multipart_direct(packet, packet_hash=packet_hash)
+
         if route_type == ROUTE_TYPE_FLOOD or route_type == ROUTE_TYPE_TRANSPORT_FLOOD:
             fwd_pkt = self.flood_forward(packet, packet_hash=packet_hash)
             if fwd_pkt is None:

repeater/packet_router.py

@@ -48,6 +48,7 @@ _EXPECTED_DROP_REASON_PREFIXES = (
     "Empty payload",
     "Path too long",
     "Invalid advert packet",
+    "Multipart",
 )
 
 

tests/test_engine.py

@@ -15,6 +15,8 @@ import pytest
 from openhop_core.protocol import Packet, PacketBuilder
 from openhop_core.protocol.constants import (
     MAX_PATH_SIZE,
+    PAYLOAD_TYPE_ACK,
+    PAYLOAD_TYPE_MULTIPART,
     PH_ROUTE_MASK,
     PH_TYPE_SHIFT,
     ROUTE_TYPE_DIRECT,
@@ -30,6 +32,10 @@ from openhop_core.protocol.packet_utils import PathUtils
 
 LOCAL_HASH = 0xAB  # repeater's own 1-byte path hash
 
+# MULTIPART is not forwarded generically — it has its own MeshCore branch
+# (forward_multipart_direct), covered by TestForwardMultipartDirect.
+_GENERIC_FORWARD_PAYLOAD_TYPES = [pt for pt in range(16) if pt != PAYLOAD_TYPE_MULTIPART]
+
 
 def _make_config(**overrides) -> dict:
     """Return a minimal valid config dict for RepeaterHandler."""
@@ -91,19 +97,26 @@ def _make_dispatcher(radio=None):
     return dispatcher
 
 
-@pytest.fixture()
-def handler():
-    """Create a RepeaterHandler with mocked external dependencies."""
-    config = _make_config()
-    dispatcher = _make_dispatcher()
+def _make_handler_with_hash(local_hash_bytes: bytes):
+    """Create a RepeaterHandler whose local path hash is exactly local_hash_bytes."""
     with (
         patch("repeater.engine.StorageCollector"),
         patch("repeater.engine.RepeaterHandler._start_background_tasks"),
     ):
         from repeater.engine import RepeaterHandler
 
-        h = RepeaterHandler(config, dispatcher, LOCAL_HASH)
-    return h
+        return RepeaterHandler(
+            _make_config(),
+            _make_dispatcher(),
+            local_hash_bytes[0],
+            local_hash_bytes=local_hash_bytes,
+        )
+
+
+@pytest.fixture()
+def handler():
+    """Create a RepeaterHandler with mocked external dependencies."""
+    return _make_handler_with_hash(bytes([LOCAL_HASH]))
 
 
 def _make_flood_packet(
@@ -347,6 +360,176 @@ class TestDirectForward:
         assert pkt.path_len == len(pkt.path) == 2
 
 
+# ===================================================================
+# 2b. forward_multipart_direct — MeshCore forwardMultipartDirect parity
+# ===================================================================
+
+
+def _make_multipart_ack_packet(
+    crc: bytes = b"\x4d\xab\xaf\x95",
+    remaining: int = 2,
+    path: bytes = None,
+    route: int = ROUTE_TYPE_DIRECT,
+    embedded_type: int = PAYLOAD_TYPE_ACK,
+) -> Packet:
+    """Build a MULTIPART wrapper: [remaining<<4 | embedded_type] followed by a CRC."""
+    if path is None:
+        path = bytes([LOCAL_HASH, 0xCC])
+    pkt = Packet()
+    pkt.header = route | (PAYLOAD_TYPE_MULTIPART << PH_TYPE_SHIFT)
+    pkt.payload = bytearray(bytes([(remaining << 4) | embedded_type]) + crc)
+    pkt.payload_len = len(pkt.payload)
+    pkt.path = bytearray(path)
+    pkt.path_len = len(path)
+    return pkt
+
+
+class TestForwardMultipartDirect:
+    """forward_multipart_direct mirrors MeshCore: regenerate the embedded ACK,
+    drop every other multipart case."""
+
+    def test_intermediate_multipart_ack_regenerates_plain_direct_ack(self, handler):
+        pkt = _make_multipart_ack_packet(remaining=2, path=bytes([LOCAL_HASH, 0xCC]))
+        result = handler.forward_multipart_direct(pkt)
+        assert result is not None
+        ack, delay = result
+        # Wrapper byte stripped, forced to a plain DIRECT ACK.
+        assert ack.get_payload_type() == PAYLOAD_TYPE_ACK
+        assert ack.get_route_type() == ROUTE_TYPE_DIRECT
+        assert bytes(ack.payload) == b"\x4d\xab\xaf\x95"
+        # This node removed from the path.
+        assert list(ack.path) == [0xCC]
+        assert ack.path_len == PathUtils.encode_path_len(1, 1)
+        # (remaining + 1) * 300 ms fragment spacing.
+        assert delay == pytest.approx(0.9)
+
+    def test_delay_scales_with_remaining_fragments(self, handler):
+        pkt = _make_multipart_ack_packet(remaining=0, path=bytes([LOCAL_HASH, 0xCC]))
+        _, delay = handler.forward_multipart_direct(pkt)
+        assert delay == pytest.approx(0.3)
+
+    def test_transport_direct_multipart_ack_regenerates_plain_direct_ack(self, handler):
+        pkt = _make_multipart_ack_packet(crc=b"\x11\x22\x33\x44", route=ROUTE_TYPE_TRANSPORT_DIRECT)
+        ack, _ = handler.forward_multipart_direct(pkt)
+        # MeshCore forces ROUTE_TYPE_DIRECT on the regenerated ACK.
+        assert ack.get_route_type() == ROUTE_TYPE_DIRECT
+        assert bytes(ack.payload) == b"\x11\x22\x33\x44"
+
+    def test_flood_multipart_not_forwarded(self, handler):
+        pkt = _make_multipart_ack_packet(route=ROUTE_TYPE_FLOOD)
+        result = handler.forward_multipart_direct(pkt)
+        assert result is None
+        assert "not direct-routed" in pkt.drop_reason
+
+    def test_final_hop_multipart_not_forwarded(self, handler):
+        pkt = _make_multipart_ack_packet(path=b"")
+        pkt.path_len = 0
+        result = handler.forward_multipart_direct(pkt)
+        assert result is None
+        assert "no path" in pkt.drop_reason
+
+    def test_wrong_next_hop_multipart_dropped(self, handler):
+        pkt = _make_multipart_ack_packet(path=bytes([0xFF, 0xCC]))
+        result = handler.forward_multipart_direct(pkt)
+        assert result is None
+        assert "not for us" in pkt.drop_reason
+
+    def test_non_ack_embedded_type_not_forwarded(self, handler):
+        pkt = _make_multipart_ack_packet(embedded_type=0x02)  # PAYLOAD_TYPE_TXT_MSG
+        result = handler.forward_multipart_direct(pkt)
+        assert result is None
+        assert "unsupported embedded type" in pkt.drop_reason
+
+    def test_short_multipart_ack_not_forwarded(self, handler):
+        pkt = _make_multipart_ack_packet(crc=b"\x01\x02\x03")  # only 4-byte payload
+        result = handler.forward_multipart_direct(pkt)
+        assert result is None
+        assert "unsupported embedded type" in pkt.drop_reason
+
+    def test_duplicate_multipart_ack_dropped(self, handler):
+        handler.forward_multipart_direct(_make_multipart_ack_packet())
+        pkt2 = _make_multipart_ack_packet()  # same CRC + path → same regenerated ACK
+        result = handler.forward_multipart_direct(pkt2)
+        assert result is None
+        assert pkt2.drop_reason == "Duplicate"
+
+    def test_dedup_key_is_regenerated_ack_before_pop(self, handler):
+        """The seen key is the plain-ACK form with the full path, matching
+        MeshCore's hasSeen()-before-removeSelfFromPath() order."""
+        pkt = _make_multipart_ack_packet(path=bytes([LOCAL_HASH, 0xCC]))
+        expected = Packet()
+        expected.header = ROUTE_TYPE_DIRECT | (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT)
+        expected.payload = bytearray(b"\x4d\xab\xaf\x95")
+        expected.payload_len = 4
+        expected.path = bytearray([LOCAL_HASH, 0xCC])
+        expected.path_len = 2
+        key = expected.calculate_packet_hash().hex().upper()
+
+        handler.forward_multipart_direct(pkt)
+        assert key in handler.seen_packets
+
+    def test_marked_do_not_retransmit_multipart_dropped(self, handler):
+        pkt = _make_multipart_ack_packet()
+        pkt.mark_do_not_retransmit()
+        result = handler.forward_multipart_direct(pkt)
+        assert result is None
+
+    def test_process_packet_dispatches_multipart(self, handler):
+        pkt = _make_multipart_ack_packet(remaining=1, path=bytes([LOCAL_HASH, 0xCC]))
+        result = handler.process_packet(pkt, snr=5.0)
+        assert result is not None
+        ack, delay = result
+        assert ack.get_payload_type() == PAYLOAD_TYPE_ACK
+        assert delay == pytest.approx(0.6)
+
+
+# Literal MeshCore-format frames (Packet::writeTo layout: header, optional 4-byte
+# transport codes, encoded path_len, path bytes, payload).  A multipart-ACK
+# payload is `[remaining<<4 | PAYLOAD_TYPE_ACK]` + 4-byte CRC; forwardMultipartDirect
+# strips that wrapper byte and re-emits a plain DIRECT ACK with this node removed
+# from the path.  These are independent wire fixtures — do NOT rebuild them through
+# Packet.write_to()/PathUtils.  remaining=2 in every case → (2+1)*300ms = 0.9s.
+FIRMWARE_MULTIPART_ACK_VECTORS = (
+    # header 0x2A = ROUTE_DIRECT | MULTIPART<<2; path_len 0x02 = 1-byte/2-hop;
+    # path AB CC (AB = us); payload 23=remaining2|ACK + CRC 95AFAB4D.
+    # out: header 0x0E = ACK<<2 | ROUTE_DIRECT; path_len 0x01; path CC; CRC.
+    pytest.param(
+        bytes([0xAB]),
+        bytes.fromhex("2A02ABCC234DABAF95"),
+        bytes.fromhex("0E01CC4DABAF95"),
+        id="direct-1-byte-2-hop",
+    ),
+    # 2-byte path width: path_len 0x42 = 2-byte/2-hop; path ABCD (us) + 1122.
+    pytest.param(
+        bytes([0xAB, 0xCD, 0xEF]),
+        bytes.fromhex("2A42ABCD1122234DABAF95"),
+        bytes.fromhex("0E4111224DABAF95"),
+        id="direct-2-byte-2-hop",
+    ),
+    # header 0x2B = ROUTE_TRANSPORT_DIRECT | MULTIPART<<2; transport codes
+    # 34127856; the regenerated ACK is forced to plain DIRECT (codes dropped).
+    pytest.param(
+        bytes([0xAB]),
+        bytes.fromhex("2B3412785602ABCC234DABAF95"),
+        bytes.fromhex("0E01CC4DABAF95"),
+        id="transport-direct-1-byte-2-hop",
+    ),
+)
+
+
+@pytest.mark.parametrize("local_hash_bytes, wire_in, wire_out", FIRMWARE_MULTIPART_ACK_VECTORS)
+def test_forward_multipart_matches_firmware_wire_vectors(local_hash_bytes, wire_in, wire_out):
+    """Parse a firmware-format multipart frame and assert the exact relayed ACK bytes."""
+    handler = _make_handler_with_hash(local_hash_bytes)
+    pkt = Packet()
+    pkt.read_from(wire_in)
+    result = handler.forward_multipart_direct(pkt)
+    assert result is not None
+    ack, delay = result
+    assert ack.write_to() == wire_out
+    assert delay == pytest.approx(0.9)
+
+
 # ===================================================================
 # 3. process_packet — route dispatch
 # ===================================================================
@@ -1644,6 +1827,36 @@ class TestPacketInjectionRouting:
         sent_pkt = handler.dispatcher.send_packet.call_args.args[0]
         assert bytes(sent_pkt.path) == b"\x44\x55"
 
+    async def test_injected_multipart_ack_forwards_regenerated_plain_ack(self, handler):
+        """An intermediate hop relays a multipart ACK as a plain DIRECT ACK off the wire."""
+        self._prepare_fast_tx(handler)
+
+        pkt = _inject_from_wire(
+            _make_multipart_ack_packet(
+                crc=b"\x4d\xab\xaf\x95", remaining=2, path=bytes([LOCAL_HASH, 0x44])
+            )
+        )
+
+        with patch("repeater.engine.asyncio.sleep", new_callable=AsyncMock):
+            await handler(pkt, {"snr": 3.0, "rssi": -82}, local_transmission=False)
+
+        assert handler.dispatcher.send_packet.call_count == 1
+        sent_pkt = handler.dispatcher.send_packet.call_args.args[0]
+        assert sent_pkt.get_payload_type() == PAYLOAD_TYPE_ACK
+        assert sent_pkt.get_route_type() == ROUTE_TYPE_DIRECT
+        assert bytes(sent_pkt.payload) == b"\x4d\xab\xaf\x95"
+        assert bytes(sent_pkt.path) == b"\x44"
+
+    async def test_injected_multipart_ack_flood_is_dropped(self, handler):
+        """MeshCore never flood-routes multipart; an intermediate hop drops it."""
+        self._prepare_fast_tx(handler)
+        pkt = _inject_from_wire(_make_multipart_ack_packet(route=ROUTE_TYPE_FLOOD, path=b"\x11"))
+
+        with patch("repeater.engine.asyncio.sleep", new_callable=AsyncMock):
+            await handler(pkt, {"snr": 3.0, "rssi": -82}, local_transmission=False)
+
+        assert handler.dispatcher.send_packet.call_count == 0
+
     async def test_direct_for_other_node_is_dropped(self, handler):
         pkt = _inject_from_wire(_make_direct_packet(payload=b"\xaa\xbb", path=b"\xfe\x44"))
         metadata = {"snr": 2.0, "rssi": -90}
@@ -1711,7 +1924,7 @@ class TestPacketInjectionRouting:
         assert len(original["duplicates"]) == 1
         assert original["duplicates"][0]["drop_reason"] == "Duplicate"
 
-    @pytest.mark.parametrize("payload_type", range(16))
+    @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES)
     async def test_all_payload_types_flood_injection_forwards(self, handler, payload_type):
         self._prepare_fast_tx(handler)
         pkt = _inject_from_wire(
@@ -1733,7 +1946,7 @@ class TestPacketInjectionRouting:
         assert sent_pkt.get_payload_type() == payload_type
         assert sent_pkt.path[-1] == LOCAL_HASH
 
-    @pytest.mark.parametrize("payload_type", range(16))
+    @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES)
     async def test_all_payload_types_direct_injection_forwards(self, handler, payload_type):
         self._prepare_fast_tx(handler)
         pkt = _inject_from_wire(
@@ -1755,7 +1968,7 @@ class TestPacketInjectionRouting:
         assert sent_pkt.get_payload_type() == payload_type
         assert bytes(sent_pkt.path) == b"\x44\x55"
 
-    @pytest.mark.parametrize("payload_type", range(16))
+    @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES)
     async def test_all_payload_types_transport_flood_injection_forwards(
         self, handler, payload_type
     ):
@@ -1781,7 +1994,7 @@ class TestPacketInjectionRouting:
         assert sent_pkt.get_payload_type() == payload_type
         assert sent_pkt.transport_codes == [0x1111, 0x2222]
 
-    @pytest.mark.parametrize("payload_type", range(16))
+    @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES)
     async def test_all_payload_types_transport_direct_injection_forwards(
         self, handler, payload_type
     ):