- Refactored ACL authentication flow with shared helpers for replay detection and session updates.

- Changed blank-password login behavior to create/manage guest sessions (when read-only is enabled), including replay protection and session timestamp tracking.
- Preserved and centralized `sync_since` handling during successful auth/session updates.
- Updated login handling to return explicit `(success, permissions)` and reset room-server sync guard state in SQLite on successful room-server login.
- Updated identity/banner output in mesh CLI from `pyMC_*` to `openHop_*` and adjusted default version fallback.
- Extended path handling to support encoded path-length semantics via `PathUtils`, with legacy fallback support for older/malformed packets.
- Added bundled ACK extraction from PATH payload extras and callback-based ACK propagation.
- Hardened room-server push logic: max-failure early skip, corrected expected ACK CRC calculation, path encoding compatibility, persisted sync/last-activity fields, and passed expected CRC + timeout into packet injection.
- Enhanced packet router ACK flow with dispatcher ACK registration helper, support for multipart ACK wrapper handling, configurable ACK wait timeout/CRC in injection, and ensured PATH helper processing always runs.
- Updated daemon wiring to pass ACK callback into `PathHelper`,
- make dispatcher dedupe configurable (prevent dbl deduping), and only register duplicate logging hook when dedupe is enabled.
- Expanded tests to cover guest blank-password replay tracking, updated room-server injector ACK expectations, and new duplicate-logging-hook gating behavior.

repeater/handler_helpers/acl.py

@@ -48,6 +48,30 @@ class ACL:
         self.allow_read_only = allow_read_only
         self.clients: Dict[bytes, ClientInfo] = {}
 
