Gate webhook updates until startup state loads and shut down gracefully on SIGTERM

bot.py

@@ -24,7 +24,9 @@ import os
 import queue
 import re
 import secrets
+import signal
 import sys
+import threading
 import time
 from pathlib import Path
 from datetime import date
@@ -1469,9 +1471,17 @@ from lumen_message_core import (
 SETWEBHOOK_MAX_ATTEMPTS = 5
 SETWEBHOOK_BACKOFF_SEC = 5.0
 
+# /webhook отвечает 503, пока старт не загрузил состояние: иначе апдейт,
+# пришедший во время загрузки, создаёт пустой чат и перезаписывает удалённую
+# историю/индекс (аудит 28.10.2026).
+_state_ready = True
+
+
 async def _webhook_startup() -> None:
+    global _state_ready
     # Синхронное дисковое чтение в потоке: иначе старт стопорит приём /webhook.
     await asyncio.to_thread(load_state_from_disk)
+    _state_ready = True
     log.info("Bot startup: webhook mode.")
     # На Space кап можно переопределить env: без этой строки в логах не видно,
     # какой лимит реально работает (разбор 26.10.2026).
@@ -1669,8 +1679,32 @@ def _require_bot_token() -> str:
     return BOT_TOKEN
 
 
+@contextlib.contextmanager
+def _capture_signals_gracefully(server: Any):
+    """SIGTERM/SIGINT: uvicorn по умолчанию пере-поднимает сигнал после
+    serve(), и finally с финальным сбросом состояния не выполняется
+    (аудит 28.10.2026). Перехватываем сигнал и только просим сервер выйти."""
+    if threading.current_thread() is not threading.main_thread():
+        yield
+        return
+    original = {}
+
+    def _stop(signum, frame):
+        server.should_exit = True
+
+    for sig in (signal.SIGINT, signal.SIGTERM):
+        with contextlib.suppress(Exception):
+            original[sig] = signal.signal(sig, _stop)
+    try:
+        yield
+    finally:
+        for sig, handler in original.items():
+            with contextlib.suppress(Exception):
+                signal.signal(sig, handler)
+
+
 async def main() -> None:
-    global bot, client
+    global bot, client, _state_ready
 
     _require_bot_token()
 
@@ -1688,6 +1722,7 @@ async def main() -> None:
     bot = Bot(token=BOT_TOKEN, session=sess)
     client = genai.Client(api_key=GEMINI_API_KEY)
 
+    _state_ready = False
     startup_task = asyncio.create_task(_webhook_startup(), name="webhook_startup")
     flush_task = asyncio.create_task(_flush_dirty_state(), name="state_flush")
     srv_config = uvicorn.Config(
@@ -1697,6 +1732,9 @@ async def main() -> None:
         app=app, host="0.0.0.0", port=7860, log_level=None, loop="asyncio", log_config=None,
     )
     server = uvicorn.Server(srv_config)
+    # Не даём uvicorn пере-поднять SIGTERM после serve(): иначе финальный
+    # сброс состояния и закрытие сессий не выполняются.
+    server.capture_signals = lambda: _capture_signals_gracefully(server)
     try:
         serve_task = asyncio.create_task(server.serve(), name="uvicorn_serve")
         done, _pending = await asyncio.wait(
@@ -1720,8 +1758,11 @@ async def main() -> None:
             if exc is not None:
                 log.error("[serve] HTTP server failed, stopping: %s", exc)
                 raise SystemExit(1) from exc
-            log.error("[serve] HTTP server exited unexpectedly, stopping")
-            raise SystemExit(1)
+            if getattr(server, "should_exit", False):
+                log.info("[serve] HTTP server stopped gracefully")
+            else:
+                log.error("[serve] HTTP server exited unexpectedly, stopping")
+                raise SystemExit(1)
     finally:
         startup_task.cancel()
         flush_task.cancel()
@@ -1737,10 +1778,21 @@ async def main() -> None:
         # Флаги — канонически в lumen_chat_state (P2): множества общие объектом,
         # bool-флаги читаем из модуля, т.к. bot-привязки после выноса stale.
         # Синхронная запись в потоке: loop уже дренирует задачи, висеть нельзя.
-        if _dirty_chat_ids or lumen_chat_state._index_dirty or _pending_chat_deletions:
-            await asyncio.to_thread(_flush_state_now)
-        if lumen_chat_state._quota_dirty:
-            await asyncio.to_thread(save_global_quota)
+        if lumen_chat_state._state_load_failed:
+            # Стартовая загрузка не удалась: не затираем удалённые данные
+            # (аудит 28.10.2026).
+            log.warning('[state] Startup load failed: skipping the final flush to keep remote data intact.')
+        else:
+            if _dirty_chat_ids or lumen_chat_state._index_dirty or _pending_chat_deletions:
+                try:
+                    await asyncio.wait_for(asyncio.to_thread(_flush_state_now), timeout=15.0)
+                except asyncio.TimeoutError:
+                    log.warning('[state] Final flush exceeded 15s; abandoning the remainder.')
+            if lumen_chat_state._quota_dirty:
+                try:
+                    await asyncio.wait_for(asyncio.to_thread(save_global_quota), timeout=10.0)
+                except asyncio.TimeoutError:
+                    log.warning('[state] Final quota save exceeded 10s; abandoning the remainder.')
         await _close_sessions()
 
 if __name__ == "__main__":

lumen_admin.py

@@ -155,6 +155,11 @@ async def webhook_handler(request: Request) -> Any:
     if not bot.WEBHOOK_SECRET or not token or not hmac.compare_digest(token.encode(), bot.WEBHOOK_SECRET.encode()):
         _log_webhook_denied("invalid secret token")
         return {"ok": False}
+    # Состояние ещё грузится: 503, чтобы Telegram повторил позже, а не создал
+    # пустой чат и не перезаписал удалённую историю (аудит 28.10.2026).
+    if not getattr(bot, "_state_ready", True):
+        log.info("[webhook] State is still loading, asking Telegram to retry")
+        return JSONResponse(status_code=503, content={"ok": False, "retry": True})
     # Тело апдейта маленькое: отсекаем мусор по заголовку до чтения.
     content_length = request.headers.get("Content-Length")
     if content_length is not None:

lumen_chat_state.py

@@ -675,6 +675,11 @@ async def _flush_dirty_state_once() -> None:
                     '[state] %d chat deletion(s) failed, will retry next cycle: %s',
                     len(failed_deletes), ", ".join(str(c) for c in sorted(failed_deletes)),
                 )
+        if _state_load_failed and _dirty_chat_ids:
+            # Удалённые истории не затёрты: файл в памяти — это либо новый чат,
+            # либо всё, что осталось после сбоя загрузки; индекс и квота при
+            # этом не пишутся (аудит A5-2, 28.10.2026).
+            log.warning('[state] Startup load failed; saving chats from memory, index stays untouched.')
         if _dirty_chat_ids:
             to_save = list(_dirty_chat_ids)
             _dirty_chat_ids.clear()

tests/test_bot_admin.py

@@ -10,7 +10,7 @@ import os
 import threading
 import time
 import pytest
-from unittest.mock import Mock
+from unittest.mock import AsyncMock, Mock
 from tests.bot_test_helpers import (
     _FakeAdminRequest,
     _FakeWebhookRequest,
@@ -24,9 +24,11 @@ def _clear_webhook_update_dedup():
     import lumen_admin
     lumen_admin._seen_update_ids.clear()
     lumen_admin._seen_update_id_set.clear()
+    bot._state_ready = True
     yield
     lumen_admin._seen_update_ids.clear()
     lumen_admin._seen_update_id_set.clear()
+    bot._state_ready = True
 
 
 def test_public_log_formatter_hashes_long_numeric_ids():
@@ -162,6 +164,33 @@ def test_webhook_handler_tracks_dispatched_task_for_shutdown():
         bot._process_raw_update = original_process
 
 
+def test_webhook_returns_503_until_state_is_ready():
+    # Апдейт во время загрузки состояния не должен создавать пустой чат и
+    # перезаписывать удалённую историю (аудит 28.10.2026).
+    original_secret = bot.WEBHOOK_SECRET
+    original_bot_obj = bot.bot
+    original_process = bot._process_raw_update
+    original_ready = bot._state_ready
+    bot.WEBHOOK_SECRET = "real-webhook-secret"
+    bot.bot = object()
+    bot._state_ready = False
+    process = AsyncMock()
+    bot._process_raw_update = process
+    try:
+        req = _FakeWebhookRequest(
+            headers={"X-Telegram-Bot-Api-Secret-Token": "real-webhook-secret"},
+            body={"update_id": 778},
+        )
+        result = asyncio.run(_run_webhook_handler(req))
+        assert getattr(result, "status_code", None) == 503
+        process.assert_not_called()
+    finally:
+        bot.WEBHOOK_SECRET = original_secret
+        bot.bot = original_bot_obj
+        bot._process_raw_update = original_process
+        bot._state_ready = original_ready
+
+
 def test_webhook_handler_ignores_duplicate_update_id():
     # Telegram повторяет доставку при таймауте/сбое: один update_id не должен
     # давать второй ответ, скачивание и трату квоты (аудит 28.10.2026).
@@ -800,6 +829,64 @@ def test_main_exits_nonzero_when_server_dies_first(monkeypatch, caplog):
         lumen_chat_state._quota_dirty = orig_quota_dirty
 
 
+def test_main_flushes_on_graceful_server_exit(monkeypatch, caplog):
+    # SIGTERM/SIGINT: сервер завершился штатно — финальный сброс состояния и
+    # закрытие сессий обязаны выполниться (аудит 28.10.2026).
+    from types import SimpleNamespace
+    from unittest.mock import AsyncMock
+    import lumen_chat_state
+    monkeypatch.setattr(bot, "_require_bot_token", lambda: "123:abc")
+    monkeypatch.setattr(bot, "Bot", lambda *a, **k: SimpleNamespace())
+    monkeypatch.setattr(bot, "IPv4AiohttpSession", lambda *a, **k: SimpleNamespace())
+    monkeypatch.setattr(bot.genai, "Client", lambda *a, **k: SimpleNamespace())
+
+    async def _never():
+        await asyncio.sleep(3600)
+
+    monkeypatch.setattr(bot, "_webhook_startup", _never)
+    monkeypatch.setattr(bot, "_flush_dirty_state", _never)
+    monkeypatch.setattr(bot, "_drain_inflight_tasks", AsyncMock())
+    monkeypatch.setattr(bot, "_close_sessions", AsyncMock())
+    monkeypatch.setattr(bot.uvicorn, "Config", lambda *a, **k: SimpleNamespace())
+
+    class _StoppedServer:
+        should_exit = True
+
+        async def serve(self):
+            return None
+
+    monkeypatch.setattr(bot.uvicorn, "Server", lambda cfg: _StoppedServer())
+    emitted = []
+    monkeypatch.setattr(bot, "_flush_state_now", lambda: emitted.append("chats"))
+    monkeypatch.setattr(bot, "save_global_quota", lambda: emitted.append("quota"))
+    orig_bot, orig_client = bot.bot, bot.client
+    orig_ready = bot._state_ready
+    orig_dirty, orig_deletions = set(bot._dirty_chat_ids), set(bot._pending_chat_deletions)
+    orig_index_dirty, orig_quota_dirty = lumen_chat_state._index_dirty, lumen_chat_state._quota_dirty
+    orig_load_failed = lumen_chat_state._state_load_failed
+    bot._dirty_chat_ids.clear()
+    bot._dirty_chat_ids.add(999001)
+    bot._pending_chat_deletions.clear()
+    lumen_chat_state._index_dirty = False
+    lumen_chat_state._quota_dirty = True
+    lumen_chat_state._state_load_failed = False
+    try:
+        with caplog.at_level(logging.INFO, logger="bot"):
+            asyncio.run(bot.main())
+        assert "[serve] HTTP server stopped gracefully" in caplog.text
+        assert "chats" in emitted and "quota" in emitted
+    finally:
+        bot.bot, bot.client = orig_bot, orig_client
+        bot._state_ready = orig_ready
+        bot._dirty_chat_ids.clear()
+        bot._dirty_chat_ids.update(orig_dirty)
+        bot._pending_chat_deletions.clear()
+        bot._pending_chat_deletions.update(orig_deletions)
+        lumen_chat_state._index_dirty = orig_index_dirty
+        lumen_chat_state._quota_dirty = orig_quota_dirty
+        lumen_chat_state._state_load_failed = orig_load_failed
+
+
 def test_probe_url_returns_ok_shape_and_redacts_secret_on_error():
     # Общий зонд /diag и /selftest: успех — статус, неуспех — текст без секрета.
     class _FakeResp: