Основные компоненты¶

LXMFBot¶

Основной класс бота, отвечающий за маршрутизацию сообщений, обработку команд и жизненный цикл бота.

from lxmfy import LXMFBot

bot = LXMFBot(
    name="MyBot",
    command_prefix="/",
    admins=set(),
    config_path=None,                 # по умолчанию "config" в рабочей директории
    reticulum_config_dir=None,        # или LXMFY_RETICULUM_CONFIG_DIR / "~/.reticulum"
    test_mode=False,                  # пропускает запуск RNS, для тестов
    log_level="INFO",                 # уровень логгера lxmfy, None не трогает логирование
    loglevel=None,                    # уровень логов RNS 0-7, None берёт конфиг reticulum

    # Анонсы
    announce=600,
    announce_immediately=True,
    announce_enabled=True,
    announce_display_name_file=None,  # файл в config_path, переопределяющий
                                      # анонсируемое имя (файл по умолчанию:
                                      # bot_display_name.txt)

    # Защита от спама
    rate_limit=5,
    cooldown=60,
    max_warnings=3,
    warning_timeout=300,

    # Коги
    cogs_dir="cogs",
    cogs_enabled=True,
    dynamic_cogs_enabled=True,
    external_cogs_enabled=True,
    external_cogs_sandbox_enabled=True,
    external_cogs_sandbox_type="auto",  # "auto", "landlock", "bwrap", "firejail", "none"
    external_cogs_timeout=30,
    hot_reloading=False,

    # Хранилище, события, права
    storage_type="json",              # "json", "sqlite", "msgpack" или "memory"
    storage_path="data",
    permissions_enabled=False,
    first_message_enabled=True,
    event_logging_enabled=True,
    max_logged_events=1000,
    event_middleware_enabled=True,

    # Безопасность
    signature_verification_enabled=False,
    require_message_signatures=False,
    require_stamps=False,             # отклоняет сообщения с невалидными stamp
    request_unknown_identities=False, # запрашивает идентичности отправителей у сети
    stamp_cost=None,                  # входящая стоимость stamp, None отключает
    include_tickets=True,             # прикладывает ticket ответа к исходящим
    identity_pinning_enabled=False,
    landlock_enabled=True,

    # Опциональные функции
    nlp_enabled=False,
    nlp_threshold=0.5,
    link_support_enabled=False,
    lxmf_commands_enabled=True,

    # Доставка
    message_persistence_enabled=True,
    message_queue_size=50,
    opportunistic_sending=True,
    direct_delivery_retries=3,
    propagation_fallback_enabled=True,
    propagation_node=None,            # хеш исходящего узла распространения
    autopeer_propagation=False,       # обнаруживает узлы распространения из анонсов
    autopeer_maxdepth=4,              # макс. глубина hops, None = без ограничения
    enable_propagation_node=False,    # запускает этого бота как узел распространения
    message_storage_limit_mb=500,     # лимит хранилища узла, только в режиме узла

    # Отложенные отправки
    pending_sends_enabled=True,       # удерживает отправки к неизвестным адресатам
    pending_sends_max=200,
    pending_sends_ttl=604800,         # 7 дней
    pending_sends_retry=300,          # секунды между проходами повторов

    # RRC
    rrc_enabled=False,
    rrc_hubs=[],
    rrc_rooms=[],
    rrc_nick=None,
    rrc_dest_name="rrc.hub",
    rrc_auto_reconnect=True,
    rrc_persist_sessions=True,
)

Все эти опции являются полями BotConfig. LXMFBot(**kwargs) пробрасывает каждый именованный аргумент, так что bot.config содержит итоговые значения.

Ключевые методы¶

• run(delay=10): Запускает главный цикл бота
• cleanup(): Сохраняет очереди, отменяет разговоры и останавливает планировщик, роутер и RNS. Вызывается автоматически при выходе из run().
• send(destination, message, title="Reply", lxmf_fields=None, stamp_cost=None, opportunistic=None, method=None, include_ticket=None, defer=None, reply_to=None, quote=None, thread=None): Отправляет сообщение адресату. stamp_cost переопределяет исходящую стоимость для этого сообщения, opportunistic переопределяет opportunistic_sending, include_ticket переопределяет include_tickets, а defer переопределяет pending_sends_enabled. reply_to, quote и thread задают поля трединга ответов.
• send_with_attachment(destination, message, attachment, title="Reply", stamp_cost=None, opportunistic=None): Отправляет сообщение с вложением
• command(name, description="No description provided", admin_only=False, permissions=None, usage=None, examples=None, category=None, aliases=None, threaded=False, rate_limit=None): Декоратор для регистрации команд. permissions переопределяет барьер DefaultPerms (ALL при admin_only, иначе USE_COMMANDS), threaded выполняет callback в рабочем потоке, rate_limit ограничивает вызовы на отправителя за окно cooldown, а usage, examples, category, aliases питают систему помощи. Команды поддерживают типизированные аргументы с автоматической конверсией.
• intent(name, examples): Декоратор для регистрации обработчиков NLP-интентов.
• nlp.export_model(): Экспортирует данные обученной NLP-модели.
• nlp.import_model(model_data): Импортирует ранее экспортированные данные NLP-модели.
• request_link(destination_hash, callback=None, app_name="lxmf", *aspects): Запрашивает RNS-соединение к адресату. Позволяет свои app_name и aspects (по умолчанию "lxmf" и "delivery").
• on_link(callback): Регистрирует обработчик входящих RNS-соединений.
• load_extension(name): Загружает модуль расширения cog по имени (напр. "cogs.utility").
• reload_extension(name): Перезагружает модуль расширения cog.
• add_cog(cog_instance): Добавляет экземпляр класса cog к боту.
• remove_cog(cog_name): Удаляет cog по имени класса.
• on_first_message(): Декоратор для обработки первых сообщений от пользователей
• on_message(): Декоратор для обработки всех сообщений (вызывается до обработки команд)
• received(function): Регистрирует callback, вызываемый с контекстом сообщения для каждого входящего сообщения, прошедшего pipeline без поглощения командой или интентом
• on_reaction(): Декоратор для обработки входящих реакций. Обработчики получают (sender, reaction), где reaction несёт ключи reaction_to, reaction_emoji и reaction_sender
• react(destination, message_hash, reaction): Отправляет реакцию на сообщение через поле LXMF FIELD_REACTION
• validate(): Выполняет проверки конфигурации бота
• get_landlock_status(): Возвращает доступность и состояние активации песочницы Landlock LSM для процесса бота
• diagnose_destination(destination, request_path=False, wait=0.0): Исследует состояние идентичности и пути для хеша адресата
• diagnose_connectivity(destination=None, request_path=False, wait=0.0): Выполняет полный doctor-отчёт и возвращает его dict-ом
• get_debugger(): Возвращает Debugger, привязанный к этому боту
• set_propagation_node(node_hash): Фиксирует исходящий узел распространения
• get_propagation_node_status(): Состояние настроенных, обнаруженных и текущего исходящего узлов распространения
• set_message_storage_limit(megabytes): Лимит хранилища при работе как узел распространения
• get_propagation_storage_stats(): Использование хранилища узла или dict с причиной недоступности
• connect_rrc(hub_hash, rooms=None, nick=None, dest_name=None, auto_reconnect=None): Подключается к RRC-хабу как клиент
• disconnect_rrc(hub_hash=None): Отключает одну или все сессии RRC-хабов
• on_rrc(callback=None): Декоратор или регистрация обработчика для событий RRC (handler(event, client, payload))
• on_delivery_event(callback=None): Подписывается на поток событий исходящей доставки, как декоратор или прямой вызов