+    def _is_replay(self, client: ClientInfo, timestamp: int) -> bool:
+        if timestamp <= client.last_timestamp:
+            logger.warning(
+                f"Possible replay attack! timestamp={timestamp}, last={client.last_timestamp}"
+            )
+            return True
+        return False
+
+    def _touch_client_session(
+        self,
+        client: ClientInfo,
+        shared_secret: bytes,
+        timestamp: int,
+        sync_since: int = None,
+    ) -> None:
+        now = int(time.time())
+        client.last_timestamp = timestamp
+        client.last_activity = now
+        client.last_login_success = now
+        client.shared_secret = shared_secret
+        if sync_since is not None:
+            client.sync_since = sync_since
+            logger.debug(f"Stored sync_since={sync_since} for client")
+
     def authenticate_client(
         self,
         client_identity: Identity,
@@ -106,13 +130,23 @@ class ACL:
         if not password:
             client = self.clients.get(pub_key)
             if client is None:
-                if self.allow_read_only:
-                    logger.info("Blank password, allowing read-only guest access")
-                    return True, PERM_ACL_GUEST
-                else:
+                if not self.allow_read_only:
                     logger.info("Blank password, sender not in ACL and read-only disabled")
                     return False, 0
-            logger.info(f"ACL-based login for {pub_key[:6].hex()}...")
+                if len(self.clients) >= self.max_clients:
+                    logger.warning("ACL full, cannot add client")
+                    return False, 0
+                client = ClientInfo(client_identity, PERM_ACL_GUEST)
+                self.clients[pub_key] = client
+                logger.info("Blank password, allowing read-only guest access")
+            else:
+                logger.info(f"ACL-based login for {pub_key[:6].hex()}...")
+
+            if self._is_replay(client, timestamp):
+                return False, 0
+            self._touch_client_session(client, shared_secret, timestamp, sync_since=sync_since)
+            if (client.permissions & PERM_ACL_ROLE_MASK) == 0:
+                client.permissions |= PERM_ACL_GUEST
             return True, client.permissions
 
         permissions = 0
@@ -140,23 +174,11 @@ class ACL:
             self.clients[pub_key] = client
             logger.info(f"Added new client {pub_key[:6].hex()}...")
 
-        if timestamp <= client.last_timestamp:
-            logger.warning(
-                f"Possible replay attack! timestamp={timestamp}, last={client.last_timestamp}"
-            )
+        if self._is_replay(client, timestamp):
             return False, 0
-
-        client.last_timestamp = timestamp
-        client.last_activity = int(time.time())
-        client.last_login_success = int(time.time())
+        self._touch_client_session(client, shared_secret, timestamp, sync_since=sync_since)
         client.permissions &= ~PERM_ACL_ROLE_MASK
         client.permissions |= permissions
-        client.shared_secret = shared_secret
-
-        # Store sync_since for room server clients
-        if sync_since is not None:
-            client.sync_since = sync_since
-            logger.debug(f"Stored sync_since={sync_since} for client")
 
         logger.info(f"Login success! Permissions: {'ADMIN' if client.is_admin() else 'GUEST'}")
         return True, client.permissions

repeater/handler_helpers/login.py

@@ -122,7 +122,7 @@ class LoginHelper:
         def auth_callback_with_context(
             client_identity, shared_secret, password, timestamp, sync_since=None
         ):
-            return identity_acl.authenticate_client(
+            success, permissions = identity_acl.authenticate_client(
                 client_identity=client_identity,
                 shared_secret=shared_secret,
                 password=password,
@@ -132,6 +132,26 @@ class LoginHelper:
                 target_identity_name=name,
                 target_identity_config=config,
             )
+            if success and identity_type == "room_server" and self.sqlite_handler is not None:
+                try:
+                    sync_kwargs = {}
+                    if sync_since is not None:
+                        sync_kwargs["sync_since"] = sync_since
+                    self.sqlite_handler.upsert_client_sync(
+                        room_hash=f"0x{hash_byte:02X}",
+                        client_pubkey=client_identity.get_public_key().hex(),
+                        pending_ack_crc=0,
+                        push_post_timestamp=0,
+                        ack_timeout_time=0,
+                        push_failures=0,
+                        last_activity=time.time(),
+                        **sync_kwargs,
+                    )
+                except Exception as e:
+                    logger.warning(
+                        f"Failed to reset room sync guard state after login for hash=0x{hash_byte:02X}: {e}"
+                    )
+            return success, permissions
 
         handler = LoginServerHandler(
             local_identity=identity,

repeater/handler_helpers/mesh_cli.py

@@ -2,7 +2,6 @@ import logging
 from pathlib import Path
 from typing import Any, Callable, Dict, Optional
 
-
 logger = logging.getLogger(__name__)
 
 
@@ -329,8 +328,8 @@ class MeshCLI:
     def _cmd_version(self) -> str:
         """Get version information."""
         role = "room_server" if self.identity_type == "room_server" else "repeater"
-        version = self.config.get("version", "1.0.0")
-        return f"pyMC_{role} v{version}"
+        version = self.config.get("version", "13")
+        return f"openHop_{role} v{version}"
 
     # ==================== Get Commands ====================
 

repeater/handler_helpers/path.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import time
 
@@ -5,14 +6,31 @@ 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_callback=None):
 
         self.acl_dict = acl_dict or {}
         self.log_fn = log_fn or logger.info
+        self.ack_received_callback = ack_received_callback
+
+    async def _register_ack_crc(self, ack_crc: int) -> None:
+        """Propagate an ACK CRC to the configured callback."""
+        if ack_crc is None:
+            return
+        callback = self.ack_received_callback
+        if callback is None:
+            return
+        try:
+            result = callback(ack_crc)
+            if asyncio.iscoroutine(result):
+                await result
+        except Exception as e:
+            logger.debug(f"ACK callback failed for CRC {ack_crc:08X}: {e}")
 
     async def process_path_packet(self, packet):
 
+        from openhop_core.protocol.constants import PAYLOAD_TYPE_ACK
         from openhop_core.protocol.crypto import CryptoUtils
+        from openhop_core.protocol.packet_utils import PathUtils
 
         try:
             if len(packet.payload) < 2:
@@ -60,30 +78,54 @@ class PathHelper:
                 return False
 
             # Parse decrypted PATH data
-            # Format: path_len(1) + path[path_len] + extra_type(1) + extra[...]
+            # Format: path_len(1) + path[path_byte_len] + extra_type(1) + extra[...]
             if len(decrypted) < 1:
                 logger.debug("Decrypted PATH data too short")
                 return False
+            path_len_byte = decrypted[0]
+            if PathUtils.is_valid_path_len(path_len_byte):
+                path_byte_len = PathUtils.get_path_byte_len(path_len_byte)
+                path_hops = PathUtils.get_path_hash_count(path_len_byte)
+            else:
+                # Legacy fallback for malformed/old packets: treat first byte as raw path bytes.
+                path_byte_len = path_len_byte
+                path_hops = path_byte_len
 
-            path_len = decrypted[0]
-            if len(decrypted) < 1 + path_len:
+            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)
             client.out_path = bytearray(path_data)
-            client.out_path_len = path_len
+            client.out_path_len = (
+                path_len_byte if PathUtils.is_valid_path_len(path_len_byte) else path_byte_len
+            )
             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_byte=0x{path_len_byte:02X}, hops={path_hops}, "
+                f"path={[hex(b) for b in path_data]}"
             )
 
