Add owner-only /selftest route-head probe with quota accounting

README.md

@@ -115,12 +115,13 @@ All three endpoints below require `Authorization: Bearer <ADMIN_PANEL_KEY>` (`/a
 | `/tts [text]` | anyone | Reads text out loud. |
 | `/lang` | anyone in DMs; group admins/owner in groups | Bot language menu (system messages; AI answers always follow your language). |
 
-Two more owner-only commands stay out of the menu on purpose, since they surface internal details that shouldn't be visible in a group chat:
+Three more owner-only commands stay out of the menu on purpose, since they surface internal details that shouldn't be visible in a group chat:
 
 | Command | Access | Purpose |
 |---|---|---|
 | `/stats` | bot owner, DMs only | Active chat count, process uptime, per-model quota usage. |
 | `/logs` | bot owner, DMs only | Sends `bot.log` with secrets redacted. |
+| `/selftest [gemini]` | bot owner, DMs only, max once a minute | Live probe of route heads (Groq, light OpenRouter, Telegram proxy, TikWM; Gemini only with the `gemini` argument). Successful probes count against quota. |
 
 In groups the bot keeps recent messages (up to 100) as context and sends them to the model providers together with the question; `/reset` (group admins/owner) clears them.
 

bot.py

@@ -665,6 +665,7 @@ from lumen_admin import (
     get_webhook_url,
     webhook_handler,
     network_diagnostics,
+    probe_url,
     export_state,
 )
 
@@ -731,6 +732,7 @@ __all__ = [
     "get_webhook_url",
     "webhook_handler",
     "network_diagnostics",
+    "probe_url",
     "export_state",
     # Имена из lumen_streaming.py код bot.py сам не читает — они нужны как `bot.X`
     # существующим тестам и `_run_route` ниже.
@@ -789,6 +791,7 @@ __all__ = [
     "ask_openrouter_text",
     "ask_openrouter_multimodal",
     "ask_groq_text",
+    "selftest_llm_head",
     "_transcribe_audio",
     "_gemini_history_contents",
     "_build_gemma_identity_contents",
@@ -1040,6 +1043,7 @@ from lumen_routes import (
     ask_openrouter_text,
     ask_openrouter_multimodal,
     ask_groq_text,
+    selftest_llm_head,
     _transcribe_audio,
     _gemini_history_contents,
     _build_gemma_identity_contents,
@@ -1190,6 +1194,7 @@ from lumen_commands import (
     handle_lang_callback,
     cmd_logs,
     cmd_stats,
+    cmd_selftest,
     _build_version,
     _webhook_info_text,
     _process_memory_text,
@@ -1205,6 +1210,7 @@ dp.message.register(cmd_lang, Command("lang"))
 dp.callback_query.register(handle_lang_callback, F.data.startswith("lang:"))
 dp.message.register(cmd_logs, Command("logs"))
 dp.message.register(cmd_stats, Command("stats"))
+dp.message.register(cmd_selftest, Command("selftest"))
 dp.callback_query.register(handle_pick_callback, F.data.startswith("pick:"))
 
 

lumen_admin.py

@@ -121,6 +121,19 @@ async def webhook_handler(request: Request) -> Any:
         log.warning("[webhook] Failed to process incoming update: %s", exc)
     return {"ok": True}
 
+async def probe_url(session: Any, url: str, *, timeout_sec: float = 6.0, redact: str = "") -> dict[str, Any]:
+    """Один GET-зонд: та же логика и тот же формат результата для /diag и /selftest."""
+    start = time.time()
+    try:
+        async with session.get(url, timeout=aiohttp.ClientTimeout(total=timeout_sec)) as resp:
+            return {"status": resp.status, "elapsed_sec": round(time.time() - start, 2), "ok": True}
+    except Exception as exc:
+        elapsed = round(time.time() - start, 2)
+        exc_str = str(exc) or repr(exc) or type(exc).__name__
+        if redact:
+            exc_str = exc_str.replace(redact, "<TOKEN>")
+        return {"error": exc_str, "elapsed_sec": elapsed, "ok": False}
+
 @app.get("/diag")
 async def network_diagnostics(request: Request) -> dict[str, Any]:
     """Проверяет исходящую сетевую доступность различных хостов из контейнера.
@@ -157,16 +170,7 @@ async def network_diagnostics(request: Request) -> dict[str, Any]:
     diag_budget = float(os.getenv("DIAG_TOTAL_BUDGET_SEC", "25"))
 
     async def _probe(name: str, url: str) -> tuple[str, dict[str, Any]]:
-        start = time.time()
-        try:
-            async with session.get(url, timeout=aiohttp.ClientTimeout(total=6.0)) as resp:
-                return name, {"status": resp.status, "elapsed_sec": round(time.time() - start, 2), "ok": True}
-        except Exception as exc:
-            elapsed = round(time.time() - start, 2)
-            exc_str = str(exc) or repr(exc) or type(exc).__name__
-            if bot_token:
-                exc_str = exc_str.replace(bot_token, "<TOKEN>")
-            return name, {"error": exc_str, "elapsed_sec": elapsed, "ok": False}
+        return name, await probe_url(session, url, timeout_sec=6.0, redact=bot_token or "")
 
     tasks = {asyncio.create_task(_probe(name, url)): name for name, url in targets.items()}
     done, pending = await asyncio.wait(tasks, timeout=diag_budget)

lumen_commands.py

@@ -648,6 +648,70 @@ async def cmd_stats(message: Message) -> None:
     await bot._tg_call(message.reply, text, parse_mode=ParseMode.HTML)
 
 
+_SELFTEST_LAST_RUN_MONOTONIC = 0.0
+SELFTEST_COOLDOWN_SEC = 60.0
+
+def _selftest_row(label: str, ok: bool, detail: str, elapsed: float) -> str:
+    """Одна строка итога без ID моделей (в группы команда не ходит вовсе)."""
+    base = f"{label}: {'OK' if ok else 'FAIL'} ({elapsed:.1f}s)"
+    return base if ok or not detail else f"{base} — {detail}"
+
+async def cmd_selftest(message: Message) -> None:
+    """Живая проверка голов маршрутов: по крошечной пробе в Groq, лёгкий OpenRouter
+    и сеть (прокси Telegram, TikWM) логикой /diag. Gemini — только по явному
+    аргументу, квота самая дефицитная. Только владелец, только личка, не чаще
+    раза в минуту. Успешные пробы честно пишутся в квоту."""
+    import bot
+    global _SELFTEST_LAST_RUN_MONOTONIC
+    if not bot._is_owner(message.from_user.id if message.from_user else None):
+        await bot._tg_call(message.reply, bot._t(message.chat.id, "stats_deny"))
+        return
+    if message.chat.type != ChatType.PRIVATE:
+        await bot._tg_call(message.reply, bot._t(message.chat.id, "selftest_group_only"))
+        return
+    now = time.monotonic()
+    wait = SELFTEST_COOLDOWN_SEC - (now - _SELFTEST_LAST_RUN_MONOTONIC)
+    if wait > 0:
+        await bot._tg_call(message.reply, bot._t(message.chat.id, "selftest_cooldown", sec=int(wait) + 1))
+        return
+    # Метку ставим до проб: повторный вызов во время долгой проверки тоже ждёт.
+    _SELFTEST_LAST_RUN_MONOTONIC = now
+    args = (getattr(message, "text", "") or "").split()
+    want_gemini = len(args) > 1 and args[1].split("@")[0].lower() == "gemini"
+    rows: list[str] = []
+    summary: list[str] = []
+    for provider, label in (("groq", "Groq"), ("openrouter", "OpenRouter")):
+        ok, detail, elapsed = await bot.selftest_llm_head(provider, chat_id=message.chat.id)
+        rows.append(_selftest_row(label, ok, detail, elapsed))
+        summary.append(f"{provider}={'ok' if ok else 'FAIL'}/{elapsed:.1f}s")
+    if want_gemini:
+        ok, detail, elapsed = await bot.selftest_llm_head("gemini", chat_id=message.chat.id)
+        rows.append(_selftest_row("Gemini", ok, detail, elapsed))
+        summary.append(f"gemini={'ok' if ok else 'FAIL'}/{elapsed:.1f}s")
+    try:
+        session = await bot._get_http_session()
+        token = bot.BOT_TOKEN or ""
+        proxy_url = bot.TELEGRAM_API_BASE_URL + "/bot" + (token[:6] if token else "x") + "/getMe"
+        net = await asyncio.gather(
+            bot.probe_url(session, proxy_url, timeout_sec=6.0, redact=token),
+            bot.probe_url(session, "https://www.tikwm.com", timeout_sec=6.0),
+        )
+    except Exception as exc:
+        net = [
+            {"ok": False, "elapsed_sec": 0.0, "error": f"{exc.__class__.__name__}"},
+            {"ok": False, "elapsed_sec": 0.0, "error": f"{exc.__class__.__name__}"},
+        ]
+    for label, res in (("TG-proxy", net[0]), ("TikWM", net[1])):
+        detail = "" if res.get("ok") else str(res.get("error") or res.get("status") or "error")[:120]
+        rows.append(_selftest_row(label, bool(res.get("ok")), detail, float(res.get("elapsed_sec") or 0.0)))
+        summary.append(f"{label}={'ok' if res.get('ok') else 'FAIL'}")
+    text = bot._t(message.chat.id, "selftest_header") + "\n" + "\n".join(rows)
+    if not want_gemini:
+        text += "\n" + bot._t(message.chat.id, "selftest_gemini_skipped")
+    await bot._tg_call(message.reply, text)
+    log.info("[selftest] chat=%s gemini_arg=%s %s", message.chat.id, want_gemini, " ".join(summary))
+
+
 _BUILD_VERSION_CACHED: str | None = None
 
 

lumen_lang.py

@@ -537,6 +537,38 @@ STRINGS: dict[str, dict[str, str]] = {
         "ru": "Эта команда показывает техническую статистику — доступна только в личных сообщениях с ботом, не в группах.",
         "uk": "Ця команда показує технічну статистику — доступна лише в особистих повідомленнях з ботом, не в групах.",
     },
+    "selftest_group_only": {
+        "be": "Гэтая каманда правярае тэхнічны стан бота — даступна толькі ў асабістых паведамленнях з ботам, не ў групах.",
+        "en": "This command checks the bot's health — available only in direct messages with the bot, not in groups.",
+        "es": "Este comando comprueba el estado técnico del bot — disponible solo en mensajes directos con el bot, no en grupos.",
+        "kk": "Бұл команда боттың техникалық күйін тексереді — тек ботпен жеке хабарламаларда қолжетімді, топтарда емес.",
+        "ru": "Эта команда проверяет техническое состояние бота — доступна только в личных сообщениях с ботом, не в группах.",
+        "uk": "Ця команда перевіряє технічний стан бота — доступна лише в особистих повідомленнях з ботом, не в групах.",
+    },
+    "selftest_cooldown": {
+        "be": "Пачакай {sec} с перад наступным запускам.",
+        "en": "Wait {sec}s before running it again.",
+        "es": "Espera {sec} s antes de volver a ejecutarlo.",
+        "kk": "Келесі іске қосу алдында {sec} с күт.",
+        "ru": "Подожди {sec} с перед следующим запуском.",
+        "uk": "Зачекай {sec} с перед наступним запуском.",
+    },
+    "selftest_header": {
+        "be": "Самаправерка:",
+        "en": "Self-test:",
+        "es": "Autoprueba:",
+        "kk": "Өзін-өзі тексеру:",
+        "ru": "Самопроверка:",
+        "uk": "Самоперевірка:",
+    },
+    "selftest_gemini_skipped": {
+        "be": "(Gemini прапушчаны — дапішы gemini, квота дэфіцытная)",
+        "en": "(Gemini skipped — add the gemini argument, its quota is scarce)",
+        "es": "(Gemini omitido — añade el argumento gemini, su cuota es escasa)",
+        "kk": "(Gemini өткізілді — gemini аргументін қос, квота тапшы)",
+        "ru": "(Gemini пропущен — допиши gemini, квота дефицитная)",
+        "uk": "(Gemini пропущено — допиши gemini, квота дефіцитна)",
+    },
     "stats_no_data": {
         "be": "  няма даных",
         "en": "  no data",
@@ -801,6 +833,10 @@ PACK_PT: dict[str, str] = {
     "logs_send_error": "Erro ao enviar os registros: {error}",
     "stats_deny": "Sem acesso a este comando.",
     "stats_group_only": "Este comando mostra estatísticas técnicas — disponível apenas em mensagens diretas com o bot, não em grupos.",
+    "selftest_group_only": "Este comando verifica o estado técnico do bot — disponível apenas em mensagens diretas com o bot, não em grupos.",
+    "selftest_cooldown": "Espera {sec}s antes da próxima execução.",
+    "selftest_header": "Autoteste:",
+    "selftest_gemini_skipped": "(Gemini ignorado — adiciona o argumento gemini, a quota dele é escassa)",
     "stats_no_data": "  sem dados",
     "stats_limit_used": " (limite esgotado)",
     "tiktok_sound_recognized": "Link de som do TikTok reconhecido.",
@@ -890,6 +926,10 @@ PACK_AR: dict[str, str] = {
     "logs_send_error": "خطأ أثناء إرسال السجلات: {error}",
     "stats_deny": "لا صلاحية لهذا الأمر.",
     "stats_group_only": "يعرض هذا الأمر الإحصاءات التقنية — متاح فقط في الرسائل الخاصة مع البوت وليس في المجموعات.",
+    "selftest_group_only": "هذا الأمر يفحص الحالة التقنية للبوت — متاح فقط في الرسائل الخاصة مع البوت وليس في المجموعات.",
+    "selftest_cooldown": "انتظر {sec} ث قبل التشغيل التالي.",
+    "selftest_header": "فحص ذاتي:",
+    "selftest_gemini_skipped": "(تم تخطي Gemini — أضف الوسيط gemini، حصته شحيحة)",
     "stats_no_data": "  لا بيانات",
     "stats_limit_used": " (الحد مستنفد)",
     "tiktok_sound_recognized": "تم التعرف على رابط صوت TikTok.",
@@ -979,6 +1019,10 @@ PACK_TR: dict[str, str] = {
     "logs_send_error": "Loglar gönderilirken hata: {error}",
     "stats_deny": "Bu komuta erişimin yok.",
     "stats_group_only": "Bu komut teknik istatistikleri gösterir — yalnızca botla direkt mesajlarda kullanılabilir, gruplarda değil.",
+    "selftest_group_only": "Bu komut botun teknik durumunu kontrol eder — yalnızca botla direkt mesajlarda kullanılabilir, gruplarda değil.",
+    "selftest_cooldown": "Sonraki çalıştırmadan önce {sec} sn bekle.",
+    "selftest_header": "Öz test:",
+    "selftest_gemini_skipped": "(Gemini atlandı — gemini argümanını ekle, kotası kıt)",
     "stats_no_data": "  veri yok",
     "stats_limit_used": " (limit doldu)",
     "tiktok_sound_recognized": "TikTok ses bağlantısı tanındı.",
@@ -1068,6 +1112,10 @@ PACK_DE: dict[str, str] = {
     "logs_send_error": "Fehler beim Senden der Logs: {error}",
     "stats_deny": "Kein Zugriff auf diesen Befehl.",
     "stats_group_only": "Dieser Befehl zeigt technische Statistiken — nur in Direktnachrichten mit dem Bot verfügbar, nicht in Gruppen.",
+    "selftest_group_only": "Dieser Befehl prüft den technischen Zustand des Bots — nur in Direktnachrichten mit dem Bot verfügbar, nicht in Gruppen.",
+    "selftest_cooldown": "Warte {sec} s vor dem nächsten Start.",
+    "selftest_header": "Selbsttest:",
+    "selftest_gemini_skipped": "(Gemini übersprungen — füge das Argument gemini hinzu, sein Kontingent ist knapp)",
     "stats_no_data": "  keine Daten",
     "stats_limit_used": " (Limit erschöpft)",
     "tiktok_sound_recognized": "TikTok-Soundlink erkannt.",
@@ -1157,6 +1205,10 @@ PACK_FR: dict[str, str] = {
     "logs_send_error": "Erreur d'envoi des logs : {error}",
     "stats_deny": "Pas d'accès à cette commande.",
     "stats_group_only": "Cette commande affiche des statistiques techniques — disponible uniquement en messages privés avec le bot, pas en groupes.",
+    "selftest_group_only": "Cette commande vérifie l'état technique du bot — disponible uniquement en messages privés avec le bot, pas en groupes.",
+    "selftest_cooldown": "Attends {sec} s avant le prochain lancement.",
+    "selftest_header": "Auto-test :",
+    "selftest_gemini_skipped": "(Gemini ignoré — ajoute l'argument gemini, son quota est limité)",
     "stats_no_data": "  pas de données",
     "stats_limit_used": " (limite épuisée)",
     "tiktok_sound_recognized": "Lien de son TikTok reconnu.",
@@ -1246,6 +1298,10 @@ PACK_IT: dict[str, str] = {
     "logs_send_error": "Errore di invio dei log: {error}",
     "stats_deny": "Nessun accesso a questo comando.",
     "stats_group_only": "Questo comando mostra statistiche tecniche — disponibile solo nei messaggi diretti con il bot, non nei gruppi.",
+    "selftest_group_only": "Questo comando verifica lo stato tecnico del bot — disponibile solo nei messaggi diretti con il bot, non nei gruppi.",
+    "selftest_cooldown": "Attendi {sec} s prima del prossimo avvio.",
+    "selftest_header": "Autotest:",
+    "selftest_gemini_skipped": "(Gemini saltato — aggiungi l'argomento gemini, la sua quota è scarsa)",
     "stats_no_data": "  nessun dato",
     "stats_limit_used": " (limite esaurito)",
     "tiktok_sound_recognized": "Link audio di TikTok riconosciuto.",
@@ -1335,6 +1391,10 @@ PACK_HI: dict[str, str] = {
     "logs_send_error": "लॉग भेजने में त्रुटि: {error}",
     "stats_deny": "इस कमांड तक पहुंच नहीं है।",
     "stats_group_only": "यह कमांड तकनीकी आंकड़े दिखाती है — केवल बॉट के साथ डायरेक्ट संदेशों में उपलब्ध है, ग्रुप में नहीं।",
+    "selftest_group_only": "यह कमांड बॉट की तकनीकी स्थिति जांचता है — केवल बॉट के साथ सीधे संदेशों में, समूहों में नहीं।",
+    "selftest_cooldown": "अगली बार चलाने से पहले {sec} सेकंड रुको।",
+    "selftest_header": "स्व-जांच:",
+    "selftest_gemini_skipped": "(Gemini छोड़ा गया — gemini आर्गुमेंट जोड़ो, उसका कोटा सीमित है)",
     "stats_no_data": "  कोई डेटा नहीं",
     "stats_limit_used": " (सीमा समाप्त)",
     "tiktok_sound_recognized": "TikTok ध्वनि लिंक पहचानी गई।",
@@ -1424,6 +1484,10 @@ PACK_ID: dict[str, str] = {
     "logs_send_error": "Gagal mengirim log: {error}",
     "stats_deny": "Tidak ada akses ke perintah ini.",
     "stats_group_only": "Perintah ini menampilkan statistik teknis — hanya tersedia di pesan langsung dengan bot, bukan di grup.",
+    "selftest_group_only": "Perintah ini memeriksa kondisi teknis bot — hanya tersedia di pesan langsung dengan bot, bukan di grup.",
+    "selftest_cooldown": "Tunggu {sec} dtk sebelum menjalankan lagi.",
+    "selftest_header": "Uji mandiri:",
+    "selftest_gemini_skipped": "(Gemini dilewati — tambahkan argumen gemini, kuotanya terbatas)",
     "stats_no_data": "  tidak ada data",
     "stats_limit_used": " (batas habis)",
     "tiktok_sound_recognized": "Tautan suara TikTok dikenali.",
@@ -1513,6 +1577,10 @@ PACK_MS: dict[str, str] = {
     "logs_send_error": "Gagal menghantar log: {error}",
     "stats_deny": "Tiada akses ke perintah ini.",
     "stats_group_only": "Perintah ini memaparkan statistik teknikal — hanya tersedia dalam mesej langsung dengan bot, bukan dalam kumpulan.",
+    "selftest_group_only": "Perintah ini memeriksa keadaan teknikal bot — hanya tersedia dalam mesej langsung dengan bot, bukan dalam kumpulan.",
+    "selftest_cooldown": "Tunggu {sec}s sebelum larian seterusnya.",
+    "selftest_header": "Ujian kendiri:",
+    "selftest_gemini_skipped": "(Gemini dilangkau — tambah argumen gemini, kuotanya terhad)",
     "stats_no_data": "  tiada data",
     "stats_limit_used": " (had habis)",
     "tiktok_sound_recognized": "Pautan bunyi TikTok dikenal pasti.",
@@ -1602,6 +1670,10 @@ PACK_VI: dict[str, str] = {
     "logs_send_error": "Lỗi khi gửi log: {error}",
     "stats_deny": "Không có quyền dùng lệnh này.",
     "stats_group_only": "Lệnh này hiện số liệu kỹ thuật — chỉ dùng được trong tin nhắn riêng với bot, không trong nhóm.",
+    "selftest_group_only": "Lệnh này kiểm tra tình trạng kỹ thuật của bot — chỉ dùng được trong tin nhắn riêng với bot, không dùng trong nhóm.",
+    "selftest_cooldown": "Đợi {sec} giây trước lần chạy tiếp theo.",
+    "selftest_header": "Tự kiểm tra:",
+    "selftest_gemini_skipped": "(Bỏ qua Gemini — thêm đối số gemini, hạn ngạch của nó khan hiếm)",
     "stats_no_data": "  không có dữ liệu",
     "stats_limit_used": " (hết hạn mức)",
     "tiktok_sound_recognized": "Đã nhận ra liên kết âm thanh TikTok.",
@@ -1691,6 +1763,10 @@ PACK_TH: dict[str, str] = {
     "logs_send_error": "เกิดข้อผิดพลาดขณะส่งล็อก: {error}",
     "stats_deny": "ไม่มีสิทธิ์ใช้คำสั่งนี้",
     "stats_group_only": "คำสั่งนี้แสดงสถิติทางเทคนิค — ใช้ได้เฉพาะในแชทส่วนตัวกับบอท ไม่ใช่ในกลุ่ม",
+    "selftest_group_only": "คำสั่งนี้ตรวจสอบสถานะทางเทคนิคของบอท — ใช้ได้เฉพาะในข้อความส่วนตัวกับบอท ไม่ใช่ในกลุ่ม",
+    "selftest_cooldown": "รอ {sec} วินาทีก่อนรันครั้งถัดไป",
+    "selftest_header": "ทดสอบตัวเอง:",
+    "selftest_gemini_skipped": "(ข้าม Gemini — เติมอาร์กิวเมนต์ gemini โควต้าของมันมีจำกัด)",
     "stats_no_data": "  ไม่มีข้อมูล",
     "stats_limit_used": " (โควต้าหมด)",
     "tiktok_sound_recognized": "รู้จักลิงก์เสียง TikTok แล้ว",
@@ -1780,6 +1856,10 @@ PACK_FA: dict[str, str] = {
     "logs_send_error": "خطا در ارسال لاگ‌ها: {error}",
     "stats_deny": "به این دستور دسترسی نداری.",
     "stats_group_only": "این دستور آمار فنی را نشان می‌دهد — فقط در پیام‌های خصوصی با ربات در دسترس است، نه در گروه‌ها.",
+    "selftest_group_only": "این دستور وضعیت فنی ربات را بررسی می‌کند — فقط در پیام‌های خصوصی با ربات، نه در گروه‌ها.",
+    "selftest_cooldown": "{sec} ثانیه تا اجرای بعدی صبر کن.",
+    "selftest_header": "خودآزمایی:",
+    "selftest_gemini_skipped": "(Gemini رد شد — آرگومان gemini را اضافه کن، سهمیه‌اش کم است)",
     "stats_no_data": "  داده‌ای نیست",
     "stats_limit_used": " (سقف تمام شد)",
     "tiktok_sound_recognized": "لینک صدای TikTok شناخته شد.",
@@ -1869,6 +1949,10 @@ PACK_UR: dict[str, str] = {
     "logs_send_error": "لاگز بھیجنے میں خرابی: {error}",
     "stats_deny": "اس کمانڈ تک رسائی نہیں ہے۔",
     "stats_group_only": "یہ کمانڈ تکنیکی اعداد دکھاتی ہے — صرف بوٹ کے ساتھ ڈائریکٹ پیغامات میں دستیاب ہے، گروپ میں نہیں۔",
+    "selftest_group_only": "یہ کمانڈ بوٹ کی تکنیکی حالت چیک کرتی ہے — صرف بوٹ کے ساتھ براہ راست پیغامات میں، گروپس میں نہیں۔",
+    "selftest_cooldown": "اگلی بار چلانے سے پہلے {sec} سیکنڈ انتظار کرو۔",
+    "selftest_header": "خود جانچ:",
+    "selftest_gemini_skipped": "(Gemini چھوڑ دیا گیا — gemini دلیل شامل کرو، اس کا کوٹہ محدود ہے)",
     "stats_no_data": "  کوئی ڈیٹا نہیں",
     "stats_limit_used": " (حد ختم)",
     "tiktok_sound_recognized": "TikTok آواز کا لنک پہچانا گیا۔",
@@ -1958,6 +2042,10 @@ PACK_UZ: dict[str, str] = {
     "logs_send_error": "Loglarni yuborishda xato: {error}",
     "stats_deny": "Bu buyruqqa kirish yo'q.",
     "stats_group_only": "Bu buyruq texnik statistikani ko'rsatadi — faqat bot bilan shaxsiy xabarlarda ochiq, guruhlarda emas.",
+    "selftest_group_only": "Bu buyruq botning texnik holatini tekshiradi — faqat bot bilan shaxsiy xabarlarda, guruhlarda emas.",
+    "selftest_cooldown": "Keyingi ishga tushirishdan oldin {sec} s kuting.",
+    "selftest_header": "O‘z-o‘zini tekshirish:",
+    "selftest_gemini_skipped": "(Gemini o‘tkazib yuborildi — gemini argumentini qo‘shing, kvotasi tanqis)",
     "stats_no_data": "  ma'lumot yo'q",
     "stats_limit_used": " (limit tugadi)",
     "tiktok_sound_recognized": "TikTok tovush havolasi tanildi.",
@@ -2047,6 +2135,10 @@ PACK_BN: dict[str, str] = {
     "logs_send_error": "লগ পাঠাতে ত্রুটি: {error}",
     "stats_deny": "এই কমান্ডে প্রবেশাধিকার নেই।",
     "stats_group_only": "এই কমান্ড প্রযুক্তিগত পরিসংখ্যান দেখায় — কেবল বটের সাথে ডাইরেক্ট মেসেজে উপলব্ধ, গ্রুপে নয়।",
+    "selftest_group_only": "এই কমান্ড বটের কারিগরি অবস্থা পরীক্ষা করে — শুধু বটের সাথে ব্যক্তিগত বার্তায়, গ্রুপে নয়।",
+    "selftest_cooldown": "পরের বার চালানোর আগে {sec} সেকেন্ড অপেক্ষা করো।",
+    "selftest_header": "স্ব-পরীক্ষা:",
+    "selftest_gemini_skipped": "(Gemini বাদ দেওয়া হয়েছে — gemini আর্গুমেন্ট যোগ করো, কোটা সীমিত)",
     "stats_no_data": "  কোনো তথ্য নেই",
     "stats_limit_used": " (সীমা শেষ)",
     "tiktok_sound_recognized": "TikTok সাউন্ড লিংক চেনা গেছে।",
@@ -2136,6 +2228,10 @@ PACK_FIL: dict[str, str] = {
     "logs_send_error": "Error sa pag-send ng mga log: {error}",
     "stats_deny": "Walang access sa command na ito.",
     "stats_group_only": "Pinapakita ng command na ito ang mga technical stat — sa direct message lang sa bot available, hindi sa grupo.",
+    "selftest_group_only": "Sinusuri ng command na ito ang teknikal na kalagayan ng bot — sa direktang mensahe lang sa bot, hindi sa mga grupo.",
+    "selftest_cooldown": "Maghintay ng {sec}s bago sa susunod na pag-run.",
+    "selftest_header": "Self-test:",
+    "selftest_gemini_skipped": "(Nilaktawan ang Gemini — idagdag ang argument na gemini, kakaunti ang quota nito)",
     "stats_no_data": "  walang data",
     "stats_limit_used": " (ubos ang limit)",
     "tiktok_sound_recognized": "Nakilala ang TikTok sound link.",
@@ -2225,6 +2321,10 @@ PACK_PL: dict[str, str] = {
     "logs_send_error": "Błąd wysyłania logów: {error}",
     "stats_deny": "Brak dostępu do tej komendy.",
     "stats_group_only": "Ta komenda pokazuje statystyki techniczne — dostępna tylko w wiadomościach prywatnych z botem, nie w grupach.",
+    "selftest_group_only": "Ta komenda sprawdza stan techniczny bota — dostępna tylko w wiadomościach prywatnych z botem, nie w grupach.",
+    "selftest_cooldown": "Poczekaj {sec} s przed kolejnym uruchomieniem.",
+    "selftest_header": "Autotest:",
+    "selftest_gemini_skipped": "(Gemini pominięte — dopisz argument gemini, limit jest niski)",
     "stats_no_data": "  brak danych",
     "stats_limit_used": " (limit wyczerpany)",
     "tiktok_sound_recognized": "Rozpoznano link do dźwięku TikToka.",
@@ -2314,6 +2414,10 @@ PACK_NL: dict[str, str] = {
     "logs_send_error": "Fout bij verzenden van logs: {error}",
     "stats_deny": "Geen toegang tot dit commando.",
     "stats_group_only": "Dit commando toont technische statistieken — alleen beschikbaar in directe berichten met de bot, niet in groepen.",
+    "selftest_group_only": "Deze opdracht controleert de technische staat van de bot — alleen beschikbaar in directe berichten met de bot, niet in groepen.",
+    "selftest_cooldown": "Wacht {sec} s voor de volgende run.",
+    "selftest_header": "Zelftest:",
+    "selftest_gemini_skipped": "(Gemini overgeslagen — voeg het argument gemini toe, zijn quotum is schaars)",
     "stats_no_data": "  geen data",
     "stats_limit_used": " (limiet op)",
     "tiktok_sound_recognized": "TikTok-geluidslink herkend.",
@@ -2403,6 +2507,10 @@ PACK_ZU: dict[str, str] = {
     "logs_send_error": "Iphutha ekuthumeleni ama-log: {error}",
     "stats_deny": "Alikho ilungelo lalo myalo.",
     "stats_group_only": "Lo myalo ubonisa izibalo zobuchwepheshe — utholakala kuphela emiyalezo eqondile ne-bot, hhayi eqenjini.",
+    "selftest_group_only": "Lo myalo uhlola isimo sobuchwepheshe se-bot — utholakala kuphela emilayezweni eqondile ne-bot, hhayi emaqenjini.",
+    "selftest_cooldown": "Linda u-{sec}s ngaphambi kokusebenza okulandelayo.",
+    "selftest_header": "Ukuzihlola:",
+    "selftest_gemini_skipped": "(I-Gemini yeqiwe — engeza i-agumenti ethi gemini, i-quota yayo yincane)",
     "stats_no_data": "  ayikho idatha",
     "stats_limit_used": " (umkhawulo uphelile)",
     "tiktok_sound_recognized": "Isixhumanisi somsindo se-TikTok saziwe.",

lumen_routes.py

@@ -357,6 +357,77 @@ async def ask_groq_text(chat_id: int, user_text: str, model_chain: list[str], *,
     bot._record_quota_usage("groq", model_trial)
     return answer
 
+_SELFTEST_PING_TEXT = "Reply with exactly: ok"
+
+def _selftest_short_error(exc: BaseException) -> str:
+    """Короткий однострочник для строки /selftest: без ID моделей и простыней."""
+    text = f"{exc.__class__.__name__}: {exc}".strip()
+    return " ".join(text.split())[:120] or exc.__class__.__name__
+
+async def selftest_llm_head(provider: str, *, chat_id: Any = None) -> tuple[bool, str, float]:
+    """Одна крошечная проба в голову маршрута (groq/openrouter/gemini) для /selftest.
+
+    Успех честно пишется в квоту тем же _record_quota_usage, что у настоящих
+    ответов; неуспех квоту не трогает (как в ask_*). Историю чата не трогаем —
+    проба не должна оседать в контексте владельца. Никогда не кидает: неуспех
+    возвращается кортежем (False, причина, секунды)."""
+    import bot
+    from lumen_router_config import _GROQ_LIGHT_ORDER
+    start = time.monotonic()
+    try:
+        if provider == "groq":
+            model = _GROQ_LIGHT_ORDER[0]
+            messages = [{"role": "system", "content": ""}, {"role": "user", "content": _SELFTEST_PING_TEXT}]
+
+            async def _groq_probe_request(payload: dict, dl: float | None):
+                return await bot._groq_request("chat/completions", "POST", json_body=payload, deadline=dl)
+
+            _ans, used = await _chat_completion_chain(
+                messages, [model], model, request_fn=_groq_probe_request,
+                provider="groq", deadline=start + bot.ROUTE_MODEL_TIMEOUT_SEC,
+                log_label="groq", chat_id=chat_id,
+            )
+            bot._record_quota_usage("groq", used)
+            return True, "", round(time.monotonic() - start, 2)
+        if provider == "openrouter":
+            model = _OR_LIGHT_ORDER[0]
+            messages = [{"role": "system", "content": ""}, {"role": "user", "content": _SELFTEST_PING_TEXT}]
+
+            async def _or_probe_request(payload: dict, dl: float | None):
+                return await bot._or_request("chat/completions", "POST", json_body=payload, deadline=dl)
+
+            _ans, used = await _chat_completion_chain(
+                messages, [model], model, request_fn=_or_probe_request,
+                provider="openrouter", deadline=start + bot.ROUTE_MODEL_TIMEOUT_SEC,
+                log_label="or", chat_id=chat_id,
+            )
+            bot._record_quota_usage("openrouter", used)
+            return True, "", round(time.monotonic() - start, 2)
+        if provider == "gemini":
+            # Квота самая дефицитная — проба минимальная: голый текст без системного
+            # промпта и истории (в отличие от ask_gemini, чистит только живость).
+            model = GEMINI_DEFAULT_CHAIN[0]
+            deadline = start + bot.ROUTE_MODEL_TIMEOUT_SEC
+            contents = [types.Content(role="user", parts=[types.Part.from_text(text=_SELFTEST_PING_TEXT)])]
+            call_contents, gconfig = bot._build_gemini_call_config(model, contents)
+            resp = await asyncio.wait_for(
+                bot.client.aio.models.generate_content(model=model, contents=call_contents, config=gconfig),
+                timeout=max(1.0, deadline - time.monotonic()),
+            )
+            ans = await bot._extract_gemini_answer_text(
+                resp, model_id=model, call_contents=call_contents, gconfig=gconfig, deadline=deadline)
+            if not ans.strip():
+                raise RuntimeError(f"Model {model} returned an empty response")
+            _gpt, _gct, _gtt = _usage_from_gemini_response(resp)
+            _log_llm_usage(chat_id=chat_id, provider="gemini", model_id=model,
+                           prompt=_gpt, completion=_gct, total=_gtt)
+            _record_model_latency(_model_speed_key("gemini", model), total_sec=time.monotonic() - start)
+            bot._record_quota_usage("gemini", model)
+            return True, "", round(time.monotonic() - start, 2)
+        raise ValueError(f"unknown selftest provider: {provider}")
+    except Exception as exc:
+        return False, _selftest_short_error(exc), round(time.monotonic() - start, 2)
+
 async def _transcribe_audio(audio_bytes: bytes, mime: str, chat_id: int, deadline: float | None = None) -> str | None:
     """Голос/аудио в текст через Groq Whisper (дешевле Gemini-квоты на порядок). None при
     любой неудаче — вызывающий код молча идёт прежним путём (аудио напрямую в Gemini).

tests/test_bot_admin.py

@@ -514,3 +514,29 @@ def test_webhook_setup_succeeds_after_retries(monkeypatch):
     assert calls["hook"] == 3
     assert sleeps == [1.5, 5.0, 10.0, 3600]
 
+
+def test_probe_url_returns_ok_shape_and_redacts_secret_on_error():
+    # Общий зонд /diag и /selftest: успех — статус, неуспех — текст без секрета.
+    class _FakeResp:
+        status = 200
+
+        async def __aenter__(self):
+            return self
+
+        async def __aexit__(self, *args):
+            return False
+
+    class _FakeSession:
+        def get(self, url, **kwargs):
+            return _FakeResp()
+
+    class _BoomSession:
+        def get(self, url, **kwargs):
+            raise RuntimeError("net down, token abc123")
+
+    ok = asyncio.run(bot.probe_url(_FakeSession(), "http://x"))
+    assert ok["ok"] is True and ok["status"] == 200 and ok["elapsed_sec"] >= 0
+    err = asyncio.run(bot.probe_url(_BoomSession(), "http://x", redact="abc123"))
+    assert err["ok"] is False
+    assert "abc123" not in err["error"] and "<TOKEN>" in err["error"]
+

tests/test_bot_commands.py

@@ -1278,3 +1278,151 @@ def test_inline_tts_send_failure_raises(monkeypatch):
             asyncio.run(bot.inline_tts(incoming, "привет"))
     finally:
         bot.chat_state.pop(chat_id, None)
+
+
+def _make_selftest_message(chat_id, user_id, text="/selftest", group=False):
+    incoming = _FakeIncomingMessage(chat_id)
+    if group:
+        incoming.chat.type = bot.ChatType.GROUP
+    incoming.from_user = SimpleNamespace(id=user_id)
+    incoming.text = text
+    return incoming
+
+
+def _setup_selftest_fakes(monkeypatch, fail_heads=()):
+    # Фейковые пробы голов и сети + захват ответа. Возвращает (calls, fake_tg_call).
+    import lumen_commands
+    calls = []
+
+    async def fake_head(provider, *, chat_id=None):
+        calls.append(provider)
+        if provider in fail_heads:
+            return False, "boom-500", 0.5
+        return True, "", 0.1
+
+    async def fake_session():
+        return SimpleNamespace()
+
+    async def fake_probe(session, url, *, timeout_sec=6.0, redact=""):
+        return {"status": 200, "elapsed_sec": 0.2, "ok": True}
+
+    async def fake_tg_call(method, *args, **kwargs):
+        fake_tg_call.text = args[0] if args else kwargs.get("text", "")
+        return SimpleNamespace()
+
+    monkeypatch.setattr(bot, "selftest_llm_head", fake_head)
+    monkeypatch.setattr(bot, "_get_http_session", fake_session)
+    monkeypatch.setattr(bot, "probe_url", fake_probe)
+    monkeypatch.setattr(bot, "_tg_call", fake_tg_call)
+    monkeypatch.setattr(lumen_commands, "_SELFTEST_LAST_RUN_MONOTONIC", 0.0)
+    return calls, fake_tg_call
+
+
+def test_cmd_selftest_denies_non_owner_without_probing(monkeypatch):
+    # Чужой не должен даже запускать пробы: иначе любой сливал бы квоту и видел внутрянку.
+    monkeypatch.setattr(bot, "OWNER_ID", 111001)
+    calls, fake_tg_call = _setup_selftest_fakes(monkeypatch)
+    incoming = _make_selftest_message(999701, 222002)
+    asyncio.run(bot.cmd_selftest(incoming))
+    assert fake_tg_call.text == bot._t(999701, "stats_deny")
+    assert calls == []
+
+
+def test_cmd_selftest_group_only_for_owner(monkeypatch):
+    # ID моделей в группы не отдаём: команда в группах только отказывает, проб нет.
+    from lumen_router_config import _GROQ_LIGHT_ORDER, _OR_LIGHT_ORDER, GEMINI_DEFAULT_CHAIN
+    monkeypatch.setattr(bot, "OWNER_ID", 333001)
+    calls, fake_tg_call = _setup_selftest_fakes(monkeypatch)
+    incoming = _make_selftest_message(999702, 333001, group=True)
+    asyncio.run(bot.cmd_selftest(incoming))
+    assert fake_tg_call.text == bot._t(999702, "selftest_group_only")
+    assert calls == []
+    for mid in list(_GROQ_LIGHT_ORDER) + list(_OR_LIGHT_ORDER) + list(GEMINI_DEFAULT_CHAIN):
+        assert mid not in fake_tg_call.text
+
+
+def test_cmd_selftest_cooldown_blocks_second_run(monkeypatch):
+    # Без паузы владелец случайно выжигал бы квоту повторными тапами.
+    monkeypatch.setattr(bot, "OWNER_ID", 444001)
+    calls, fake_tg_call = _setup_selftest_fakes(monkeypatch)
+    asyncio.run(bot.cmd_selftest(_make_selftest_message(999703, 444001)))
+    assert calls == ["groq", "openrouter"]
+    asyncio.run(bot.cmd_selftest(_make_selftest_message(999703, 444001)))
+    assert calls == ["groq", "openrouter"]
+    assert fake_tg_call.text == bot._t(999703, "selftest_cooldown", sec=60)
+
+
+def test_cmd_selftest_default_skips_gemini_and_hides_model_ids(monkeypatch):
+    # Дефолт без gemini бережёт самую дефицитную квоту; ID моделей в ответе нет.
+    from lumen_router_config import _GROQ_LIGHT_ORDER, _OR_LIGHT_ORDER, GEMINI_DEFAULT_CHAIN
+    monkeypatch.setattr(bot, "OWNER_ID", 555001)
+    calls, fake_tg_call = _setup_selftest_fakes(monkeypatch)
+    asyncio.run(bot.cmd_selftest(_make_selftest_message(999704, 555001)))
+    assert calls == ["groq", "openrouter"]
+    assert bot._t(999704, "selftest_gemini_skipped") in fake_tg_call.text
+    for mid in list(_GROQ_LIGHT_ORDER) + list(_OR_LIGHT_ORDER) + list(GEMINI_DEFAULT_CHAIN):
+        assert mid not in fake_tg_call.text
+
+
+def test_cmd_selftest_with_gemini_arg_probes_gemini(monkeypatch):
+    # Явный аргумент включает пробу Gemini.
+    monkeypatch.setattr(bot, "OWNER_ID", 666001)
+    calls, fake_tg_call = _setup_selftest_fakes(monkeypatch)
+    asyncio.run(bot.cmd_selftest(_make_selftest_message(999705, 666001, "/selftest gemini")))
+    assert calls == ["groq", "openrouter", "gemini"]
+    assert bot._t(999705, "selftest_gemini_skipped") not in fake_tg_call.text
+
+
+def test_cmd_selftest_reports_failed_head(monkeypatch):
+    # Упавшая голова — строка FAIL с причиной, а не молчание или падение команды.
+    monkeypatch.setattr(bot, "OWNER_ID", 777001)
+    calls, fake_tg_call = _setup_selftest_fakes(monkeypatch, fail_heads=("groq",))
+    asyncio.run(bot.cmd_selftest(_make_selftest_message(999706, 777001)))
+    assert "Groq: FAIL" in fake_tg_call.text
+    assert "boom-500" in fake_tg_call.text
+    assert "OpenRouter: OK" in fake_tg_call.text
+
+
+def test_selftest_llm_head_openrouter_success_targets_head_and_counts_quota(monkeypatch):
+    # Честный учёт: успешная проба идёт ровно в голову лёгкого маршрута и +1 в квоту.
+    import lumen_model_speed
+    import lumen_routes
+    from lumen_router_config import _OR_LIGHT_ORDER
+    head = _OR_LIGHT_ORDER[0]
+    seen = {}
+
+    async def fake_or(path, method="GET", *, json_body=None, deadline=None):
+        seen["model"] = (json_body or {}).get("model")
+        return {"choices": [{"message": {"content": "ok"}}],
+                "usage": {"prompt_tokens": 4001, "completion_tokens": 2, "total_tokens": 4003}}
+
+    monkeypatch.setattr(bot, "_or_request", fake_or)
+    before = ((bot.GLOBAL_QUOTA.get("openrouter") or {}).get(head) or {}).get("used") or 0
+    ema_before = dict(lumen_model_speed._latency_ema)
+    try:
+        ok, detail, elapsed = asyncio.run(lumen_routes.selftest_llm_head("openrouter", chat_id=1))
+    finally:
+        lumen_model_speed._latency_ema.clear()
+        lumen_model_speed._latency_ema.update(ema_before)
+    assert ok is True and detail == "" and elapsed >= 0
+    assert seen["model"] == head
+    after = ((bot.GLOBAL_QUOTA.get("openrouter") or {}).get(head) or {}).get("used") or 0
+    assert after == before + 1
+    bot.GLOBAL_QUOTA["openrouter"][head]["used"] = before
+
+
+def test_selftest_llm_head_openrouter_error_returns_false_without_quota(monkeypatch):
+    # Неуспех квоту не трогает (как в ask_*): иначе ошибка стоила бы как ответ.
+    import lumen_routes
+    from lumen_router_config import _OR_LIGHT_ORDER
+    head = _OR_LIGHT_ORDER[0]
+
+    async def boom(path, method="GET", *, json_body=None, deadline=None):
+        raise RuntimeError("boom-429")
+
+    monkeypatch.setattr(bot, "_or_request", boom)
+    before = ((bot.GLOBAL_QUOTA.get("openrouter") or {}).get(head) or {}).get("used") or 0
+    ok, detail, elapsed = asyncio.run(lumen_routes.selftest_llm_head("openrouter", chat_id=1))
+    assert ok is False and "boom-429" in detail and elapsed >= 0
+    after = ((bot.GLOBAL_QUOTA.get("openrouter") or {}).get(head) or {}).get("used") or 0
+    assert after == before