pyMC_Repeater/repeater/engine.py

import asyncio
import copy
import logging
import math
import secrets
import time
from collections import OrderedDict, deque
from collections.abc import Mapping
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Tuple

from openhop_core.node.handlers.base import BaseHandler
from openhop_core.protocol import Packet
from openhop_core.protocol.constants import (
    MAX_PATH_SIZE,
    PAYLOAD_TYPE_ACK,
    PAYLOAD_TYPE_ADVERT,
    PAYLOAD_TYPE_ANON_REQ,
    PAYLOAD_TYPE_MULTIPART,
    PAYLOAD_TYPE_TRACE,
    PH_ROUTE_MASK,
    PH_TYPE_SHIFT,
    ROUTE_TYPE_DIRECT,
    ROUTE_TYPE_FLOOD,
    ROUTE_TYPE_TRANSPORT_DIRECT,
    ROUTE_TYPE_TRANSPORT_FLOOD,
)
from openhop_core.protocol.packet_utils import (
    PacketHashingUtils,
    PacketHeaderUtils,
    PathUtils,
    flood_rx_metrics,
    packet_score,
)

from repeater.airtime import AirtimeManager
from repeater.config import fabric_origin_tx
from repeater.data_acquisition import StorageCollector
from repeater.modem_config import normalize_modem_config, redact_modem_tokens_in_place
from repeater.neighbour_links import NeighbourLinkTracker
from repeater.policy_engine import PolicyDecision, PolicyEngine

logger = logging.getLogger("RepeaterHandler")

NOISE_FLOOR_INTERVAL = 30.0  # seconds

LOOP_DETECT_OFF = "off"
LOOP_DETECT_MINIMAL = "minimal"
LOOP_DETECT_MODERATE = "moderate"
LOOP_DETECT_STRICT = "strict"

# Thresholds for flood loop detection, keyed by loop mode and path-hash width.
# MeshCore treats 4-byte hashes as reserved, so only 1-3 byte widths are valid here.
LOOP_DETECT_MAX_COUNTERS = {
    LOOP_DETECT_MINIMAL: {1: 4, 2: 2, 3: 1},
    LOOP_DETECT_MODERATE: {1: 2, 2: 1, 3: 1},
    LOOP_DETECT_STRICT: {1: 1, 2: 1, 3: 1},
}


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


@dataclass
class RadioTxResult:
    """Outcome of one physical transmission through one radio endpoint."""

    radio_id: Optional[str]
    success: bool
    metadata: Optional[dict] = None
    error: Optional[BaseException] = None


@dataclass
class FanoutTxResult:
    """Per-egress outcomes of one logical transmission, in attempt order.

    One logical packet can go out through several Fabric radios; each egress
    succeeds or fails on its own, and the logical send succeeds when any does.
    """

    results: list = field(default_factory=list)

    @property
    def any_success(self) -> bool:
        return any(result.success for result in self.results)

    @property
    def all_success(self) -> bool:
        return bool(self.results) and all(result.success for result in self.results)

    @property
    def successful_radio_ids(self) -> list:
        return [r.radio_id for r in self.results if r.success and r.radio_id]

    @property
    def failed_radio_ids(self) -> list:
        return [r.radio_id for r in self.results if not r.success and r.radio_id]

    @property
    def primary(self) -> Optional[RadioTxResult]:
        """The first successful egress, else the first one attempted."""
        for result in self.results:
            if result.success:
                return result
        return self.results[0] if self.results else None


class RepeaterHandler(BaseHandler):
    @staticmethod
    def payload_type() -> int:

        return 0xFF  # Special marker (not a real payload type)

    def __init__(
        self,
        config: dict,
        dispatcher,
        local_hash: int,
        *,
        local_hash_bytes=None,
        send_advert_func=None,
        periodic_advert_tick_func=None,
    ):

        self.config = config
        self.dispatcher = dispatcher
        self.local_hash = local_hash
        self.local_hash_bytes = local_hash_bytes or bytes([local_hash])
        self.send_advert_func = send_advert_func
        self.periodic_advert_tick_func = periodic_advert_tick_func
        self.airtime_mgr = AirtimeManager(config)
        self.policy_engine = PolicyEngine.from_runtime_config(config)
        self.seen_packets = OrderedDict()
        self.cache_ttl = max(
            300, config.get("repeater", {}).get("cache_ttl", 3600)
        )  # Min 5 min, default 1 hour
        self.max_cache_size = 1000
        self.max_duplicates_per_packet = 20
        self.tx_delay_factor = config.get("delays", {}).get("tx_delay_factor", 1.0)
        self.direct_tx_delay_factor = config.get("delays", {}).get("direct_tx_delay_factor", 0.5)
        # These are airtime multipliers, not seconds: TX delay is a random value in
        # [0, 5 * airtime * factor] (matching MeshCore getRetransmitDelay). Flood and
        # direct share the formula and differ only in the factor.
        logger.info(
            "TX delay factors (airtime multipliers, not seconds): "
            f"flood random 0-5*airtime*F (F={self.tx_delay_factor}), "
            f"direct random 0-5*airtime*F (F={self.direct_tx_delay_factor})"
        )
        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
        )
        self.last_advert_time = time.time()
        self.loop_detect_mode = self._normalize_loop_detect_mode(
            config.get("mesh", {}).get("loop_detect", LOOP_DETECT_OFF)
        )
        self.neighbour_link_tracker = NeighbourLinkTracker(config)
        # (packet hash, path) -> recent physical sends, each with its radio and
        # timing. Two radios sharing one frequency hear each other send; that echo
        # ends in our own path hash and must not make this node its own neighbour.
        self._recent_own_tx: OrderedDict = OrderedDict()

        radio = dispatcher.radio if dispatcher else None
        if radio:
            self.radio_config = {
                "spreading_factor": getattr(radio, "spreading_factor", 8),
                "bandwidth": getattr(radio, "bandwidth", 125000),
                "coding_rate": getattr(radio, "coding_rate", 8),
                "preamble_length": getattr(radio, "preamble_length", 32),
                "frequency": getattr(radio, "frequency", 915000000),
                "tx_power": getattr(radio, "tx_power", 14),
            }
            logger.info(
                f"radio settings: SF={self.radio_config['spreading_factor']}, "
                f"BW={self.radio_config['bandwidth']}Hz, CR={self.radio_config['coding_rate']}"
            )
        else:
            raise RuntimeError("Radio object not available - cannot initialize repeater")

        # Statistics tracking for dashboard
        self.rx_count = 0
        self.forwarded_count = 0
        self.dropped_count = 0
        self.max_recent_packets = 50
        self.recent_packets = deque(maxlen=self.max_recent_packets)
        self._recent_hash_index = {}
        self.start_time = time.time()
        # Flood/direct and duplicate counters (for GET_STATUS / firmware RepeaterStats)
        self.recv_flood_count = 0
        self.recv_direct_count = 0
        self.sent_flood_count = 0
        self.sent_direct_count = 0
        self.flood_dup_count = 0
        self.direct_dup_count = 0

        # Storage collector for persistent packet logging
        try:
            local_identity = dispatcher.local_identity if dispatcher else None
            self.storage = StorageCollector(config, local_identity, repeater_handler=self)
            logger.info("StorageCollector initialized successfully")
        except Exception as e:
            logger.error(f"Failed to initialize StorageCollector: {e}")
            self.storage = None

        # Initialize background timer tracking
        self.last_noise_measurement = time.time()
        self.last_cache_cleanup = time.time()
        self.last_db_cleanup = time.time()
        self.noise_floor_interval = NOISE_FLOOR_INTERVAL  # 30 seconds
        self._background_task = None
        self._cached_noise_floor = None
        self._last_crc_error_count = 0  # Track radio counter for delta persistence

        # Cache transport keys for efficient lookup
        self._transport_keys_cache = None
        self._transport_keys_cache_time = 0
        self._transport_keys_cache_ttl = 60  # Cache for 60 seconds

        # Serialise all radio TX calls.
        #
        # Background: since the queue loop dispatches each packet as an
        # asyncio.create_task, multiple _route_packet coroutines can have their
        # TX delay timers running concurrently — which is the intended behaviour
        # (firmware nodes do the same with a hardware timer).  However, the
        # LoRa radio is half-duplex: it can only transmit one packet at a time.
        # Without serialisation, two tasks whose delay timers expire near-
        # simultaneously both call dispatcher.send_packet, interleaving SPI/serial
        # commands to the radio and both passing the LBT check before either has
        # actually transmitted.
        #
        # _tx_lock is acquired after each delay sleep and held for the entire
        # send_packet call.  Delays still run concurrently; only the radio
        # access is serialised.  This also eliminates the TOCTOU gap in duty-cycle
        # enforcement — see schedule_retransmit / delayed_send for details.
        self._tx_lock = asyncio.Lock()

        self._start_background_tasks()

    async def __call__(
        self, packet: Packet, metadata: Optional[dict] = None, local_transmission: bool = False
    ) -> bool:

        if metadata is None:
            metadata = {}

        # Only count as receive when packet came from the radio (not locally injected)
        if not local_transmission:
            self.rx_count += 1
            route_type = packet.header & PH_ROUTE_MASK
            if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD):
                self.recv_flood_count += 1
            elif route_type in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT):
                self.recv_direct_count += 1
            try:
                rx_airtime_ms = self.airtime_mgr.calculate_airtime(packet.get_raw_length())
                self.airtime_mgr.record_rx(rx_airtime_ms)
            except Exception as exc:
                logger.debug(f"Failed to record RX airtime: {exc}")

        route_type = packet.header & PH_ROUTE_MASK
        pkt_hash_full = packet.calculate_packet_hash().hex().upper()
        snr = metadata.get("snr", 0.0)
        rssi = metadata.get("rssi", 0)

        # TX mode: forward (repeat on), monitor (no repeat, tenants can TX), no_tx (all TX off)
        mode = self.config.get("repeater", {}).get("mode", "forward")
        if mode not in ("forward", "monitor", "no_tx"):
            mode = "forward"
        allow_forward = mode == "forward"
        allow_local_tx = mode != "no_tx"

        policy_context = {
            "route_type": route_type,
            "payload_type": packet.get_payload_type()
            if hasattr(packet, "get_payload_type")
            else None,
            "payload_length": len(packet.payload or b""),
            "path_hash_size": packet.get_path_hash_size()
            if hasattr(packet, "get_path_hash_size")
            else None,
            "hop_count": packet.get_path_hash_count()
            if hasattr(packet, "get_path_hash_count")
            else None,
            "rssi": metadata.get("rssi", 0),
            "snr": metadata.get("snr", 0.0),
            "rx_radio_id": metadata.get("rx_radio_id")
            or getattr(packet, "_rx_radio_id", None)
            or getattr(packet, "rx_radio_id", None),
            "local_transmission": local_transmission,
            "mode": mode,
        }
        prechecked_decision = metadata.get("_policy_precheck_decision")
        if isinstance(prechecked_decision, PolicyDecision):
            policy_decision = prechecked_decision
        else:
            policy_decision = self.policy_engine.evaluate(packet, policy_context)
        policy_reason = None

        if policy_decision.matched:
            logger.info(policy_decision.reason)

        if policy_decision.action == "drop":
            allow_forward = False
            allow_local_tx = False
            policy_reason = self._policy_drop_reason(policy_decision)

        if logger.isEnabledFor(logging.DEBUG):
            logger.debug(
                f"RX packet: header=0x{packet.header:02x}, payload_len={len(packet.payload or b'')}, "
                f"path_len={len(packet.path) if packet.path else 0}, "
                f"rssi={metadata.get('rssi', 'N/A')}, snr={metadata.get('snr', 'N/A')}, mode={mode}"
            )

        # clone the packet to avoid modifying the original
        processed_packet = copy.deepcopy(packet)

        transmitted = False
        tx_delay_ms = 0.0
        drop_reason = None
        tx_radio_id = None
        lbt_attempts = 0
        lbt_backoff_delays_ms = None
        lbt_channel_busy = False
        rx_radio_id = (
            metadata.get("rx_radio_id")
            or getattr(packet, "_rx_radio_id", None)
            or getattr(packet, "rx_radio_id", None)
        )
        tx_radio_ids_sent = None
        if local_transmission and not rx_radio_id:
            # Originated here: companion, repeater/room identity, protocol reply.
            tx_radio_ids = self._resolve_origin_tx_radio_ids()
        else:
            # Heard on RF and being repeated. A helper re-emitting a received
            # packet through the local path (TRACE forwarding) still carries its
            # ingress radio, so it resolves as the relay it is.
            tx_radio_ids = self._resolve_relay_tx_radio_ids(rx_radio_id)

        original_path_hashes = packet.get_path_hashes_hex()
        path_hash_size = packet.get_path_hash_size()

        if not local_transmission:
            payload_type = (
                packet.get_payload_type() if hasattr(packet, "get_payload_type") else None
            )
            frame_len = packet.get_raw_length() if hasattr(packet, "get_raw_length") else 0
            score = flood_rx_metrics(
                frame_len,
                snr,
                self.radio_config["spreading_factor"],
                self.radio_config["bandwidth"],
                self.radio_config["coding_rate"],
                self.radio_config["preamble_length"],
            ).score
            if not self._is_own_echo(pkt_hash_full, original_path_hashes, rx_radio_id):
                self.neighbour_link_tracker.observe(
                    packet,
                    route_type=route_type,
                    payload_type=payload_type,
                    rssi=rssi,
                    snr=snr,
                    score=score,
                    is_duplicate=(pkt_hash_full in self.seen_packets),
                    rx_radio_id=rx_radio_id,
                )

        # Process for forwarding (skip if repeat disabled or if this is a local transmission).
        # Pass pkt_hash_full so flood_forward / direct_forward don't recompute SHA-256.
        result = (
            None
            if (not allow_forward or local_transmission)
            else self.process_packet(processed_packet, snr, packet_hash=pkt_hash_full)
        )
        forwarded_path_hashes = None

        # For local transmissions, create a direct transmission result (if local TX allowed)
        if local_transmission and allow_local_tx:
            # Mark local packet as seen to prevent duplicate processing when received back
            self.mark_seen(packet, packet_hash=pkt_hash_full)
            # Calculate transmission delay for local packets
            delay = self._calculate_tx_delay(packet, snr)
            result = (packet, delay)
            forwarded_path_hashes = packet.get_path_hashes_hex()
            if logger.isEnabledFor(logging.DEBUG):
                logger.debug(f"Local transmission: calculated delay {delay:.3f}s")

        if result:
            fwd_pkt, delay = result
            tx_delay_ms = delay * 1000.0

            # 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, and
            # every packet from one forwarding decision shares its egress set.
            # A fan-out decides send-time normalisation (flood scope, hash mode)
            # up front; do it before metering so every duty-cycle gate sees the
            # bytes that actually go on air.
            fanout = bool(tx_radio_ids) and len(tx_radio_ids) > 1
            extra_tx_tasks = []
            for extra_pkt, extra_delay in getattr(result, "extras", ()):
                if fanout:
                    self._freeze_tx_normalisation(extra_pkt)
                extra_airtime_ms = self.airtime_mgr.calculate_airtime(extra_pkt.get_raw_length())
                extra_tx_tasks.append(
                    (
                        extra_pkt,
                        await self._schedule_egress_tx(
                            extra_pkt, extra_delay, extra_airtime_ms, tx_radio_ids
                        ),
                    )
                )

            # Check duty-cycle before scheduling TX
            if fanout:
                self._freeze_tx_normalisation(fwd_pkt)
            airtime_ms = self.airtime_mgr.calculate_airtime(fwd_pkt.get_raw_length())

            can_tx, wait_time = self.airtime_mgr.can_transmit(airtime_ms)

            # LBT metadata (set after any TX path that awaits send)
            tx_metadata = None
            lbt_attempts = 0
            lbt_backoff_delays_ms = None
            lbt_channel_busy = False

            if not can_tx and not local_transmission:
                logger.warning(
                    f"Duty-cycle limit exceeded. Airtime={airtime_ms:.1f}ms, "
                    f"wait={wait_time:.1f}s before retry"
                )
                self.dropped_count += 1
                drop_reason = DropReason.DUTY_CYCLE_LIMIT
            else:
                if can_tx:
                    failure_reason = "TX failed"
                    failure_label = "Local TX failed"
                else:
                    # Defer local TX until duty cycle allows instead of dropping
                    logger.info(
                        f"Duty-cycle limit: deferring local TX by {wait_time:.1f}s "
                        f"(airtime={airtime_ms:.1f}ms)"
                    )
                    delay += wait_time
                    failure_reason = "TX failed (deferred)"
                    failure_label = "Deferred local TX failed"
                tx_task = await self._schedule_egress_tx(
                    fwd_pkt,
                    delay,
                    airtime_ms,
                    tx_radio_ids,
                    local_transmission=local_transmission,
                )
                try:
                    tx_result = await self._await_egress_tx(tx_task, fwd_pkt)
                except Exception as e:
                    transmitted = False
                    drop_reason = failure_reason
                    logger.warning(f"{failure_label}: {e}")
                    raise
                # One logical forward, however many radios carried it: the
                # per-radio sent_flood/sent_direct counters track physical sends.
                if tx_result.any_success:
                    self.forwarded_count += 1
                    transmitted = True
                else:
                    transmitted = False
                    drop_reason = failure_reason
                    self.dropped_count += 1

                if len(tx_result.results) > 1 and not tx_result.all_success:
                    logger.warning(
                        "Fan-out %s TX failure: %s successful=%s failed=%s",
                        "partial" if tx_result.any_success else "total",
                        f"rx_radio={rx_radio_id}" if rx_radio_id else "origin=local",
                        tx_result.successful_radio_ids,
                        tx_result.failed_radio_ids,
                    )
                tx_radio_ids_sent = tx_result.successful_radio_ids or None

                # Scalar TX fields describe the primary egress.
                primary = tx_result.primary
                if primary is not None and primary.success:
                    tx_radio_id = primary.radio_id
                tx_metadata = primary.metadata if primary is not None else None
                if tx_metadata:
                    lbt_attempts = tx_metadata.get("lbt_attempts", 0)
                    lbt_backoff_delays_ms = tx_metadata.get("lbt_backoff_delays_ms", [])
                    lbt_channel_busy = tx_metadata.get("lbt_channel_busy", False)

                    if lbt_attempts > 0:
                        total_lbt_delay = sum(lbt_backoff_delays_ms)
                        logger.info(
                            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_pkt, extra_task in extra_tx_tasks:
                try:
                    if (await self._await_egress_tx(extra_task, extra_pkt)).any_success:
                        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 DropReason.NO_TX_MODE
            elif not allow_forward:
                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(
                    processed_packet, packet_hash=pkt_hash_full
                )
                if logger.isEnabledFor(logging.DEBUG):
                    logger.debug(f"Packet not forwarded: {drop_reason}")

        # Extract packet type and route from header
        if not hasattr(packet, "header") or packet.header is None:
            logger.error(f"Packet missing header attribute! Packet: {packet}")
            payload_type = 0
            route_type = 0
        else:
            header_info = PacketHeaderUtils.parse_header(packet.header)
            payload_type = header_info["payload_type"]
            route_type = header_info["route_type"]
            if logger.isEnabledFor(logging.DEBUG):
                logger.debug(
                    f"Packet header=0x{packet.header:02x}, type={payload_type}, route={route_type}"
                )

        # Check if this is a duplicate
        is_dupe = pkt_hash_full in self.seen_packets and not transmitted

        # Set drop reason for duplicates and count flood vs direct dups
        if is_dupe and drop_reason is None:
            drop_reason = DropReason.DUPLICATE
        if is_dupe:
            if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD):
                self.flood_dup_count += 1
            elif route_type in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT):
                self.direct_dup_count += 1

        # Expose effective drop reason to PacketRouter.
        # Some Packet implementations are slot-based and cannot accept dynamic attrs.
        metadata["_repeater_drop_reason"] = drop_reason
        try:
            setattr(packet, "_repeater_drop_reason", drop_reason)
        except (AttributeError, TypeError):
            pass

        display_hashes = (
            original_path_hashes if original_path_hashes else packet.get_path_hashes_hex()
        )
        path_hash = self._path_hash_display(display_hashes)
        src_hash, dst_hash = self._packet_record_src_dst(packet, payload_type)

        # Record packet for charts
        packet_record = self._build_packet_record(
            packet,
            payload_type,
            route_type,
            rssi,
            snr,
            original_path_hashes,
            path_hash_size,
            path_hash,
            src_hash,
            dst_hash,
            transmitted=transmitted,
            drop_reason=drop_reason,
            is_duplicate=is_dupe,
            forwarded_path=forwarded_path_hashes,
            tx_delay_ms=tx_delay_ms,
            rx_radio_id=rx_radio_id,
            tx_radio_id=tx_radio_id,
            tx_radio_ids=tx_radio_ids_sent,
            lbt_attempts=lbt_attempts,
            lbt_backoff_delays_ms=lbt_backoff_delays_ms,
            lbt_channel_busy=lbt_channel_busy,
            packet_hash=pkt_hash_full,
        )

        # Store packet record to persistent storage
        # Skip mqtt only for invalid packets (not duplicates or operational drops)
        if self.storage:
            try:
                # Only skip mqtt for actual invalid/bad packets
                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:
                logger.error(f"Failed to store packet record: {e}")

        # If this is a duplicate, try to attach it to the original packet
        if is_dupe and len(self.recent_packets) > 0:
            prev_pkt = self._recent_hash_index.get(packet_record["packet_hash"])
            if prev_pkt is not None:
                # Add duplicate to original packet's duplicate list
                if "duplicates" not in prev_pkt:
                    prev_pkt["duplicates"] = []
                if len(prev_pkt["duplicates"]) < self.max_duplicates_per_packet:
                    prev_pkt["duplicates"].append(packet_record)
                # Don't add duplicate to main list, just track in original
            else:
                # Original not found, add as regular packet
                self._append_recent_packet(packet_record)
        else:
            # Not a duplicate or first occurrence
            self._append_recent_packet(packet_record)

        return transmitted

    @staticmethod
    def _policy_drop_reason(decision: PolicyDecision) -> str:
        if decision.rule_id is None:
            return "Policy blocked packet"
        return f"Policy blocked packet (rule {decision.rule_id})"

    def log_trace_record(self, packet_record: dict) -> None:
        """Manually log a packet trace record (used by external callers)"""
        self._append_recent_packet(packet_record)

        self.rx_count += 1
        if packet_record.get("transmitted", False):
            self.forwarded_count += 1
        else:
            self.dropped_count += 1

        # Store to persistent storage (same as __call__ does)
        if self.storage:
            try:
                self.storage.record_packet(packet_record)
            except Exception as e:
                logger.error(f"Failed to store packet record: {e}")

    def record_packet_only(self, packet: Packet, metadata: dict) -> None:
        """Record a packet for UI/storage without running forwarding or duplicate logic.

        Used by the packet router for injection-only types (ANON_REQ, ACK, PATH, etc.)
        so they still appear in the web UI.

        TRACE packets are excluded: TraceHelper.log_trace_record stores the real trace path;
        packet.path on TRACE holds SNR bytes, not routing hashes.
        """
        if not self.storage:
            return
        rssi = metadata.get("rssi", 0)
        snr = metadata.get("snr", 0.0)
        if not hasattr(packet, "header") or packet.header is None:
            logger.debug("record_packet_only: packet missing header, skipping")
            return
        header_info = PacketHeaderUtils.parse_header(packet.header)
        payload_type = header_info["payload_type"]
        route_type = header_info["route_type"]
        if payload_type == PAYLOAD_TYPE_TRACE:
            return
        original_path_hashes = packet.get_path_hashes_hex()
        path_hash_size = packet.get_path_hash_size()
        path_hash = self._path_hash_display(original_path_hashes)
        src_hash, dst_hash = self._packet_record_src_dst(packet, payload_type)
        packet_record = self._build_packet_record(
            packet,
            payload_type,
            route_type,
            rssi,
            snr,
            original_path_hashes,
            path_hash_size,
            path_hash,
            src_hash,
            dst_hash,
            rx_radio_id=(
                metadata.get("rx_radio_id")
                or getattr(packet, "_rx_radio_id", None)
                or getattr(packet, "rx_radio_id", None)
            ),
            packet_hash=packet.calculate_packet_hash().hex().upper(),
        )
        try:
            self.storage.record_packet(packet_record, skip_mqtt_if_invalid=False)
        except Exception as e:
            logger.error(f"Failed to store packet record (record_packet_only): {e}")
            return
        self._append_recent_packet(packet_record)

    # Another radio of this node hears a send only while it is on the air or
    # shortly after, allowing for a remote modem's delivery delay.
    OWN_ECHO_MARGIN_SECONDS = 2.0
    # A send that never reports back stops counting after this long.
    OWN_ECHO_INFLIGHT_LIMIT_SECONDS = 60.0
    OWN_ECHO_MAX_ENTRIES = 256

    def _note_own_send_started(self, packet: Packet, radio_id: Optional[str]):
        """Record a physical send about to go on air; return a handle for its outcome.

        Only on a node with two or more radios: a lone radio cannot hear itself.
        """
        _, radio_ids = self._fabric_endpoints()
        if not radio_ids or len(radio_ids) < 2:
            return None
        path_hashes = packet.get_path_hashes_hex()
        if not path_hashes:
            return None
        key = (packet.calculate_packet_hash().hex().upper(), tuple(path_hashes))
        record = {"radio_id": radio_id, "started": time.monotonic(), "finished": None}
        sends = self._recent_own_tx.pop(key, [])
        sends.append(record)
        self._recent_own_tx[key] = sends
        while len(self._recent_own_tx) > self.OWN_ECHO_MAX_ENTRIES:
            self._recent_own_tx.popitem(last=False)
        return key, record

    def _note_own_send_finished(self, handle, packet: Packet, *, sent: bool) -> None:
        """Close a send: keep it for its echo window, or forget it if nothing went on air."""
        if handle is None:
            return
        key, record = handle
        if record["finished"] is not None:
            return
        record["finished"] = time.monotonic()
        if not sent:
            sends = [r for r in self._recent_own_tx.get(key, []) if r is not record]
            if sends:
                self._recent_own_tx[key] = sends
            else:
                self._recent_own_tx.pop(key, None)
            return
        if record["radio_id"] is None:
            metadata = getattr(packet, "_tx_metadata", None)
            if isinstance(metadata, dict):
                record["radio_id"] = metadata.get("radio_id") or metadata.get("tx_radio_id")

    def _own_send_is_live(self, record: dict, now: float) -> bool:
        if record["finished"] is None:
            return now - record["started"] <= self.OWN_ECHO_INFLIGHT_LIMIT_SECONDS
        return now - record["finished"] <= self.OWN_ECHO_MARGIN_SECONDS

    def _is_own_echo(self, packet_hash: str, path_hashes, rx_radio_id) -> bool:
        """True when another radio of this node is hearing our own transmission.

        The reception must carry exactly the path we sent, arrive on a radio other
        than the one that sent it, and arrive while that send is on the air or
        within OWN_ECHO_MARGIN_SECONDS of it finishing. A neighbour repeating our
        copy appends its own hash; one whose 1-byte hash collides with ours is only
        mistaken for an echo if it repeats the same packet on the same path inside
        that window.
        """
        if rx_radio_id is None or not path_hashes or not self._recent_own_tx:
            return False
        now = time.monotonic()
        while self._recent_own_tx:
            sends = next(iter(self._recent_own_tx.values()))
            if any(self._own_send_is_live(record, now) for record in sends):
                break
            self._recent_own_tx.popitem(last=False)
        for record in self._recent_own_tx.get((packet_hash, tuple(path_hashes)), ()):
            if not self._own_send_is_live(record, now):
                continue
            sender = record["radio_id"]
            if sender is None or str(sender) != str(rx_radio_id):
                return True
        return False

    def record_duplicate(self, packet: Packet, rssi: int = 0, snr: float = 0.0) -> None:
        """Record a known-duplicate packet for UI/storage visibility without forwarding.

        Called by the raw_packet_subscriber path so that path variants blocked
        by the Dispatcher's payload-based dedup still appear in the UI.
        """
        self.rx_count += 1
        route_type = packet.header & PH_ROUTE_MASK
        if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD):
            self.recv_flood_count += 1
            self.flood_dup_count += 1
        elif route_type in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT):
            self.recv_direct_count += 1
            self.direct_dup_count += 1

        header_info = PacketHeaderUtils.parse_header(packet.header)
        payload_type = header_info["payload_type"]
        route_type_parsed = header_info["route_type"]

        original_path_hashes = packet.get_path_hashes_hex()
        path_hash_size = packet.get_path_hash_size()
        path_hash = self._path_hash_display(original_path_hashes)
        src_hash, dst_hash = self._packet_record_src_dst(packet, payload_type)
        pkt_hash_full = packet.calculate_packet_hash().hex().upper()

        frame_len = packet.get_raw_length() if hasattr(packet, "get_raw_length") else 0
        score = flood_rx_metrics(
            frame_len,
            snr,
            self.radio_config["spreading_factor"],
            self.radio_config["bandwidth"],
            self.radio_config["coding_rate"],
            self.radio_config["preamble_length"],
        ).score
        rx_radio_id = getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None)
        if not self._is_own_echo(pkt_hash_full, original_path_hashes, rx_radio_id):
            self.neighbour_link_tracker.observe(
                packet,
                route_type=route_type,
                payload_type=payload_type,
                rssi=float(rssi),
                snr=float(snr),
                score=score,
                is_duplicate=True,
                rx_radio_id=rx_radio_id,
            )

        packet_record = self._build_packet_record(
            packet,
            payload_type,
            route_type_parsed,
            rssi,
            snr,
            original_path_hashes,
            path_hash_size,
            path_hash,
            src_hash,
            dst_hash,
            transmitted=False,
            drop_reason=DropReason.DUPLICATE,
            is_duplicate=True,
            rx_radio_id=(
                getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None)
            ),
            packet_hash=pkt_hash_full,
        )

        if self.storage:
            try:
                self.storage.record_packet(packet_record, skip_mqtt_if_invalid=False)
            except Exception as e:
                logger.error(f"Failed to store duplicate record: {e}")

        # Group under original in recent_packets
        if len(self.recent_packets) > 0:
            prev_pkt = self._recent_hash_index.get(packet_record["packet_hash"])
            if prev_pkt is not None:
                if "duplicates" not in prev_pkt:
                    prev_pkt["duplicates"] = []
                prev_pkt["duplicates"].append(packet_record)
            else:
                self._append_recent_packet(packet_record)
        else:
            self._append_recent_packet(packet_record)

    def cleanup_cache(self):

        now = time.time()
        expired = [k for k, ts in self.seen_packets.items() if now - ts > self.cache_ttl]
        for k in expired:
            del self.seen_packets[k]

    def _path_hash_display(self, display_hashes) -> Optional[str]:
        """Build path hash string for packet record from path hashes list."""
        if not display_hashes:
            return None
        display = display_hashes[:8]
        if len(display_hashes) > 8:
            display = list(display) + ["..."]
        return "[" + ", ".join(display) + "]"

    def _packet_record_src_dst(
        self, packet: Packet, payload_type: int
    ) -> Tuple[Optional[str], Optional[str]]:
        """Return (src_hash, dst_hash) for packet_record from packet and payload_type."""
        src_hash = None
        dst_hash = None
        payload = getattr(packet, "payload", None)
        if payload_type in [0x00, 0x01, 0x02, 0x08]:
            if payload and len(payload) >= 2:
                dst_hash = f"{payload[0]:02X}"
                src_hash = f"{payload[1]:02X}"
        elif payload_type == PAYLOAD_TYPE_ADVERT:
            if payload and len(payload) >= 1:
                src_hash = f"{payload[0]:02X}"
        elif payload_type == PAYLOAD_TYPE_ANON_REQ:
            if payload and len(payload) >= 1:
                dst_hash = f"{payload[0]:02X}"
        return (src_hash, dst_hash)

    def _build_packet_record(
        self,
        packet: Packet,
        payload_type: int,
        route_type: int,
        rssi: int,
        snr: float,
        original_path_hashes,
        path_hash_size: int,
        path_hash: Optional[str],
        src_hash: Optional[str],
        dst_hash: Optional[str],
        *,
        transmitted: bool = False,
        drop_reason: Optional[str] = None,
        is_duplicate: bool = False,
        forwarded_path=None,
        tx_delay_ms: float = 0.0,
        rx_radio_id: Optional[str] = None,
        tx_radio_id: Optional[str] = None,
        tx_radio_ids=None,
        lbt_attempts: int = 0,
        lbt_backoff_delays_ms=None,
        lbt_channel_busy: bool = False,
        packet_hash: Optional[str] = None,
    ) -> dict:
        """Build a single packet_record dict for storage and recent_packets.

        ``tx_radio_id`` is the primary successful egress; ``tx_radio_ids`` lists
        every radio that successfully transmitted this logical packet.
        """
        pkt_hash = packet_hash or packet.calculate_packet_hash().hex().upper()
        payload = getattr(packet, "payload", None)
        payload_len = len(payload or b"")
        upstream_hash, upstream_hash_size = self.neighbour_link_tracker.get_upstream_peer_identity(
            packet,
            route_type,
            payload_type,
            path_hashes=original_path_hashes,
            path_hash_size=path_hash_size,
        )

        # LoRa time-on-air for this packet (Semtech reference formula).
        # Computed once here so every downstream consumer (MQTT, SQLite, Glass,
        # websocket) reads the same value instead of recomputing or shipping
        # zeros. The same calculator is used by the RX accumulator and the
        # duty-cycle gate, keeping reporting and metering aligned.
        airtime_ms = 0.0
        try:
            raw_len = packet.get_raw_length() if hasattr(packet, "get_raw_length") else 0
            if raw_len > 0:
                airtime_ms = float(self.airtime_mgr.calculate_airtime(raw_len))
        except Exception as e:
            logger.debug(f"Could not compute airtime for packet record: {e}")

        return {
            "timestamp": time.time(),
            "header": (
                f"0x{packet.header:02X}"
                if hasattr(packet, "header") and packet.header is not None
                else None
            ),
            "payload": payload.hex() if payload else None,
            "payload_length": len(payload) if payload else 0,
            "type": payload_type,
            "route": route_type,
            "length": payload_len,
            "rssi": rssi,
            "snr": snr,
            "score": flood_rx_metrics(
                packet.get_raw_length() if hasattr(packet, "get_raw_length") else 0,
                snr,
                self.radio_config["spreading_factor"],
                self.radio_config["bandwidth"],
                self.radio_config["coding_rate"],
                self.radio_config["preamble_length"],
            ).score,
            "tx_delay_ms": tx_delay_ms,
            "rx_radio_id": rx_radio_id
            or getattr(packet, "_rx_radio_id", None)
            or getattr(packet, "rx_radio_id", None),
            "tx_radio_id": tx_radio_id,
            "tx_radio_ids": list(tx_radio_ids) if tx_radio_ids else None,
            "airtime_ms": airtime_ms,
            "transmitted": transmitted,
            "is_duplicate": is_duplicate,
            "packet_hash": pkt_hash[:16],
            "drop_reason": drop_reason,
            "path_hash": path_hash,
            "src_hash": src_hash,
            "dst_hash": dst_hash,
            "original_path": original_path_hashes or None,
            "forwarded_path": forwarded_path,
            "path_hash_size": path_hash_size,
            "upstream_hash": upstream_hash,
            "upstream_hash_size": upstream_hash_size,
            "raw_packet": packet.write_to().hex() if hasattr(packet, "write_to") else None,
            "lbt_attempts": lbt_attempts,
            "lbt_backoff_delays_ms": lbt_backoff_delays_ms,
            "lbt_channel_busy": lbt_channel_busy,
        }

    def _append_recent_packet(self, packet_record: dict) -> None:
        """Append packet to bounded recent list and keep hash index aligned."""
        if len(self.recent_packets) >= self.max_recent_packets:
            oldest = self.recent_packets.popleft()
            oldest_hash = oldest.get("packet_hash") if isinstance(oldest, dict) else None
            if oldest_hash and self._recent_hash_index.get(oldest_hash) is oldest:
                del self._recent_hash_index[oldest_hash]

        self.recent_packets.append(packet_record)
        pkt_hash = packet_record.get("packet_hash") if isinstance(packet_record, dict) else None
        if pkt_hash:
            self._recent_hash_index[pkt_hash] = packet_record

    def _get_drop_reason(self, packet: Packet, packet_hash: Optional[str] = None) -> str:

        if self.is_duplicate(packet, packet_hash=packet_hash):
            return DropReason.DUPLICATE

        if not packet or not packet.payload:
            return DropReason.EMPTY_PAYLOAD

        if len(packet.path or []) > MAX_PATH_SIZE:
            return DropReason.PATH_TOO_LONG

        route_type = packet.header & PH_ROUTE_MASK

        if route_type == ROUTE_TYPE_FLOOD:
            # Check if unscoped flood policy blocked it
            unscoped_flood_allow = self.config.get("mesh", {}).get(
                "unscoped_flood_allow", self.config.get("mesh", {}).get("global_flood_allow", True)
            )
            if not unscoped_flood_allow:
                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 DropReason.DIRECT_NO_PATH
            next_hop = bytes(packet.path[:hash_size])
            if next_hop != self.local_hash_bytes[:hash_size]:
                return DropReason.DIRECT_NOT_FOR_US

        # Default reason
        return "Unknown"

    def is_duplicate(self, packet: Packet, packet_hash: Optional[str] = None) -> bool:
        """Return True if this packet has already been seen.

        Accepts an optional pre-computed packet_hash to avoid a redundant SHA-256
        when the caller (e.g. __call__ → process_packet → flood/direct_forward)
        has already calculated the hash.  Falls back to computing it if not provided.

        INVARIANT: this method is synchronous with no await points.  The caller
        (process_packet / __call__) relies on is_duplicate + mark_seen being
        effectively atomic within the asyncio event loop.  Do NOT add any await
        here without revisiting that invariant.
        """
        pkt_hash = packet_hash or packet.calculate_packet_hash().hex().upper()
        return pkt_hash in self.seen_packets

    def mark_seen(self, packet: Packet, packet_hash: Optional[str] = None):

        pkt_hash = packet_hash or packet.calculate_packet_hash().hex().upper()
        self.seen_packets[pkt_hash] = time.time()

        if len(self.seen_packets) > self.max_cache_size:
            self.seen_packets.popitem(last=False)

    def validate_packet(self, packet: Packet) -> Tuple[bool, str]:

        if not packet or not packet.payload:
            return False, DropReason.EMPTY_PAYLOAD

        if packet.get_path_hash_size() > 3:
            return False, "Reserved path hash size is invalid"

        if len(packet.path or []) > MAX_PATH_SIZE:
            return (
                False,
                f"Path length {len(packet.path or [])} exceeds MAX_PATH_SIZE ({MAX_PATH_SIZE})",
            )

        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()
            if normalized in {
                LOOP_DETECT_OFF,
                LOOP_DETECT_MINIMAL,
                LOOP_DETECT_MODERATE,
                LOOP_DETECT_STRICT,
            }:
                return normalized
        return LOOP_DETECT_OFF

    def _get_loop_detect_mode(self) -> str:
        return self.loop_detect_mode

    def _is_flood_looped(self, packet: Packet, mode: Optional[str] = None) -> bool:
        mode = mode or self._get_loop_detect_mode()
        if mode == LOOP_DETECT_OFF:
            return False

        max_counters = LOOP_DETECT_MAX_COUNTERS.get(mode)
        if max_counters is None:
            return False

        hash_size = packet.get_path_hash_size()
        if hash_size not in max_counters:
            return False

        max_counter = max_counters[hash_size]
        hop_count = packet.get_path_hash_count()
        path = packet.path or bytearray()
        local_hash = self.local_hash_bytes[:hash_size]
        local_count = sum(
            1
            for i in range(hop_count)
            if bytes(path[i * hash_size : (i + 1) * hash_size]) == local_hash
        )
        return local_count >= max_counter

    def _check_transport_codes(self, packet: Packet) -> Tuple[bool, str]:

        if not self.storage:
            logger.warning("Transport code check failed: no storage available")
            return False, "No storage available for transport key validation"

        try:
            from openhop_core.protocol.transport_keys import calc_transport_code

            # Check cache validity
            current_time = time.time()
            if (
                self._transport_keys_cache is None
                or current_time - self._transport_keys_cache_time > self._transport_keys_cache_ttl
            ):
                # Refresh cache
                self._transport_keys_cache = self.storage.get_transport_keys()
                self._transport_keys_cache_time = current_time

            transport_keys = self._transport_keys_cache

            if not transport_keys:
                return False, "No transport keys configured"

            # Check if packet has transport codes
            if not packet.has_transport_codes():
                return False, "No transport codes present"

            transport_code_0 = packet.transport_codes[0]  # First transport code

            # Check packet against each transport key
            for key_record in transport_keys:
                transport_key_encoded = key_record.get("transport_key")
                key_name = key_record.get("name", "unknown")
                flood_policy = key_record.get("flood_policy", "deny")

                if not transport_key_encoded:
                    continue

                try:
                    import base64

                    transport_key = base64.b64decode(transport_key_encoded)
                    expected_code = calc_transport_code(transport_key, packet)
                    if transport_code_0 == expected_code:
                        if logger.isEnabledFor(logging.DEBUG):
                            logger.debug(
                                f"Transport code validated for key '{key_name}' with policy '{flood_policy}'"
                            )

                        # Update last_used timestamp for this key
                        try:
                            key_id = key_record.get("id")
                            if key_id:
                                self.storage.update_transport_key(
                                    key_id=key_id, last_used=time.time()
                                )
                                if logger.isEnabledFor(logging.DEBUG):
                                    logger.debug(
                                        f"Updated last_used timestamp for transport key '{key_name}'"
                                    )
                        except Exception as e:
                            logger.warning(
                                f"Failed to update last_used for transport key '{key_name}': {e}"
                            )

                        # Check flood policy for this key
                        if flood_policy == "allow":
                            return True, ""
                        else:
                            return False, f"Transport key '{key_name}' flood policy denied"

                except Exception as e:
                    logger.warning(f"Error checking transport key '{key_name}': {e}")
                    continue

            # No matching transport code found
            if logger.isEnabledFor(logging.DEBUG):
                logger.debug(
                    f"Transport code 0x{transport_code_0:04X} denied (checked {len(transport_keys)} keys)"
                )
            return False, "No matching transport code"

        except Exception as e:
            logger.error(f"Transport code validation error: {e}")
            return False, f"Transport code validation error: {e}"

    def flood_forward(self, packet: Packet, packet_hash: Optional[str] = None) -> Optional[Packet]:
        """Forward a FLOOD packet, appending our hash to the path.

        INVARIANT: purely synchronous — no await points.  The is_duplicate +
        mark_seen pair is atomic within the asyncio event loop.  Do NOT add any
        await here without revisiting that invariant in __call__ / process_packet.
        """
        # Validate
        valid, reason = self.validate_packet(packet)
        if not valid:
            packet.drop_reason = reason
            return None

        # Check if packet is marked do-not-retransmit
        if packet.is_marked_do_not_retransmit():
            # Check if packet has custom drop reason
            if not packet.drop_reason:
                packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT
            return None

        # Check unscoped flood policy
        unscoped_flood_allow = self.config.get("mesh", {}).get(
            "unscoped_flood_allow", self.config.get("mesh", {}).get("global_flood_allow", True)
        )
        route_type = packet.header & PH_ROUTE_MASK
        if route_type == ROUTE_TYPE_FLOOD:
            if not unscoped_flood_allow:
                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 = DropReason.TRANSPORT_CODE_NOT_ALLOWED
                return None

        mode = self._get_loop_detect_mode()
        if self._is_flood_looped(packet, 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 = DropReason.DUPLICATE
            return None

        if packet.path is None:
            packet.path = bytearray()
        elif not isinstance(packet.path, bytearray):
            packet.path = bytearray(packet.path)

        hash_size = packet.get_path_hash_size()
        hop_count = packet.get_path_hash_count()

        if self.max_flood_hops > 0 and 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 = 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 = DropReason.PATH_EXCEEDS_MAX_SIZE
            return None

        self.mark_seen(packet, packet_hash=packet_hash)

        # Append hash_size bytes from our public key prefix
        packet.path.extend(self.local_hash_bytes[:hash_size])
        packet.path_len = PathUtils.encode_path_len(hash_size, hop_count + 1)

        # This is a transit packet received over RF. Its existing scope decision
        # is authoritative: a plain flood must remain plain, and a transport
        # flood must retain the originator's codes when the shared dispatcher
        # performs its send-time scope resolution.
        packet._flood_scope_applied = True

        return packet

    def direct_forward(self, packet: Packet, packet_hash: Optional[str] = None) -> Optional[Packet]:
        """Forward a DIRECT packet, removing the first hop from the path.

        INVARIANT: purely synchronous — no await points.  The is_duplicate +
        mark_seen pair is atomic within the asyncio event loop.  Do NOT add any
        await here without revisiting that invariant in __call__ / process_packet.
        """
        # Validate packet (empty payload, oversized path, etc.)
        valid, reason = self.validate_packet(packet)
        if not valid:
            packet.drop_reason = reason
            return None

        # Check if packet is marked do-not-retransmit
        if packet.is_marked_do_not_retransmit():
            if not packet.drop_reason:
                packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT
            return None

        hash_size = packet.get_path_hash_size()
        hop_count = packet.get_path_hash_count()

        # Check if we're the next hop
        if not packet.path or len(packet.path) < hash_size:
            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 = 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 = DropReason.DUPLICATE
            return None

        self.mark_seen(packet, packet_hash=packet_hash)

        # Remove first hash entry (hash_size bytes)
        packet.path = bytearray(packet.path[hash_size:])
        packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1)

        return packet

    # Multipart ACKs are spaced ~300ms apart per remaining fragment, matching
    # MeshCore's forwardMultipartDirect (`(remaining + 1) * 300`).
    MULTIPART_ACK_SPACING_MS = 300

    def forward_multipart_direct(
        self, packet: Packet, 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
        packet.  Only a multipart-ACK at an intermediate direct hop is relayed,
        and it is relayed by regenerating the embedded ACK as a plain DIRECT ACK
        (the wrapper byte stripped) rather than repeating the wrapper.  Every
        other multipart case — flood routing, a final hop, or a non-ACK embedded
        type — is dropped.  Returns 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 —
        the is_duplicate + mark_seen pair must stay atomic within the event loop.
        """
        # MeshCore only forwards multipart on the direct-route branch; the flood
        # switch case updates ACK state locally but never re-routes it.
        if not packet.is_route_direct():
            packet.drop_reason = f"{DropReason.MULTIPART}: not direct-routed"
            return None

        valid, reason = self.validate_packet(packet)
        if not valid:
            packet.drop_reason = reason
            return None

        if packet.is_marked_do_not_retransmit():
            packet.drop_reason = packet.drop_reason or DropReason.MARKED_DO_NOT_RETRANSMIT
            return None

        payload = packet.payload or b""
        remaining = payload[0] >> 4
        embedded_type = payload[0] & 0x0F
        # Only multipart-ACKs are relayed; MeshCore leaves other types for future
        # use and does not forward them.
        if embedded_type != PAYLOAD_TYPE_ACK or len(payload) < 5:
            packet.drop_reason = f"{DropReason.MULTIPART}: unsupported embedded type"
            return None

        hash_size = packet.get_path_hash_size()
        hop_count = packet.get_path_hash_count()

        # We must be the next hop (a final-hop multipart has no remaining path and
        # is handled locally, not forwarded).
        if not packet.path or len(packet.path) < hash_size:
            packet.drop_reason = 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

        # 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, 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
        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, 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:
        """Reception-quality score in [0, 1] via the shared core scorer
        (MeshCore RadioLibWrappers packetScoreInt)."""
        return packet_score(snr, spreading_factor, packet_len)

    def _calculate_tx_delay(self, packet: Packet, snr: float = 0.0) -> float:

        packet_len = packet.get_raw_length()
        airtime_ms = self.airtime_mgr.calculate_airtime(packet_len)

        if packet.is_route_flood():
            delay_factor = self.tx_delay_factor
        elif packet.is_route_direct():
            delay_factor = self.direct_tx_delay_factor
        else:
            delay_factor = self.direct_tx_delay_factor

        # MeshCore uses the same random window for flood and direct variants:
        # choose a value from 0 through 5 * airtime * factor.
        base_delay_ms = airtime_ms * delay_factor
        random_mult = secrets.randbelow(5001) / 1000.0
        delay_s = (base_delay_ms * random_mult) / 1000.0

        # OpenHop's optional score gate is applied only after the 50 ms threshold.
        if delay_s >= 0.05 and self.use_score_for_tx:
            score = self.calculate_packet_score(snr, packet_len)
            # Higher score = shorter delay: max(0.2, 1.0 - score)
            # score 1.0 → multiplier 0.2 (20% of original)
            # score 0.0 → multiplier 1.0 (100% of original)
            score_multiplier = max(0.2, 1.0 - score)
            delay_s = delay_s * score_multiplier
            if logger.isEnabledFor(logging.DEBUG):
                logger.debug(
                    f"Congestion detected (delay >= 50ms), score={score:.2f}, "
                    f"delay multiplier={score_multiplier:.2f}"
                )

        # Cap at 5 seconds maximum
        delay_s = min(delay_s, 5.0)

        if logger.isEnabledFor(logging.DEBUG):
            logger.debug(
                f"Route={'FLOOD' if packet.is_route_flood() else 'DIRECT'}, "
                f"len={packet_len}B, airtime={airtime_ms:.1f}ms, delay={delay_s:.3f}s"
            )

        return delay_s

    def process_packet(
        self,
        packet: Packet,
        snr: float = 0.0,
        packet_hash: Optional[str] = None,
    ) -> Optional[Tuple[Packet, float]]:
        """Route a received packet to flood_forward or direct_forward.

        packet_hash is the pre-computed SHA-256 hex string from __call__.
        Passing it here avoids recomputing the hash in flood_forward /
        direct_forward / is_duplicate / mark_seen — reducing SHA-256 calls
        from 3 per forwarded packet to 1.
        """
        route_type = packet.header & PH_ROUTE_MASK

        # MeshCore routes multipart traffic through its own branch, never the
        # generic flood/direct path — it regenerates embedded ACKs and drops
        # everything else.  Its delay is fixed by the fragment count, so return
        # the (packet, delay) pair directly instead of _calculate_tx_delay.
        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)
            if fwd_pkt is None:
                return None
            delay = self._calculate_tx_delay(fwd_pkt, snr)
            return fwd_pkt, delay

        elif route_type == ROUTE_TYPE_DIRECT or route_type == ROUTE_TYPE_TRANSPORT_DIRECT:
            fwd_pkt = self.direct_forward(packet, packet_hash=packet_hash)
            if fwd_pkt is None:
                return None
            delay = self._calculate_tx_delay(fwd_pkt, snr)
            return fwd_pkt, delay

        else:
            packet.drop_reason = f"Unknown route type: {route_type}"
            return None

    # A local transmit whose radio link has dropped waits this long for it to come
    # back before giving up, polling at LOCAL_TX_LINK_POLL_S. Kept well inside the
    # companion client's own command timeout (meshcore_py: 15s), which is already
    # partly spent on the TX delay by the time we get here.
    DEFAULT_LOCAL_TX_LINK_WAIT_S = 5.0
    # Hard ceiling regardless of configuration: the companion client is already
    # waiting on this send, and an unbounded (or infinite) value would hold its
    # command open forever rather than reporting a failure it can act on.
    MAX_LOCAL_TX_LINK_WAIT_S = 30.0
    LOCAL_TX_LINK_POLL_S = 0.1

    def _local_tx_link_wait_s(self) -> float:
        """Configured ceiling for waiting out a radio-link outage on a local TX."""
        delays = self.config.get("delays") or {}
        try:
            wait = float(
                delays.get("local_tx_link_wait_seconds", self.DEFAULT_LOCAL_TX_LINK_WAIT_S)
            )
        except (TypeError, ValueError):
            return self.DEFAULT_LOCAL_TX_LINK_WAIT_S
        if not math.isfinite(wait):
            logger.warning(
                "local_tx_link_wait_seconds=%r is not a finite number; using %.1fs",
                wait,
                self.DEFAULT_LOCAL_TX_LINK_WAIT_S,
            )
            return self.DEFAULT_LOCAL_TX_LINK_WAIT_S
        return min(max(0.0, wait), self.MAX_LOCAL_TX_LINK_WAIT_S)

    def _resolve_tx_radio(self, radio_id: Optional[str] = None):
        """The radio a send will actually use, or None when that cannot be told.

        Attribute access on a FabricRadio passes through to its *default* endpoint
        (rf_fabric/fabric.py), so reading link state off it while the send targets
        another radio_id answers about the wrong radio -- suppressing a healthy one,
        or waving through a dead one. Resolve the named endpoint instead, and return
        None rather than guess: no answer leaves the send to proceed as before.
        """
        radio = getattr(self.dispatcher, "radio", None)
        if radio is None:
            return None
        fabric = getattr(radio, "fabric", None)
        if fabric is None and hasattr(radio, "get_radio"):
            fabric = radio  # an RFFabric handed to the dispatcher directly
        radios = getattr(fabric, "radios", None) if fabric is not None else None
        # isinstance, not truthiness: RFFabric.radios is an OrderedDict, so this
        # recognises a real multi-endpoint fabric without mistaking a stand-in
        # attribute on a single-radio wrapper for one.
        if not isinstance(radios, Mapping) or len(radios) <= 1:
            return radio  # one radio, and it is the one that transmits
        if not radio_id:
            # The fabric picks its endpoint at send time, so there is nothing here
            # to inspect. Say so rather than answer about whichever radio happens
            # to be the default -- a wrong "down" would suppress a healthy send.
            return None
        getter = getattr(fabric, "get_radio", None)
        if getter is None:
            return None
        try:
            return getter(str(radio_id))
        except Exception:
            return None

    def _radio_link_down(self, radio_id: Optional[str] = None) -> bool:
        """True when the radio this send will use reports its link unusable.

        Only radios that track a link state answer this -- a KISS modem over serial
        does, a directly-driven SX1262 has none to lose -- so it reads False for the
        rest and leaves their timing untouched.
        """
        radio = self._resolve_tx_radio(radio_id)
        if radio is None:
            return False
        # Identity checks, not truthiness: a radio that does not track link state
        # answers with something that is neither True nor False (or nothing at all),
        # and that has to read as "no opinion" rather than "down".
        if getattr(radio, "is_connected", None) is False:
            return True
        return getattr(radio, "is_degraded", None) is True

    async def _await_radio_link(self, timeout: float, radio_id: Optional[str] = None) -> bool:
        """Wait up to *timeout* seconds for the radio link, True if it is usable."""
        if not self._radio_link_down(radio_id):
            return True
        if timeout <= 0:
            return False
        logger.info("Local TX: radio link down, waiting up to %.1fs for it", timeout)
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            await asyncio.sleep(self.LOCAL_TX_LINK_POLL_S)
            if not self._radio_link_down(radio_id):
                logger.info("Local TX: radio link back, transmitting now")
                return True
        logger.warning("Local TX: radio link still down after %.1fs, giving up", timeout)
        return False

    async def schedule_retransmit(
        self,
        fwd_pkt: Packet,
        delay: float,
        airtime_ms: float = 0.0,
        local_transmission: bool = False,
        preferred_tx_radio_id: Optional[str] = None,
    ):
        """Schedule a packet retransmission with delay and return the task.

        If local_transmission is True and the first send fails, retry once after
        a short delay (handles transient radio/LBT failures).
        """

        async def delayed_send():
            await asyncio.sleep(delay)

            # Each attempt gets its own lock acquisition so the 1-second retry
            # backoff (local_transmission only) happens OUTSIDE the lock.
            # Holding _tx_lock across asyncio.sleep(1.0) would block every other
            # queued TX task for the full backoff period.
            #
            # Loop runs once for relayed packets, twice for local_transmission:
            #   attempt 0 — initial try (no pre-sleep)
            #   attempt 1 — retry after 1s backoff outside the lock
            for attempt in range(2 if local_transmission else 1):
                if attempt > 0 and not self._radio_link_down(preferred_tx_radio_id):
                    # A transient failure with the link still up: back off OUTSIDE
                    # the lock — other tasks can transmit here.
                    logger.info("Retrying local TX in 1s (lock released during backoff)...")
                    await asyncio.sleep(1.0)

                # A dropped link is not a transient failure to back off from. A KISS
                # modem that falls off USB re-enumerates and re-handshakes in ~2s, so
                # the flat backoff above would land inside the outage and spend the
                # retry on a link that is still down — which is exactly what turned
                # one modem hiccup into a failed message. Wait for the link instead.
                if local_transmission and not await self._await_radio_link(
                    self._local_tx_link_wait_s(), preferred_tx_radio_id
                ):
                    return False

                async with self._tx_lock:
                    # ── Authoritative duty-cycle gate ──────────────────────────
                    # The upfront can_transmit() call in __call__ is advisory: it
                    # avoids scheduling packets obviously over budget, but cannot
                    # prevent a race between tasks whose delay timers expire nearly
                    # simultaneously.  Both pass the advisory check before either
                    # records airtime, then both attempt to transmit.
                    #
                    # Inside _tx_lock only one task runs at a time.  The check and
                    # record_tx() are effectively atomic — no TOCTOU window.
                    # Re-checked every attempt because airtime state may change
                    # while we wait for the lock or sleep through backoff.
                    if airtime_ms > 0:
                        can_tx_now, _ = self.airtime_mgr.can_transmit(airtime_ms)
                        if not can_tx_now:
                            logger.warning(
                                "Packet dropped at TX time: duty-cycle exceeded (airtime=%.1fms)",
                                airtime_ms,
                            )
                            return False

                    # Registered as the send starts: another radio of this node can
                    # hear it while it is still on the air.
                    own_send = self._note_own_send_started(fwd_pkt, preferred_tx_radio_id)
                    try:
                        if preferred_tx_radio_id:
                            try:
                                sent = await self.dispatcher.send_packet(
                                    fwd_pkt,
                                    wait_for_ack=False,
                                    radio_id=preferred_tx_radio_id,
                                )
                            except TypeError:
                                sent = await self.dispatcher.send_packet(
                                    fwd_pkt, wait_for_ack=False
                                )
                        else:
                            sent = await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False)
                        self._note_own_send_finished(own_send, fwd_pkt, sent=bool(sent))
                        if not sent:
                            logger.warning(
                                "Retransmit failed (attempt %d): dispatcher returned false",
                                attempt + 1,
                            )
                            if local_transmission and attempt == 0:
                                continue
                            return False
                        self._record_packet_sent(fwd_pkt)
                        if airtime_ms > 0:
                            self.airtime_mgr.record_tx(airtime_ms)
                        packet_size = fwd_pkt.get_raw_length()
                        logger.info(
                            f"Retransmitted packet ({packet_size} bytes, "
                            f"{airtime_ms:.1f}ms airtime)"
                        )
                        return True
                    except Exception as e:
                        self._note_own_send_finished(own_send, fwd_pkt, sent=False)
                        logger.error(f"Retransmit failed (attempt {attempt + 1}): {e}")
                        if local_transmission and attempt == 0:
                            pass  # release lock, outer loop sleeps, then retries
                        else:
                            raise
            return False

        return asyncio.create_task(delayed_send())

    async def schedule_retransmit_fanout(
        self,
        fwd_pkt: Packet,
        delay: float,
        airtime_ms: float,
        radio_ids,
        *,
        local_transmission: bool = False,
    ):
        """Schedule one logical packet as one physical TX per radio; return the task.

        The task resolves to a ``FanoutTxResult``. Each egress runs through
        ``schedule_retransmit`` — duty-cycle gate, TX lock, local retry and
        per-radio link wait — so the sends stay serialised and one radio's
        failure never stops another's.

        The first egress transmits ``fwd_pkt`` itself, so TX-time side effects
        (flood scope, ``_tx_metadata``) land where callers already look. Every
        other egress gets its own copy, so one send's metadata never overwrites
        another's.
        """
        radio_ids = tuple(radio_ids)
        self._freeze_tx_normalisation(fwd_pkt)
        packets = [fwd_pkt] + [copy.deepcopy(fwd_pkt) for _ in radio_ids[1:]]

        async def run() -> FanoutTxResult:
            await asyncio.sleep(delay)
            # One shared delay, then the per-egress sends are created in order with
            # no further sleep. When every radio is ready they reach the TX lock in
            # egress order (asyncio runs ready tasks and grants the lock FIFO), so
            # the primary egress gets the first claim on the duty-cycle budget. A
            # local send whose radio link is down waits outside the lock, so a
            # healthy radio later in the order is not held up by it and may go first.
            tasks = [
                await self.schedule_retransmit(
                    pkt,
                    0.0,
                    airtime_ms,
                    local_transmission=local_transmission,
                    preferred_tx_radio_id=radio_id,
                )
                for pkt, radio_id in zip(packets, radio_ids)
            ]
            outcomes = await asyncio.gather(*tasks, return_exceptions=True)
            results = []
            for radio_id, pkt, outcome in zip(radio_ids, packets, outcomes):
                if isinstance(outcome, BaseException):
                    logger.warning("Fan-out TX via %s failed: %s", radio_id, outcome)
                    results.append(RadioTxResult(radio_id, False, error=outcome))
                    continue
                metadata = getattr(pkt, "_tx_metadata", None) if outcome else None
                results.append(
                    RadioTxResult(
                        radio_id,
                        bool(outcome),
                        metadata=metadata if isinstance(metadata, dict) else None,
                    )
                )
            return FanoutTxResult(results)

        return asyncio.create_task(run())

    def _freeze_tx_normalisation(self, pkt: Packet) -> None:
        """Apply the dispatcher's send-time normalisation once and mark it decided.

        The dispatcher resolves flood scope and the default path-hash mode at
        send time. Fan-out sends the same logical packet several times, possibly
        seconds apart (link wait, retry backoff), so a live change to the node's
        default region or hash mode in between would put different bytes on each
        radio. Deciding both on the logical packet before it is copied keeps
        every egress, and the packet the router echoes, identical.
        """
        for hook in ("_apply_flood_scope", "_apply_default_path_hash_mode"):
            apply = getattr(self.dispatcher, hook, None)
            if callable(apply):
                apply(pkt)
        pkt._flood_scope_applied = True
        pkt._path_hash_mode_applied = True

    async def _schedule_egress_tx(
        self,
        fwd_pkt: Packet,
        delay: float,
        airtime_ms: float,
        tx_radio_ids: Optional[Tuple[str, ...]],
        *,
        local_transmission: bool = False,
    ):
        """Schedule a TX over the resolved egress set.

        One egress (or none, leaving the choice to the fabric) keeps the plain
        ``schedule_retransmit`` path, whose task resolves to a bool and raises on
        a relay send error; more than one fans out.
        """
        if tx_radio_ids and len(tx_radio_ids) > 1:
            return await self.schedule_retransmit_fanout(
                fwd_pkt,
                delay,
                airtime_ms,
                tx_radio_ids,
                local_transmission=local_transmission,
            )
        return await self.schedule_retransmit(
            fwd_pkt,
            delay,
            airtime_ms,
            local_transmission=local_transmission,
            preferred_tx_radio_id=tx_radio_ids[0] if tx_radio_ids else None,
        )

    @staticmethod
    async def _await_egress_tx(tx_task, fwd_pkt: Packet) -> FanoutTxResult:
        """Await a task from ``_schedule_egress_tx`` as a ``FanoutTxResult``.

        A single-egress send error propagates, exactly as it did before fan-out.
        """
        outcome = await tx_task
        if isinstance(outcome, FanoutTxResult):
            return outcome
        metadata = getattr(fwd_pkt, "_tx_metadata", None)
        metadata = metadata if isinstance(metadata, dict) else None
        radio_id = (metadata.get("radio_id") or metadata.get("tx_radio_id")) if metadata else None
        return FanoutTxResult([RadioTxResult(radio_id, bool(outcome), metadata=metadata)])

    def _fabric_cfg(self) -> dict:
        fabric_cfg = self.config.get("fabric")
        return fabric_cfg if isinstance(fabric_cfg, dict) else {}

    def _fabric_endpoints(self):
        """Return ``(fabric, radio_ids)`` for a multi-radio fabric, else ``(None, None)``."""
        radio = getattr(self.dispatcher, "radio", None)
        fabric = getattr(radio, "fabric", None)
        radios = getattr(fabric, "radios", None)
        # isinstance, not truthiness: RFFabric.radios is an OrderedDict, and a
        # stand-in attribute on a non-fabric radio must not read as endpoints.
        if not isinstance(radios, Mapping):
            return None, None
        return fabric, [str(rid) for rid in radios.keys()]

    def _resolve_relay_tx_radio_ids(self, rx_radio_id: Optional[str]) -> Optional[Tuple[str, ...]]:
        """Egress radios for a packet heard on ``rx_radio_id`` and being repeated.

        Applies only when fabric.tx_mode=bridge and exactly two radios are
        registered: the bridge target (the other radio) comes first, followed by
        the ingress radio itself when fabric.repeat_on_ingress is true. For all
        other layouts return None and let the dispatcher/fabric choose via
        existing logic.
        """
        fabric_cfg = self._fabric_cfg()
        if str(fabric_cfg.get("tx_mode", "default")).strip().lower() != "bridge":
            return None
        if not rx_radio_id:
            return None

        _, ids = self._fabric_endpoints()
        if ids is None or len(ids) != 2:
            return None

        rx_id = str(rx_radio_id)
        if rx_id not in ids:
            return None
        bridge_id = ids[1] if ids[0] == rx_id else ids[0]
        if fabric_cfg.get("repeat_on_ingress") is True:
            return (bridge_id, rx_id)
        return (bridge_id,)

    def _resolve_bridge_pair_tx_radio(self, rx_radio_id: Optional[str]) -> Optional[str]:
        """Return deterministic pair TX radio for bridge mode (A->B, B->A)."""
        radio_ids = self._resolve_relay_tx_radio_ids(rx_radio_id)
        return radio_ids[0] if radio_ids else None

    def _resolve_origin_tx_radio_ids(self) -> Optional[Tuple[str, ...]]:
        """Egress radios for a packet this node originates.

        With fabric.origin_tx=all and exactly two radios, every radio, the
        default radio first. Otherwise None: the existing fabric selection
        (default radio / tx_mode selector) stays authoritative.
        """
        if fabric_origin_tx(self._fabric_cfg()) != "all":
            return None
        fabric, ids = self._fabric_endpoints()
        if ids is None or len(ids) != 2:
            return None
        default_id = getattr(fabric, "default_radio_id", None)
        if isinstance(default_id, str) and default_id in ids:
            ids.remove(default_id)
            ids.insert(0, default_id)
        return tuple(ids)

    def _record_packet_sent(self, packet: Packet) -> None:
        """Record a packet send for flood/direct stats (forwarded and originated)."""
        route = getattr(packet, "header", 0) & PH_ROUTE_MASK
        if route in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD):
            self.sent_flood_count += 1
        elif route in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT):
            self.sent_direct_count += 1

    def get_noise_floor(self) -> Optional[float]:
        try:
            radio = self.dispatcher.radio if self.dispatcher else None
            if radio and hasattr(radio, "get_noise_floor"):
                return radio.get_noise_floor()
            return None
        except Exception as e:
            logger.debug(f"Failed to get noise floor: {e}")
            return None

    def get_cached_noise_floor(self) -> Optional[float]:
        """Return the last asynchronously-sampled noise floor value."""
        return self._cached_noise_floor

    def get_stats(self) -> dict:
        runtime_config = normalize_modem_config(self.config, warn=False)
        redact_modem_tokens_in_place(runtime_config)

        uptime_seconds = time.time() - self.start_time

        # Get config sections
        repeater_config = self.config.get("repeater", {})
        duty_cycle_config = self.config.get("duty_cycle", {})
        delays_config = self.config.get("delays", {})

        max_airtime_ms = duty_cycle_config.get("max_airtime_per_minute", 3600)
        max_duty_cycle_percent = (max_airtime_ms / 60000) * 100  # 60000ms = 1 minute

        # Calculate actual hourly rates (packets in last 3600 seconds)
        now = time.time()
        packets_last_hour = [p for p in self.recent_packets if now - p["timestamp"] < 3600]
        rx_per_hour = len(packets_last_hour)
        forwarded_per_hour = sum(1 for p in packets_last_hour if p.get("transmitted", False))

        # Use cached value sampled by the background timer to avoid serial I/O on stats requests.
        noise_floor_dbm = self.get_cached_noise_floor()

        # Get CRC error count from radio hardware
        radio = self.dispatcher.radio if self.dispatcher else None
        crc_error_count = getattr(radio, "crc_error_count", 0) if radio else 0

        # Get neighbors from database
        neighbors = self.storage.get_neighbors() if self.storage else {}

        # Format local_hash respecting path_hash_mode
        phm = self.config.get("mesh", {}).get("path_hash_mode", 0)
        _bc = {0: 1, 1: 2, 2: 3}.get(phm, 1)
        _hc = _bc * 2
        _val = int.from_bytes(bytes(self.local_hash_bytes[:_bc]), "big")
        local_hash_str = f"0x{_val:0{_hc}x}"

        stats = {
            "local_hash": local_hash_str,
            "duplicate_cache_size": len(self.seen_packets),
            "cache_ttl": self.cache_ttl,
            "rx_count": self.rx_count,
            "forwarded_count": self.forwarded_count,
            "dropped_count": self.dropped_count,
            "recv_flood_count": self.recv_flood_count,
            "recv_direct_count": self.recv_direct_count,
            "sent_flood_count": self.sent_flood_count,
            "sent_direct_count": self.sent_direct_count,
            "flood_dup_count": self.flood_dup_count,
            "direct_dup_count": self.direct_dup_count,
            "rx_per_hour": rx_per_hour,
            "forwarded_per_hour": forwarded_per_hour,
            "recent_packets": list(self.recent_packets),
            "neighbors": neighbors,
            "uptime_seconds": uptime_seconds,
            "noise_floor_dbm": noise_floor_dbm,
            "crc_error_count": crc_error_count,
            # Add configuration data
            "config": {
                "node_name": repeater_config.get("node_name", "Unknown"),
                "repeater": {
                    "mode": repeater_config.get("mode", "forward"),
                    "use_score_for_tx": repeater_config.get("use_score_for_tx", False),
                    "score_threshold": repeater_config.get("score_threshold", 0.3),
                    "send_advert_interval_hours": repeater_config.get(
                        "send_advert_interval_hours", 10
                    ),
                    "direct_advert_interval_hours": repeater_config.get(
                        "direct_advert_interval_hours", 0
                    ),
                    "latitude": repeater_config.get("latitude", 0.0),
                    "longitude": repeater_config.get("longitude", 0.0),
                    "max_flood_hops": repeater_config.get("max_flood_hops", 64),
                    "advert_interval_minutes": repeater_config.get("advert_interval_minutes", 120),
                    "advert_rate_limit": repeater_config.get("advert_rate_limit", {}),
                    "advert_penalty_box": repeater_config.get("advert_penalty_box", {}),
                    "advert_adaptive": repeater_config.get("advert_adaptive", {}),
                },
                "radio": self.config.get(
                    "radio", {}
                ),  # Read from live config, not cached radio_config
                "radios": runtime_config.get("radios") or [],
                "fabric": self.config.get("fabric") or {},
                "radio_type": runtime_config.get("radio_type"),
                "sx1262": self.config.get("sx1262", {}),
                "ch341": self.config.get("ch341", {}),
                "kiss": self.config.get("kiss", {}),
                "modem_usb": runtime_config.get("modem_usb", {}),
                "modem_tcp": runtime_config.get("modem_tcp", {}),
                "duty_cycle": {
                    "max_airtime_percent": max_duty_cycle_percent,
                    "enforcement_enabled": duty_cycle_config.get("enforcement_enabled", True),
                },
                "delays": {
                    "tx_delay_factor": delays_config.get("tx_delay_factor", 1.0),
                    "direct_tx_delay_factor": delays_config.get("direct_tx_delay_factor", 0.5),
                    "rx_delay_base": delays_config.get("rx_delay_base", 0.0),
                },
                "web": self.config.get("web", {}),  # Include web configuration
                "mesh": {
                    "loop_detect": self.config.get("mesh", {}).get("loop_detect", "off"),
                    "unscoped_flood_allow": self.config.get("mesh", {}).get(
                        "unscoped_flood_allow",
                        self.config.get("mesh", {}).get("global_flood_allow", True),
                    ),
                    "path_hash_mode": self.config.get("mesh", {}).get("path_hash_mode", 0),
                },
                "mqtt_brokers": self.config.get("mqtt_brokers", {}),
            },
            "public_key": None,
        }
        # Add airtime stats
        stats.update(self.airtime_mgr.get_stats())
        return stats

    def _start_background_tasks(self):
        if self._background_task is None:
            self._background_task = asyncio.create_task(self._background_timer_loop())
            logger.info("Background timer started for noise floor and adverts")

    async def _background_timer_loop(self):
        try:
            while True:
                current_time = time.time()

                # Check noise floor recording (every 30 seconds)
                if current_time - self.last_noise_measurement >= self.noise_floor_interval:
                    await self._record_noise_floor_async()
                    await self._record_crc_errors_async()
                    self.last_noise_measurement = current_time

                # Check advert scheduling.
                # Prefer daemon-managed scheduling so repeater + room-server
                # intervals can share one loop and stagger transmissions.
                if self.periodic_advert_tick_func:
                    await self.periodic_advert_tick_func()
                elif self.send_advert_interval_hours > 0 and self.send_advert_func:
                    interval_seconds = self.send_advert_interval_hours * 3600
                    if current_time - self.last_advert_time >= interval_seconds:
                        await self._send_periodic_advert_async()
                        self.last_advert_time = current_time

                # Prune expired entries from duplicate detection cache (every 60s)
                if current_time - self.last_cache_cleanup >= 60.0:
                    self.cleanup_cache()
                    self.last_cache_cleanup = current_time

                # Prune old SQLite data (check every 6 hours)
                if current_time - self.last_db_cleanup >= 21600:
                    if self.storage:
                        try:
                            retention_cfg = self.config.get("storage", {}).get("retention", {})
                            retention_days = retention_cfg.get("sqlite_cleanup_days", 31)
                            companion_events_days = retention_cfg.get("companion_events_days", 31)
                            self.storage.cleanup_old_data(
                                days=retention_days,
                                companion_events_days=companion_events_days,
                            )
                            logger.info("Cleaned up SQLite data older than %d days", retention_days)
                        except Exception as e:
                            logger.warning(f"SQLite cleanup failed: {e}")
                    self.last_db_cleanup = current_time

                # Sleep for 5 seconds before next check
                await asyncio.sleep(5.0)

        except asyncio.CancelledError:
            logger.info("Background timer loop cancelled")
            raise
        except Exception as e:
            logger.error(f"Error in background timer loop: {e}")
            # Restart the timer after a delay
            await asyncio.sleep(30)
            self._background_task = asyncio.create_task(self._background_timer_loop())

    async def _record_noise_floor_async(self):
        if not self.storage:
            return

        try:
            # Run in executor so KISS modem's blocking _send_command (up to 5s timeout)
            # does not block the event loop and hang the process / delay Ctrl+C.
            loop = asyncio.get_running_loop()
            noise_floor = await loop.run_in_executor(None, self.get_noise_floor)
            if noise_floor is not None:
                self._cached_noise_floor = noise_floor
                self.storage.record_noise_floor(noise_floor)
                logger.debug(f"Recorded noise floor: {noise_floor} dBm")
            else:
                logger.debug("Unable to read noise floor from radio")
        except Exception as e:
            logger.error(f"Error recording noise floor: {e}")

    async def _record_crc_errors_async(self):
        """Persist CRC error delta from the radio hardware counter."""
        if not self.storage:
            return

        try:
            radio = self.dispatcher.radio if self.dispatcher else None
            current = getattr(radio, "crc_error_count", 0) if radio else 0
            delta = current - self._last_crc_error_count
            if delta > 0:
                self.storage.record_crc_errors(delta)
                logger.debug(f"Recorded {delta} CRC errors (total: {current})")
            self._last_crc_error_count = current
        except Exception as e:
            logger.error(f"Error recording CRC errors: {e}")

    async def _send_periodic_advert_async(self):
        logger.info(
            f"Periodic advert timer triggered (interval: {self.send_advert_interval_hours}h)"
        )
        try:
            if self.send_advert_func:
                success = await self.send_advert_func()
                if success:
                    logger.info("Periodic advert sent successfully")
                else:
                    logger.warning("Failed to send periodic advert")
            else:
                logger.debug("No send_advert_func configured")
        except Exception as e:
            logger.error(f"Error sending periodic advert: {e}")

    def reload_runtime_config(self):
        """Reload runtime configuration from self.config (called after live config updates)."""
        try:
            # Refresh delay factors
            self.tx_delay_factor = self.config.get("delays", {}).get("tx_delay_factor", 1.0)
            self.direct_tx_delay_factor = self.config.get("delays", {}).get(
                "direct_tx_delay_factor", 0.5
            )

            # Refresh repeater settings
            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)
            self.neighbour_link_tracker.refresh_config(self.config)
            self.loop_detect_mode = self._normalize_loop_detect_mode(
                self.config.get("mesh", {}).get("loop_detect", LOOP_DETECT_OFF)
            )

            with self.neighbour_link_tracker.lock:
                now_monotonic = time.monotonic()
                self.neighbour_link_tracker.purge_expired_locked(now_monotonic)
                while (
                    len(self.neighbour_link_tracker.links) > self.neighbour_link_tracker.max_entries
                ):
                    self.neighbour_link_tracker.evict_stalest_locked()

            # Radio hardware apply and AirtimeManager modulation refresh are
            # handled by ConfigManager after a successful live radio update.

            logger.info("Runtime configuration reloaded successfully")
        except Exception as e:
            logger.error(f"Error reloading runtime config: {e}")

    def cleanup(self):
        if self._background_task and not self._background_task.done():
            self._background_task.cancel()
            logger.info("Background timer task cancelled")

        if self.storage:
            try:
                self.storage.close()
                logger.info("StorageCollector closed successfully")
            except Exception as e:
                logger.error(f"Error closing StorageCollector: {e}")

    def __del__(self):
        try:
            self.cleanup()
        except Exception as exc:
            logger.debug(f"Engine cleanup during __del__ failed: {exc}")