Merge pull request #362 from agessaman/fix/all-the-things

config.yaml.example

@@ -35,6 +35,11 @@ repeater:
   # Enable quality-based packet filtering and adaptive delays
   use_score_for_tx: false
 
+  # Multi-ack redundancy (0 = off, 1 = on; matches MeshCore's multi.acks pref).
+  # When on, a relayed routed ACK is preceded by a MULTIPART-wrapped copy so
+  # ACKs survive lossy links. Also settable via the mesh CLI: set multi.acks
+  multi_acks: 0
+
   # Score threshold for quality monitoring (future use)
   # Currently reserved for potential future features like dashboard alerts,
   # proactive statistics collection, or advanced filtering strategies.

repeater/engine.py

@@ -4,6 +4,7 @@ import logging
 import secrets
 import time
 from collections import OrderedDict, deque
+from enum import Enum
 from typing import Optional, Tuple
 
 from openhop_core.node.handlers.base import BaseHandler
@@ -23,6 +24,7 @@ from openhop_core.protocol.constants import (
     ROUTE_TYPE_TRANSPORT_FLOOD,
 )
 from openhop_core.protocol.packet_utils import (
+    PacketHashingUtils,
     PacketHeaderUtils,
     PathUtils,
     flood_rx_metrics,
@@ -52,6 +54,57 @@ LOOP_DETECT_MAX_COUNTERS = {
 }
 
 
+class DropReason(str, Enum):
+    """Canonical, non-alarming reasons a packet was intentionally not retransmitted.
+
+    Single source of truth shared with the packet router, replacing the string
+    prefixes that used to be duplicated there. Members subclass ``str`` so they
+    compare equal to their text, JSON-serialize, and persist to SQLite as plain
+    strings unchanged. Where the engine still needs a detail suffix (hop counts,
+    loop mode, multipart sub-case) it formats the member into a larger string;
+    the router accepts both the bare member and any string that begins with one.
+    """
+
+    DUPLICATE = "Duplicate"
+    MAX_FLOOD_HOPS = "Max flood hops limit reached"
+    PATH_HOP_COUNT_MAX = "Path hop count at maximum"
+    PATH_EXCEEDS_MAX_SIZE = "Path would exceed MAX_PATH_SIZE"
+    DIRECT_NO_PATH = "Direct: no path"
+    DIRECT_NOT_FOR_US = "Direct: not for us"
+    UNSCOPED_FLOOD_DISABLED = "Unscoped flood policy disabled"
+    TRANSPORT_CODE_NOT_ALLOWED = "Transport code not allowed to flood"
+    FLOOD_LOOP_DETECTED = "FLOOD loop detected"
+    MARKED_DO_NOT_RETRANSMIT = "Marked do not retransmit"
+    REPEAT_DISABLED = "Repeat disabled"
+    NO_TX_MODE = "No TX mode"
+    DUTY_CYCLE_LIMIT = "Duty cycle limit"
+    EMPTY_PAYLOAD = "Empty payload"
+    PATH_TOO_LONG = "Path too long"
+    INVALID_ADVERT = "Invalid advert packet"
+    MULTIPART = "Multipart"
+
+    # Python 3.11+ formats a (str, Enum) member as "DropReason.X" under str()/%s;
+    # return the value so log lines and stored records keep the plain reason text.
+    def __str__(self) -> str:
+        return self.value
+
+
+class ForwardResult(tuple):
+    """A forwarding decision that unpacks as a plain ``(packet, delay_seconds)`` pair.
+
+    ``extras`` carries additional ``(packet, delay_seconds)`` transmissions that
+    accompany the primary packet: MeshCore's routeDirectRecvAcks queues its
+    optional multi-ack redundancy copies ahead of the plain ACK at the same
+    scheduled time, so the caller must create the extra TX tasks before the
+    primary one.
+    """
+
+    def __new__(cls, packet: Packet, delay_s: float, extras=()):
+        result = super().__new__(cls, (packet, delay_s))
+        result.extras = tuple(extras)
+        return result
+
+
 class RepeaterHandler(BaseHandler):
     @staticmethod
     def payload_type() -> int:
@@ -93,6 +146,7 @@ class RepeaterHandler(BaseHandler):
         )
         self.use_score_for_tx = config.get("repeater", {}).get("use_score_for_tx", False)
         self.score_threshold = config.get("repeater", {}).get("score_threshold", 0.3)
+        self.multi_acks = self._normalize_multi_acks(config)
         self.max_flood_hops = config.get("repeater", {}).get("max_flood_hops", 64)
         self.send_advert_interval_hours = config.get("repeater", {}).get(
             "send_advert_interval_hours", 10
@@ -314,6 +368,16 @@ class RepeaterHandler(BaseHandler):
             # Capture the forwarded path (after modification)
             forwarded_path_hashes = fwd_pkt.get_path_hashes_hex()
 
+            # MeshCore queues multi-ack redundancy copies ahead of the primary
+            # ACK at the same scheduled time, so create their TX tasks first.
+            # Each task re-checks the duty-cycle gate before transmitting.
+            extra_tx_tasks = []
+            for extra_pkt, extra_delay in getattr(result, "extras", ()):
+                extra_airtime_ms = self.airtime_mgr.calculate_airtime(extra_pkt.get_raw_length())
+                extra_tx_tasks.append(
+                    await self.schedule_retransmit(extra_pkt, extra_delay, extra_airtime_ms)
+                )
+
             # Check duty-cycle before scheduling TX
             airtime_ms = self.airtime_mgr.calculate_airtime(fwd_pkt.get_raw_length())
 
@@ -368,7 +432,7 @@ class RepeaterHandler(BaseHandler):
                         f"wait={wait_time:.1f}s before retry"
                     )
                     self.dropped_count += 1
-                    drop_reason = "Duty cycle limit"
+                    drop_reason = DropReason.DUTY_CYCLE_LIMIT
             else:
                 tx_task = await self.schedule_retransmit(
                     fwd_pkt, delay, airtime_ms, local_transmission=local_transmission
@@ -399,13 +463,22 @@ class RepeaterHandler(BaseHandler):
                             f"LBT: {lbt_attempts} attempts, {total_lbt_delay:.0f}ms delay, "
                             f"backoffs={lbt_backoff_delays_ms}"
                         )
+
+            # Redundancy copies ride alongside the primary result: collect them
+            # without letting a failed extra change the primary outcome.
+            for extra_task in extra_tx_tasks:
+                try:
+                    if await extra_task:
+                        self.forwarded_count += 1
+                except Exception as e:
+                    logger.warning(f"Multi-ack redundancy TX failed: {e}")
         else:
             self.dropped_count += 1
             # Determine drop reason
             if local_transmission and not allow_local_tx:
-                drop_reason = policy_reason or "No TX mode"
+                drop_reason = policy_reason or DropReason.NO_TX_MODE
             elif not allow_forward:
-                drop_reason = policy_reason or "Repeat disabled"
+                drop_reason = policy_reason or DropReason.REPEAT_DISABLED
             else:
                 # Check if packet has a specific drop reason set by handlers
                 drop_reason = processed_packet.drop_reason or self._get_drop_reason(
@@ -433,7 +506,7 @@ class RepeaterHandler(BaseHandler):
 
         # Set drop reason for duplicates and count flood vs direct dups
         if is_dupe and drop_reason is None:
-            drop_reason = "Duplicate"
+            drop_reason = DropReason.DUPLICATE
         if is_dupe:
             if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD):
                 self.flood_dup_count += 1
@@ -482,7 +555,11 @@ class RepeaterHandler(BaseHandler):
         if self.storage:
             try:
                 # Only skip mqtt for actual invalid/bad packets
-                invalid_reasons = ["Invalid advert packet", "Empty payload", "Path too long"]
+                invalid_reasons = (
+                    DropReason.INVALID_ADVERT,
+                    DropReason.EMPTY_PAYLOAD,
+                    DropReason.PATH_TOO_LONG,
+                )
                 skip_mqtt = drop_reason in invalid_reasons if drop_reason else False
                 self.storage.record_packet(packet_record, skip_mqtt_if_invalid=skip_mqtt)
             except Exception as e:
@@ -631,7 +708,7 @@ class RepeaterHandler(BaseHandler):
             src_hash,
             dst_hash,
             transmitted=False,
-            drop_reason="Duplicate",
+            drop_reason=DropReason.DUPLICATE,
             is_duplicate=True,
             packet_hash=pkt_hash_full,
         )
@@ -795,13 +872,13 @@ class RepeaterHandler(BaseHandler):
     def _get_drop_reason(self, packet: Packet, packet_hash: Optional[str] = None) -> str:
 
         if self.is_duplicate(packet, packet_hash=packet_hash):
-            return "Duplicate"
+            return DropReason.DUPLICATE
 
         if not packet or not packet.payload:
-            return "Empty payload"
+            return DropReason.EMPTY_PAYLOAD
 
         if len(packet.path or []) > MAX_PATH_SIZE:
-            return "Path too long"
+            return DropReason.PATH_TOO_LONG
 
         route_type = packet.header & PH_ROUTE_MASK
 
@@ -811,15 +888,15 @@ class RepeaterHandler(BaseHandler):
                 "unscoped_flood_allow", self.config.get("mesh", {}).get("global_flood_allow", True)
             )
             if not unscoped_flood_allow:
-                return "Unscoped flood policy disabled"
+                return DropReason.UNSCOPED_FLOOD_DISABLED
 
         if route_type == ROUTE_TYPE_DIRECT:
             hash_size = packet.get_path_hash_size()
             if not packet.path or len(packet.path) < hash_size:
-                return "Direct: no path"
+                return DropReason.DIRECT_NO_PATH
             next_hop = bytes(packet.path[:hash_size])
             if next_hop != self.local_hash_bytes[:hash_size]:
-                return "Direct: not for us"
+                return DropReason.DIRECT_NOT_FOR_US
 
         # Default reason
         return "Unknown"
@@ -850,7 +927,7 @@ class RepeaterHandler(BaseHandler):
     def validate_packet(self, packet: Packet) -> Tuple[bool, str]:
 
         if not packet or not packet.payload:
-            return False, "Empty payload"
+            return False, DropReason.EMPTY_PAYLOAD
 
         if packet.get_path_hash_size() > 3:
             return False, "Reserved path hash size is invalid"
@@ -863,6 +940,16 @@ class RepeaterHandler(BaseHandler):
 
         return True, ""
 
+    @staticmethod
+    def _normalize_multi_acks(config: dict) -> int:
+        """Read ``repeater.multi_acks``, constrained to 0..1 like MeshCore's
+        CommonCLI prefs load."""
+        try:
+            value = int(config.get("repeater", {}).get("multi_acks", 0))
+        except (TypeError, ValueError):
+            return 0
+        return max(0, min(1, value))
+
     def _normalize_loop_detect_mode(self, mode) -> str:
         if isinstance(mode, str):
             normalized = mode.strip().lower()
@@ -1006,7 +1093,7 @@ class RepeaterHandler(BaseHandler):
         if packet.is_marked_do_not_retransmit():
             # Check if packet has custom drop reason
             if not packet.drop_reason:
-                packet.drop_reason = "Marked do not retransmit"
+                packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT
             return None
 
         # Check unscoped flood policy
@@ -1016,24 +1103,24 @@ class RepeaterHandler(BaseHandler):
         route_type = packet.header & PH_ROUTE_MASK
         if route_type == ROUTE_TYPE_FLOOD:
             if not unscoped_flood_allow:
-                packet.drop_reason = "Unscoped flood policy disabled"
+                packet.drop_reason = DropReason.UNSCOPED_FLOOD_DISABLED
                 return None
 
         # Check transport scopes flood policy
         if route_type == ROUTE_TYPE_TRANSPORT_FLOOD:
             allowed, check_reason = self._check_transport_codes(packet)
             if not allowed:
-                packet.drop_reason = "Transport code not allowed to flood"
+                packet.drop_reason = DropReason.TRANSPORT_CODE_NOT_ALLOWED
                 return None
 
         mode = self._get_loop_detect_mode()
         if self._is_flood_looped(packet, mode):
-            packet.drop_reason = f"FLOOD loop detected ({mode})"
+            packet.drop_reason = f"{DropReason.FLOOD_LOOP_DETECTED} ({mode})"
             return None
 
         # Suppress duplicates — pass pre-computed hash to avoid a second SHA-256.
         if self.is_duplicate(packet, packet_hash=packet_hash):
-            packet.drop_reason = "Duplicate"
+            packet.drop_reason = DropReason.DUPLICATE
             return None
 
         if packet.path is None:
@@ -1045,17 +1132,17 @@ class RepeaterHandler(BaseHandler):
         hop_count = packet.get_path_hash_count()
 
         if self.max_flood_hops > 0 and hop_count >= self.max_flood_hops:
-            packet.drop_reason = f"Max flood hops limit reached ({hop_count}/{self.max_flood_hops})"
+            packet.drop_reason = f"{DropReason.MAX_FLOOD_HOPS} ({hop_count}/{self.max_flood_hops})"
             return None
 
         # path_len encodes hop count in 6 bits (0-63); adding ourselves must not exceed 63
         if hop_count >= 63:
-            packet.drop_reason = "Path hop count at maximum (63), cannot append"
+            packet.drop_reason = f"{DropReason.PATH_HOP_COUNT_MAX} (63), cannot append"
             return None
 
         # Check path won't exceed MAX_PATH_SIZE after append
         if (hop_count + 1) * hash_size > MAX_PATH_SIZE:
-            packet.drop_reason = "Path would exceed MAX_PATH_SIZE"
+            packet.drop_reason = DropReason.PATH_EXCEEDS_MAX_SIZE
             return None
 
         self.mark_seen(packet, packet_hash=packet_hash)
@@ -1082,7 +1169,7 @@ class RepeaterHandler(BaseHandler):
         # Check if packet is marked do-not-retransmit
         if packet.is_marked_do_not_retransmit():
             if not packet.drop_reason:
-                packet.drop_reason = "Marked do not retransmit"
+                packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT
             return None
 
         hash_size = packet.get_path_hash_size()
@@ -1090,17 +1177,17 @@ class RepeaterHandler(BaseHandler):
 
         # Check if we're the next hop
         if not packet.path or len(packet.path) < hash_size:
-            packet.drop_reason = "Direct: no path"
+            packet.drop_reason = DropReason.DIRECT_NO_PATH
             return None
 
         next_hop = bytes(packet.path[:hash_size])
         if next_hop != self.local_hash_bytes[:hash_size]:
-            packet.drop_reason = "Direct: not for us"
+            packet.drop_reason = DropReason.DIRECT_NOT_FOR_US
             return None
 
         # Suppress duplicates — pass pre-computed hash to avoid a second SHA-256.
         if self.is_duplicate(packet, packet_hash=packet_hash):
-            packet.drop_reason = "Duplicate"
+            packet.drop_reason = DropReason.DUPLICATE
             return None
 
         self.mark_seen(packet, packet_hash=packet_hash)
@@ -1116,8 +1203,8 @@ class RepeaterHandler(BaseHandler):
     MULTIPART_ACK_SPACING_MS = 300
 
     def forward_multipart_direct(
-        self, packet: Packet, packet_hash: Optional[str] = None
-    ) -> Optional[Tuple[Packet, float]]:
+        self, packet: Packet, snr: float = 0.0, packet_hash: Optional[str] = None
+    ) -> Optional["ForwardResult"]:
         """Forward a MULTIPART packet the way MeshCore's forwardMultipartDirect does.
 
         MeshCore never relays the multipart wrapper as an ordinary directed
@@ -1125,7 +1212,8 @@ class RepeaterHandler(BaseHandler):
         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
+        type — is dropped.  Returns a ``ForwardResult`` — ``(ack_packet,
+        delay_seconds)`` plus any multi-ack redundancy ``extras`` — when a
         regenerated ACK should be sent, otherwise ``None``.
 
         INVARIANT: purely synchronous, mirroring flood_forward / direct_forward —
@@ -1134,7 +1222,7 @@ class RepeaterHandler(BaseHandler):
         # 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"
+            packet.drop_reason = f"{DropReason.MULTIPART}: not direct-routed"
             return None
 
         valid, reason = self.validate_packet(packet)
@@ -1143,7 +1231,7 @@ class RepeaterHandler(BaseHandler):
             return None
 
         if packet.is_marked_do_not_retransmit():
-            packet.drop_reason = packet.drop_reason or "Marked do not retransmit"
+            packet.drop_reason = packet.drop_reason or DropReason.MARKED_DO_NOT_RETRANSMIT
             return None
 
         payload = packet.payload or b""
@@ -1152,7 +1240,7 @@ class RepeaterHandler(BaseHandler):
         # 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"
+            packet.drop_reason = f"{DropReason.MULTIPART}: unsupported embedded type"
             return None
 
         hash_size = packet.get_path_hash_size()
@@ -1161,33 +1249,127 @@ class RepeaterHandler(BaseHandler):
         # 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"
+            packet.drop_reason = DropReason.DIRECT_NO_PATH
             return None
         if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]:
-            packet.drop_reason = "Direct: not for us"
+            packet.drop_reason = DropReason.DIRECT_NOT_FOR_US
             return None
 
+        # Dedupe before regenerating, on MeshCore's exact seen key: hasSeen()
+        # runs on a copy that keeps the MULTIPART payload type but carries the
+        # unwrapped ACK payload (the `remaining` count is stripped).  Keeping
+        # that key distinct from the plain ACK's is what lets a multi-ack
+        # redundancy pair — MULTIPART copy plus plain ACK — survive every hop
+        # instead of the second half being swallowed as a duplicate.  The
+        # pre-computed hash belongs to the wrapper as received, so recompute.
+        seen_key = (
+            PacketHashingUtils.calculate_packet_hash(
+                PAYLOAD_TYPE_MULTIPART, packet.path_len, bytes(payload[1:])
+            )
+            .hex()
+            .upper()
+        )
+        if self.is_duplicate(packet, packet_hash=seen_key):
+            packet.drop_reason = DropReason.DUPLICATE
+            return None
+        self.mark_seen(packet, packet_hash=seen_key)
+
         # 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.
+        # drop the multipart header byte and force the ACK payload/route type, then
+        # remove this node from the path (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.transport_codes = [0, 0]
         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
+        extras, delay_ms = self._multi_ack_extras(
+            packet, float((remaining + 1) * self.MULTIPART_ACK_SPACING_MS), snr
+        )
+        return ForwardResult(packet, delay_ms / 1000.0, extras)
+
+    def _multi_ack_extras(
+        self, ack_packet: Packet, base_delay_ms: float, snr: float
+    ) -> Tuple[list, float]:
+        """Build MeshCore's optional multi-ack redundancy copies for a relayed ACK.
+
+        With ``multi_acks`` enabled, routeDirectRecvAcks precedes the plain ACK
+        with a MULTIPART-wrapped copy (``remaining<<4 | ACK`` prefix byte) on the
+        already self-removed path, pushing the accumulated delay out by a direct
+        retransmit delay + 300 ms per copy; the plain ACK itself slides to the
+        final accumulated delay.  Returns ``(extras, plain_ack_delay_ms)``.
+        """
+        extra = self.multi_acks
+        extras = []
+        delay_ms = base_delay_ms
+        while extra > 0:
+            delay_ms += (
+                self._calculate_tx_delay(ack_packet, snr) * 1000.0 + self.MULTIPART_ACK_SPACING_MS
+            )
+            wrapped = Packet()
+            wrapped.header = (PAYLOAD_TYPE_MULTIPART << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT
+            wrapped.payload = bytearray(
+                bytes([(extra << 4) | PAYLOAD_TYPE_ACK]) + bytes(ack_packet.payload)
+            )
+            wrapped.payload_len = len(wrapped.payload)
+            wrapped.path = bytearray(ack_packet.path)
+            wrapped.path_len = ack_packet.path_len
+            extras.append((wrapped, delay_ms / 1000.0))
+            extra -= 1
+        return extras, delay_ms
+
+    def forward_routed_ack(
+        self, packet: Packet, snr: float = 0.0, packet_hash: Optional[str] = None
+    ) -> Optional["ForwardResult"]:
+        """Relay a routed ACK the way MeshCore's routeDirectRecvAcks does.
+
+        An ACK at an intermediate direct hop is not repeated verbatim: MeshCore
+        regenerates it (createAck) as a plain DIRECT ACK — transport codes
+        dropped, every other header bit cleared — removes this node from the
+        path, and transmits it with zero retransmit delay rather than the
+        generic direct forwarding delay.  With ``multi_acks`` enabled, the plain
+        ACK is preceded by a MULTIPART-wrapped redundancy copy and both slide
+        out by a direct retransmit delay + 300 ms.
+
+        INVARIANT: purely synchronous, mirroring direct_forward — the
+        is_duplicate + mark_seen pair must stay atomic within the event loop.
+        """
+        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 DropReason.MARKED_DO_NOT_RETRANSMIT
+            return None
+
+        hash_size = packet.get_path_hash_size()
+        hop_count = packet.get_path_hash_count()
+
+        # A final-hop ACK (no remaining path) is consumed locally, never relayed.
+        if not packet.path or len(packet.path) < hash_size:
+            packet.drop_reason = DropReason.DIRECT_NO_PATH
+            return None
+        if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]:
+            packet.drop_reason = DropReason.DIRECT_NOT_FOR_US
+            return None
+
+        # The packet hash covers only the payload type and payload, so the
+        # received form and the regenerated plain ACK share one key — matching
+        # MeshCore's hasSeen() on the incoming ACK before removeSelfFromPath().
+        if self.is_duplicate(packet, packet_hash=packet_hash):
+            packet.drop_reason = DropReason.DUPLICATE
+            return None
+        self.mark_seen(packet, packet_hash=packet_hash)
+
+        packet.header = (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT
+        packet.transport_codes = [0, 0]
+        packet.path = bytearray(packet.path[hash_size:])
+        packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1)
+
+        extras, delay_ms = self._multi_ack_extras(packet, 0.0, snr)
+        return ForwardResult(packet, delay_ms / 1000.0, extras)
 
     @staticmethod
     def calculate_packet_score(snr: float, packet_len: int, spreading_factor: int = 8) -> float:
@@ -1257,8 +1439,18 @@ class RepeaterHandler(BaseHandler):
         # 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)
+        payload_type = packet.get_payload_type()
+        if payload_type == PAYLOAD_TYPE_MULTIPART:
+            return self.forward_multipart_direct(packet, snr, packet_hash=packet_hash)
+
+        # Routed ACKs also get their own MeshCore branch: an intermediate direct
+        # hop re-emits the ACK immediately (routeDirectRecvAcks) instead of the
+        # delayed generic direct forward.  Flood ACKs stay on the generic path.
+        if payload_type == PAYLOAD_TYPE_ACK and route_type in (
+            ROUTE_TYPE_DIRECT,
+            ROUTE_TYPE_TRANSPORT_DIRECT,
+        ):
+            return self.forward_routed_ack(packet, snr, 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)
@@ -1600,6 +1792,7 @@ class RepeaterHandler(BaseHandler):
             repeater_config = self.config.get("repeater", {})
             self.use_score_for_tx = repeater_config.get("use_score_for_tx", False)
             self.score_threshold = repeater_config.get("score_threshold", 0.3)
+            self.multi_acks = self._normalize_multi_acks(self.config)
             self.send_advert_interval_hours = repeater_config.get("send_advert_interval_hours", 10)
             self.cache_ttl = repeater_config.get("cache_ttl", 60)
             self.max_flood_hops = repeater_config.get("max_flood_hops", 64)

repeater/handler_helpers/advert.py

@@ -15,6 +15,8 @@ from typing import Dict, Optional, Tuple
 
 from openhop_core.node.handlers.advert import AdvertHandler
 
+from repeater.engine import DropReason
+
 logger = logging.getLogger("AdvertHelper")
 
 
@@ -551,7 +553,7 @@ class AdvertHelper:
             if not advert_data or not advert_data.get("valid"):
                 logger.warning("Invalid advert packet received, dropping.")
                 packet.mark_do_not_retransmit()
-                packet.drop_reason = "Invalid advert packet"
+                packet.drop_reason = DropReason.INVALID_ADVERT
                 return
 
             # Extract data from parsed advert

repeater/packet_router.py

@@ -5,7 +5,6 @@ import time
 from openhop_core.node.handlers.ack import AckHandler
 from openhop_core.node.handlers.advert import AdvertHandler
 from openhop_core.node.handlers.control import ControlHandler
-from openhop_core.node.handlers.group_text import GroupTextHandler
 from openhop_core.node.handlers.login_response import LoginResponseHandler
 from openhop_core.node.handlers.login_server import LoginServerHandler
 from openhop_core.node.handlers.multipart import MultipartAckHandler
@@ -16,11 +15,13 @@ 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,
     PH_ROUTE_MASK,
     ROUTE_TYPE_DIRECT,
     ROUTE_TYPE_TRANSPORT_DIRECT,
 )
 
+from repeater.engine import DropReason
 from repeater.policy_engine import PolicyDecision, PolicyEngine
 
 logger = logging.getLogger("PacketRouter")
@@ -29,27 +30,11 @@ logger = logging.getLogger("PacketRouter")
 # so the client is not spammed with duplicate telemetry when the mesh delivers multiple copies.
 _COMPANION_DEDUPE_TTL_SEC = 60.0
 
-# Drop reasons that are normal policy outcomes and should not be warning-level.
-# TODO: create Enum in engine for drop reasons and use it here and in engine instead of string matching.
-_EXPECTED_DROP_REASON_PREFIXES = (
-    "Duplicate",
-    "Max flood hops limit reached",
-    "Path hop count at maximum",
-    "Path would exceed MAX_PATH_SIZE",
-    "Direct: no path",
-    "Direct: not for us",
-    "Unscoped flood policy disabled",
-    "Transport code not allowed to flood",
-    "FLOOD loop detected",
-    "Marked do not retransmit",
-    "Repeat disabled",
-    "No TX mode",
-    "Duty cycle limit",
-    "Empty payload",
-    "Path too long",
-    "Invalid advert packet",
-    "Multipart",
-)
+# Normal policy/forwarding drop outcomes (logged at debug, not warning). The engine
+# is the single source of truth via DropReason; string-form acceptance is retained
+# because records and the recent_packets fallback still carry detailed text variants
+# (e.g. "Max flood hops limit reached (5/5)").
+_EXPECTED_DROP_REASONS = frozenset(DropReason)
 
 
 def _companion_dedup_key(packet) -> str | None:
@@ -77,10 +62,14 @@ def _is_direct_intermediate_hop(packet) -> bool:
     )
 
 
-def _is_expected_drop_reason(reason: str | None) -> bool:
+def _is_expected_drop_reason(reason) -> bool:
+    # Membership is the fast path for a DropReason member or its exact text; the
+    # startswith scan covers detailed variants that embed a suffix.
+    if reason in _EXPECTED_DROP_REASONS:
+        return True
     if not isinstance(reason, str) or not reason:
         return False
-    return any(reason.startswith(prefix) for prefix in _EXPECTED_DROP_REASON_PREFIXES)
+    return any(reason.startswith(expected) for expected in _EXPECTED_DROP_REASONS)
 
 
 def _drop_reason_from_recent_packets(handler, packet) -> str | None:
@@ -181,11 +170,28 @@ class PacketRouter:
             if exc is not None:
                 logger.error("_route_packet raised: %s", exc, exc_info=exc)
 
-    def _should_deliver_path_to_companions(self, packet) -> bool:
-        """Return True if this PATH/protocol-response should be delivered to companions (first of duplicates)."""
+    def _was_delivered_to_companions(self, packet) -> bool:
+        """Pure check: True if this PATH/protocol-response was already delivered (unexpired).
+
+        Does not mutate delivery state, so callers can deliver first and only record
+        the packet as delivered once a bridge has actually received it.
+        """
         key = _companion_dedup_key(packet)
         if not key:
-            return True
+            return False
+        expiry = self._companion_delivered.get(key)
+        return expiry is not None and expiry > time.time()
+
+    def _mark_delivered_to_companions(self, packet) -> None:
+        """Record that this PATH/protocol-response reached companions.
+
+        Called only after a bridge has received the packet, so a delivery that
+        raised for every bridge is retried on the next copy instead of being
+        suppressed for the full dedupe TTL.
+        """
+        key = _companion_dedup_key(packet)
+        if not key:
+            return
         now = time.time()
         # Prune expired entries only when the dict grows large, avoiding a full
         # dict comprehension on every packet.  200 entries × 60 s TTL means a
@@ -196,9 +202,13 @@ class PacketRouter:
             self._companion_delivered = {
                 k: v for k, v in self._companion_delivered.items() if v > now
             }
-        if key in self._companion_delivered:
-            return False
         self._companion_delivered[key] = now + _COMPANION_DEDUPE_TTL_SEC
+
+    def _should_deliver_path_to_companions(self, packet) -> bool:
+        """Atomic check-and-mark: True on the first of duplicate PATH/protocol-responses."""
+        if self._was_delivered_to_companions(packet):
+            return False
+        self._mark_delivered_to_companions(packet)
         return True
 
     def _policy_companion_decision(self, packet, metadata: dict) -> PolicyDecision | None:
@@ -263,6 +273,25 @@ class PacketRouter:
             return {}
         return companion_bridges
 
+    async def _fan_out_to_bridges(self, packet, bridges, *, context: str) -> bool:
+        """Offer packet to each bridge; True if any bridge authenticated it.
+
+        Accepts a dict of bridges — pass a single-entry dict for targeted delivery
+        to the bridge that owns ``dest_hash``. A bridge that raises is logged and
+        skipped; ``result.authenticated`` is read directly (every bridge returns a
+        HandlerResult) so a broken contract surfaces instead of being hidden.
+        """
+        authenticated = False
+        for bridge in bridges.values():
+            try:
+                result = await bridge.process_received_packet(packet)
+            except Exception as e:
+                logger.debug("Companion bridge %s error: %s", context, e)
+                continue
+            if result.authenticated is True:
+                authenticated = True
+        return authenticated
+
     async def _consume_via_local_candidates(
         self, packet, metadata: dict, dest_hash, helper, process_method_name: str
     ) -> bool:
@@ -537,11 +566,7 @@ class PacketRouter:
             # Also feed adverts to companion bridges (for contact/path updates),
             # but keep policy drop final just like the other companion paths.
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
-            for bridge in companion_bridges.values():
-                try:
-                    await bridge.process_received_packet(packet)
-                except Exception as e:
-                    logger.debug(f"Companion bridge advert error: {e}")
+            await self._fan_out_to_bridges(packet, companion_bridges, context="advert")
 
         elif payload_type == LoginServerHandler.payload_type():
             # Route ANON_REQ/login to the local identity that owns it. The on-air
@@ -566,11 +591,7 @@ class PacketRouter:
             # 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)
-            for bridge in companion_bridges.values():
-                try:
-                    await bridge.process_received_packet(packet)
-                except Exception as e:
-                    logger.debug(f"Companion bridge ACK error: {e}")
+            await self._fan_out_to_bridges(packet, companion_bridges, context="ACK")
 
         elif payload_type == MultipartAckHandler.payload_type():
             # MULTIPART ACK wrapper: low nibble of first byte is embedded payload type.
@@ -610,18 +631,22 @@ class PacketRouter:
             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:
-                if self._should_deliver_path_to_companions(packet):
-                    result = await companion_bridges[dest_hash].process_received_packet(packet)
-                    consumed = consumed or getattr(result, "authenticated", False) is True
-            elif companion_bridges and self._should_deliver_path_to_companions(packet):
+                if not self._was_delivered_to_companions(packet):
+                    consumed = (
+                        await self._fan_out_to_bridges(
+                            packet, {dest_hash: companion_bridges[dest_hash]}, context="PATH"
+                        )
+                        or consumed
+                    )
+                    self._mark_delivered_to_companions(packet)
+            elif companion_bridges and not self._was_delivered_to_companions(packet):
                 # Dest not in bridges: path-return with ephemeral dest (e.g. multi-hop login).
                 # Deliver to all bridges; each will try to decrypt and ignore if not relevant.
-                for bridge in companion_bridges.values():
-                    try:
-                        result = await bridge.process_received_packet(packet)
-                        consumed = consumed or getattr(result, "authenticated", False) is True
-                    except Exception as e:
-                        logger.debug(f"Companion bridge PATH error: {e}")
+                consumed = (
+                    await self._fan_out_to_bridges(packet, companion_bridges, context="PATH")
+                    or consumed
+                )
+                self._mark_delivered_to_companions(packet)
                 logger.debug(
                     "PATH dest=0x%02x (anon) delivered to %d bridge(s) for matching",
                     dest_hash or 0,
@@ -643,23 +668,15 @@ class PacketRouter:
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
             local_hash = getattr(self.daemon, "local_hash", None)
             if dest_hash is not None and dest_hash in companion_bridges:
-                try:
-                    result = await companion_bridges[dest_hash].process_received_packet(packet)
-                    consumed = consumed or getattr(result, "authenticated", False) is True
-                    logger.info(
-                        "RESPONSE dest=0x%02x delivered to companion bridge",
-                        dest_hash,
-                    )
-                except Exception as e:
-                    logger.debug(f"Companion bridge RESPONSE error: {e}")
+                consumed = await self._fan_out_to_bridges(
+                    packet, {dest_hash: companion_bridges[dest_hash]}, context="RESPONSE"
+                )
+                logger.info("RESPONSE dest=0x%02x delivered to companion bridge", dest_hash)
             elif dest_hash == local_hash and companion_bridges:
                 # Response addressed to this repeater (e.g. path-based reply to first hop)
-                for bridge in companion_bridges.values():
-                    try:
-                        result = await bridge.process_received_packet(packet)
-                        consumed = consumed or getattr(result, "authenticated", False) is True
-                    except Exception as e:
-                        logger.debug(f"Companion bridge RESPONSE error: {e}")
+                consumed = await self._fan_out_to_bridges(
+                    packet, companion_bridges, context="RESPONSE"
+                )
                 logger.info(
                     "RESPONSE dest=0x%02x (local) delivered to %d companion bridge(s)",
                     dest_hash,
@@ -669,12 +686,9 @@ class PacketRouter:
                 # Dest not in bridges and not local: likely ANON_REQ response (dest = ephemeral
                 # sender hash). Deliver to all bridges; each will try to decrypt and ignore if
                 # not relevant (firmware-like behavior, works with multiple companion bridges).
-                for bridge in companion_bridges.values():
-                    try:
-                        result = await bridge.process_received_packet(packet)
-                        consumed = consumed or getattr(result, "authenticated", False) is True
-                    except Exception as e:
-                        logger.debug(f"Companion bridge RESPONSE error: {e}")
+                consumed = await self._fan_out_to_bridges(
+                    packet, companion_bridges, context="RESPONSE"
+                )
                 logger.debug(
                     "RESPONSE dest=0x%02x (anon) delivered to %d bridge(s) for matching",
                     dest_hash or 0,
@@ -691,23 +705,16 @@ class PacketRouter:
 
         elif payload_type == ProtocolResponseHandler.payload_type():
             # PAYLOAD_TYPE_PATH (0x08): protocol responses (telemetry, binary, etc.).
-            # Deliver at most once per logical packet so the client is not spammed with duplicates.
-            # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop.
+            # Deliver at most once per logical packet so the client is not spammed with duplicates,
+            # but always deliver at a final hop (we are the destination). Do not set
+            # processed_by_injection for a middle hop so the packet still reaches engine forwarding.
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
-            if companion_bridges and self._should_deliver_path_to_companions(packet):
-                for bridge in companion_bridges.values():
-                    try:
-                        await bridge.process_received_packet(packet)
-                    except Exception as e:
-                        logger.debug(f"Companion bridge RESPONSE error: {e}")
-            if companion_bridges and _is_direct_final_hop(packet):
-                # DIRECT with empty path: we're the final hop; ensure delivery to all bridges (anon)
-                if not self._should_deliver_path_to_companions(packet):
-                    for bridge in companion_bridges.values():
-                        try:
-                            await bridge.process_received_packet(packet)
-                        except Exception as e:
-                            logger.debug(f"Companion bridge RESPONSE (final hop) error: {e}")
+            final_hop = _is_direct_final_hop(packet)
+            if companion_bridges and (final_hop or not self._was_delivered_to_companions(packet)):
+                await self._fan_out_to_bridges(packet, companion_bridges, context="RESPONSE")
+                self._mark_delivered_to_companions(packet)
+            if companion_bridges and final_hop:
+                # DIRECT with empty path: we're the final hop, so consume after delivery.
                 processed_by_injection = True
                 self._record_for_ui(packet, metadata)
 
@@ -732,36 +739,22 @@ class PacketRouter:
                 if companion_bridges and _is_direct_final_hop(packet):
                     # OpenHop release hygiene: an empty-path DIRECT has no next hop,
                     # so consume after offering it to bridges even without MAC ownership.
-                    for bridge in companion_bridges.values():
-                        try:
-                            await bridge.process_received_packet(packet)
-                        except Exception as e:
-                            logger.debug(f"Companion bridge REQ (final hop) error: {e}")
+                    await self._fan_out_to_bridges(packet, companion_bridges, context="REQ")
                     processed_by_injection = True
                     self._record_for_ui(packet, metadata)
 
-        elif payload_type == GroupTextHandler.payload_type():
+        elif payload_type == PAYLOAD_TYPE_GRP_TXT:
             # GRP_TXT: pass to all companions (they filter by channel); still forward.
             # Policy drop is final and blocks companion delivery.
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
-            if companion_bridges:
-                for bridge in companion_bridges.values():
-                    try:
-                        await bridge.process_received_packet(packet)
-                    except Exception as e:
-                        logger.debug(f"Companion bridge GRP_TXT error: {e}")
+            await self._fan_out_to_bridges(packet, companion_bridges, context="GRP_TXT")
 
         elif payload_type == PAYLOAD_TYPE_GRP_DATA:
             # MeshCore forwards direct packets with remaining hops before payload
             # handling. Otherwise, companions authenticate and filter channels.
             if not _is_direct_intermediate_hop(packet):
                 companion_bridges = self._companion_bridges_for_packet(packet, metadata)
-                if companion_bridges:
-                    for bridge in companion_bridges.values():
-                        try:
-                            await bridge.process_received_packet(packet)
-                        except Exception as e:
-                            logger.debug(f"Companion bridge GRP_DATA error: {e}")
+                await self._fan_out_to_bridges(packet, companion_bridges, context="GRP_DATA")
 
         # Only pass to repeater engine if not already processed by injection
         # Skip engine for packets we injected for TX (already sent; avoid double-send/double-count)

tests/test_engine.py

@@ -38,6 +38,13 @@ LOCAL_HASH = 0xAB  # repeater's own 1-byte path hash
 # (forward_multipart_direct), covered by TestForwardMultipartDirect.
 _GENERIC_FORWARD_PAYLOAD_TYPES = [pt for pt in range(16) if pt != PAYLOAD_TYPE_MULTIPART]
 
+# On direct routes, ACK also leaves the generic path — routed ACKs are
+# regenerated by forward_routed_ack (TestForwardRoutedAck); flood ACKs stay
+# generic.
+_GENERIC_DIRECT_FORWARD_PAYLOAD_TYPES = [
+    pt for pt in _GENERIC_FORWARD_PAYLOAD_TYPES if pt != PAYLOAD_TYPE_ACK
+]
+
 
 def _make_config(**overrides) -> dict:
     """Return a minimal valid config dict for RepeaterHandler."""
@@ -477,21 +484,47 @@ class TestForwardMultipartDirect:
         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()
+    def test_dedup_key_keeps_multipart_type_with_unwrapped_payload(self, handler):
+        """The seen key matches MeshCore's hasSeen() copy: MULTIPART payload
+        type over the unwrapped ACK payload, checked before removeSelfFromPath.
+        The `remaining` count is stripped, so re-wraps of the same ACK dedupe."""
+        from openhop_core.protocol.packet_utils import PacketHashingUtils
+
+        pkt = _make_multipart_ack_packet(remaining=2, path=bytes([LOCAL_HASH, 0xCC]))
+        key = (
+            PacketHashingUtils.calculate_packet_hash(
+                PAYLOAD_TYPE_MULTIPART, pkt.path_len, b"\x4d\xab\xaf\x95"
+            )
+            .hex()
+            .upper()
+        )
 
         handler.forward_multipart_direct(pkt)
         assert key in handler.seen_packets
 
+        # A re-wrap of the same ACK with a different remaining count is a duplicate.
+        rewrap = _make_multipart_ack_packet(remaining=1, path=bytes([LOCAL_HASH, 0xCC]))
+        assert handler.forward_multipart_direct(rewrap) is None
+        assert rewrap.drop_reason == "Duplicate"
+
+    def test_multipart_relay_does_not_block_plain_ack_relay(self, handler):
+        """A multi-ack redundancy pair (MULTIPART copy + plain ACK) must survive
+        the hop: MeshCore's seen keys differ by payload type, so relaying the
+        MULTIPART copy first must not swallow the plain ACK as a duplicate."""
+        wrapper = _make_multipart_ack_packet(remaining=1, path=bytes([LOCAL_HASH, 0xCC]))
+        assert handler.forward_multipart_direct(wrapper) is not None
+
+        plain = _make_direct_packet(
+            payload=b"\x4d\xab\xaf\x95",
+            path=bytes([LOCAL_HASH, 0xCC]),
+            payload_type=PAYLOAD_TYPE_ACK,
+        )
+        result = handler.forward_routed_ack(plain)
+        assert result is not None
+        ack, delay = result
+        assert delay == pytest.approx(0.0)
+        assert bytes(ack.payload) == b"\x4d\xab\xaf\x95"
+
     def test_marked_do_not_retransmit_multipart_dropped(self, handler):
         pkt = _make_multipart_ack_packet()
         pkt.mark_do_not_retransmit()
@@ -554,6 +587,259 @@ def test_forward_multipart_matches_firmware_wire_vectors(local_hash_bytes, wire_
     assert delay == pytest.approx(0.9)
 
 
+# ===================================================================
+# 2c. forward_routed_ack — MeshCore routeDirectRecvAcks parity
+# ===================================================================
+
+
+def _make_direct_ack_packet(
+    crc: bytes = b"\x4d\xab\xaf\x95",
+    path: bytes = None,
+    route: int = ROUTE_TYPE_DIRECT,
+    transport_codes=None,
+) -> Packet:
+    """Build a routed ACK: 4-byte CRC payload on a direct route."""
+    if path is None:
+        path = bytes([LOCAL_HASH, 0xCC])
+    pkt = Packet()
+    pkt.header = route | (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT)
+    pkt.payload = bytearray(crc)
+    pkt.payload_len = len(pkt.payload)
+    pkt.path = bytearray(path)
+    pkt.path_len = len(path)
+    if transport_codes is not None:
+        pkt.transport_codes = list(transport_codes)
+    return pkt
+
+
+class TestForwardRoutedAck:
+    """forward_routed_ack mirrors MeshCore: regenerate the ACK as a plain DIRECT
+    packet and relay it immediately, never through the generic delayed forward."""
+
+    def test_intermediate_ack_relays_with_zero_delay(self, handler):
+        pkt = _make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC]))
+        result = handler.forward_routed_ack(pkt)
+        assert result is not None
+        ack, delay = result
+        assert delay == pytest.approx(0.0)
+        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)
+
+    def test_transport_direct_ack_regenerated_as_plain_direct(self, handler):
+        pkt = _make_direct_ack_packet(
+            route=ROUTE_TYPE_TRANSPORT_DIRECT, transport_codes=(0x1234, 0x5678)
+        )
+        result = handler.forward_routed_ack(pkt)
+        assert result is not None
+        ack, delay = result
+        assert delay == pytest.approx(0.0)
+        # createAck rebuilds the header: plain DIRECT, transport codes dropped.
+        assert ack.get_route_type() == ROUTE_TYPE_DIRECT
+        assert not ack.has_transport_codes()
+        assert ack.transport_codes == [0, 0]
+
+    def test_final_hop_ack_not_relayed(self, handler):
+        pkt = _make_direct_ack_packet(path=b"")
+        pkt.path_len = 0
+        result = handler.forward_routed_ack(pkt)
+        assert result is None
+        assert "no path" in pkt.drop_reason
+
+    def test_wrong_next_hop_ack_dropped(self, handler):
+        pkt = _make_direct_ack_packet(path=bytes([0xFE, 0xCC]))
+        result = handler.forward_routed_ack(pkt)
+        assert result is None
+        assert "not for us" in pkt.drop_reason
+
+    def test_duplicate_ack_dropped(self, handler):
+        handler.forward_routed_ack(_make_direct_ack_packet())
+        pkt2 = _make_direct_ack_packet()
+        result = handler.forward_routed_ack(pkt2)
+        assert result is None
+        assert pkt2.drop_reason == "Duplicate"
+
+    def test_marked_do_not_retransmit_ack_dropped(self, handler):
+        pkt = _make_direct_ack_packet()
+        pkt.mark_do_not_retransmit()
+        result = handler.forward_routed_ack(pkt)
+        assert result is None
+
+    def test_process_packet_dispatches_direct_ack(self, handler):
+        pkt = _make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC]))
+        result = handler.process_packet(pkt, snr=5.0)
+        assert result is not None
+        ack, delay = result
+        # The routed-ACK branch relays immediately; the generic direct forward
+        # would have applied a randomized direct TX delay instead.
+        assert delay == pytest.approx(0.0)
+        assert ack.get_route_type() == ROUTE_TYPE_DIRECT
+
+    def test_process_packet_keeps_flood_ack_on_generic_path(self, handler):
+        pkt = _make_flood_packet(
+            payload=b"\x4d\xab\xaf\x95", path=b"\x11", payload_type=PAYLOAD_TYPE_ACK
+        )
+        result = handler.process_packet(pkt, snr=5.0)
+        assert result is not None
+        fwd, _ = result
+        # Generic flood forwarding appends this node's hash.
+        assert fwd.get_route_type() == ROUTE_TYPE_FLOOD
+        assert list(fwd.path) == [0x11, LOCAL_HASH]
+
+
+# Literal MeshCore-format routed-ACK frames (Packet::writeTo layout).  MeshCore's
+# routeDirectRecvAcks re-emits through createAck: the header is rebuilt from
+# scratch (payload type ACK, route DIRECT, version bits cleared, transport codes
+# dropped) and this node is removed from the path.  Independent wire fixtures —
+# do NOT rebuild them through Packet.write_to()/PathUtils.
+FIRMWARE_ROUTED_ACK_VECTORS = (
+    # header 0x0E = ACK<<2 | ROUTE_DIRECT; path_len 0x02 = 1-byte/2-hop;
+    # path AB CC (AB = us); CRC 95AFAB4D.  out: path popped to CC.
+    pytest.param(
+        bytes([0xAB]),
+        bytes.fromhex("0E02ABCC4DABAF95"),
+        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("0E42ABCD11224DABAF95"),
+        bytes.fromhex("0E4111224DABAF95"),
+        id="direct-2-byte-2-hop",
+    ),
+    # header 0x0F = ACK<<2 | ROUTE_TRANSPORT_DIRECT; transport codes 34127856;
+    # the regenerated ACK is forced to plain DIRECT (codes dropped).
+    pytest.param(
+        bytes([0xAB]),
+        bytes.fromhex("0F3412785602ABCC4DABAF95"),
+        bytes.fromhex("0E01CC4DABAF95"),
+        id="transport-direct-1-byte-2-hop",
+    ),
+    # header 0x4E = version bit set on a direct ACK; createAck clears it.
+    pytest.param(
+        bytes([0xAB]),
+        bytes.fromhex("4E02ABCC4DABAF95"),
+        bytes.fromhex("0E01CC4DABAF95"),
+        id="direct-version-bits-cleared",
+    ),
+)
+
+
+@pytest.mark.parametrize("local_hash_bytes, wire_in, wire_out", FIRMWARE_ROUTED_ACK_VECTORS)
+def test_forward_routed_ack_matches_firmware_wire_vectors(local_hash_bytes, wire_in, wire_out):
+    """Parse a firmware-format routed ACK and assert the exact relayed bytes."""
+    handler = _make_handler_with_hash(local_hash_bytes)
+    pkt = Packet()
+    pkt.read_from(wire_in)
+    result = handler.forward_routed_ack(pkt)
+    assert result is not None
+    ack, delay = result
+    assert ack.write_to() == wire_out
+    assert delay == pytest.approx(0.0)
+
+
+class TestMultiAckRedundancy:
+    """multi_acks=1 adds MeshCore's MULTIPART-wrapped redundancy copy ahead of
+    the plain ACK; both slide out by the direct retransmit delay + 300 ms."""
+
+    def test_default_off_produces_no_extras(self, handler):
+        result = handler.forward_routed_ack(_make_direct_ack_packet())
+        assert result.extras == ()
+
+    def test_extra_precedes_plain_ack_at_same_delay(self, handler):
+        handler.multi_acks = 1
+        pkt = _make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC]))
+        with patch.object(handler, "_calculate_tx_delay", return_value=0.25):
+            result = handler.forward_routed_ack(pkt)
+        assert result is not None
+        ack, delay = result
+        # delay = direct retransmit delay (0.25s) + 300ms, shared by both copies.
+        assert delay == pytest.approx(0.55)
+        assert len(result.extras) == 1
+        wrapped, wrapped_delay = result.extras[0]
+        assert wrapped_delay == pytest.approx(0.55)
+        # MULTIPART wrapper: (remaining=1)<<4 | ACK prefix byte + the CRC, on the
+        # already self-removed path, plain DIRECT route.
+        assert wrapped.get_payload_type() == PAYLOAD_TYPE_MULTIPART
+        assert wrapped.get_route_type() == ROUTE_TYPE_DIRECT
+        assert bytes(wrapped.payload) == b"\x13\x4d\xab\xaf\x95"
+        assert bytes(wrapped.path) == bytes(ack.path)
+        assert wrapped.path_len == ack.path_len
+        # Exact wire bytes: header 0x2A = MULTIPART<<2 | DIRECT.
+        assert wrapped.write_to() == bytes.fromhex("2A01CC134DABAF95")
+
+    def test_multipart_relay_also_gets_extra(self, handler):
+        handler.multi_acks = 1
+        pkt = _make_multipart_ack_packet(remaining=2, path=bytes([LOCAL_HASH, 0xCC]))
+        with patch.object(handler, "_calculate_tx_delay", return_value=0.25):
+            result = handler.forward_multipart_direct(pkt)
+        assert result is not None
+        ack, delay = result
+        # base (remaining+1)*300ms = 0.9s, plus 0.25s + 300ms for the extra.
+        assert delay == pytest.approx(1.45)
+        assert len(result.extras) == 1
+        _, wrapped_delay = result.extras[0]
+        assert wrapped_delay == pytest.approx(1.45)
+
+    def test_multi_acks_clamped_like_firmware(self):
+        # MeshCore CommonCLI constrains multi_acks to 0..1 on prefs load.
+        handler = _make_handler_with_hash(bytes([LOCAL_HASH]))
+        handler.config.setdefault("repeater", {})["multi_acks"] = 5
+        handler.reload_runtime_config()
+        assert handler.multi_acks == 1
+
+        handler.config["repeater"]["multi_acks"] = "bogus"
+        handler.reload_runtime_config()
+        assert handler.multi_acks == 0
+
+        handler.config["repeater"]["multi_acks"] = -3
+        handler.reload_runtime_config()
+        assert handler.multi_acks == 0
+
+    def test_init_reads_repeater_multi_acks(self):
+        with (
+            patch("repeater.engine.StorageCollector"),
+            patch("repeater.engine.RepeaterHandler._start_background_tasks"),
+        ):
+            from repeater.engine import RepeaterHandler
+
+            handler = RepeaterHandler(
+                _make_config(repeater={"multi_acks": 1}),
+                _make_dispatcher(),
+                LOCAL_HASH,
+                local_hash_bytes=bytes([LOCAL_HASH]),
+            )
+        assert handler.multi_acks == 1
+
+    @pytest.mark.asyncio
+    async def test_call_transmits_wrapped_copy_before_plain_ack(self, handler):
+        handler.multi_acks = 1
+        handler.airtime_mgr.calculate_airtime = MagicMock(return_value=20.0)
+        handler.airtime_mgr.can_transmit = MagicMock(return_value=(True, 0.0))
+        handler.airtime_mgr.record_tx = MagicMock()
+        handler.airtime_mgr.record_rx = MagicMock()
+        handler.dispatcher.send_packet = AsyncMock(return_value=True)
+
+        pkt = _inject_from_wire(_make_direct_ack_packet(path=bytes([LOCAL_HASH, 0xCC])))
+        with (
+            patch.object(handler, "_calculate_tx_delay", return_value=0.0),
+            patch("repeater.engine.asyncio.sleep", new_callable=AsyncMock),
+        ):
+            await handler(pkt, {"snr": 3.0, "rssi": -80}, local_transmission=False)
+
+        assert handler.dispatcher.send_packet.call_count == 2
+        first = handler.dispatcher.send_packet.call_args_list[0].args[0]
+        second = handler.dispatcher.send_packet.call_args_list[1].args[0]
+        # MeshCore queues the MULTIPART redundancy copy ahead of the plain ACK.
+        assert first.write_to() == bytes.fromhex("2A01CC134DABAF95")
+        assert second.write_to() == bytes.fromhex("0E01CC4DABAF95")
+        assert handler.forwarded_count == 2
+
+
 # ===================================================================
 # 3. process_packet — route dispatch
 # ===================================================================
@@ -2163,7 +2449,7 @@ class TestPacketInjectionRouting:
         assert sent_pkt.get_payload_type() == payload_type
         assert sent_pkt.path[-1] == LOCAL_HASH
 
-    @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES)
+    @pytest.mark.parametrize("payload_type", _GENERIC_DIRECT_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(
@@ -2211,7 +2497,7 @@ class TestPacketInjectionRouting:
         assert sent_pkt.get_payload_type() == payload_type
         assert sent_pkt.transport_codes == [0x1111, 0x2222]
 
-    @pytest.mark.parametrize("payload_type", _GENERIC_FORWARD_PAYLOAD_TYPES)
+    @pytest.mark.parametrize("payload_type", _GENERIC_DIRECT_FORWARD_PAYLOAD_TYPES)
     async def test_all_payload_types_transport_direct_injection_forwards(
         self, handler, payload_type
     ):

tests/test_identities_endpoint_payload.py

@@ -0,0 +1,127 @@
+"""Regression: /api/identities payload shape with the registered repeater identity.
+
+Since the daemon registers its own default identity in the IdentityManager
+(so companion/room-server collisions against the repeater's hash byte are
+caught), the endpoint's raw ``registered`` list carries a
+``repeater:repeater`` entry and ``total_registered`` counts it. The web UI
+is unaffected — it renders only the per-entry ``registered`` boolean on
+configured room servers/companions, which match on ``room_server:``/
+``companion:``-prefixed names — but external API consumers see the new
+entry, so this test pins the intended payload.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import MagicMock
+
+import cherrypy
+import pytest
+
+from openhop_core.protocol import LocalIdentity
+
+from repeater.identity_manager import IdentityManager
+from repeater.web.api_endpoints import APIEndpoints
+
+
+@pytest.fixture
+def cherrypy_ctx(monkeypatch):
+    request = SimpleNamespace(method="GET", params={}, json={})
+    response = SimpleNamespace(headers={}, status=200)
+    monkeypatch.setattr(cherrypy, "request", request, raising=False)
+    monkeypatch.setattr(cherrypy, "response", response, raising=False)
+    return request, response
+
+
+def _distinct_identities(count):
+    """LocalIdentities with pairwise-distinct first hash bytes."""
+    picked = []
+    seen = set()
+    while len(picked) < count:
+        identity = LocalIdentity()
+        hash_byte = identity.get_public_key()[0]
+        if hash_byte not in seen:
+            seen.add(hash_byte)
+            picked.append(identity)
+    return picked
+
+
+def _make_api(config, identity_manager):
+    api = APIEndpoints.__new__(APIEndpoints)
+    api.config = config
+    api.daemon_instance = SimpleNamespace(identity_manager=identity_manager)
+    api.send_advert_func = None
+    api.event_loop = None
+    api.stats_getter = None
+    api._config_path = "/tmp/test-config.yaml"
+    api.config_manager = MagicMock()
+    return api
+
+
+def test_registered_list_includes_repeater_identity(cherrypy_ctx):
+    repeater_id, companion_id = _distinct_identities(2)
+
+    config = {
+        "identities": {
+            "companions": [
+                {
+                    "name": "phone",
+                    "identity_key": companion_id.get_private_key().hex(),
+                    "settings": {},
+                }
+            ],
+            "room_servers": [],
+        }
+    }
+    manager = IdentityManager(config)
+    assert manager.register_identity("repeater", repeater_id, config, "repeater")
+    assert manager.register_identity("phone", companion_id, {}, "companion")
+
+    api = _make_api(config, manager)
+    payload = api.identities()
+
+    assert payload["success"] is True
+    data = payload["data"]
+    registered = data["registered"]
+    assert data["total_registered"] == len(registered) == 2
+
+    by_name = {entry["name"]: entry for entry in registered}
+    repeater_entry = by_name["repeater:repeater"]
+    assert repeater_entry["type"] == "repeater"
+    assert repeater_entry["hash"] == f"0x{repeater_id.get_public_key()[0]:02X}"
+    assert repeater_entry["public_key"] == repeater_id.get_public_key().hex()
+
+    companion_entry = by_name["companion:phone"]
+    assert companion_entry["type"] == "companion"
+
+    # The configured-companion view (what the web UI renders) matches the
+    # companion by its prefixed name and is not disturbed by the repeater
+    # entry: it still reports the companion as registered.
+    assert data["total_configured_companions"] == 1
+    ui_entry = data["configured_companions"][0]
+    assert ui_entry["name"] == "phone"
+    assert ui_entry["registered"] is True
+
+
+def test_name_repeater_is_reserved_by_the_default_identity(cherrypy_ctx):
+    """Registering the default repeater identity reserves the bare name
+    "repeater": a room server or companion configured with that name is
+    rejected by the collision rules, and the endpoint reports it
+    unregistered rather than silently matching the repeater's entry."""
+    repeater_id, room_id = _distinct_identities(2)
+
+    config = {
+        "identities": {
+            "room_servers": [{"name": "repeater", "identity_key": "aa" * 32, "settings": {}}],
+            "companions": [],
+        }
+    }
+    manager = IdentityManager(config)
+    assert manager.register_identity("repeater", repeater_id, config, "repeater")
+    assert not manager.register_identity("repeater", room_id, {}, "room_server")
+
+    api = _make_api(config, manager)
+    data = api.identities()["data"]
+
+    assert data["total_registered"] == 1
+    room_entry = data["configured"][0]
+    assert room_entry["registered"] is False
+    assert room_entry["hash"] is None