+            # Handle bundled ACK in PATH extra section.
+            ack_crc = None
+            extra_start = 1 + path_byte_len
+            if len(decrypted) > extra_start:
+                extra_type = decrypted[extra_start] & 0x0F
+                extra_payload = decrypted[extra_start + 1 :]
+                if extra_type == PAYLOAD_TYPE_ACK and len(extra_payload) >= 4:
+                    ack_crc = int.from_bytes(extra_payload[:4], "little")
+                    logger.info(
+                        f"PATH bundled ACK extracted for client 0x{src_hash:02X}: CRC={ack_crc:08X}"
+                    )
+
+            if ack_crc is not None:
+                await self._register_ack_crc(ack_crc)
             # 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")
 
@@ -328,6 +329,12 @@ class RoomServer:
 
             if sync_state:
                 failures = sync_state.get("push_failures", 0)
+                if failures >= MAX_PUSH_FAILURES:
+                    logger.debug(
+                        f"Room '{self.room_name}': Client 0x{client_info.id.get_public_key()[0]:02X} "
+                        f"at max failures ({failures}), skipping push"
+                    )
+                    return False
                 if failures > 0:
                     # Apply exponential backoff
                     backoff_idx = min(failures, len(RETRY_BACKOFF_SCHEDULE) - 1)
@@ -356,11 +363,9 @@ class RoomServer:
             plaintext = (
                 timestamp.to_bytes(4, "little") + bytes([flags]) + author_prefix + message_bytes
             )
-
-            # Calculate expected ACK (same algorithm as openhop_core)
-            attempt = 0
-            pack_data = PacketBuilder._pack_timestamp_data(timestamp, attempt, message_bytes)
-            ack_hash = CryptoUtils.sha256(pack_data + client_info.id.get_public_key())[:4]
+            # Calculate expected ACK (MeshCore signed text):
+            # sha256(timestamp + flags + author_prefix + text + recipient_pubkey)[:4]
+            ack_hash = CryptoUtils.sha256(plaintext + client_info.id.get_public_key())[:4]
             expected_ack_crc = int.from_bytes(ack_hash, "little")
 
             # Determine routing based on stored out_path
@@ -378,29 +383,52 @@ class RoomServer:
 
             # Add stored path for direct routing
             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_len = client_info.out_path_len
+                if PathUtils.is_valid_path_len(client_info.out_path_len):
+                    path_byte_len = PathUtils.get_path_byte_len(client_info.out_path_len)
+                    packet.path = bytearray(client_info.out_path[:path_byte_len])
+                    packet.path_len = client_info.out_path_len
+                else:
+                    # Legacy fallback: treat stored path as 1-byte-hop path and clamp to
+                    # valid encoded range (0-63 hops).
+                    legacy_hops = min(len(client_info.out_path), 63)
+                    packet.path = bytearray(client_info.out_path[:legacy_hops])
+                    packet.path_len = legacy_hops
 
             # Calculate ACK timeout
             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
+                if PathUtils.is_valid_path_len(client_info.out_path_len):
+                    path_len = PathUtils.get_path_hash_count(client_info.out_path_len)
+                else:
+                    path_len = min(len(client_info.out_path), 63)
                 ack_timeout = (
                     PUSH_TIMEOUT_BASE_MS + PUSH_ACK_TIMEOUT_FACTOR_MS * (path_len + 1)
                 ) / 1000.0
 
             # Update client sync state with pending ACK
+            current_sync_since = (
+                sync_state.get("sync_since", 0)
+                if sync_state
+                else getattr(client_info, "sync_since", 0)
+            )
             self.db.upsert_client_sync(
                 room_hash=f"0x{self.room_hash:02X}",
                 client_pubkey=client_info.id.get_public_key().hex(),
+                sync_since=current_sync_since,
                 pending_ack_crc=expected_ack_crc,
                 push_post_timestamp=post["post_timestamp"],
                 ack_timeout_time=time.time() + ack_timeout,
+                last_activity=time.time(),
             )
             # Send packet (dispatcher will track ACK automatically)
             # 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_crc=expected_ack_crc,
+                ack_timeout_s=ack_timeout,
+            )
 
             # SAFETY: Release transmission lock AFTER send completes
             self.global_limiter.release()
@@ -460,7 +488,7 @@ class RoomServer:
                     pending_ack_crc=0,
                 )
 
-                if failures >= 3:
+                if failures >= MAX_PUSH_FAILURES:
                     logger.warning(
                         f"Room '{self.room_name}': Client 0x{client_pubkey[0]:02X} "
                         f"has {failures} consecutive failures"
@@ -625,7 +653,7 @@ class RoomServer:
                             )
                             continue
 
-                        if push_failures >= 3:
+                        if push_failures >= MAX_PUSH_FAILURES:
                             logger.debug(
                                 f"Skipping client 0x{client.id.get_public_key()[0]:02X} (max failures)"
                             )

repeater/main.py

@@ -178,8 +178,12 @@ class RepeaterDaemon:
             from openhop_core import LocalIdentity
             from openhop_core.node.dispatcher import Dispatcher
 
-            self.dispatcher = Dispatcher(self.radio)
+            dedupe_enabled = bool(
+                self.config.get("repeater", {}).get("dispatcher_dedupe_enabled", False)
+            )
+            self.dispatcher = Dispatcher(self.radio, dedupe_enabled=dedupe_enabled)
             logger.info("Dispatcher initialized")
+            logger.info("Dispatcher dedupe enabled: %s", dedupe_enabled)
 
             # Initialize Identity Manager for additional identities (e.g., room servers)
             self.identity_manager = IdentityManager(self.config)
@@ -373,6 +377,11 @@ class RepeaterDaemon:
             self.path_helper = PathHelper(
                 acl_dict=self.login_helper.get_acl_dict(),  # Per-identity ACLs
                 log_fn=logger.info,
+                ack_received_callback=(
+                    self.dispatcher._register_ack_received
+                    if self.dispatcher and hasattr(self.dispatcher, "_register_ack_received")
+                    else None
+                ),
             )
             logger.info("PATH packet processing helper initialized")
 
@@ -403,9 +412,7 @@ class RepeaterDaemon:
                 n,
             )
 
-            # Subscribe to parsed packets (pre-dedup) so duplicate path variants
-            # still appear in the web UI even though the Dispatcher blocks them.
-            self.dispatcher.add_raw_packet_subscriber(self._on_raw_packet_for_dedup_logging)
+            self._register_duplicate_logging_hook(dedupe_enabled)
 
             # When trace reaches final node, push PUSH_CODE_TRACE_DATA (0x89) to companion clients (firmware onTraceRecv)
             self.trace_helper.on_trace_complete = self._on_trace_complete_for_companions
@@ -909,6 +916,14 @@ class RepeaterDaemon:
             except Exception as e:
                 logger.debug("Push RX raw to companion: %s", e)
 