Атрибуты¶

• config: Итоговая BotConfig
• commands, cogs: Реестры команд и когов
• storage: Активный backend хранилища
• scheduler: TaskScheduler для cron-подобных задач
• events: EventManager для обработчиков событий и dispatch
• middleware: MiddlewareManager для middleware команд
• permissions: PermissionManager для ролей и флагов
• spam_protection: SpamProtection для ограничений частоты, предупреждений и банов
• signature_manager: Слой политики подписей
• nlp: Классификатор интентов (матчит только при nlp_enabled)
• delivery: DeliveryTracker, поток исходящих событий
• conversations: ConversationManager для вопросов msg.ask
• rrc: RRCManager для сессий с несколькими хабами, None пока RRC не включён или connect_rrc() не выполнен
• local: RNS.Destination бота (его LXMF-адрес это bot.local.hash)

Анонсируемое имя¶

Имя, которое видят пиры, берётся из name, но для анонсов существуют два переопределения. Если задан announce_display_name_file и такой файл существует в config_path, его содержимое побеждает. Иначе читается bot_display_name.txt в config_path, если он есть. В обоих случаях присвоение bot.name = "Новое имя" ресинхронизирует анонсируемое имя во время выполнения.

Это позволяет операторам переименовывать бота без изменения кода, а анонсируемое имя может отличаться от внутреннего имени в конфигурации.

Структурированные команды через поля LXMF¶

Боты могут принимать команды, отправленные через FIELD_COMMANDS (0x09) LXMF, и автоматически отвечать с FIELD_RESULTS (0x0A). Это даёт структурированные запросы и ответы наряду с обычными текстовыми командами.

Входящие FIELD_COMMANDS разбираются и направляются через тот же реестр команд, что и текстовые, с общими проверками прав, разбором аргументов по аннотациям, потоками и middleware.

from lxmfy import LXMFBot, FIELD_COMMANDS, FIELD_RESULTS, pack_result, unpack_commands

bot = LXMFBot(name="FieldBot")

@bot.command(name="status", description="Return bot status")
def status_cmd(ctx):
    # ctx.fields содержит исходный словарь полей LXMF
    # ctx.request_id устанавливается автоматически, если команда его включала
    ctx.reply("Bot is online")

# Отправка структурированной команды из другого клиента LXMF:
# lxm.fields[FIELD_COMMANDS] = {"command": "status", "args": [], "request_id": "abc123"}
# router.handle_outbound(lxm)

# Ответ бота автоматически включает FIELD_RESULTS с ответом и request_id.

Чтобы отключить обработку полевых команд, установите lxmf_commands_enabled=False в BotConfig.

Реакции¶

Реакции передаются полем FIELD_REACTION (0x40) LXMF поверх пустого сообщения. pack_reaction и unpack_reaction собирают и разбирают это поле.

from lxmfy import pack_reaction, unpack_reaction

# Отправка реакции на сообщение
bot.react(destination_hash, message_hash_hex, "\U0001F44D")

# Приём реакций
@bot.on_reaction()
def on_reaction(sender, reaction):
    # reaction["reaction_to"]  — hex-хэш целевого сообщения
    # reaction["reaction_emoji"] — текст реакции (до 16 символов)
    # reaction["reaction_sender"] — отправитель
    print(f"{sender} reacted {reaction['reaction_emoji']} to {reaction['reaction_to']}")
    return True

Текст реакции обрезается до 16 печатаемых символов. Исходное поле остаётся доступным в ctx.fields и msg.fields для совместимости.

Трединг ответов¶

Ответы могут нести поля LXMF FIELD_REPLY_TO (0x30), FIELD_REPLY_QUOTE (0x31) и FIELD_THREAD (0x08). Клиенты, рендерящие треды, как MeshChatX и Sideband, показывают их как настоящие цитированные ответы вместо плоских сообщений.

msg.reply() тредит автоматически: устанавливает FIELD_REPLY_TO в хеш входящего сообщения и FIELD_THREAD в корень разговора.

@bot.command("status")
def status(msg):
    msg.reply("all systems nominal")          # тредованный ответ
    msg.reply("flat", reply_to=None)          # отказаться от трединга
    msg.reply("noted", quote=True)            # цитирует входящий текст

Для отправок, не являющихся ответами, передавайте поля явно:

bot.send(dest, "see above", reply_to=msg_hash_hex, quote="earlier text")

Входящие ответы парсятся в контекст сообщения:

@bot.command("ctx")
def ctx_cmd(msg):
    msg.reply_to      # hex-хеш, на который отвечает это сообщение, или None
    msg.reply_quote   # цитируемый текст ответа, или None
    msg.thread        # hex-хеш корня треда, или None

pack_reply(message_hash, quote=..., thread=...) и unpack_reply(fields) экспортируются для ручной обработки полей.

Разговоры¶

Команды могут задавать отправителю вопрос и трактовать его следующее сообщение как ответ, вместо диспетчеризации как команды:

@bot.command("report")
def report(msg):
    title = msg.ask("Report title?", timeout=300)
    if title is None:
        msg.reply("Timed out.")
        return
    body = msg.ask("Describe the issue.", timeout=600)
    if body is None:
        msg.reply("Timed out.")
        return
    msg.reply(f"Filed: {title.content}")

msg.ask(prompt, timeout=..., validator=...) блокирует обработчик, пока не придёт ответ, не сработает timeout или разговор не будет отменён. Возвращает Answer с content, fields, hash, sender и шорткатом reply(text).

Валидатор отклоняет плохие ответы и переспрашивает:

num = msg.ask(
    "Pick a number",
    validator=lambda a: None if a.content.isdigit() else "Digits only",
)

В асинхронных обработчиках команд используйте await msg.ask_async(...). Для долгих ожиданий или когда может быть много открытых разговоров, используйте стиль колбеков, чтобы ни один поток не оставался припаркованным:

msg.ask(
    "Send the log file",
    on_answer=lambda ans: ans.reply("received"),
    on_timeout=lambda sender: bot.send(sender, "Too slow."),
    timeout=3600,
)

Заметки:

• Отправка зарегистрированной команды во время ожидания вопроса отменяет вопрос и выполняет команду. У пользователей всегда есть выход.
• bot.conversations.pending_count() и bot.conversations.cancel(sender) открывают реестр для диагностики и административных инструментов.
• Реестр ограничен 1024 ожидающими вопросами. ask возвращает None, когда он заполнен.
• Блокирующий ask паркует поток доставки, обрабатывающий это сообщение. Это безопасно для прямых доставок, но боты, синхронизирующие большие пачки с узла распространения, должны предпочитать колбеки on_answer.

Хранилище¶

Фреймворк предоставляет три бэкенда хранилища:

JSONStorage¶

from lxmfy import JSONStorage

storage = JSONStorage("data")

SQLiteStorage¶

from lxmfy import SQLiteStorage

storage = SQLiteStorage("data/bot.db")

MsgPackStorage¶

from lxmfy.storage import MsgPackStorage

storage = MsgPackStorage("data") # {key}.msgpack files

MemoryStorage¶

from lxmfy.storage import MemoryStorage

storage = MemoryStorage() # Полностью в памяти

Команды¶

Регистрация и обработка команд:

@bot.command(name="hello", description="Says hello")
def hello(ctx):
    ctx.reply(f"Hello {ctx.sender}!")

Метаданные помощи и контроль доступа приходят из дополнительных kwarg-ов декоратора:

from lxmfy import DefaultPerms

@bot.command(
    name="purge",
    description="Clear stored data",
    permissions=DefaultPerms.MANAGE_MESSAGES,
    usage="/purge <key>",
    examples=["/purge cache"],
    category="Admin",
    aliases=["clear"],
)
def purge(ctx, key: str):
    bot.storage.delete(key)
    ctx.reply(f"Deleted {key}")

permissions переопределяет барьер по умолчанию: USE_COMMANDS для обычных команд, ALL для admin_only. category группирует команду в выводе /help. aliases являются только метаданными помощи: алиасы показываются пользователям, но не регистрируются для dispatch, так что /clear не выполнит purge, если только вы не зарегистрируете его как вторую команду.

Аргументы с аннотациями типов¶

Команды автоматически разбирают и преобразуют аргументы на основе аннотаций типов в функции-обработчике.

@bot.command(name="add", description="Adds two numbers")
def add(ctx, a: int, b: int):
    result = a + b
    ctx.reply(f"The result is {result}")

Ограничения частоты для отдельной команды¶

Ограничивает, как часто один отправитель может вызывать команду в рамках глобального окна cooldown. При достижении отклоняется только вызов, никаких предупреждений или банов.

@bot.command(name="report", rate_limit=3)
def report(ctx):
    # каждый отправитель может вызвать это 3 раза за период cooldown
    ...

Требует permissions_enabled=True, как и глобальное ограничение. Пользователи с ролью admin или BYPASS_SPAM пропускают проверку.

Защита от спама¶

bot.spam_protection обеспечивает глобальное ограничение: отправитель может послать rate_limit сообщений в рамках каждого окна cooldown. Превышение добавляет предупреждение и отклоняет сообщение. На max_warnings отправителя банят. Предупреждения истекают через warning_timeout секунд без нарушений.

Проверки спама выполняются внутри события message_received и требуют permissions_enabled=True.

bot = LXMFBot(
    name="GuardedBot",
    permissions_enabled=True,
    rate_limit=5,        # сообщений за окно cooldown
    cooldown=60,         # длительность окна в секундах
    max_warnings=3,      # предупреждений до бана
    warning_timeout=300, # секунд до сброса предупреждений
)

# Снять бан вручную
bot.spam_protection.unban(sender_hash)

Предупреждения, баны и счётчики персистятся в настроенном backend хранилища, так что баны переживают рестарты. Отправители с BYPASS_SPAM никогда не ограничиваются и не банятся. Покомандное rate_limit в @bot.command мягче: оно только отклоняет вызов и никогда не предупреждает и не банит.

Система помощи¶

Фреймворк включает интерактивный генератор помощи с категоризированными меню на основе метаданных когов и команд.

# Команда help регистрируется автоматически.
# Пользователи могут использовать '/help' или '/help <command>'

Потоковые команды¶

Для долгих или блокирующих операций, которые не взаимодействуют напрямую с Reticulum Network Stack, команды можно запускать в отдельном потоке, чтобы бот оставался отзывчивым.

import time

@bot.command(name="long_task", description="Performs a long-running task in a separate thread", threaded=True)
def long_task_command(ctx):
    ctx.reply("Starting a long task... please wait.")
    time.sleep(10) # Выполняется в отдельном потоке
    ctx.reply("Long task completed!")

События¶

Система событий для обработки различных событий бота:

@bot.events.on("message_received", EventPriority.HIGHEST)
def handle_message(event):
    # event.data несёт полезную нагрузку, напр. sender и message
    event.cancel()  # останавливает последующих обработчиков и дальнейшую обработку

Обработчики выполняются в порядке EventPriority: HIGHEST, HIGH, NORMAL, LOW. Сама проверка спама является обработчиком message_received с HIGHEST, так что отмена этого события является механизмом, которым ограничитель отбрасывает сообщения.

Диспетчеризируйте свои события:

from lxmfy import Event

bot.events.dispatch(Event("order_placed", data={"user": ctx.sender}))

event_logging_enabled, max_logged_events и event_middleware_enabled существуют в BotConfig, но не подключены: события не пишутся в хранилище, а bot.events.use() является stub. Считайте их зарезервированными.

Тестирование¶

lxmfy.testing.TestBot является LXMFBot, предварительно настроенный для тестов. Никакой экземпляр Reticulum не запускается. Входящие сообщения проходят настоящий pipeline получения (middleware, проверки спама, права, dispatch), а исходящие отправки захватываются для утверждений.

from lxmfy import TestBot

def test_ping():
    with TestBot() as bot:
        @bot.command("ping")
        def ping(msg):
            msg.reply("pong")

        sent = bot.receive("/ping", sender="alice")
        assert sent[0].content == "pong"
        assert sent[0].destination == bot.sender_hex("alice")

• bot.receive(content, sender=..., fields=..., message_hash=...) вводит сообщение и возвращает созданные объекты SentMessage. Отправители именованные: "alice" мапится на стабильный фейковый хеш, или передайте hex-хеш адресата напрямую.
• bot.drain() вытягивает исходящие сообщения из очереди. bot.outbox накапливает всё отправленное. bot.last_sent(sender=...) достаёт последнее.
• bot.wait_sent(n, timeout=...) ждёт потоковые команды.
• bot.receive_later(content, sender=..., delay=...) отвечает на блокирующие вызовы msg.ask из daemon-потока.
• fake_message(content, source_hash=..., ...) строит входящее сообщение для прямого вызова bot._message_received.

SentMessage оборачивает каждое захваченное исходящее сообщение: destination (hex), content, title, fields, method, include_ticket, stamp_cost и raw для базового объекта.

Тестовый набор репозитория также включает сценарии надёжности и стресса. Запускайте их тестовым раннером репозитория.

Расширенный набор проверок надёжности¶

Фреймворк включает обширный набор автоматизированных тестов для суровых условий:

• Manifold Testing: проверка математической топологии векторного пространства намерений NLP.
• Chaos Engineering: симуляция битовой гнили, отказа SD-карты и порчи хранилища.
• Temporal Drift: проверка устойчивости к скачкам системных часов (±1 год).
• Leak Detection: долгосрочное отслеживание памяти, файловых дескрипторов и потоков.

Права доступа¶

Система прав для контроля доступа к возможностям бота:

from lxmfy import DefaultPerms

@bot.command(name="admin", description="Admin command", admin_only=True)
def admin_command(ctx):
    if ctx.is_admin:
        ctx.reply("Admin command executed")

Включите через permissions_enabled=True. Флаги DefaultPerms:

• USE_BOT, SEND_MESSAGES, USE_COMMANDS: базовый доступ
• MANAGE_MESSAGES, MANAGE_COMMANDS, MANAGE_USERS: повышенный
• BYPASS_RATELIMIT, BYPASS_SPAM, VIEW_ADMIN_COMMANDS: специальные
• VIEW_EVENTS, MANAGE_EVENTS, BYPASS_EVENT_CHECKS: система событий
• NONE, ALL: шорткаты

bot.permissions управляет ролями и назначениями:

bot.permissions.create_role("moderator", DefaultPerms.MANAGE_MESSAGES | DefaultPerms.BYPASS_SPAM)
bot.permissions.assign_role(user_hash, "moderator")
bot.permissions.remove_role(user_hash, "moderator")
bot.permissions.has_permission(user_hash, DefaultPerms.USE_COMMANDS)

Роли и назначения персистятся в настроенном backend хранилища. Существуют две встроенные роли, которые нельзя удалить: user (по умолчанию) и admin, автоматически выдаваемая каждому хешу в admins.

Middleware¶

Система middleware для обработки сообщений и событий:

from lxmfy import MiddlewareType

@bot.middleware.register(MiddlewareType.PRE_COMMAND)
def pre_command_middleware(ctx):
    # ctx оборачивает контекст сообщения, ctx.cancelled отбрасывает его
    if "spamword" in ctx.data.content:
        ctx.cancel()

Три точки pipeline выполняют middleware:

• PRE_COMMAND: до dispatch команд, после проверок спама. Если цепочка возвращает None, сообщение полностью отменяется.
• POST_COMMAND: после callback команды (включая потоковые команды, срабатывающие в рабочем потоке).
• PRE_EVENT: до dispatch события message_received.

POST_EVENT, REQUEST и RESPONSE существуют в MiddlewareType, но ничто в pipeline их пока не выполняет.

Вложения¶

Поддержка отправки файлов, изображений и аудио:

from lxmfy import Attachment, AttachmentType

attachment = Attachment(
    type=AttachmentType.IMAGE,
    name="image.jpg",
    data=image_data,
    format="jpg"
)
bot.send_with_attachment(destination, "Here's an image", attachment)

Внешний вид иконки (поле LXMF)¶

Боту можно назначить пользовательскую иконку, которую отображают совместимые клиенты LXMF. Используется LXMF.FIELD_ICON_APPEARANCE.

from lxmfy import IconAppearance, pack_icon_appearance_field
import LXMF # Требуется для LXMF.FIELD_ICON_APPEARANCE

# Определяем внешний вид иконки
icon_data = IconAppearance(
    icon_name="smart_toy",  # Имя из Material Symbols
    fg_color=b'\xFF\xFF\xFF',  # Белый передний план (3 байта)
    bg_color=b'\x4A\x90\xE2'   # Синий фон (3 байта)
)

# Упаковываем в формат поля LXMF
icon_lxmf_field = pack_icon_appearance_field(icon_data)

# Отправляем сообщение с этой иконкой
bot.send(
    destination_hash_str,
    "Hello from your friendly bot!",
    title="Bot Message",
    lxmf_fields=icon_lxmf_field
)

# Можно комбинировать с другими полями, например с вложениями:
# attachment_field = pack_attachment(some_attachment)
# combined_fields = {**icon_lxmf_field, **attachment_field}
# bot.send(destination, "Message with icon and attachment", lxmf_fields=combined_fields)

Планировщик¶

Система планирования задач:

@bot.scheduler.schedule(name="daily_task", cron_expr="0 0 * * *")
def daily_task():
    # Выполняется ежедневно в полночь
    pass

Подписи¶

LXMFy предоставляет настройки для встроенной криптографической подписи и проверки сообщений LXMF:

from lxmfy import LXMFBot

bot = LXMFBot(
    name="SecureBot",
    signature_verification_enabled=True,  # Включить проверку подписей
    require_message_signatures=False      # True — отклонять неподписанные сообщения
)

Сами криптографические операции выполняет LXMF/RNS, а не LXMFy.

Песочница Landlock LSM¶

На ядрах Linux с поддержкой Landlock (5.13+) LXMFy может ограничивать доступ к файловой системе для процесса бота и для внешних скриптовых когов.

Песочница процесса бота

Когда landlock_enabled=True (по умолчанию) и бот не запущен в test_mode, бот вызывает apply_landlock_sandbox() при инициализации. Системные каталоги доступны только для чтения; хранилище бота, конфигурация, коги, конфигурация Reticulum и временные пути остаются записываемыми.

bot = LXMFBot(
    name="SecureBot",
    landlock_enabled=True,
)

status = bot.get_landlock_status()
# ключи status: landlock_kernel_supported, landlock_requested,
# landlock_auto_enabled, landlock_disabled_by_env, landlock_active

Переопределение через окружение

• LXMFY_LANDLOCK=0: отключить Landlock даже на поддерживаемых ядрах
• LXMFY_LANDLOCK=1: попытаться применить Landlock в Linux независимо от автоопределения
• не задано: следовать landlock_enabled и автоопределению ядра

Песочница внешних когов

Скриптовые коги используют external_cogs_sandbox_type. В режиме auto предпочитается Landlock, так как ему не нужны внешние инструменты. Полный список вариантов песочницы — в руководстве Создание ботов.

Закрепление идентификатора¶

LXMFy поддерживает опциональное закрепление идентификатора (pinning) для защиты от подмены при ротации или компрометации идентификатора. Когда включено, бот «закрепляет» LXMF-адрес за первым увиденным открытым ключом.

bot = LXMFBot(
    identity_pinning_enabled=True
)

Методы SignatureManager¶

SignatureManager доступен как bot.signature_manager, когда signature_verification_enabled=True:

• should_verify_message(sender): определить, нужно ли проверять сообщение от данного отправителя
• handle_unsigned_message(sender, message_hash): обработать сообщения без валидной подписи согласно политике

Как работают подписи LXMF¶

LXMF автоматически подписывает все исходящие сообщения идентификатором RNS отправителя во время операции pack(). При получении сообщений LXMF проверяет подписи и предоставляет:

• message.signature_validated: булево значение валидности подписи
• message.unverified_reason: код причины при неудачной проверке (например, SIGNATURE_INVALID, SOURCE_UNKNOWN)

LXMFy использует эти встроенные свойства LXMF для применения политики подписей бота.

Доставка сообщений¶

LXMFy предоставляет расширенные возможности доставки, включая узлы распространения и автоматические повторы:

Узлы распространения¶

Отправка сообщений через конкретные узлы распространения для повышения надёжности в сети Reticulum:

# Настройте узел распространения один раз на уровне конфигурации/времени выполнения
bot.set_propagation_node("<propagation_node_hash>")

# Отправка с настроенным поведением доставки
bot.send(
    destination_hash,
    "Message content"
)

# Хэш узла распространения должен быть валидным узлом распространения LXMF
# в сети Reticulum

Узел также можно задать при создании через propagation_node="<hash>", или бот может находить узлы сам:

bot = LXMFBot(
    name="AutoBot",
    autopeer_propagation=True, # учит узлы из анонсов
    autopeer_maxdepth=4,       # игнорирует узлы глубже 4 hops
)

bot.get_propagation_node_status() сообщает о ручном узле, обнаруженных узлах и исходящем узле, используемом в данный момент.

Автоматические повторы¶

Настройка автоматических повторных попыток для неудачных прямых доставок:

bot = LXMFBot(
    name="ReliableBot",
    direct_delivery_retries=5,  # Повторять прямую доставку до 5 раз
    propagation_fallback_enabled=True
)

bot.send(destination_hash, "Important message")

# direct_delivery_retries по умолчанию равно 3
# Логика повторов автоматически обрабатывает обратные вызовы доставки

Система повторов отслеживает попытки доставки по каждому получателю и автоматически повторяет неудачные доставки. Успешные доставки сбрасывают счётчик повторов для этого получателя.

Отложенные отправки¶

Отправка адресату, чью идентичность узел ещё не слышал, обычно сразу проваливается. С pending_sends_enabled (по умолчанию) сообщение вместо этого удерживается в хранилище, а затем автоматически выталкивается, когда адресат анонсируется или при периодическом проходе.

