From 1e38096818e099be2b6f757b2ea3caa4df3a2e06 Mon Sep 17 00:00:00 2001 From: Tayo Date: Thu, 8 Oct 2026 18:00:34 +0700 Subject: [PATCH 1/2] fix(dispatch): route updates via sub-observers instead of dp.update() The aiogram migration registered the single update entrypoint with @dp.update(). That handler never ran: Dispatcher.__init__ registers its own internal _listen_update on dp.update first, and the observer stops at the first matching handler, so every update fell through to _listen_update, found no router handlers, and was logged 'not handled'. The bot was deaf from the deploy moment. Register the entrypoint on the four concrete sub-observers instead (message, edited_message, callback_query, chat_member), which have no internal handlers, and recover the full Update via the event_update kwarg that _listen_update sets. Also rename the shared-state kwarg from state to app_state: FSMContextMiddleware (always registered on dp.update) injects its own FSMContext as 'state' on update handlers, which would have shadowed our AppState even if the handler had run. build_dispatcher() centralizes the wiring so tests can pin it; tests/test_main_wiring.py feeds synthetic updates through a real Dispatcher and asserts dispatch_update is reached with our AppState. --- AGENTS.md | 3 +- src/bot/dispatch.py | 2 +- src/bot/main.py | 76 +++++++++++++++++++-------- tests/test_main_wiring.py | 105 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 163 insertions(+), 23 deletions(-) create mode 100644 tests/test_main_wiring.py diff --git a/AGENTS.md b/AGENTS.md index 67d262f..c29912f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -263,7 +263,8 @@ Time threshold → Auto-restrict via scheduler (parallel path) - **Python 3.11+** with type hints - **Imports**: stdlib → third-party → local - **Async/await**: All handlers are async -- **aiogram 3.x**: Handlers are `async (update: Update, context: HandlerContext)`; updates flow through the manual group-ordered dispatch loop in `bot/dispatch.py` (`AppState` carries shared state, injected via `dp.start_polling(..., state=state)`) +- **aiogram 3.x**: Handlers are `async (update: Update, context: HandlerContext)`; updates flow through the manual group-ordered dispatch loop in `bot/dispatch.py` (`AppState` carries shared state, injected via `dp.start_polling(..., app_state=app_state)`) +- **Dispatcher wiring gotchas** (learned the hard way, 2026-10-08): (1) never register the update entrypoint on `dp.update()` — `Dispatcher.__init__` registers its internal `_listen_update` there first and the observer stops at the first matching handler, so a user `@dp.update()` handler silently never runs (every update logs "not handled"); register on the concrete sub-observers (`dp.message()`, `dp.edited_message()`, `dp.callback_query()`, `dp.chat_member()`) and recover the full `Update` via the `event_update` kwarg. (2) name the shared-state kwarg `app_state`, never `state` — `FSMContextMiddleware` injects its own `FSMContext` as `state` on update handlers and shadows yours. `build_dispatcher()` in `main.py` centralizes this wiring; `tests/test_main_wiring.py` feeds synthetic updates through a real `Dispatcher` to guard it - **Logging**: Use `logfire` via stdlib `logging.getLogger(__name__)` - **Error handling**: Catch specific exceptions (`TelegramNetworkError`), log, return gracefully - **No inline comments** unless code is complex diff --git a/src/bot/dispatch.py b/src/bot/dispatch.py index c66dfc8..1274b22 100644 --- a/src/bot/dispatch.py +++ b/src/bot/dispatch.py @@ -56,7 +56,7 @@ class AppState: """Explicit shared runtime state (replaces PTB ``application.bot_data``). Injected into the aiogram dispatcher via - ``dp.start_polling(bot, state=state)`` and handed to every handler + ``dp.start_polling(bot, app_state=state)`` and handed to every handler inside :class:`HandlerContext`. """ diff --git a/src/bot/main.py b/src/bot/main.py index 652e78a..aad4da6 100644 --- a/src/bot/main.py +++ b/src/bot/main.py @@ -12,7 +12,7 @@ import logfire from aiogram import Bot, Dispatcher -from aiogram.types import ErrorEvent, Update +from aiogram.types import ErrorEvent, TelegramObject, Update from apscheduler.schedulers.asyncio import AsyncIOScheduler from bot.config import get_settings @@ -104,17 +104,17 @@ async def error_handler(event: ErrorEvent) -> None: """ await handle_bot_error(event.update, event.exception) -async def on_shutdown(bot: Bot, state: AppState) -> None: +async def on_shutdown(bot: Bot, app_state: AppState) -> None: """Release shared resources after the bot stops polling.""" _ = bot - if state.scheduler is not None: - state.scheduler.shutdown() + if app_state.scheduler is not None: + app_state.scheduler.shutdown() logger.info("on_shutdown: scheduler stopped") await close_client() logger.info("on_shutdown: classifier HTTP client closed") -async def on_startup(bot: Bot, state: AppState) -> None: +async def on_startup(bot: Bot, app_state: AppState) -> None: """ Startup handler: resolve bot username, start the scheduler, fetch and cache group admin IDs. @@ -127,32 +127,73 @@ async def on_startup(bot: Bot, state: AppState) -> None: logger.info("Starting on_startup: resolving bot username, starting scheduler, fetching admin IDs") me = await bot.get_me() - state.bot_username = me.username + app_state.bot_username = me.username logger.info(f"Bot username resolved: @{me.username}") - if state.scheduler is not None: - state.scheduler.start() + if app_state.scheduler is not None: + app_state.scheduler.start() logger.info("Scheduler started") registry = get_group_registry() # Use preload_admin_ids which preserves cache on failures - await preload_admin_ids(state) + await preload_admin_ids(app_state) # Record start time for /status uptime - state.start_time = time.monotonic() + app_state.start_time = time.monotonic() # Preload trusted users cache db = get_database() trusted_ids = db.get_trusted_user_ids() - state.trusted_user_ids = trusted_ids + app_state.trusted_user_ids = trusted_ids logger.info(f"Loaded {len(trusted_ids)} trusted user(s) into cache") # Recover pending captcha verifications for groups with captcha enabled has_captcha = any(gc.captcha_enabled for gc in registry.all_groups()) if has_captcha: logger.info("Recovering pending captcha verifications from database") - await recover_pending_captchas(state) + await recover_pending_captchas(app_state) + + +def build_dispatcher() -> Dispatcher: + """Build the aiogram Dispatcher with lifecycle, error, and update wiring. + + The single update entrypoint runs the manual group-ordered dispatch + (``dispatch_update``) for every incoming update. + + Two aiogram behaviors shape this wiring — do not "simplify" it: + + 1. Never register the entrypoint on ``dp.update()``. ``Dispatcher.__init__`` + registers its own internal ``_listen_update`` handler there first, and + ``TelegramEventObserver.trigger()`` stops at the first matching + handler, so a user ``@dp.update()`` handler never runs and every update + is logged "not handled". The four concrete sub-observers below have no + internal handlers, so the entrypoint is always reached. + 2. The shared-state kwarg is named ``app_state``, not ``state``. + ``FSMContextMiddleware`` (always registered on ``dp.update``) injects + its own ``FSMContext`` as ``state`` on update handlers, which would + shadow ours. + """ + dp = Dispatcher() + dp.startup.register(on_startup) + dp.shutdown.register(on_shutdown) + dp.errors.register(error_handler) + + @dp.message() + @dp.edited_message() + @dp.callback_query() + @dp.chat_member() + async def _route_update( + event: TelegramObject, + bot: Bot, + app_state: AppState, + event_update: Update, + ) -> None: + # event_update is the full Update (aiogram sets it in _listen_update); + # dispatch_update needs the Update wrapper, not the bare event. + await dispatch_update(event_update, bot, app_state) + + return dp def main() -> None: """ @@ -212,14 +253,7 @@ def main() -> None: # Wire up the dispatcher: startup/shutdown lifecycle, error handler, # and a single update entrypoint that runs the group-ordered dispatch. - dp = Dispatcher() - dp.startup.register(on_startup) - dp.shutdown.register(on_shutdown) - dp.errors.register(error_handler) - - @dp.update() - async def _route_update(update: Update, bot: Bot, state: AppState) -> None: - await dispatch_update(update, bot, state) + dp = build_dispatcher() logger.info(f"Starting bot polling for {group_count} group(s)") logger.info("All handlers registered successfully") @@ -227,7 +261,7 @@ async def _route_update(update: Update, bot: Bot, state: AppState) -> None: asyncio.run( dp.start_polling( bot, - state=state, + app_state=state, allowed_updates=["message", "edited_message", "callback_query", "chat_member"], ) ) diff --git a/tests/test_main_wiring.py b/tests/test_main_wiring.py new file mode 100644 index 0000000..9895258 --- /dev/null +++ b/tests/test_main_wiring.py @@ -0,0 +1,105 @@ +"""Regression tests for the aiogram dispatcher wiring in bot.main. + +Background: the aiogram migration initially registered the single update +entrypoint with ``@dp.update()``. That handler never ran — ``Dispatcher`` +registers its own internal ``_listen_update`` on ``dp.update`` first and the +observer stops at the first matching handler — so the bot was deaf from the +deploy moment (every update logged "not handled"). Additionally the shared +state kwarg was named ``state``, which aiogram's ``FSMContextMiddleware`` +shadows with its own ``FSMContext`` on update handlers. + +These tests wire a real ``Dispatcher`` exactly like production +(``build_dispatcher()`` + ``feed_update``) and assert the entrypoint is +reached with the real ``AppState``. +""" + +from datetime import UTC, datetime +from unittest.mock import patch + +from aiogram import Bot +from aiogram.types import CallbackQuery, Chat, Message, Update, User + +import bot.main as main_module +from bot.dispatch import AppState +from bot.main import build_dispatcher + + +def _make_bot() -> Bot: + return Bot(token="123456:ABC-DEF1234ghIkl-zyx57W2v1u123ew11") + + +def _make_message_update(update_id: int) -> Update: + return Update( + update_id=update_id, + message=Message( + message_id=1, + date=datetime.now(UTC), + chat=Chat(id=-1001234567890, type="supergroup"), + from_user=User(id=111, is_bot=False, first_name="Uji"), + text="/status", + ), + ) + + +def _make_callback_update(update_id: int) -> Update: + return Update( + update_id=update_id, + callback_query=CallbackQuery( + id="cb1", + from_user=User(id=111, is_bot=False, first_name="Uji"), + chat_instance="ci1", + data="verify:1:2", + ), + ) + + +async def test_message_update_reaches_dispatch_with_app_state(): + bot = _make_bot() + try: + app_state = AppState(bot=bot) + dp = build_dispatcher() + update = _make_message_update(update_id=4242) + + calls: list = [] + with patch.object(main_module, "dispatch_update") as mock_dispatch: + mock_dispatch.side_effect = lambda u, b, s: calls.append((u, b, s)) + await dp.feed_update(bot, update, app_state=app_state) + + assert len(calls) == 1 + received_update, received_bot, received_state = calls[0] + assert received_update.update_id == 4242 + assert received_bot is bot + # Must be our AppState, not aiogram's FSMContext injected as `state`. + assert received_state is app_state + finally: + await bot.session.close() + + +async def test_callback_query_update_reaches_dispatch(): + bot = _make_bot() + try: + app_state = AppState(bot=bot) + dp = build_dispatcher() + update = _make_callback_update(update_id=777) + + calls: list = [] + with patch.object(main_module, "dispatch_update") as mock_dispatch: + mock_dispatch.side_effect = lambda u, b, s: calls.append((u, b, s)) + await dp.feed_update(bot, update, app_state=app_state) + + assert len(calls) == 1 + assert calls[0][0].update_id == 777 + assert calls[0][2] is app_state + finally: + await bot.session.close() + + +async def test_build_dispatcher_registers_entrypoint_on_sub_observers(): + """The entrypoint must live on the concrete observers, never dp.update.""" + dp = build_dispatcher() + # dp.update keeps only aiogram's internal _listen_update. + assert len(dp.update.handlers) == 1 + assert len(dp.message.handlers) == 1 + assert len(dp.edited_message.handlers) == 1 + assert len(dp.callback_query.handlers) == 1 + assert len(dp.chat_member.handlers) == 1 From a8fe4c9c16c5fb37693bb1ecb545b6e0788e75db Mon Sep 17 00:00:00 2001 From: Tayo Date: Thu, 8 Oct 2026 18:07:08 +0700 Subject: [PATCH 2/2] chore(dispatch): trace logging for every dispatch path Every update now emits an update_id-tagged trail through the dispatch loop: entry (update_id, kind, chat, user, command), per matched handler (label, plugin, group), StopPropagation / error-handler decisions, and 'unhandled by all groups' when nothing matches. Per-spec filter misses stay at DEBUG. handle_bot_error and the wiring entrypoint (_route_update) also tag their lines with update_id. This makes the 2026-10-08 deaf-bot failure mode visible in Logfire: updates arriving with no dispatch lines (or aiogram 'not handled' with zero 'handled') is now directly queryable, and the monitor script (logfire_check.py) gained a liveness check on exactly that signal. --- src/bot/dispatch.py | 121 +++++++++++++++++++++++++++++++++++++++-- src/bot/main.py | 5 ++ tests/test_dispatch.py | 52 ++++++++++++++++++ 3 files changed, 174 insertions(+), 4 deletions(-) diff --git a/src/bot/dispatch.py b/src/bot/dispatch.py index 1274b22..af5e3b1 100644 --- a/src/bot/dispatch.py +++ b/src/bot/dispatch.py @@ -238,6 +238,49 @@ def update_kind(update: Update) -> str | None: return None +def _update_chat(update: Update): + """Return the chat attached to an update, regardless of kind.""" + if update.message is not None: + return update.message.chat + if update.edited_message is not None: + return update.edited_message.chat + if update.callback_query is not None and update.callback_query.message is not None: + return update.callback_query.message.chat + if update.chat_member is not None: + return update.chat_member.chat + return None + + +def _chat_label(update: Update) -> str: + chat = _update_chat(update) + if chat is None: + return "?" + return f"{chat.type}:{chat.id}" + + +def _update_user_id(update: Update) -> int | str: + """Return the acting user id for an update, regardless of kind.""" + if update.message is not None and update.message.from_user is not None: + return update.message.from_user.id + if update.edited_message is not None and update.edited_message.from_user is not None: + return update.edited_message.from_user.id + if update.callback_query is not None and update.callback_query.from_user is not None: + return update.callback_query.from_user.id + if update.chat_member is not None and update.chat_member.from_user is not None: + return update.chat_member.from_user.id + return "?" + + +def _command_suffix(update: Update) -> str: + """Append the command text (if any) to the dispatch entry log line.""" + message = _message_or_none(update) + if message is not None and message.text: + text = message.text.strip().split()[0] + if text.startswith("/"): + return f" cmd={text[:40]}" + return "" + + # --------------------------------------------------------------------------- # Filter predicates (replacements for PTB filters) # --------------------------------------------------------------------------- @@ -426,10 +469,14 @@ async def handle_bot_error(update: Update | None, exc: BaseException) -> None: aiogram's ``TelegramNetworkError``) are transient and logged at warning level; everything else is logged as an error with traceback. """ + update_id = update.update_id if update is not None else "?" if isinstance(exc, TelegramNetworkError): - logger.warning(f"Telegram network error: {exc}") + logger.warning(f"dispatch: update_id={update_id} telegram network error: {exc}") return - logger.error("Unhandled exception in update handler:", exc_info=exc) + logger.error( + f"dispatch: update_id={update_id} unhandled exception in update handler:", + exc_info=exc, + ) # --------------------------------------------------------------------------- @@ -455,14 +502,43 @@ async def dispatch_update(update: Update, bot: Bot, state: AppState) -> None: """Dispatch one update through the handler groups (PTB semantics). See the module docstring for the exact ordering/stop rules. + + Logging: every update emits an entry line (update_id, kind, chat, user) + plus one line per matched handler and per stop/error decision, so the + full path of an update is traceable in Logfire. Per-spec filter + evaluations stay at DEBUG to bound volume. """ kind = update_kind(update) + update_id = update.update_id if kind is None: + logger.info("dispatch: update_id=%s ignored (unsupported event type)", update_id) return + logger.info( + "dispatch: update_id=%s kind=%s chat=%s user_id=%s%s", + update_id, + kind, + _chat_label(update), + _update_user_id(update), + _command_suffix(update), + ) + context = HandlerContext(bot=bot, state=state) - for _group, specs in _grouped_specs(state): + grouped = _grouped_specs(state) + if not grouped: + logger.warning("dispatch: update_id=%s no handlers registered", update_id) + return + + handled_any = False + for group, specs in grouped: + logger.debug( + "dispatch: update_id=%s evaluating group=%s (%d specs)", + update_id, + group, + len(specs), + ) + matched_in_group = False for spec in specs: if kind not in spec.update_kinds: continue @@ -470,21 +546,58 @@ async def dispatch_update(update: Update, bot: Bot, state: AppState) -> None: matched = spec.check(update) except Exception: logger.warning( - f"Filter error in handler '{spec.label}', skipping", + "dispatch: update_id=%s filter error in handler '%s', skipping", + update_id, + spec.label, exc_info=True, ) continue if not matched: + logger.debug( + "dispatch: update_id=%s handler '%s' filter did not match", + update_id, + spec.label, + ) continue + logger.info( + "dispatch: update_id=%s matched handler='%s' plugin='%s' group=%s", + update_id, + spec.label, + spec.plugin_name, + group, + ) if spec.command is not None: message = _message_or_none(update) context.args = parse_command_args(message) if message else [] try: await spec.callback(update, context) except StopPropagation: + logger.info( + "dispatch: update_id=%s StopPropagation raised by '%s', done", + update_id, + spec.label, + ) return except Exception as exc: # noqa: BLE001 - PTB parity: route to error handler, try next + logger.info( + "dispatch: update_id=%s handler '%s' raised %s, routed to error handler", + update_id, + spec.label, + type(exc).__name__, + ) await handle_bot_error(update, exc) continue # First matching handler wins within a group. + logger.info( + "dispatch: update_id=%s handler '%s' completed, first-match-wins for group=%s", + update_id, + spec.label, + group, + ) + handled_any = True + matched_in_group = True break + if not matched_in_group: + logger.debug("dispatch: update_id=%s no match in group=%s", update_id, group) + if not handled_any: + logger.info("dispatch: update_id=%s unhandled by all groups", update_id) diff --git a/src/bot/main.py b/src/bot/main.py index aad4da6..ba6a051 100644 --- a/src/bot/main.py +++ b/src/bot/main.py @@ -191,6 +191,11 @@ async def _route_update( ) -> None: # event_update is the full Update (aiogram sets it in _listen_update); # dispatch_update needs the Update wrapper, not the bare event. + logger.info( + "wiring: received update_id=%s via %s", + event_update.update_id, + type(event).__name__, + ) await dispatch_update(event_update, bot, app_state) return dp diff --git a/tests/test_dispatch.py b/tests/test_dispatch.py index 9f37e9c..e279622 100644 --- a/tests/test_dispatch.py +++ b/tests/test_dispatch.py @@ -622,3 +622,55 @@ def test_iter_and_len(self): assert "admin_ids" in keys assert "custom" in keys assert len(ctx.bot_data) == len(keys) + + +class TestDispatchLogging: + """Every dispatch path emits a traceable log line (update_id-tagged).""" + + def _state_with(self, specs): + state = AppState() + state.plugin_handlers = { + "p": {"handler_group": 0, "handlers": specs}, + } + return state + + async def test_entry_and_match_logged(self, bot, caplog): + import logging + calls = [] + state = self._state_with([_spec("plug", 0, ("message",), _yes, _cb("h", calls))]) + with caplog.at_level(logging.INFO, logger="bot.dispatch"): + await dispatch_update(_make_update(_make_message("/status")), bot, state) + text = caplog.text + assert "dispatch: update_id=1 kind=message" in text + assert "cmd=/status" in text + assert "matched handler='plug'" in text + assert "completed, first-match-wins" in text + + async def test_unhandled_logged(self, bot, caplog): + import logging + state = self._state_with([_spec("plug", 0, ("message",), _no, _cb("h", []))]) + with caplog.at_level(logging.INFO, logger="bot.dispatch"): + await dispatch_update(_make_update(), bot, state) + assert "dispatch: update_id=1 unhandled by all groups" in caplog.text + + async def test_stop_propagation_logged(self, bot, caplog): + import logging + calls = [] + state = self._state_with([_spec("plug", 0, ("message",), _yes, _cb("h", calls, stop=True))]) + with caplog.at_level(logging.INFO, logger="bot.dispatch"): + await dispatch_update(_make_update(), bot, state) + assert "StopPropagation raised by 'plug'" in caplog.text + + async def test_unsupported_update_ignored_logged(self, bot, caplog): + import logging + state = AppState() + with caplog.at_level(logging.INFO, logger="bot.dispatch"): + await dispatch_update(Update(update_id=9), bot, state) + assert "dispatch: update_id=9 ignored (unsupported event type)" in caplog.text + + async def test_no_handlers_registered_logged(self, bot, caplog): + import logging + state = AppState() + with caplog.at_level(logging.INFO, logger="bot.dispatch"): + await dispatch_update(_make_update(), bot, state) + assert "dispatch: update_id=1 no handlers registered" in caplog.text