meshcore-bot/modules/core.py

#!/usr/bin/env python3
"""
Core MeshCore Bot functionality
Contains the main bot class and message processing logic
"""

import asyncio
import atexit
import configparser
import contextlib  # noqa: F401  importable from this module on dev
import contextvars  # noqa: F401  importable from this module on dev
import functools  # noqa: F401  importable from this module on dev
import json
import logging
import signal
import sqlite3
import struct
import threading
import time
from collections.abc import Callable
from logging.handlers import RotatingFileHandler  # noqa: F401  importable from this module on dev
from pathlib import Path
from typing import Any

import colorlog  # noqa: F401  importable from this module on dev

# Import the official meshcore package
import meshcore  # noqa: F401  (tests patch modules.core.meshcore.MeshCore)
from meshcore import EventType

from .admin_server import BotAdminServer
from .channel_manager import ChannelManager
from .command_manager import CommandManager
from .config_reload import ConfigReloadMixin
from .db_manager import AsyncDBManager, DBManager
from .device_setup import DeviceSetupMixin
from .feed_manager import FeedManager
from .i18n import Translator
from .logging_setup import configure_bot_logging, configure_meshcore_loggers, meshcore_log_level
from .message_handler import MessageHandler
from .radio_link import RadioLinkMixin, _radio_session_held, _serialize_command_frames  # noqa: F401
from .radio_offline import RadioOfflineBreaker

# Import our modules
from .rate_limiter import BotTxRateLimiter, ChannelRateLimiter, NominatimRateLimiter, PerUserRateLimiter, RateLimiter
from .repeater_manager import RepeaterManager
from .scheduler import MessageScheduler
from .service_plugin_loader import ServicePluginLoader
from .service_supervisor import ServiceSupervisorMixin
from .solar_conditions import set_config
from .transmission_tracker import TransmissionTracker
from .utils import resolve_path
from .web_viewer.integration import WebViewerIntegration