+    def _register_duplicate_logging_hook(self, dedupe_enabled: bool) -> None:
+        """Register pre-dedup duplicate logging only when dispatcher dedupe is active."""
+        if not self.dispatcher or not dedupe_enabled:
+            return
+        # When dispatcher dedupe is disabled, duplicates still flow through
+        # router -> repeater_handler and are already recorded there.
+        self.dispatcher.add_raw_packet_subscriber(self._on_raw_packet_for_dedup_logging)
+
     def _on_raw_packet_for_dedup_logging(self, pkt, data: bytes, analysis: dict) -> None:
         """Record duplicate packets for UI visibility.
 

repeater/packet_router.py

@@ -8,6 +8,7 @@ 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
 from openhop_core.node.handlers.path import PathHandler
 from openhop_core.node.handlers.protocol_request import ProtocolRequestHandler
 from openhop_core.node.handlers.protocol_response import ProtocolResponseHandler
@@ -261,6 +262,16 @@ class PacketRouter:
             except Exception as e:
                 logger.debug("Record for UI failed: %s", e)
 
+    async def _register_ack_with_dispatcher(self, ack_crc: int, context: str) -> None:
+        """Best-effort ACK CRC registration with the dispatcher waiter path."""
+        dispatcher = getattr(self.daemon, "dispatcher", None)
+        if dispatcher is None or not hasattr(dispatcher, "_register_ack_received"):
+            return
+        try:
+            await dispatcher._register_ack_received(ack_crc)
+        except Exception as e:
+            logger.debug("Dispatcher %s registration error: %s", context, e)
+
     async def enqueue(self, packet):
         """Add packet to router queue."""
         if self.queue.full():
@@ -271,7 +282,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,
+        expected_crc=None,
+        origin_hash=None,
+        ack_timeout_s: float = 5.0,
+    ):
         try:
             metadata = {
                 "rssi": getattr(packet, "rssi", 0),
@@ -327,11 +345,20 @@ 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)
+                            wait_crc = (
+                                expected_crc if expected_crc is not None else packet.get_crc()
+                            )
+                            wait_timeout = (
+                                float(ack_timeout_s)
+                                if isinstance(ack_timeout_s, (int, float)) and ack_timeout_s > 0
+                                else 5.0
+                            )
+                            ack_ok = await dispatcher.wait_for_ack(wait_crc, timeout=wait_timeout)
                             if not ack_ok:
                                 logger.warning(
-                                    "Injected packet ACK timeout (crc=%08X)", expected_crc
+                                    "Injected packet ACK timeout (crc=%08X, timeout=%.1fs)",
+                                    wait_crc,
+                                    wait_timeout,
                                 )
                                 return False
                         except Exception as e:
@@ -463,6 +490,10 @@ class PacketRouter:
                 self._record_for_ui(packet, metadata)
 
         elif payload_type == AckHandler.payload_type():
+            # Ensure ACK CRC reaches dispatcher waiter path even when only router fallback is active.
+            if len(getattr(packet, "payload", b"")) >= 4:
+                ack_crc = int.from_bytes(packet.payload[:4], "little")
+                await self._register_ack_with_dispatcher(ack_crc, "ACK")
             # 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)
@@ -472,6 +503,15 @@ class PacketRouter:
                 except Exception as e:
                     logger.debug(f"Companion bridge ACK error: {e}")
 
+        elif payload_type == MultipartAckHandler.payload_type():
+            # MULTIPART ACK wrapper: low nibble of first byte is embedded payload type.
+            if (
+                len(getattr(packet, "payload", b"")) >= 5
+                and (packet.payload[0] & 0x0F) == AckHandler.payload_type()
+            ):
+                ack_crc = int.from_bytes(packet.payload[1:5], "little")
+                await self._register_ack_with_dispatcher(ack_crc, "multi-ACK")
+
         elif payload_type == TextMessageHandler.payload_type():
             dest_hash = packet.payload[0] if packet.payload else None
             companion_bridges = self._companion_bridges_for_packet(packet, metadata)
@@ -486,6 +526,13 @@ class PacketRouter:
                     self._record_for_ui(packet, metadata)
 
         elif payload_type == PathHandler.payload_type():
+            # Always let PathHelper inspect/decrypt PATH first so out_path and bundled ACK state
+            # are updated even when companion routing fan-out also happens for this packet.
+            if self.daemon.path_helper:
+                try:
+                    await self.daemon.path_helper.process_path_packet(packet)
+                except Exception as e:
+                    logger.debug(f"Path helper processing error: {e}")
             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:
@@ -506,8 +553,6 @@ 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 payload_type == LoginResponseHandler.payload_type():
             # PAYLOAD_TYPE_RESPONSE (0x01): payload is dest_hash(1)+src_hash(1)+encrypted.

tests/test_handler_helpers_acl_advert.py

@@ -40,6 +40,21 @@ def test_acl_blank_password_guest_rules_and_room_server_password_requirements():
     )
     assert ok is True
     assert perms == PERM_ACL_GUEST
+    guest_client = acl.get_client(identity.get_public_key())
+    assert guest_client is not None
+    assert guest_client.permissions == PERM_ACL_GUEST
+    assert guest_client.shared_secret == b"secret"
+    assert guest_client.last_timestamp == 10
+
+    # Blank-password logins now track replay/timestamp just like password logins.
+    replay_ok, replay_perms = acl.authenticate_client(
+        client_identity=identity,
+        shared_secret=b"secret",
+        password="",
+        timestamp=10,
+    )
+    assert replay_ok is False
+    assert replay_perms == 0
 
     acl_ro_disabled = ACL(allow_read_only=False)
     ok2, perms2 = acl_ro_disabled.authenticate_client(

tests/test_handler_helpers_room_server.py

@@ -149,7 +149,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_crc=67305985,
+        ack_timeout_s=10.0,
+    )
     rs._handle_ack_received.assert_awaited_once_with(
         client.id.get_public_key(), post["post_timestamp"]
     )

tests/test_main_py_coverage.py

@@ -76,6 +76,20 @@ def test_get_stats_includes_public_key_gps_sensors_and_radio_state():
     assert stats["radio_error"] == "missing device"
 
 
+def test_register_duplicate_logging_hook_only_when_dispatcher_dedup_enabled():
+    daemon = RepeaterDaemon(_base_config(), radio=object())
+    dispatcher = SimpleNamespace(add_raw_packet_subscriber=MagicMock())
+    daemon.dispatcher = dispatcher
+
+    daemon._register_duplicate_logging_hook(False)
+    dispatcher.add_raw_packet_subscriber.assert_not_called()
+
+    daemon._register_duplicate_logging_hook(True)
+    dispatcher.add_raw_packet_subscriber.assert_called_once_with(
+        daemon._on_raw_packet_for_dedup_logging
+    )
+
+
 def test_detect_container_from_proc_env_and_fallback_path():
     with patch("builtins.open", MagicMock()) as open_mock:
         open_mock.return_value.__enter__.return_value.read.return_value = b"container=docker"