bot = LXMFBot(
    pending_sends_enabled=True,
    pending_sends_max=200,     # самые старые удержанные сообщения выпадают сверх этого
    pending_sends_ttl=604800,  # удержанные сообщения истекают через 7 дней
    pending_sends_retry=300,   # секунды между проходами в run()
)

# Переопределение для отдельной отправки
bot.send(dest, "hold this", defer=True)
bot.send(dest, "send or drop", defer=False)

Удержанные отправки появляются как held (unknown peers) в административной команде /queue и создают события deferred в трекере доставки.

Персистентность сообщений¶

Исходящие сообщения можно сохранять на диск, чтобы они доставлялись даже после перезапуска бота. Персистентность включена по умолчанию. Очередь исходящих в памяти ограничена (message_queue_size, по умолчанию 50) и при переполнении отбрасывает самое старое сообщение. Невалидные хэши получателей не восстанавливаются.

bot = LXMFBot(
    message_persistence_enabled=True,
    message_queue_size=50,
)

Stamp-ы и ticket-ы¶

Стоимость stamp заставляет отправителей выполнять proof-of-work до того, как их сообщение будет принято, что сдерживает нежелательный трафик. LXMFy открывает обе стороны механизма.

bot = LXMFBot(
    stamp_cost=16,          # требует эту входящую стоимость stamp
    require_stamps=True,    # отклоняет сообщения с невалидными stamp
    include_tickets=True,   # позволяет пирам отвечать без генерации stamp
)

• stamp_cost является входящим требованием. Исходящая стоимость отправки по-прежнему берётся из анонса пира, если только вы не передадите stamp_cost= в bot.send().
• include_tickets (по умолчанию True) прикладывает ticket ответа к исходящим сообщениям, так что пир с требованием stamp может ответить без оплаты. Переопределяйте для отдельной отправки через include_ticket=.
• request_unknown_identities=True запрашивает у сети идентичность отправителя, когда сообщение приходит из неизвестного источника, что помогает проверкам stamp и подписей разрешиться вместо слепого провала.

Контроль во время выполнения лежит в разделе «Управление роутером»: set_inbound_stamp_cost, enforce_stamps, ignore_stamps, generate_ticket и методы инспекции ticket-ов.

Работа как узел распространения¶

Бот может одновременно быть узлом распространения LXMF, храня сообщения для пиров, которые офлайн:

bot = LXMFBot(
    enable_propagation_node=True,
    message_storage_limit_mb=500,
)

bot.set_message_storage_limit(750)
stats = bot.get_propagation_storage_stats()

get_propagation_node_status() работает для обеих ролей: сообщает об исходящем узле, который этот бот использует, и о том, является ли он сам узлом. Связанные контроли: announce_propagation_node() рекламирует узел, set_retain_on_node() сохраняет доставленные сообщения на нём, а allow_control_identity() / disallow_control_identity() управляют тем, какие идентичности могут использовать контрольный канал узла.

События доставки¶

bot.delivery записывает ограниченный поток событий исходящего жизненного цикла, чтобы наблюдать поток сообщений без чтения логов. Стадии: queued, deferred, dispatched, delivered, failed, cancelled, dropped. Недавний хвост персистится в хранилище и восстанавливается при запуске.

@bot.on_delivery_event()
def watch(event):
    print(event["stage"], event.get("destination"), event.get("reason"))

# Или прямая инспекция
recent = bot.delivery.recent(20)
failures = bot.delivery.recent(stage="failed")
to_peer = bot.delivery.recent(destination="aa11bb...")

Каждое событие является dict-ом с ts, stage и опционально destination, message_id, hash, method, attempts, reason, title.

Админы получают команду /delivery [limit], рендерящую ту же хронологию в чате, а lxmfy debug показывает сводку хронологии доставки в проверках pipeline отправки.

Встроенные административные команды¶

Эти команды регистрируются автоматически и требуют, чтобы отправитель был в admins, когда права включены:

Управление роутером¶

Тонкие обёртки над базовым LXMRouter для управления отправителями, ticket-ами, управления исходящей очередью и синхронизации узла распространения. Все принимают хеши адресатов как hex-строки и возвращают False, когда роутер не работает (например в test_mode).

Управление отправителями (входящее)

• ignore_destination(destination) / unignore_destination(destination) / is_ignored(destination): Отбрасывает входящие сообщения от отправителя
• allow_destination(destination) / disallow_destination(destination): Управление whitelist, когда роутер работает в режиме allow-list
• prioritise_destination(destination) / unprioritise_destination(destination): Список приоритетных отправителей
• set_inbound_stamp_cost(stamp_cost): Требует стоимость stamp на входящих сообщениях (None очищает)
• enforce_stamps() / ignore_stamps(): Переключатели требования входящих stamp-ов

Ticket-ы

• generate_ticket(destination, expiry=None): Выдаёт входящий stamp-ticket для отправителя
• get_inbound_tickets(destination): Ticket-ы, удерживаемые для отправителя
• get_outbound_ticket(destination) / get_outbound_ticket_expiry(destination) / get_outbound_stamp_cost(destination): Состояние исходящего ticket-а, узнанное из сети

Исходящая очередь

• outbound_queue(): Снимок ожидающих исходящих сообщений
• get_outbound_progress(lxm_hash): Прогресс доставки для хеша сообщения или None
• cancel_outbound(message_id): Удаляет сообщение из очереди до доставки
• delivery_link_available(destination): Существует ли активное RNS-соединение к адресату

Входящая очередь

• has_message(message_hash): Был ли уже доставлен входящий LXM-хеш
• inbound_count(): Активные входящие передачи ресурсов в процессе
• inbound_transfers(): Снимок каждой передачи с хешем, размером, прогрессом и статусом
• cancel_inbound(resource_hash): Прерывает активную входящую передачу
• cancel_all_inbound(): Прерывает все активные входящие передачи, возвращает отменённое количество

Обнаружение пиров

Метаданные анонсов для адресатов, которых слышал этот узел:

• get_peer_app_data(destination): Сырые байты анонсированных app_data
• get_peer_lxmf_data(destination): Декодированные метаданные LXMF-анонса (display_name, stamp_cost, capabilities) или None, когда пир не анонсировал валидные данные LXMF
• get_peer_announce(destination): Полная запись анонса с hops, received_at, interface и app_data
• list_peer_announces(limit=100): Все услышанные анонсы, новейшие первыми

Распространение

• sync_propagation_node(max_messages=None): Тянет сообщения с настроенного узла распространения
• cancel_propagation_sync(): Останавливает текущую синхронизацию
• get_propagation_stats(): Состояние передачи и лимиты узла или None
• set_retain_on_node(retain): Сохраняет доставленные сообщения на узле
• announce_propagation_node(): Анонсирует этот узел как узел распространения
• allow_control_identity(destination) / disallow_control_identity(destination): Whitelist контрольного канала распространения

Приём

• ingest_lxm_uri(uri): Импортирует сообщение URI lxm:// во входящую очередь

Диагностика¶

Когда сообщения не текут, отладчик проверяет весь путь вместо угадывания: конфигурацию Reticulum, состояние общего экземпляра, интерфейсы, идентичность, поведение анонсов, конфигурацию доставки и pipeline отправки.

lxmfy debug                          # полный doctor-отчёт, сохранённый в файл
lxmfy debug probe <hash> --request-path --wait 30
lxmfy debug send <hash>              # отслеживает тестовую отправку
lxmfy debug receive                  # проверяет готовность приёма
lxmfy debug compare <hash_a> <hash_b>
lxmfy debug tips                     # исправления распространённых сбоев

Отчёты по умолчанию редактируются ради приватности: домашние пути и хеши обрезаются. --json выдаёт машиночитаемый вывод, -o ФАЙЛ записывает отчёт, --no-save пропускает файл, --no-privacy сохраняет полные значения для локального использования, а --no-color или NO_COLOR отключает ANSI-вывод.

Те же проверки можно вызывать из кода:

report = bot.diagnose_connectivity()            # doctor-отчёт как dict
probe = bot.diagnose_destination(               # исследование идентичности и пути
    "<peer_hash>", request_path=True, wait=30,
)

debugger = bot.get_debugger()                   # полный API
checks = debugger.check_send_pipeline()
verdict = debugger.run_doctor(destination)

lxmfy/debugger.py также экспортирует самостоятельный помощник diagnose_destination(hash, ...) плюс типы отчётов CheckResult, DestinationProbe, DoctorReport и MessageDebugger для собственных инструментов.

Обработчики сообщений¶

LXMFy предоставляет декораторы для обработки разных типов входящих сообщений:

Обработчик первого сообщения¶

Обработка первого сообщения от каждого пользователя:

@bot.on_first_message()
def welcome_user(sender, message):
    content = message.content.decode("utf-8")
    bot.send(sender, f"Welcome! You said: {content}")
    return True  # Верните True, чтобы остановить дальнейшую обработку

Общий обработчик сообщений¶

Обработка всех входящих сообщений до обработки команд:

@bot.on_message()
def handle_all_messages(sender, message):
    content = message.content.decode("utf-8").strip()

    # Своя логика здесь
    if content.startswith("echo:"):
        bot.send(sender, content[5:])
        return True  # Остановить дальнейшую обработку

    return False  # Перейти к обработке команд

Обработчики сообщений вызываются в таком порядке: 1. обработчик первого сообщения (если это первое сообщение от отправителя) 2. общие обработчики (зарегистрированные через on_message()) 3. обработка команд (если сообщение начинается с префикса команды)

Резервный callback¶

bot.received(fn) регистрирует callback, выполняемый в конце pipeline для сообщений, которые ничто другое не поглотило: ни обработчик первого сообщения, ни обработчик on_message с возвратом True, ни подходящая команда или NLP-интент. Callback получает тот же контекст сообщения, что и команды, с msg.sender, msg.content, msg.reply() и прочим.

@bot.received
def fallback(msg):
    msg.reply("Sorry, I did not understand that.")

Используйте как универсальный обработчик для свободного текста.

Reticulum Relay Chat (RRC)¶

Боты могут подключаться к хабам RRC по RNS Link с конвертами CBOR. Пакет: lxmfy.rrc.

Опции BotConfig¶

• rrc_enabled (bool, по умолчанию False): подключать настроенные хабы при запуске
• rrc_hubs (список hex-хэшей): хэши получателей хабов
• rrc_rooms (список строк): комнаты для авто-join после WELCOME
• rrc_nick (строка или None): никнейм в HELLO и сообщениях комнат
• rrc_dest_name (строка, по умолчанию "rrc.hub"): имя получателя для построения destination хаба
• rrc_auto_reconnect (bool, по умолчанию True): переподключение после потери link
• rrc_persist_sessions (bool, по умолчанию True): сохранять хабы и комнаты между перезапусками
• reticulum_config_dir (строка или None): каталог конфигурации Reticulum. Также задаётся через LXMFY_RETICULUM_CONFIG_DIR. Используйте ту же конфигурацию, что у MeshChatX (часто ~/.reticulum), чтобы анонсы хаба были видны.

Пример¶

from lxmfy import LXMFBot, RRCMessage

bot = LXMFBot(
    name="RoomBot",
    reticulum_config_dir="~/.reticulum",
    rrc_enabled=True,
    rrc_hubs=["664fc0e8d2e448658e37bb3f34e6c88f"],
    rrc_rooms=["general"],
    rrc_nick="RoomBot",
)

@bot.on_rrc
def on_rrc(event, client, payload):
    if event == "msg" and isinstance(payload, RRCMessage) and payload.mention:
        client.send_message(payload.room, f"Hi {payload.nick}")

# API времени выполнения
# bot.connect_rrc(hub_hash, rooms=["general"])
# bot.rrc.send_message("general", "hello")
# bot.rrc.send_notice("general", "notice")
# bot.rrc.send_action("general", "waves")
# bot.rrc.join("ops")
# bot.rrc.part("ops")
# bot.rrc.status()
# bot.disconnect_rrc()

Экспортируемые типы¶

• RRCClient: сессия с одним хабом
• RRCManager: менеджер нескольких хабов (bot.rrc)
• RRCMessage: данные события комнаты (kind, room, text, nick, src, mention, ...)
• RRC_VERSION: константа версии проводного протокола
• DEFAULT_DEST_NAME: Имя адресата хаба по умолчанию ("rrc.hub")
• make_envelope, encode_envelope, decode_envelope, validate_envelope, normalize_room: Помощники wire-формата для инструментов, говорящих с хабами напрямую

Типичные события, передаваемые обработчикам @bot.on_rrc: status, welcome, joined, parted, msg, notice, action, motd, error и rtt.

Шаблоны¶

Фреймворк включает несколько готовых шаблонов ботов:

EchoBot¶

Простой echo-бот, повторяющий сообщения:

from lxmfy.templates import EchoBot

bot = EchoBot()
bot.run()

NoteBot¶

Бот для заметок с хранилищем JSON:

from lxmfy.templates import NoteBot

bot = NoteBot()
bot.run()

ReminderBot¶

Бот-напоминалка с хранилищем SQLite:

from lxmfy.templates import ReminderBot

bot = ReminderBot()
bot.run()

RRCBot¶

RRC-бот для комнат: подключается к настроенным хабам и отвечает на @упоминания. По умолчанию хаб 664fc0e8d2e448658e37bb3f34e6c88f, комната #general и ~/.reticulum, если доступен.

from lxmfy.templates import RRCBot

bot = RRCBot(
    hubs=["664fc0e8d2e448658e37bb3f34e6c88f"],
    rooms=["general"],
    nick="RRCBot",
    reticulum_config_dir="~/.reticulum",
)
bot.run()

Инструменты CLI¶

Фреймворк предоставляет инструменты командной строки для управления ботами. Запуск lxmfy без аргументов открывает интерактивное меню.

# Интерактивный scaffold полного проекта
lxmfy init mybot                 # директория проекта, bot.py, cogs/, README
lxmfy init --here --yes          # текущая директория, принимает все значения по умолчанию

# Создать один файл бота
lxmfy create mybot
lxmfy create --template echo mybot
lxmfy create --template rrc my_rrc_bot
lxmfy create mybot --no-cogs     # пропускает пакет cogs
lxmfy create --output dir/bot.py --name MyBot

# Запустить бота из шаблона
lxmfy run echo
lxmfy run rrc
lxmfy run reminder --name "MyReminder"

# Диагностировать соединение (см. раздел Диагностика)
lxmfy debug
lxmfy debug probe <hash> --request-path --wait 30

# Тестировать проверку подписей с сообщением
lxmfy signatures test

# Включить проверку подписей
lxmfy signatures enable

# Выключить проверку подписей
lxmfy signatures disable

lxmfy init спрашивает имя проекта, шаблон, backend хранилища, префикс команд и хеши админов. Каждый вопрос принимает вместо этого флаг: --dir, --bot-name, --template, --storage, --prefix, --admins, --no-cogs, --force, --yes. На не-TTY stdin берутся значения по умолчанию.

Шаблоны для create и run: basic, echo, reminder, note, cogtest, rrc.

Обработка ошибок¶

Перехватывайте завершение работы и ошибки выполнения вокруг bot.run():

try:
    bot.run()
except KeyboardInterrupt:
    bot.cleanup()
except Exception as e:
    logger.error(f"Error running bot: {str(e)}")

Справочник модулей¶

Генерируется из строк документации исходного кода.

 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531

class LXMFBot(
    AnnounceMixin,
    DispatchMixin,
    CogMixin,
    InboundMixin,
    OutboundMixin,
    LinkMixin,
    RRCMixin,
    PropagationMixin,
):
    """Main bot class for handling LXMF messages and commands.

    This class manages the bot's lifecycle, including:
    - Message routing and delivery
    - Command registration and execution
    - Cog (extension) loading and management
    - Spam protection
    - Admin privileges
    """

    def __init__(self, name: str | None = None, **kwargs: Any):
        """Initialize a new LXMFBot instance.

        Args:
            name: Optional bot name, same as the name config key.
            **kwargs: Override default configuration settings

        """
        if name is not None:
            kwargs["name"] = name
        self.config = BotConfig(**kwargs)
        self.commands = {}
        self.cogs = {}
        self.first_message_handlers = []
        self.message_handlers = []
        self.reaction_handlers = []
        self.delivery_callbacks = []
        self.receipts = []
        self._receive_lock = threading.Lock()
        self._delivery_lock = threading.Lock()
        self._last_pending_flush = 0.0
        queue_size = max(1, int(getattr(self.config, "message_queue_size", 50) or 50))
        self.queue = Queue(maxsize=queue_size)
        self.announce_time = 600
        self.router = None
        self.local = None
        self.logger = logging.getLogger("lxmfy")
        self._configure_logging()
        self.thread_pool = ThreadPoolExecutor(
            max_workers=5,
        )  # For offloading CPU-bound or blocking I/O tasks
        self.scheduler = TaskScheduler(self)  # Initialize the scheduler

        if self.config.config_path:
            self.config_path = self.config.config_path
        else:
            self.config_path = os.path.join(os.getcwd(), "config")

        os.makedirs(self.config_path, exist_ok=True)
        if self.config.test_mode and not self.config.reticulum_config_dir:
            self.reticulum_config_dir = os.path.abspath(self.config_path)
        else:
            self.reticulum_config_dir = resolve_reticulum_config_dir(
                self.config.reticulum_config_dir,
                self.config_path,
            )
        os.makedirs(self.reticulum_config_dir, exist_ok=True)
        if not self.config.test_mode and is_isolated_reticulum_dir(
            self.reticulum_config_dir,
            self.config_path,
        ):
            ensure_isolated_share_instance_disabled(self.reticulum_config_dir)

        if self.config.storage_type == "json":
            self.storage = Storage(JSONStorage(self.config.storage_path))
        elif self.config.storage_type == "sqlite":
            self.storage = Storage(SQLiteStorage(self.config.storage_path))
        elif self.config.storage_type == "msgpack":
            self.storage = Storage(MsgPackStorage(self.config.storage_path))
        elif self.config.storage_type == "memory":
            self.storage = Storage(MemoryStorage())
        else:
            raise ValueError(
                f"Unknown storage_type {self.config.storage_type!r}; "
                "expected 'json', 'sqlite', 'msgpack', or 'memory'",
            )

        self.admins = set(self.config.admins or [])
        self.permissions = PermissionManager(
            storage=self.storage,
            enabled=self.config.permissions_enabled,
            admins=self.admins,
        )

        self.events = EventManager(self.storage)

        self._register_builtin_events()

        self.middleware = MiddlewareManager()

        self.cogs_dir = os.path.join(self.config_path, self.config.cogs_dir)
        if self.config.cogs_enabled:
            os.makedirs(self.cogs_dir, exist_ok=True)
            init_file = os.path.join(self.cogs_dir, "__init__.py")
            if not os.path.exists(init_file):
                open(init_file, "w", encoding="utf-8").close()

        self.transport = Transport(self, self.storage)
        self.delivery = DeliveryTracker(storage=self.storage)
        self.delivery.load_persisted()
        self.spam_protection = SpamProtection(
            storage=self.storage,
            bot=self,
            rate_limit=self.config.rate_limit,
            cooldown=self.config.cooldown,
            max_warnings=self.config.max_warnings,
            warning_timeout=self.config.warning_timeout,
        )

        self.delivery_attempts = {}
        self._load_delivery_attempts()
        self.conversations = ConversationManager(self)

        self.landlock_active = False
        if not self.config.test_mode:
            storage_dir = os.path.abspath(
                os.path.expanduser(self.config.storage_path),
            )
            self.landlock_active = apply_landlock_sandbox(
                storage_dir=storage_dir,
                reticulum_config_dir=self.reticulum_config_dir,
                config_dir=self.config_path,
                cogs_dir=self.cogs_dir,
                config_enabled=self.config.landlock_enabled,
            )

        identity_file = os.path.join(self.config_path, "identity")

        self._owns_reticulum = False
        if not self.config.test_mode:
            # Initialize Reticulum (will raise exception if already running)
            if RNS.Reticulum.get_instance() is None:
                try:
                    RNS.Reticulum(
                        configdir=self.reticulum_config_dir,
                        loglevel=self.config.loglevel,
                    )
                    self._owns_reticulum = True
                except OSError:
                    if RNS.Reticulum.get_instance() is None:
                        raise

            if not os.path.isfile(identity_file):
                RNS.log("No Primary Identity file found, creating new...", RNS.LOG_INFO)
                identity = RNS.Identity(True)
                identity.to_file(identity_file)
            identity = RNS.Identity.from_file(identity_file)
            if identity is None:
                raise RuntimeError(f"Failed to load identity from {identity_file}")
            self.identity = identity
            RNS.log("Loaded identity from file", RNS.LOG_INFO)

            self.router = LXMRouter(
                identity=self.identity,
                storagepath=self.config_path,
                autopeer=self.config.autopeer_propagation,
                autopeer_maxdepth=self.config.autopeer_maxdepth,
                enforce_stamps=self.config.require_stamps,
            )
            self.local = self.router.register_delivery_identity(
                self.identity,
                display_name=self.config.name,
                stamp_cost=self.config.stamp_cost,
            )
            if self.local is None:
                raise RuntimeError("Failed to register delivery identity")
            self._sync_delivery_display_name()
            self.router.register_delivery_callback(self._message_received)
            self.local.set_link_established_callback(
                self._delivery_link_established,
            )

            if self.config.pending_sends_enabled:
                RNS.Transport.register_announce_handler(
                    PendingSendAnnounceHandler(self),
                )

        self._configure_propagation()

        if self.local:
            RNS.log(
                f"LXMF Router ready to receive on: {RNS.prettyhexrep(self.local.hash)}",
                RNS.LOG_INFO,
            )
        else:
            # Test mode - create mock components
            if os.path.isfile(identity_file):
                identity = RNS.Identity.from_file(identity_file)
                if identity is None:
                    raise RuntimeError(
                        f"Failed to load identity from {identity_file}",
                    )
                self.identity = identity
            else:
                self.identity = RNS.Identity()  # Create a basic identity for testing
                if self.config.config_path:
                    self.identity.to_file(identity_file)

            self.router = None
            self.local = None

        self.announce_enabled = self.config.announce_enabled
        self.announce_time = self.config.announce

        if self.announce_enabled and not self.config.test_mode:
            if self.announce_time > 0:
                # Schedule the announce task. announce_now throttles on the
                # announce interval file, so a sub-minute cron step still
                # respects announce_time.
                minutes = max(1, self.announce_time // 60)
                self.scheduler.add_task(
                    "announce_task",
                    self.announce_now,
                    f"*/{minutes} * * * *",  # Convert seconds to minutes for cron
                )
            if self.config.announce_immediately:
                self.announce_now(force=True)
                RNS.log("Initial announce sent", RNS.LOG_INFO)

        self.hot_reloading = self.config.hot_reloading
        self.command_prefix = self.config.command_prefix

        self.help_system = HelpSystem(self)
        register_admin_commands(self)

        self.nlp = IntentClassifier(threshold=self.config.nlp_threshold)
        self.intents = {}  # {intent_name: callback}

        self.link_handlers = []
        self.links = {}  # {dest_hash: Link}

        self.rrc = None
        self.rrc_handlers = []
        self._init_rrc()

        self.signature_manager = SignatureManager(
            self,
            verification_enabled=self.config.signature_verification_enabled,
            require_signatures=self.config.require_message_signatures,
            request_unknown_identities=self.config.request_unknown_identities,
        )

        self._load_persisted_queue()

        if self.config.cogs_enabled:
            load_cogs_from_directory(self)

    def _configure_logging(self) -> None:
        level = self.config.log_level
        if level is None:
            return
        if isinstance(level, str):
            level = logging.getLevelNamesMapping().get(level.upper())
            if level is None:
                raise ValueError(
                    f"Unknown log_level {self.config.log_level!r}",
                )
        logger = logging.getLogger("lxmfy")
        if not logger.handlers:
            handler = logging.StreamHandler()
            handler.setFormatter(
                logging.Formatter(
                    "%(asctime)s [%(levelname)s] %(name)s: %(message)s",
                    "%H:%M:%S",
                ),
            )
            logger.addHandler(handler)
        logger.setLevel(level)

    def get_debugger(self):
        """Return a Debugger bound to this bot."""
        from .debugger import Debugger

        return Debugger(bot=self)

    def diagnose_destination(
        self,
        destination: str,
        *,
        request_path: bool = False,
        wait: float = 0.0,
    ) -> dict:
        """Probe path and identity for a destination hash.

        Args:
            destination: Hex destination hash.
            request_path: Whether to request a path if missing.
            wait: Seconds to wait for a path after requesting.

        Returns:
            Dict describing identity/path status and hints.

        """
        return (
            self.get_debugger()
            .probe_destination(
                destination,
                request_path=request_path,
                wait=wait,
            )
            .to_dict()
        )

    def diagnose_connectivity(
        self,
        destination: str | None = None,
        *,
        request_path: bool = False,
        wait: float = 0.0,
    ) -> dict:
        """Run a connectivity doctor report for this bot.

        Args:
            destination: Optional peer hash to include in send diagnosis.
            request_path: Request a path when probing destination.
            wait: Seconds to wait for path discovery.

        Returns:
            Structured doctor report dict.

        """
        return (
            self.get_debugger()
            .run_doctor(
                destination,
                request_path=request_path,
                wait=wait,
            )
            .to_dict()
        )

    def on_delivery_event(self, callback=None):
        """Subscribe to the outbound delivery event stream.

        Usable as bot.on_delivery_event(fn) or as a decorator
        @bot.on_delivery_event(). Subscribers receive event dicts with
        ts, stage (queued, deferred, dispatched, delivered, failed,
        cancelled, dropped), and destination/message_id/hash/reason
        fields when available.
        """
        if callback is None:
            return self.delivery.subscribe
        return self.delivery.subscribe(callback)

    def get_landlock_status(self) -> dict[str, bool]:
        """Return Landlock LSM sandbox availability and activation state."""
        return landlock_status_dict(
            active=self.landlock_active,
            config_enabled=self.config.landlock_enabled,
        )

    def run(self, delay=10):
        """Run the bot"""
        self.scheduler.start()  # Start the scheduler
        try:
            while True:
                # Process outgoing queue with a timeout to prevent hanging
                while not self.queue.empty():
                    try:
                        lxm = self.queue.get(block=False)
                    except Exception:
                        break
                    try:
                        if self.router:
                            self.router.handle_outbound(lxm)
                            self.delivery.record(
                                "dispatched",
                                destination=_delivery_dest_hex(lxm),
                                message_id=_delivery_id_hex(lxm),
                                hash=_delivery_hex(lxm),
                            )
                        self._persist_queue()
                    except Exception as e:
                        self.logger.exception(
                            "Outbound send failed, requeueing",
                        )
                        self.delivery.record(
                            "failed",
                            destination=_delivery_dest_hex(lxm),
                            reason=str(e)[:120],
                        )
                        if not self._enqueue_outbound(lxm):
                            self.logger.exception(
                                "Failed to requeue after outbound error"
                            )
                        break

                retry_interval = max(
                    0,
                    int(getattr(self.config, "pending_sends_retry", 300) or 0),
                )
                if retry_interval and (
                    time.time() - self._last_pending_flush >= retry_interval
                ):
                    self._last_pending_flush = time.time()
                    self._flush_pending_sends()

                time.sleep(delay)

        except KeyboardInterrupt:
            pass
        finally:
            self.cleanup()

    def request_page(
        self,
        destination_hash: str,
        page_path: str,
        field_data: dict | None = None,
    ) -> dict:
        """Request a page from a destination.

        Args:
            destination_hash: The destination hash.
            page_path: The path to the page.
            field_data: Optional field data to send with the request.

        Returns:
            The response from the destination.

        """
        try:
            dest_hash_bytes = bytes.fromhex(destination_hash)
            return self.transport.request_page(dest_hash_bytes, page_path, field_data)
        except Exception:
            self.logger.exception("Error requesting page")
            raise

    def cleanup(self):
        """Clean up resources."""
        RNS.log("Cleaning up LXMFBot...", RNS.LOG_DEBUG)
        try:
            self._persist_queue()
        except Exception:
            self.logger.exception("Failed to persist queue during cleanup")
        self.conversations.cancel_all()
        if hasattr(self, "rrc") and self.rrc:
            try:
                self.rrc.shutdown()
            except Exception:
                self.logger.exception("RRC shutdown failed")
        self.transport.cleanup()
        self.thread_pool.shutdown(wait=False)
        self.scheduler.stop()
        if hasattr(self, "router") and self.router:
            try:
                self.router.exit_handler()
            except Exception as e:
                self.logger.debug("Router exit handler failed: %s", e)

        # Ensure Reticulum exits cleanly, but only if this bot started it.
        # A shared instance may still be serving other bots or tests.
        if not self.config.test_mode and self._owns_reticulum:
            try:
                RNS.Reticulum.exit_handler()
            except Exception as e:
                self.logger.debug("Reticulum exit handler failed: %s", e)
        RNS.log("LXMFBot cleanup complete", RNS.LOG_DEBUG)

    def validate(self) -> str:
        """Run validation checks and return formatted results."""
        results = validate_bot(self)
        return format_validation_results(results)