class MeshCoreBot(ServiceSupervisorMixin, RadioLinkMixin, RadioOfflineBreaker, DeviceSetupMixin, ConfigReloadMixin):
    """MeshCore Bot using official meshcore package.

    This class handles the core functionality of the bot, including connection management,
    message processing initialization, and module coordination.
    """

    def __init__(self, config_file: str = "config.ini"):
        self.config_file = config_file
        # Reload writers are serialized, while readers rely on the atomic
        # replacement of this parser reference.  Never mutate a published
        # ConfigParser in place: readers do not take this lock.
        self._config_reload_lock = threading.RLock()
        self.config = configparser.ConfigParser()
        self.load_config()

        # Setup logging
        self.setup_logging()

        # Connection
        self.meshcore = None
        self.connected = False
        self.connection_time = None  # Track when connection was established to skip old cached messages

        # Volatile: DM-only admin command (channelpause) toggles this; not persisted across restarts.
        self.channel_responses_enabled = True

        # Bot start time for uptime tracking
        self.start_time = time.time()

        # Initialize database manager first (needed by plugins)
        db_path = self.config.get('Bot', 'db_path', fallback='meshcore_bot.db')

        # Resolve database path (relative paths resolved from bot root, absolute paths used as-is)
        db_path = resolve_path(db_path, self.bot_root)

        self.logger.info(f"Initializing database manager with database: {db_path}")
        try:
            self.db_manager = DBManager(self, db_path)
            self.async_db_manager = AsyncDBManager(str(db_path), self.logger)
            self.logger.info("Database manager initialized successfully")
        except (OSError, ValueError, sqlite3.Error) as e:
            self.logger.error(f"Failed to initialize database manager: {e}")
            raise

        # Set length of prefix
        self.prefix_bytes = self.config.getint("Bot", "prefix_bytes", fallback=1)
        self.prefix_hex_chars = self.prefix_bytes * 2
        self.logger.info(f"Prefix mode: {self.prefix_bytes} bytes ({self.prefix_hex_chars} hex chars)")

        # Store start time in database for web viewer access
        try:
            self.db_manager.set_bot_start_time(self.start_time)
            self.logger.info("Bot start time stored in database")
        except (OSError, sqlite3.Error, AttributeError) as e:
            self.logger.warning(f"Could not store start time in database: {e}")

        # Notify if Web_Viewer uses a different database (split-DB setup)
        if self.config.has_section('Web_Viewer') and self.config.has_option('Web_Viewer', 'db_path'):
            wv_raw = self.config.get('Web_Viewer', 'db_path').strip()
            if wv_raw:
                wv_path = Path(resolve_path(wv_raw, self.bot_root)).resolve()
                bot_path = Path(self.db_manager.db_path).resolve()
                if wv_path != bot_path:
                    self.logger.warning(
                        "Web viewer database path differs from bot database: viewer=%s, bot=%s. "
                        "For shared repeater/graph and packet stream data, set [Web_Viewer] db_path to the same as [Bot] db_path or remove it to use the bot database. See docs/web-viewer.md (migrating from a separate database).",
                        wv_path, bot_path
                    )

        # Initialize web viewer integration (after database manager)
        try:
            self.web_viewer_integration = WebViewerIntegration(self)
            self.logger.info("Web viewer integration initialized")

            # Register cleanup handler for web viewer
            atexit.register(self._cleanup_web_viewer)
        except (OSError, ValueError, AttributeError, ImportError) as e:
            self.logger.error("Web viewer integration failed: %s", e)
            self.web_viewer_integration = None

        # Admin HTTP server (optional — [Admin] section)
        self._admin_server: BotAdminServer | None = None
        if self.config.getboolean('Admin', 'enabled', fallback=False):
            admin_port = self.config.getint('Admin', 'port', fallback=5001)
            admin_token = self.config.get('Admin', 'token', fallback='')
            if admin_token:
                self._admin_server = BotAdminServer(self, admin_port, admin_token)
            else:
                self.logger.warning("Admin server enabled but no token configured — skipping")

        # Initialize modules
        self.rate_limiter = RateLimiter(
            self.config.getint('Bot', 'rate_limit_seconds', fallback=10)
        )
        self.bot_tx_rate_limiter = BotTxRateLimiter(
            self.config.getfloat('Bot', 'bot_tx_rate_limit_seconds', fallback=1.0)
        )
        # Per-user rate limiter: minimum seconds between replies to the same user (key = pubkey or name)
        self.per_user_rate_limit_enabled = self.config.getboolean(
            'Bot', 'per_user_rate_limit_enabled', fallback=True
        )
        self.per_user_rate_limiter = PerUserRateLimiter(
            seconds=self.config.getfloat('Bot', 'per_user_rate_limit_seconds', fallback=5.0),
            max_entries=1000
        )
        # Nominatim rate limiter: 1.1 seconds between requests (Nominatim policy: max 1 req/sec)
        self.nominatim_rate_limiter = NominatimRateLimiter(
            self.config.getfloat('Bot', 'nominatim_rate_limit_seconds', fallback=1.1)
        )
        # Per-channel rate limiter: loaded from [Rate_Limits] channel.<name>_seconds keys
        self.channel_rate_limiter = self._load_channel_rate_limiter()
        self.tx_delay_ms = self.config.getint('Bot', 'tx_delay_ms', fallback=250)

        # Radio health, set before any command or service plugin is built, since
        # their constructors may read is_radio_offline / is_radio_zombie. Zombie: the firmware stopped acting on commands and only a
        # power cycle recovers it. Offline: repeated send timeouts. The probe
        # timestamp starts on the first health-loop pass, hence None until then.
        self._radio_zombie_detected = False
        self._radio_fail_count = 0
        self._tcp_probe_fail_count = 0
        self._radio_offline = False
        self._send_consecutive_failures = 0
        self._last_radio_probe: float | None = None
        self._last_health_update = 0.0

        # Initialize translator for localization BEFORE CommandManager
        # This ensures translated keywords are available when commands are loaded
        try:
            default_local_translations = self._default_local_translation_path(self.config)
            if self.config.has_section('Localization'):
                language = self.config.get('Localization', 'language', fallback='en')
                translation_path = self.config.get('Localization', 'translation_path', fallback='translations/')
                local_translation_path = self.config.get(
                    'Localization', 'local_translation_path', fallback=default_local_translations
                )
            else:
                language = 'en'
                translation_path = 'translations/'
                local_translation_path = default_local_translations
            self.translation_path = translation_path
            self.local_translation_path = local_translation_path
            self._translator_cache: dict[str, Any] = {}
            self.translator = Translator(language, translation_path, local_translation_path)
            self._translator_cache[language] = self.translator
            self.logger.info(f"Localization initialized: {language}")
        except (OSError, ValueError, FileNotFoundError, json.JSONDecodeError) as e:
            self.logger.warning(f"Failed to initialize translator: {e}")
            # Create a dummy translator that just returns keys
            class DummyTranslator:
                language = 'en'
                base_language = 'en'

                def translate(self, key, **kwargs):
                    return key

                def get_value(self, key):
                    return None

                def get_available_languages(self):
                    return []

            self.translator = DummyTranslator()
            self.translation_path = 'translations/'
            # get_translator() reads both paths when it builds a per-language
            # translator, so neither may be left unset here.
            self.local_translation_path = 'local/translations/'
            self._translator_cache = {}

        # Initialize solar conditions configuration
        set_config(self.config)

        self.message_handler = MessageHandler(self)
        self.command_manager = CommandManager(self)

        # Regional flood-scope tallies, and the opt-in warning they can drive.
        try:
            from .region_warning import RegionWarningMonitor
            self.region_warning_monitor = RegionWarningMonitor(self)
        except Exception as e:
            self.logger.warning(f"Failed to initialize region warning monitor: {e}")
            self.region_warning_monitor = None

        # Initialize transmission tracker for monitoring TX success
        try:
            self.transmission_tracker = TransmissionTracker(self)
            self.logger.info("Transmission tracker initialized")
        except Exception as e:
            self.logger.warning(f"Failed to initialize transmission tracker: {e}")
            self.transmission_tracker = None

        # Load max_channels from config (default 40, MeshCore supports up to 40 channels)
        max_channels = self.config.getint('Bot', 'max_channels', fallback=40)
        self.channel_manager = ChannelManager(self, max_channels=max_channels)

        # Callbacks invoked when the bot sends a channel message (e.g. Discord/Telegram bridges)
        self.channel_sent_listeners: list[Callable] = []

        self.scheduler = MessageScheduler(self)

        # Initialize feed manager
        self.logger.info("Initializing feed manager")
        try:
            self.feed_manager = FeedManager(self)
            self.logger.info("Feed manager initialized successfully")
        except (OSError, ValueError, AttributeError, ImportError, configparser.NoSectionError) as e:
            self.logger.warning(f"Failed to initialize feed manager: {e}")
            self.feed_manager = None

        # Initialize repeater manager
        self.logger.info("Initializing repeater manager")
        try:
            self.repeater_manager = RepeaterManager(self)
            self.logger.info("Repeater manager initialized successfully")
        except (OSError, ValueError, AttributeError) as e:
            self.logger.error(f"Failed to initialize repeater manager: {e}")
            raise

        # Initialize mesh graph for path validation
        self.logger.info("Initializing mesh graph")
        try:
            from .mesh_graph import MeshGraph
            self.mesh_graph = MeshGraph(self)
            self.logger.info("Mesh graph initialized successfully")

            # Register cleanup handler for mesh graph (independent of web viewer)
            # This ensures pending graph writes are flushed during shutdown
            atexit.register(self._cleanup_mesh_graph)
        except (OSError, ValueError, AttributeError, ImportError) as e:
            self.logger.warning(f"Failed to initialize mesh graph: {e}")
            self.mesh_graph = None

        # Initialize service plugin loader and load all services
        self.logger.info("Initializing service plugin loader")
        try:
            self.service_loader = ServicePluginLoader(
                self, local_services_dir=str(self._local_root / "service_plugins")
            )
            self.services = self.service_loader.load_all_services()
            self.logger.info(f"Service plugin loader initialized with {len(self.services)} service(s)")
        except (OSError, ImportError, AttributeError, ValueError) as e:
            self.logger.error(f"Failed to initialize service plugin loader: {e}")
            self.service_loader = None
            self.services = {}

        # Backward compatibility: expose packet_capture_service for existing code
        # This allows code that references self.packet_capture_service to continue working
        # Try to find by service name first, then by class name
        self.packet_capture_service = None
        for service_name, service_instance in self.services.items():
            if (service_name == 'packetcapture' or
                service_instance.__class__.__name__ == 'PacketCaptureService'):
                self.packet_capture_service = service_instance
                break

        # Reload translated keywords for all commands now that translator is available
        # This ensures keywords are loaded even if translator wasn't ready during command init
        if hasattr(self, 'command_manager') and hasattr(self, 'translator'):
            for _cmd_name, cmd_instance in self.command_manager.commands.items():
                if hasattr(cmd_instance, '_load_translated_keywords'):
                    cmd_instance._load_translated_keywords()

        # Advert tracking
        self.last_advert_time = None

        # Clock sync tracking
        self.last_clock_sync_time = None

        # Shutdown event for graceful shutdown
        self._shutdown_event = threading.Event()

        # Idempotent async shutdown (see stop()); lock created when event loop is available
        self._shutdown_lock: asyncio.Lock | None = None
        self._shutdown_complete = False

        # Service plugin restart state (name -> timestamp of last failed restart)
        self._service_restart_failures: dict[str, float] = {}
        self._service_restarting: set = set()

        # Transport reconnect (serial/BLE/TCP) — lock created when event loop runs
        self._transport_reconnect_lock: asyncio.Lock | None = None
        self._transport_reconnect_in_progress = False
        # Web-viewer reboot/reconnect ops in flight (a count, since they can overlap)
        self._radio_relinks_in_progress = 0


        # Serialize host->radio commands: one companion frame in flight at a
        # time, with a minimum inter-command gap so the firmware's single
        # serial loop can drain its RX buffer between frames. Prevents the
        # USB-CDC overrun / parser-desync failure mode. Lock is created lazily
        # once an event loop is running (see _get_radio_cmd_lock).
        self._radio_cmd_lock: asyncio.Lock | None = None
        self._radio_cmd_last_ts: float = 0.0
        self._radio_cmd_min_interval = max(
            0.0,
            self.config.getfloat(
                'Connection', 'command_min_interval_ms', fallback=30.0
            ) / 1000.0,
        )

    def _default_local_translation_path(self, config: configparser.ConfigParser) -> str:
        """Default local catalog directory: ``<local_dir_path>/translations``.

        ``local_dir_path`` already selects where an operator's own commands, service
        plugins and config overlay live, so the local translation catalog belongs in
        that same tree rather than in a second, separately-configured location. The
        result is absolute, so it does not depend on the process's cwd.
        """
        local_dir = config.get('Bot', 'local_dir_path', fallback='local')
        return str(Path(resolve_path(local_dir, self.bot_root)) / 'translations')

    @property
    def bot_root(self) -> Path:
        """Get bot root directory (where config.ini is located)"""
        return Path(self.config_file).parent.resolve()


    @property
    def keep_running(self) -> bool:
        """True while the main loop and scheduler thread should stay alive.

        ``connected`` alone is not enough: it drops to False while a transport
        reconnect or a web-viewer reboot/reconnect re-establishes the link, and
        treating that window as a stop kills the bot on every transport blip.
        A reconnect that gives up leaves ``connected`` False and clears its
        in-progress flag, which still ends the loops.
        """
        if self._shutdown_event.is_set():
            return False
        return bool(
            self.connected
            or self._transport_reconnect_in_progress
            or self._radio_relinks_in_progress
        )

    def load_config(self) -> None:
        """Load configuration from file.

        Reads the configuration file specified in self.config_file. If the file
        does not exist, a default configuration is created first.
        """
        if not Path(self.config_file).exists():
            self.create_default_config()

        self.config, self._local_root = self._read_config_snapshot()

    def _read_config_snapshot(self) -> tuple[configparser.ConfigParser, Path]:
        """Read base and local overlay into a new, unpublished parser.

        ``ConfigParser.read`` mutates its receiver.  Keeping that receiver
        private until both files have parsed guarantees that concurrent readers
        only ever observe a complete old or complete new snapshot.
        """
        snapshot = configparser.ConfigParser()
        loaded = snapshot.read(self.config_file, encoding="utf-8")
        if not loaded:
            raise FileNotFoundError(self.config_file)

        # The overlay location is selected by the base file.  An overlay cannot
        # silently relocate itself midway through the same read operation.
        local_root = Path(
            resolve_path(
                snapshot.get("Bot", "local_dir_path", fallback="local"),
                self.bot_root,
            )
        )
        local_config = local_root / "config.ini"
        if local_config.exists():
            snapshot.read(local_config, encoding="utf-8")
        return snapshot, local_root

    def _get_radio_settings(
        self, config: configparser.ConfigParser | None = None
    ) -> dict[str, Any]:
        """Get current radio/connection settings from config.

        Returns:
            Dict[str, Any]: Dictionary containing all radio-related settings.
        """
        source = config if config is not None else self.config
        return {
            'connection_type': source.get('Connection', 'connection_type', fallback='ble').lower(),
            'serial_port': source.get('Connection', 'serial_port', fallback=''),
            'ble_device_name': source.get('Connection', 'ble_device_name', fallback=''),
            'hostname': source.get('Connection', 'hostname', fallback=''),
            'tcp_port': source.getint('Connection', 'tcp_port', fallback=5000),
            'timeout': source.getint('Connection', 'timeout', fallback=30),
            # radio_debug intentionally excluded — only needs a reconnect, not a full restart
        }

    def _load_channel_rate_limiter(
        self, config: configparser.ConfigParser | None = None
    ) -> ChannelRateLimiter:
        """Build a ChannelRateLimiter from [Rate_Limits] channel.<name>_seconds keys."""
        limits: dict[str, float] = {}
        source = config if config is not None else self.config
        if source.has_section('Rate_Limits'):
            for key, value in source.items('Rate_Limits'):
                if key.startswith('channel.') and key.endswith('_seconds'):
                    channel_name = key[len('channel.'):-len('_seconds')]
                    try:
                        # Normalize now; limiter will also normalize at use-time.
                        limits[channel_name.strip().lower()] = float(value)
                    except ValueError:
                        self.logger.warning(f"Invalid channel rate limit for {key}: {value!r}")
        return ChannelRateLimiter(limits)

    def _available_translation_codes(self) -> set[str]:
        """Return concrete locale codes from filesystem or bundled catalogs."""
        try:
            return {
                code
                for code in self.translator.get_available_languages()
                if code
            }
        except (AttributeError, OSError) as e:
            self.logger.debug("Could not enumerate translation files: %s", e)
            return set()

    def get_translator(self, language: str) -> Any:
        """Return a cached translator without changing the bot-wide default."""
        if not language:
            return self.translator
        available_codes = self._available_translation_codes()
        resolved_language = language
        if language not in available_codes:
            locale_matches = sorted(
                code
                for code in available_codes
                if code.replace("_", "-").split("-", 1)[0] == language
            )
            if locale_matches:
                resolved_language = locale_matches[0]
        cached = self._translator_cache.get(resolved_language)
        if cached is not None:
            return cached
        try:
            translator = Translator(
                resolved_language, self.translation_path, self.local_translation_path
            )
        except (OSError, ValueError, FileNotFoundError, json.JSONDecodeError) as e:
            self.logger.warning(
                "Failed to build translator for %r: %s", resolved_language, e
            )
            return self.translator
        self._translator_cache[resolved_language] = translator
        return translator

    def available_languages(self) -> set[str]:
        """Return detectable base languages from filesystem or bundled catalogs."""
        available = {
            language.replace("_", "-").split("-", 1)[0]
            for language in self._available_translation_codes()
            if language
        }
        available.add(
            getattr(self.translator, "base_language", None)
            or getattr(self.translator, "language", "en")
        )
        return available

    _COMMAND_CONFIG_STATE = (
        "keywords",
        "custom_syntax",
        "banned_users",
        "monitor_channels",
        "channel_keywords",
        "command_prefixes",
        "require_command_prefix",
        "_command_prefix_default",
        "flood_scope_allow_global",
        "flood_scope_keys",
        "plugin_loader",
        "commands",
    )

    def reload_config(self) -> tuple[bool, str]:
        """Reload configuration from file without restarting the bot.

        This method reloads the configuration file and updates all components
        that depend on it. It will reject the reload if radio/connection settings
        have changed, as those require a full restart.

        Returns:
            Tuple[bool, str]: (success, message) tuple indicating if reload succeeded
                and a descriptive message.
        """
        try:
            with self._config_reload_lock:
                old_config = self.config
                if not Path(self.config_file).exists():
                    return (False, "Config file not found")
                new_config, new_local_root = self._read_config_snapshot()
                uninterpolatable = self._validate_config_snapshot(new_config)
                if uninterpolatable:
                    self.logger.warning(
                        "Config values with a bare '%%' (fine where they are read raw, such as "
                        "templates; use '%%%%' elsewhere): %s",
                        ", ".join(uninterpolatable),
                    )

                old_radio_settings = self._get_radio_settings(old_config)
                new_radio_settings = self._get_radio_settings(new_config)
                if old_radio_settings != new_radio_settings:
                    changed_settings = [
                        f"{key}: {old_radio_settings[key]} -> {new_radio_settings[key]}"
                        for key in old_radio_settings
                        if old_radio_settings[key] != new_radio_settings[key]
                    ]
                    return (
                        False,
                        "Radio settings changed. Restart required. Changes: "
                        + ", ".join(changed_settings),
                    )

                restart_changes = self._restart_only_config_changes(old_config, new_config)
                if restart_changes:
                    return (
                        False,
                        "Startup-only or cached service settings changed. Restart required: "
                        + ", ".join(restart_changes),
                    )

                # Typed reads validate the final merged snapshot and prepare all
                # independent replacements before the live reference is changed.
                new_rate_limiter = RateLimiter(
                    new_config.getint('Bot', 'rate_limit_seconds', fallback=10)
                )
                new_bot_tx_rate_limiter = BotTxRateLimiter(
                    new_config.getfloat('Bot', 'bot_tx_rate_limit_seconds', fallback=1.0)
                )
                new_per_user_enabled = new_config.getboolean(
                    'Bot', 'per_user_rate_limit_enabled', fallback=True
                )
                new_per_user_rate_limiter = PerUserRateLimiter(
                    seconds=new_config.getfloat(
                        'Bot', 'per_user_rate_limit_seconds', fallback=5.0
                    ),
                    max_entries=1000,
                )
                new_nominatim_rate_limiter = NominatimRateLimiter(
                    new_config.getfloat(
                        'Bot', 'nominatim_rate_limit_seconds', fallback=1.1
                    )
                )
                new_channel_rate_limiter = self._load_channel_rate_limiter(new_config)
                new_tx_delay_ms = new_config.getint('Bot', 'tx_delay_ms', fallback=250)
                new_max_channels = new_config.getint('Bot', 'max_channels', fallback=40)

                new_language = new_config.get('Localization', 'language', fallback='en')
                new_translation_path = new_config.get(
                    'Localization', 'translation_path', fallback='translations/'
                )
                new_local_translation_path = new_config.get(
                    'Localization',
                    'local_translation_path',
                    fallback=self._default_local_translation_path(new_config),
                )
                new_translator = Translator(
                    new_language, new_translation_path, new_local_translation_path
                )
                new_translator_cache = {new_language: new_translator}

                candidate = {
                    "config": new_config,
                    "_local_root": new_local_root,
                    "rate_limiter": new_rate_limiter,
                    "bot_tx_rate_limiter": new_bot_tx_rate_limiter,
                    "per_user_rate_limit_enabled": new_per_user_enabled,
                    "per_user_rate_limiter": new_per_user_rate_limiter,
                    "nominatim_rate_limiter": new_nominatim_rate_limiter,
                    "channel_rate_limiter": new_channel_rate_limiter,
                    "tx_delay_ms": new_tx_delay_ms,
                    "translation_path": new_translation_path,
                    "local_translation_path": new_local_translation_path,
                    "_translator_cache": new_translator_cache,
                    "translator": new_translator,
                }
                old_state = {name: getattr(self, name) for name in candidate}
                old_command_config_state = self._command_config_state(self.command_manager)
                old_max_channels = self.channel_manager.max_channels

                scheduler_apply_started = False
                try:
                    # Atomic complete-snapshot publication.  Component reference
                    # swaps follow under the single-writer lock and are all
                    # restored if any component rejects the candidate.
                    for name, value in candidate.items():
                        setattr(self, name, value)
                    # Commands and nested delegates require the real bot. They
                    # are therefore constructed after candidate publication,
                    # inside the rollback boundary, rather than against a
                    # facade that could leak into nested objects.
                    new_command_manager = CommandManager(self)
                    old_plugin_failures = (
                        self.command_manager.plugin_loader.get_failed_plugins()
                    )
                    new_plugin_failures = (
                        new_command_manager.plugin_loader.get_failed_plugins()
                    )
                    introduced_failures = {
                        name: reason
                        for name, reason in new_plugin_failures.items()
                        if old_plugin_failures.get(name) != reason
                    }
                    if introduced_failures:
                        names = ", ".join(sorted(introduced_failures))
                        raise ValueError(
                            f"Command plugin reload failed for: {names}"
                        )
                    self._apply_command_config_state(
                        self.command_manager,
                        self._command_config_state(new_command_manager),
                    )
                    self.channel_manager.max_channels = new_max_channels
                    set_config(new_config)

                    if getattr(self, 'region_warning_monitor', None):
                        self.region_warning_monitor.reload_config()

                    if hasattr(self, 'scheduler'):
                        scheduler_apply_started = True
                        self.scheduler.setup_scheduled_messages()
                        self.logger.info("Scheduler config reloaded")
                except (Exception, SystemExit):
                    for name, value in old_state.items():
                        setattr(self, name, value)
                    self._apply_command_config_state(self.command_manager, old_command_config_state)
                    self.channel_manager.max_channels = old_max_channels
                    set_config(old_config)
                    if getattr(self, 'region_warning_monitor', None):
                        self.region_warning_monitor.reload_config()
                    # setup_scheduled_messages may have stopped the previous
                    # APScheduler before failing. Rebuild it against old config.
                    if scheduler_apply_started and hasattr(self, 'scheduler'):
                        try:
                            self.scheduler.setup_scheduled_messages()
                        except Exception as rollback_error:
                            self.logger.error(
                                "Scheduler rollback failed: %s", rollback_error
                            )
                    raise

                self.logger.info("Configuration reloaded successfully")
                return (
                    True,
                    "Configuration reloaded successfully. Active service plugin and "
                    "process settings remain startup-only and require restart.",
                )

        except (Exception, SystemExit) as e:
            error_msg = f"Error reloading configuration: {e}"
            self.logger.error(error_msg)
            import traceback
            self.logger.error(traceback.format_exc())
            return (False, error_msg)

    def create_default_config(self) -> None:
        """Create default configuration file.

        Writes the packaged ``modules/templates/default_config.ini`` (standard
        settings with comments explaining each option) to ``self.config_file``.
        """
        default_config = (Path(__file__).parent / "templates" / "default_config.ini").read_text(encoding="utf-8")
        with open(self.config_file, 'w') as f:
            f.write(default_config)
        # Note: Using print here since logger may not be initialized yet
        print(f"Created default config file: {self.config_file}")

    def setup_logging(self) -> None:
        """Setup logging configuration.

        Configures the logging system based on settings in the config file (see
        :func:`modules.logging_setup.configure_bot_logging`), then installs the
        shutdown signal handlers.
        If [Logging] section is missing, uses defaults (console/journal only, no file).
        """
        self.logger, self._log_formatter = configure_bot_logging(self.config, self.bot_root)

        # Setup signal handlers for graceful shutdown
        self._setup_signal_handlers()

    def _configure_meshcore_debug_logging(self, enable: bool) -> None:
        """Route meshcore library output through the bot's handlers.

        When *enable* is True the meshcore loggers are set to DEBUG and share
        all of the bot's handlers (console + rotating file), so raw-protocol
        lines appear in the log file tagged as DEBUG.

        When *enable* is False the loggers revert to the ``meshcore_log_level``
        from config with a console-only StreamHandler (same as setup_logging).
        """
        if enable:
            configure_meshcore_loggers(logging.DEBUG, None, shared_handlers=list(self.logger.handlers))
        else:
            configure_meshcore_loggers(meshcore_log_level(self.config), getattr(self, '_log_formatter', None))

    def _setup_signal_handlers(self) -> None:
        """Setup signal handlers for graceful shutdown.

        Registers handlers for SIGTERM and SIGINT to ensure the bot can
        clean up resources and disconnect properly when stopped.
        SIGHUP is intentionally handled by the process entrypoint so it can
        perform in-process config reload without triggering shutdown.
        """
        def signal_handler(signum, frame):
            self.logger.info(f"Received shutdown signal {signum}, initiating graceful shutdown...")
            # Set shutdown event to break main loop
            self._shutdown_event.set()
            # Reflect the disconnected state for cleanup and status reporting
            self.connected = False

        # Register signal handlers
        signal.signal(signal.SIGTERM, signal_handler)
        signal.signal(signal.SIGINT, signal_handler)


    async def wait_for_contacts(self) -> None:
        """Wait for contacts to be loaded from the device.

        Polls the device for contact list or waits for automatic loading.
        Times out after 30 seconds if contacts are not loaded.
        """
        self.logger.info("Waiting for contacts to load...")

        # Try to manually load contacts first
        try:
            from meshcore_cli.meshcore_cli import next_cmd
            self.logger.info("Manually requesting contacts from device...")
            result = await next_cmd(self.meshcore, ["contacts"])
            self.logger.info(f"Contacts command result: {len(result) if result else 0} contacts")
        except (OSError, AttributeError, ValueError) as e:
            self.logger.warning(f"Error manually loading contacts: {e}")

        # MeshCore.contacts is a property that always exists (possibly empty).
        self.logger.info(f"Contacts loaded: {len(self.meshcore.contacts)} contacts")

    async def setup_message_handlers(self) -> None:
        """Setup event handlers for messages.

        Registers callbacks for various meshcore events including contact messages,
        channel messages, RF data, and raw data packets.
        """
        # Handle contact messages (DMs)
        async def on_contact_message(event, metadata=None):
            await self.message_handler.handle_contact_message(event, metadata)

        # Handle channel messages
        async def on_channel_message(event, metadata=None):
            await self.message_handler.handle_channel_message(event, metadata)

        # Handle RF log data for SNR information
        async def on_rf_data(event, metadata=None):
            await self.message_handler.handle_rf_log_data(event, metadata)

        # Handle raw data events (full packet data)
        async def on_raw_data(event, metadata=None):
            await self.message_handler.handle_raw_data(event, metadata)

        # Handle new contact events
        async def on_new_contact(event, metadata=None):
            await self.message_handler.handle_new_contact(event, metadata)

        async def on_disconnected(event, metadata=None):
            payload = event.payload if isinstance(event.payload, dict) else {}
            reason = payload.get('reason', 'unknown')
            await self._schedule_transport_reconnect(reason)

        # Subscribe to events
        self.meshcore.subscribe(EventType.DISCONNECTED, on_disconnected)
        self.meshcore.subscribe(EventType.CONTACT_MSG_RECV, on_contact_message)
        self.meshcore.subscribe(EventType.CHANNEL_MSG_RECV, on_channel_message)
        self.meshcore.subscribe(EventType.RX_LOG_DATA, on_rf_data)

        # Subscribe to RAW_DATA events for full packet data
        self.meshcore.subscribe(EventType.RAW_DATA, on_raw_data)

        # Note: Debug mode commands are not available in current meshcore-cli version
        # The meshcore library handles debug output automatically when needed

        # Start auto message fetching
        await self.meshcore.start_auto_message_fetching()

        # Delay NEW_CONTACT subscription to ensure device is fully ready
        self.logger.info("Delaying NEW_CONTACT subscription to ensure device readiness...")
        await asyncio.sleep(5)  # Wait 5 seconds for device to be fully ready

        # Subscribe to NEW_CONTACT events for automatic contact management
        self.meshcore.subscribe(EventType.NEW_CONTACT, on_new_contact)
        self.logger.info("NEW_CONTACT subscription active - ready to receive new contact events")

        self.logger.info("Message handlers setup complete")

    async def start(self) -> None:
        """Start the bot.

        Initiates the connection to the node, sets up scheduling, services,
        and starts the main execution loop.
        """
        self.logger.info("Starting MeshCore Bot...")

        # Store reference to main event loop for scheduler thread access
        self.main_event_loop = asyncio.get_running_loop()

        # Suppress noisy "Task exception was never retrieved" warnings that
        # originate from malformed/truncated MeshCore packets.  IndexError and
        # struct.error are raised deep inside meshcore_parser.parsePacketPayload
        # and are not programming errors we can fix on this side.
        _default_handler = self.main_event_loop.get_exception_handler()

        def _loop_exception_handler(
            loop: asyncio.AbstractEventLoop,
            context: dict[str, Any],
        ) -> None:
            exc = context.get("exception")
            if isinstance(exc, (IndexError, struct.error)):
                self.logger.debug(
                    "Suppressed meshcore parser exception in asyncio task: %s: %s",
                    type(exc).__name__,
                    exc,
                )
                return
            if _default_handler is not None:
                _default_handler(loop, context)
            else:
                loop.default_exception_handler(context)

        self.main_event_loop.set_exception_handler(_loop_exception_handler)

        if self._shutdown_lock is None:
            self._shutdown_lock = asyncio.Lock()

        # Mark bot as initializing so the web viewer can show a status banner
        try:
            self.db_manager.set_metadata('bot.initializing', 'true')
        except Exception as e:
            self.logger.debug("Could not set bot.initializing metadata: %s", e)

        # The offline flag in bot_metadata is only written on transitions, so a
        # 'true' left by an earlier run would keep the viewer's Radio Offline
        # banner up forever: this process starts with no send failures and never
        # transitions. Publish the fresh state so the stored flag is this run's.
        self._publish_offline_state()

        # Start web viewer early (before radio connect) so operators can see the
        # initializing banner and monitor startup progress
        if self.web_viewer_integration and self.web_viewer_integration.enabled:
            self.web_viewer_integration.start_viewer()
            self.logger.info("Web viewer started (early, before radio connect)")

        # Start the inbound webhook service early too (before radio connect). Unlike
        # the other service plugins, this one accepts inbound connections — starting
        # it late leaves a window (proportional to how long radio connect() takes)
        # where external callers get connection-refused instead of a clear response.
        # The handler itself gates on self.connected and returns 503 until the mesh
        # link is actually ready, so it's safe to bind before we're connected.
        webhook_service = self.services.get('webhook')
        if webhook_service is not None and getattr(webhook_service, 'enabled', False):
            await self._start_service_at_boot(
                'webhook', webhook_service,
                started="Service 'webhook' started (early, before radio connect)",
                failed="Failed to start service 'webhook' early",
            )

        # Connect to MeshCore node
        if not await self.connect():
            self.logger.error("Failed to connect to MeshCore node")
            try:
                self.db_manager.set_metadata('bot.initializing', 'false')
            except Exception:
                pass
            return

        # Bot is now connected — clear the initializing flag
        try:
            self.db_manager.set_metadata('bot.initializing', 'false')
        except Exception as e:
            self.logger.debug("Could not clear bot.initializing metadata: %s", e)

        # Update transmission tracker bot prefix now that we're connected
        if hasattr(self, 'transmission_tracker') and self.transmission_tracker:
            self.transmission_tracker._update_bot_prefix()

        # Setup scheduled messages
        self.scheduler.setup_scheduled_messages()

        # Initialize feed manager (if enabled)
        if self.feed_manager:
            await self.feed_manager.initialize()

        # Start scheduler thread
        self.scheduler.start()

        # Start admin server if configured
        if self._admin_server is not None:
            self._admin_server.start()
            self.logger.info(
                "Admin server started on http://127.0.0.1:%d",
                self.config.getint('Admin', 'port', fallback=5001),
            )

        # Web viewer already started early above (before radio connect)

        # Send startup advert if enabled
        await self.send_startup_advert()

        # Start all loaded services (webhook already started early, before radio connect)
        for service_name, service_instance in self.services.items():
            if service_name == 'webhook':
                continue
            await self._start_service_at_boot(
                service_name, service_instance,
                started=f"Service '{service_name}' started",
                failed=f"Failed to start service '{service_name}'",
            )

        # Start command queue processor if needed
        self.command_manager._start_queue_processor()

        # Keep running
        self.logger.info("Bot is running. Press Ctrl+C to stop.")
        try:
            while self.keep_running:
                # Before the transport check, so a viewer clear works while disconnected.
                if self._radio_offline_sync_due():
                    await asyncio.to_thread(self._sync_radio_offline_from_metadata, check_due=False)

                # Backup: meshcore transport dropped (DISCONNECTED event is primary)
                if self.meshcore and not self.meshcore.is_connected:
                    await self._schedule_transport_reconnect('poll_detected')
                    await asyncio.sleep(5)
                    continue

                # Monitor web viewer process and health (never restart during shutdown)
                if (
                    self.web_viewer_integration
                    and self.web_viewer_integration.enabled
                    and self.connected
                    and not self._shutdown_event.is_set()
                ):
                    # Check if process died
                    if (self.web_viewer_integration and
                        self.web_viewer_integration.viewer_process and
                        self.web_viewer_integration.viewer_process.poll() is not None):
                        try:
                            self.logger.warning("Web viewer process died, restarting...")
                        except (AttributeError, TypeError):
                            print("Web viewer process died, restarting...")
                        self.web_viewer_integration.restart_viewer()

                    # Simple health check for web viewer
                    if (self.web_viewer_integration and
                        not self.web_viewer_integration.is_viewer_healthy()):
                        try:
                            self.logger.warning("Web viewer health check failed, restarting...")
                            self.web_viewer_integration.restart_viewer()
                        except (AttributeError, TypeError) as e:
                            print(f"Web viewer health check failed: {e}")

                # Periodically probe radio responsiveness
                # Skip entirely once a zombie is confirmed — only a power cycle
                # can recover the firmware; probing just generates log noise.
                if not self._radio_zombie_detected:
                    if self._last_radio_probe is None:
                        self._last_radio_probe = time.time()
                    probe_interval = self._radio_probe_interval_seconds()
                    if time.time() - self._last_radio_probe >= probe_interval:
                        self._last_radio_probe = time.time()
                        asyncio.create_task(self._probe_radio_health())

                # Periodically update system health in database (every 30 seconds)
                if time.time() - self._last_health_update >= 30:
                    try:
                        await self.get_system_health()  # This stores it in the database
                        self._last_health_update = time.time()
                    except Exception as e:
                        self.logger.debug(f"Error updating system health: {e}")

                    # Service health check and restart
                    restart_backoff = self.config.getint(
                        'Bot', 'service_restart_backoff_seconds', fallback=300
                    )
                    self._restart_unhealthy_services(time.time(), restart_backoff)

                await asyncio.sleep(5)  # Check every 5 seconds
        except KeyboardInterrupt:
            self.logger.info("Received interrupt signal")
        # Shutdown is owned by the entrypoint (e.g. meshcore_bot.run_bot finally)
        # so stop() runs exactly once; see stop() idempotency guard.

    async def stop(self) -> None:
        """Stop the bot.

        Performs graceful shutdown by stopping services, scheduling, and
        disconnecting from the mesh node.
        """
        if self._shutdown_lock is None:
            try:
                self._shutdown_lock = asyncio.Lock()
            except RuntimeError:
                self._shutdown_lock = None

        if self._shutdown_lock is not None:
            async with self._shutdown_lock:
                if self._shutdown_complete:
                    try:
                        self.logger.debug("stop() skipped — shutdown already completed")
                    except (AttributeError, TypeError):
                        pass
                    return
                await self._stop_inner()
        else:
            if self._shutdown_complete:
                return
            await self._stop_inner()

    async def _stop_inner(self) -> None:
        """Run one full shutdown pass (caller holds _shutdown_lock when used)."""
        try:
            try:
                self.logger.info("Stopping MeshCore Bot...")
            except (AttributeError, TypeError):
                print("Stopping MeshCore Bot...")

            self._shutdown_event.set()
            self.connected = False
            self._update_radio_connected_metadata(False)

            # Shutdown mesh graph first to flush pending writes
            if hasattr(self, 'mesh_graph') and self.mesh_graph:
                try:
                    self.mesh_graph.shutdown()
                except Exception as e:
                    self.logger.warning(f"Error shutting down mesh graph: {e}")

            # Stop feed manager
            if self.feed_manager:
                await self.feed_manager.stop()

            # Stop all loaded services
            for service_name, service_instance in self.services.items():
                try:
                    await service_instance.stop()
                    self.logger.info(f"Service '{service_name}' stopped")
                except Exception as e:
                    self.logger.error(f"Failed to stop service '{service_name}': {e}")

            # Stop web viewer with proper shutdown sequence
            if self.web_viewer_integration:
                # Web viewer has simpler shutdown
                self.web_viewer_integration.stop_viewer()
                try:
                    self.logger.info("Web viewer stopped")
                except (AttributeError, TypeError):
                    print("Web viewer stopped")

            # Wait for scheduler thread to exit (it checks self.connected)
            if hasattr(self, 'scheduler') and self.scheduler and self.scheduler.scheduler_thread:
                self.scheduler.join(timeout=5.0)

            if self.meshcore:
                disconnect_timeout = self.config.getfloat('Bot', 'disconnect_timeout_seconds', fallback=10.0)
                try:
                    await asyncio.wait_for(self.meshcore.disconnect(), timeout=disconnect_timeout)
                except asyncio.TimeoutError:
                    self.logger.warning(
                        "MeshCore disconnect timed out after %.1fs; continuing shutdown",
                        disconnect_timeout,
                    )
                except Exception as e:
                    self.logger.warning("Error during meshcore disconnect: %s", e)

            # Finish queued repeat-count writes now that no more packets arrive
            tracker = getattr(self, 'transmission_tracker', None)
            if tracker:
                try:
                    await asyncio.to_thread(tracker.close)
                except Exception as e:
                    self.logger.warning("Error finishing transmission tracker writes: %s", e)

            try:
                self.logger.info("Bot stopped")
            except (AttributeError, TypeError):
                print("Bot stopped")
        finally:
            self._shutdown_complete = True


    def _cleanup_web_viewer(self) -> None:
        """Cleanup web viewer resources on exit.

        Called by atexit handler to ensure the web viewer process is terminated
        properly when the bot shuts down.
        """
        try:
            if hasattr(self, 'web_viewer_integration') and self.web_viewer_integration:
                self.web_viewer_integration.stop_viewer()
        except (OSError, AttributeError, TypeError, ValueError):
            pass  # Do not log; stream may be closed during atexit

    def _cleanup_mesh_graph(self) -> None:
        """Cleanup mesh graph resources on exit.

        Called by atexit handler to ensure graph state is persisted
        properly when the bot shuts down.
        """
        try:
            if hasattr(self, 'mesh_graph') and self.mesh_graph:
                self.mesh_graph.shutdown()
        except (OSError, AttributeError, TypeError, ValueError):
            pass  # Do not log; stream may be closed during atexit

    def key_prefix(self, public_key: str) -> str:
        return public_key[:self.prefix_hex_chars]

    def is_valid_prefix(self, prefix: str) -> bool:
        return len(prefix) == self.prefix_hex_chars