Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion AGENTS.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
123 changes: 118 additions & 5 deletions src/bot/dispatch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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`.
"""

Expand Down Expand Up @@ -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)
# ---------------------------------------------------------------------------
Expand Down Expand Up @@ -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,
)


# ---------------------------------------------------------------------------
Expand All @@ -455,36 +502,102 @@ 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
try:
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)
81 changes: 60 additions & 21 deletions src/bot/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.
Expand All @@ -127,32 +127,78 @@ 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.
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

def main() -> None:
"""
Expand Down Expand Up @@ -212,22 +258,15 @@ 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")

asyncio.run(
dp.start_polling(
bot,
state=state,
app_state=state,
allowed_updates=["message", "edited_message", "callback_query", "chat_member"],
)
)
Expand Down
52 changes: 52 additions & 0 deletions tests/test_dispatch.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Loading
Loading