From bc18d8fde579a491e7df010510ed8471cd19ecc1 Mon Sep 17 00:00:00 2001 From: Pouri Date: Sat, 3 Oct 2026 19:29:07 +0330 Subject: [PATCH 1/6] daemon: attach the update handler when the client is built, so incoming updates reach the bus SessionManager.ensure() called the ready hook straight after AccountSession.start(), which only schedules the supervisor. The client did not exist yet, so _attach_handlers returned early and no incoming update ever reached the bus: jobs, webhooks and watch saw the daemon's own sends only, and every gateway job stopped with matched=0 skipped=0. The session now calls an on_client hook from the supervisor right after it builds the client, before the private-API hooks that can raise. The client is kept across reconnects, so the handler is attached once. --- tests/test_daemon_updates.py | 63 ++++++++++++++++++++++++++++++++++++ tlgr/daemon/app.py | 4 +-- tlgr/daemon/session.py | 27 ++++++++++++++++ tlgr/daemon/sessions.py | 8 ++--- 4 files changed, 95 insertions(+), 7 deletions(-) create mode 100644 tests/test_daemon_updates.py diff --git a/tests/test_daemon_updates.py b/tests/test_daemon_updates.py new file mode 100644 index 0000000..49c5431 --- /dev/null +++ b/tests/test_daemon_updates.py @@ -0,0 +1,63 @@ +"""Incoming updates reach the daemon's event bus. + +Until 2.0.1 none did: the daemon attached its update handler straight after +`AccountSession.start()`, which only schedules the supervisor, so it always +found `session.client is None` and returned. Jobs, webhooks and `watch` saw +the daemon's own sends and nothing else, and every job stopped with +`matched=0 skipped=0`. +""" + +from __future__ import annotations + +import asyncio + +from telethon.tl import types + +from tlgr.daemon.session import SessionState + + +async def _online(daemon, alias: str): + session = await daemon.sessions.ensure(alias) + deadline = asyncio.get_running_loop().time() + 2.0 + while session.state != SessionState.ONLINE: + if asyncio.get_running_loop().time() > deadline: + raise AssertionError(f"{alias} never came online: {session.state}") + await asyncio.sleep(0.01) + return session + + +def _incoming(message_id: int = 7) -> types.UpdateNewChannelMessage: + message = types.Message( + id=message_id, + peer_id=types.PeerChannel(channel_id=1234), + date=None, + message="hello", + ) + return types.UpdateNewChannelMessage(message=message, pts=1, pts_count=1) + + +class TestIncomingUpdates: + async def test_an_incoming_message_is_numbered_on_the_bus(self, daemon, stub_account): + session = await _online(daemon, stub_account) + before = daemon.bus.latest_seq(stub_account) + + await session.client.feed(_incoming()) + + assert daemon.bus.latest_seq(stub_account) == before + 1 + [envelope], _gap = daemon.bus.replay(stub_account, before) + assert envelope.type == "message_new" + assert envelope.chat_id == -1000000001234 + + async def test_the_handler_is_attached_once_across_reconnects(self, daemon, stub_account): + """The client outlives a reconnect, so a second handler would double every event.""" + session = await _online(daemon, stub_account) + handlers = len(session.client._handlers) + + await session.client.disconnect() + deadline = asyncio.get_running_loop().time() + 5.0 + while session.reconnects == 0 or session.state != SessionState.ONLINE: + if asyncio.get_running_loop().time() > deadline: + raise AssertionError("the session never reconnected") + await asyncio.sleep(0.01) + + assert len(session.client._handlers) == handlers diff --git a/tlgr/daemon/app.py b/tlgr/daemon/app.py index 49eb58f..423c41b 100644 --- a/tlgr/daemon/app.py +++ b/tlgr/daemon/app.py @@ -96,7 +96,7 @@ def __init__( self.paths, self.config, accounts=self.accounts, - on_session_ready=self._attach_handlers, + on_client=self._attach_handlers, **factory_kwargs, ) self.webhook_config = load_webhook_config(self.paths.base) @@ -141,7 +141,7 @@ def token(self) -> str | None: # -- session plumbing -------------------------------------------------- async def _attach_handlers(self, session: Any) -> None: - """Feed a newly connected account's updates into the bus. + """Feed an account's updates into the bus, once its client is built. The handler body is deliberately tiny: normalise, number, fan out. Any real work happens on a bus worker lane, because with diff --git a/tlgr/daemon/session.py b/tlgr/daemon/session.py index a7d6bd8..4718149 100644 --- a/tlgr/daemon/session.py +++ b/tlgr/daemon/session.py @@ -153,6 +153,7 @@ def __init__( client_factory: Callable[[Path, ClientOptions], Any] = build_client, on_state: Callable[[str, str, int | None], None] | None = None, on_event: Callable[[str, Any], None] | None = None, + on_client: Callable[[AccountSession], Awaitable[None]] | None = None, state_save_interval: int = 60, presence: str = "off", resync_depth: int = 50, @@ -166,6 +167,7 @@ def __init__( self._factory = client_factory self._on_state = on_state self._on_event = on_event + self._on_client = on_client self.state_save_interval = state_save_interval self.presence = presence self.resync_depth = resync_depth @@ -307,6 +309,7 @@ async def _connect_once(self) -> None: """Connect, authorise, catch up, then wait for the connection to end.""" if self.client is None: self.client = self._factory(self.session_path, self.options) + await self._client_built() self._install_hooks() await self.client.connect() @@ -371,6 +374,30 @@ async def catch_up(self) -> None: # -- hooks ------------------------------------------------------------- + async def _client_built(self) -> None: + """Hand the new client to whoever feeds its updates into the bus. + + This runs here, inside the supervisor, because the client does not + exist until the first connect attempt builds it. The daemon used to + attach its handler straight after `start()`, which only schedules the + supervisor, so it always found `client is None` and no incoming update + ever reached a job, a webhook or `watch`. The client is built once and + kept across reconnects, so this runs once per session. + """ + if self._on_client is None: + return + try: + await self._on_client(self) + except asyncio.CancelledError: + raise + except Exception as exc: + log.warning( + "could not attach update handlers for %s: %s", + self.alias, + exc, + extra={"account": self.alias}, + ) + def _install_hooks(self) -> None: compat.install_reconnect_hook(self.client, self._on_reconnect) compat.install_too_long_hook(self.client, self._on_too_long) diff --git a/tlgr/daemon/sessions.py b/tlgr/daemon/sessions.py index 4bd3e53..e51c1be 100644 --- a/tlgr/daemon/sessions.py +++ b/tlgr/daemon/sessions.py @@ -40,13 +40,13 @@ def __init__( *, accounts: AccountManager | None = None, client_factory: Any = build_client, - on_session_ready: Any = None, + on_client: Any = None, ) -> None: self.paths = paths self.config = config self.accounts = accounts or AccountManager(paths.base) self._factory = client_factory - self._on_ready = on_session_ready + self._on_client = on_client self._sessions: dict[str, AccountSession] = {} self._limiters: dict[str, RateLimiter] = {} self._locks: dict[str, asyncio.Lock] = {} @@ -133,6 +133,7 @@ def on_state(state: str, reason: str, user_id: int | None) -> None: options=self._options(alias), client_factory=self._factory, on_state=on_state, + on_client=self._on_client, state_save_interval=self.config.daemon.state_save_interval, presence=self.config.presence.mode, resync_depth=self.config.daemon.resync_depth, @@ -160,9 +161,6 @@ async def ensure(self, alias: str) -> AccountSession: session = self._make(alias) await session.start() self._sessions[alias] = session - if self._on_ready is not None: - with contextlib.suppress(Exception): - await self._on_ready(session) return session async def connect_all(self, aliases: list[str]) -> dict[str, str]: From e5a4eb018b1e32832d029368d0588f01d9b895fb Mon Sep 17 00:00:00 2001 From: Pouri Date: Sat, 3 Oct 2026 19:29:46 +0330 Subject: [PATCH 2/6] compat: install the too-long hook on a per-client MessageBox subclass MessageBox declares __slots__, so assigning apply_difference on the instance raised AttributeError on every account's first connect. The account logged 'degraded', burned one reconnect, and then ran without the hook, because the retry reuses the client and skips _install_hooks. The box now moves onto a subclass built for that client. It adds no slots, so __class__ assignment is allowed, and other clients keep the plain class. The new tests use Telethon's real MessageBox; the fake client has none, which is why this was never caught. --- tests/test_telethon_compat.py | 56 +++++++++++++++++++++++++++++++++++ tlgr/core/telethon_compat.py | 43 ++++++++++++++++----------- 2 files changed, 82 insertions(+), 17 deletions(-) create mode 100644 tests/test_telethon_compat.py diff --git a/tests/test_telethon_compat.py b/tests/test_telethon_compat.py new file mode 100644 index 0000000..617cf28 --- /dev/null +++ b/tests/test_telethon_compat.py @@ -0,0 +1,56 @@ +"""`install_too_long_hook` against Telethon's real `MessageBox`. + +The fake client has no message box, so nothing caught that `MessageBox` +declares `__slots__` and refuses instance assignment: every account logged +`AttributeError: 'MessageBox' object attribute 'apply_difference' is +read-only` on its first connect, burned one reconnect, and ran without the +hook. +""" + +from __future__ import annotations + +import contextlib +import logging + +from telethon._updates.messagebox import MessageBox +from telethon.tl import types + +from tlgr.core import telethon_compat as compat + + +class _Client: + def __init__(self) -> None: + self._message_box = MessageBox(logging.getLogger("test.messagebox")) + + +def _too_long() -> types.updates.DifferenceTooLong: + return types.updates.DifferenceTooLong(pts=10) + + +class TestTooLongHook: + def test_it_installs_on_a_real_message_box(self): + client = _Client() + assert compat.install_too_long_hook(client, lambda scope, channel: None) + assert isinstance(client._message_box, MessageBox) + + def test_it_reports_a_global_gap(self): + client = _Client() + seen: list[tuple[str, int | None]] = [] + compat.install_too_long_hook(client, lambda scope, channel: seen.append((scope, channel))) + + # Telethon's own handling of the diff is not under test. + with contextlib.suppress(Exception): + client._message_box.apply_difference(_too_long(), None) + + assert seen == [(compat.TOO_LONG_GLOBAL, None)] + + def test_one_clients_hook_does_not_fire_for_another(self): + hooked, plain = _Client(), _Client() + seen: list[str] = [] + compat.install_too_long_hook(hooked, lambda scope, channel: seen.append(scope)) + + with contextlib.suppress(Exception): + plain._message_box.apply_difference(_too_long(), None) + + assert seen == [] + assert type(plain._message_box) is MessageBox diff --git a/tlgr/core/telethon_compat.py b/tlgr/core/telethon_compat.py index 6d221cd..a4ed4a6 100644 --- a/tlgr/core/telethon_compat.py +++ b/tlgr/core/telethon_compat.py @@ -141,9 +141,14 @@ def install_too_long_hook(client: Any, callback: Callable[[str, int | None], Any Telethon consumes both inside `MessageBox` and delivers nothing to handlers, so a client that was offline long enough to blow the server's difference window silently skips history (checklist 9). The hook wraps - `MessageBox.apply_difference` / `apply_channel_difference` **on this - client's own box instance**, never on the class, so one account's hook - cannot fire for another's. + `MessageBox.apply_difference` / `apply_channel_difference` **for this + client's own box only**, so one account's hook cannot fire for another's. + + `MessageBox` declares `__slots__`, so the methods cannot be replaced on + the instance: that raised `AttributeError: ... is read-only` on every + account's first connect. Instead the box is moved onto a subclass built + for this one client. The subclass adds no slots, so the layout matches and + `__class__` assignment is allowed. `callback(scope, channel_id)` is called with `TOO_LONG_GLOBAL`/`None` or `TOO_LONG_CHANNEL`/``. @@ -153,38 +158,42 @@ def install_too_long_hook(client: Any, callback: Callable[[str, int | None], Any _warn_once("_message_box") return False - installed = False + base = type(box) + overrides: dict[str, Any] = {"__slots__": ()} - original_global = getattr(box, "apply_difference", None) + original_global = getattr(base, "apply_difference", None) if callable(original_global): - def apply_difference(diff: Any, chat_hashes: Any, _orig: Any = original_global) -> Any: + def apply_difference(self: Any, diff: Any, chat_hashes: Any) -> Any: if type(diff).__name__ == "DifferenceTooLong": _safe(callback, TOO_LONG_GLOBAL, None) - return _orig(diff, chat_hashes) + return original_global(self, diff, chat_hashes) - box.apply_difference = apply_difference - installed = True + overrides["apply_difference"] = apply_difference else: # pragma: no cover - present in 1.44 _warn_once("MessageBox.apply_difference") - original_channel = getattr(box, "apply_channel_difference", None) + original_channel = getattr(base, "apply_channel_difference", None) if callable(original_channel): - def apply_channel_difference( - request: Any, diff: Any, chat_hashes: Any, _orig: Any = original_channel - ) -> Any: + def apply_channel_difference(self: Any, request: Any, diff: Any, chat_hashes: Any) -> Any: if type(diff).__name__ == "ChannelDifferenceTooLong": channel = getattr(request, "channel", None) _safe(callback, TOO_LONG_CHANNEL, getattr(channel, "channel_id", None)) - return _orig(request, diff, chat_hashes) + return original_channel(self, request, diff, chat_hashes) - box.apply_channel_difference = apply_channel_difference - installed = True + overrides["apply_channel_difference"] = apply_channel_difference else: # pragma: no cover - present in 1.44 _warn_once("MessageBox.apply_channel_difference") - return installed + if len(overrides) == 1: + return False + try: + box.__class__ = type(base.__name__, (base,), overrides) + except TypeError as exc: + _warn_once("MessageBox subclassing", str(exc)) + return False + return True def _safe(callback: Callable[..., Any], *args: Any) -> None: From 528872844c1e3e3993be3654e25cf0b500d86bdd Mon Sep 17 00:00:00 2001 From: Pouri Date: Sat, 3 Oct 2026 19:32:06 +0330 Subject: [PATCH 3/6] gateway: build Telethon events from bus updates and resolve @chat refs, so jobs can match Two faults hid behind the dispatch bug fixed earlier: - The bus hands a job the raw TL update (UpdateNewChannelMessage), but every filter and action reads a high-level event: event.chat_id, is_private, reply(). The job now builds the event with the builder a Telethon handler would have used, attaches entities and client the way _dispatch_update does, and applies the builder's filter. That also restores NewMessage(incoming=True), so an auto-reply job no longer answers the account's own messages. - filter_chat_id skips '@name' refs on the promise that the gateway resolves them, and nothing did, so 'chat_id: "@channel"' never matched, in v1 either. setup() now resolves them through the job client, and logs a ref it cannot resolve. The fake client gains _mb_entity_cache, _self_id and parse_mode, which Telethon's events read, so the new tests drive real builders and TL updates. --- tests/fake_telethon.py | 11 ++ tests/test_gateway_bus.py | 244 ++++++++++++++++++++++++++++++++++++++ tlgr/gateway/engine.py | 136 +++++++++++++++++---- 3 files changed, 368 insertions(+), 23 deletions(-) create mode 100644 tests/test_gateway_bus.py diff --git a/tests/fake_telethon.py b/tests/fake_telethon.py index c108848..ec06535 100644 --- a/tests/fake_telethon.py +++ b/tests/fake_telethon.py @@ -1429,6 +1429,17 @@ def __init__(self, world: World | None = None, session: Any = None, **_: Any) -> self._handlers: list[tuple[Any, Any]] = [] self.disconnected: asyncio.Future[None] = asyncio.get_event_loop().create_future() self._requests = 0 + # What Telethon's events and messages read off a client: `_set_client` + # looks entities up here, `build()` takes `_self_id`, and + # `Message.text` unparses through `parse_mode` (None means plain text). + self.parse_mode = None + from telethon._updates import EntityCache + + self._mb_entity_cache = EntityCache(self_id=self.world.me.id, self_bot=False) + + @property + def _self_id(self) -> int: + return int(self.world.me.id) # -- connection -------------------------------------------------------- diff --git a/tests/test_gateway_bus.py b/tests/test_gateway_bus.py new file mode 100644 index 0000000..bc70dec --- /dev/null +++ b/tests/test_gateway_bus.py @@ -0,0 +1,244 @@ +"""A gateway job fed from the daemon's event bus, with real TL updates. + +The bus hands a job the raw TL `Update*` beside the envelope, while every +filter and action reads a high-level Telethon event (`event.chat_id`, +`event.is_private`, `event.reply()`). These tests drive a job the way the +daemon does, so a shape mismatch fails here instead of in production, where +it showed up as `matched=0` for days. +""" + +from __future__ import annotations + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +from fake_telethon import FakeTelegramClient, World +from telethon.tl import types + +from tlgr.gateway.config import _parse_job +from tlgr.gateway.engine import Gateway + +SOURCE = 1234 +SOURCE_MARKED = -1000000001234 +ALICE = 4242 + + +class _Bus: + def __init__(self) -> None: + self.handlers: list = [] + + def add_handler(self, handler) -> None: + self.handlers.append(handler) + + def remove_handler(self, handler) -> None: + self.handlers.remove(handler) + + async def deliver(self, update, *, event_type: str = "message_new", account: str = "Neo"): + envelope = SimpleNamespace(type=event_type, account=account) + for handler in list(self.handlers): + await handler(envelope, update) + + +class _JobClient: + def __init__(self, world: World, chats: dict[str, int]) -> None: + self.client = FakeTelegramClient(world) + self.client.forward_messages = AsyncMock() + self.client.send_message = AsyncMock() + self.resolve_chat = AsyncMock(side_effect=lambda ref: chats[ref]) + + +def _channel_post(text: str = "breaking", *, channel: int = SOURCE, out: bool = False): + message = types.Message( + id=5, peer_id=types.PeerChannel(channel_id=channel), date=None, message=text, out=out + ) + return types.UpdateNewChannelMessage(message=message, pts=1, pts_count=1) + + +def _private(text: str = "hi", *, out: bool = False): + message = types.Message( + id=6, + peer_id=types.PeerUser(user_id=ALICE), + from_id=None if out else types.PeerUser(user_id=ALICE), + date=None, + message=text, + out=out, + ) + return types.UpdateNewMessage(message=message, pts=1, pts_count=1) + + +async def _running(job: Gateway, bus: _Bus) -> None: + job.start() + for _ in range(100): + if bus.handlers: + return + await asyncio.sleep(0.01) + raise AssertionError("the job never subscribed to the bus") + + +@pytest.fixture +def bus() -> _Bus: + return _Bus() + + +@pytest.fixture +async def client() -> _JobClient: + return _JobClient(World(), {"@source": SOURCE_MARKED, "@archive": -1000000009999}) + + +class TestForwardFromTheBus: + async def test_a_post_in_the_watched_channel_is_forwarded(self, bus, client): + config = _parse_job( + { + "name": "archive", + "account": "Neo", + "filters": {"chat_id": "@source"}, + "actions": [{"forward": {"to": ["@archive"]}}], + } + ) + job = Gateway(config, client, bus=bus) + await _running(job, bus) + try: + await bus.deliver(_channel_post()) + finally: + await job.stop() + + assert job._stats == {"matched": 1, "skipped": 0, "errors": 0} + client.client.forward_messages.assert_awaited_once() + assert client.client.forward_messages.await_args.args[0] == -1000000009999 + + async def test_a_post_elsewhere_is_skipped(self, bus, client): + config = _parse_job( + { + "name": "archive", + "account": "Neo", + "filters": {"chat_id": "@source"}, + "actions": [{"forward": {"to": ["@archive"]}}], + } + ) + job = Gateway(config, client, bus=bus) + await _running(job, bus) + try: + await bus.deliver(_channel_post(channel=777)) + finally: + await job.stop() + + assert job._stats["matched"] == 0 + assert job._stats["skipped"] == 1 + client.client.forward_messages.assert_not_awaited() + + async def test_the_processed_copy_is_sent_with_the_rewritten_text(self, bus, client): + config = _parse_job( + { + "name": "rename", + "account": "Neo", + "filters": {"chat_id": "@source"}, + "actions": [ + { + "forward": { + "to": ["@archive"], + "processors": [ + {"type": "regex", "pattern": "wbnet", "replacement": "VPEON"} + ], + } + } + ], + } + ) + job = Gateway(config, client, bus=bus) + await _running(job, bus) + try: + await bus.deliver(_channel_post("join wbnet today")) + finally: + await job.stop() + + client.client.send_message.assert_awaited_once_with(-1000000009999, "join VPEON today") + + async def test_another_accounts_post_is_ignored(self, bus, client): + config = _parse_job( + { + "name": "archive", + "account": "Neo", + "filters": {"chat_id": "@source"}, + "actions": [{"forward": {"to": ["@archive"]}}], + } + ) + job = Gateway(config, client, bus=bus) + await _running(job, bus) + try: + await bus.deliver(_channel_post(), account="other") + finally: + await job.stop() + + assert job._stats == {"matched": 0, "skipped": 0, "errors": 0} + + +class TestReplyFromTheBus: + @pytest.fixture + def config(self): + return _parse_job( + { + "name": "away", + "account": "Neo", + "filters": {"chat_type": "private"}, + "actions": [{"reply": "away"}], + } + ) + + async def test_an_incoming_private_message_is_answered(self, bus, client, config): + client.client.send_message = AsyncMock() + job = Gateway(config, client, bus=bus) + await _running(job, bus) + try: + await bus.deliver(_private()) + finally: + await job.stop() + + assert job._stats["matched"] == 1 + client.client.send_message.assert_awaited_once() + + async def test_the_accounts_own_message_is_not_answered(self, bus, client, config): + """`new_message` means incoming, as it did when jobs used `NewMessage(incoming=True)`.""" + job = Gateway(config, client, bus=bus) + await _running(job, bus) + try: + await bus.deliver(_private(out=True)) + finally: + await job.stop() + + assert job._stats == {"matched": 0, "skipped": 0, "errors": 0} + client.client.send_message.assert_not_awaited() + + +class TestChatRefResolution: + async def test_numeric_refs_are_left_alone(self, bus, client): + config = _parse_job( + { + "name": "ids", + "account": "Neo", + "filters": {"chat_id": [SOURCE_MARKED, "@source"]}, + "actions": [{"forward": {"to": ["@archive"]}}], + } + ) + job = Gateway(config, client, bus=bus) + await job.setup() + + assert config.filters.filter_value == [SOURCE_MARKED, SOURCE_MARKED] + client.resolve_chat.assert_awaited_once_with("@source") + + async def test_an_unresolvable_ref_is_logged_and_kept(self, bus, client, caplog): + client.resolve_chat = AsyncMock(side_effect=ValueError("no such chat")) + config = _parse_job( + { + "name": "typo", + "account": "Neo", + "filters": {"chat_id": "@nope"}, + "actions": [{"forward": {"to": ["@archive"]}}], + } + ) + job = Gateway(config, client, bus=bus) + await job.setup() + + assert config.filters.filter_value == "@nope" + assert "cannot resolve chat @nope" in caplog.text diff --git a/tlgr/gateway/engine.py b/tlgr/gateway/engine.py index 3abb83e..5b65c83 100644 --- a/tlgr/gateway/engine.py +++ b/tlgr/gateway/engine.py @@ -10,9 +10,13 @@ order is preserved, the update loop is never blocked, and a job that falls behind is bounded rather than unbounded. -The pipeline itself is unchanged, and filters still read the raw Telethon -event, which is why the bus carries it beside the normalised envelope. The -full move to model-based filters belongs to the updates group (PR-4). +The pipeline itself is unchanged, and filters still read a high-level +Telethon event (`NewMessage.Event` and friends). The bus carries the raw TL +update beside the normalised envelope, so the job builds that event itself +with the same builder a Telethon handler would have used, including +`NewMessage(incoming=True)`, so a job never answers the account's own +messages. The full move to model-based filters belongs to the updates group +(PR-4). Without a bus — a unit test, or a daemon that has not started one — the job falls back to registering Telethon handlers exactly as v1 did. @@ -21,12 +25,14 @@ from __future__ import annotations import asyncio +import inspect import logging +from typing import Any from telethon import events from tlgr.actions import get_action -from tlgr.filters.compose import evaluate +from tlgr.filters.compose import FilterNode, Op, evaluate from tlgr.gateway.config import ActionConfig, GatewayConfig from tlgr.gateway.event import Event from tlgr.jobs.base import BaseJob @@ -49,8 +55,8 @@ def __init__(self, gw: GatewayConfig) -> None: self.account = gw.account -#: v1's job event names → the bus taxonomy, and back for the pipeline, which -#: still labels envelopes with v1's names. The expansion is the taxonomy's own +#: v1's job event names → the bus taxonomy. The pipeline still labels +#: envelopes with v1's names, taken from the builder that matched. The expansion is the taxonomy's own #: alias table (`core.eventtypes.ALIASES`), so a job and a `watch` accept the #: same words; a job may also name any v2 type directly. def _bus_types(names: list[str]) -> set[str]: @@ -62,16 +68,6 @@ def _bus_types(names: list[str]) -> set[str]: return wanted -_V1_TYPE_MAP = { - "message_new": "new_message", - "message_edited": "message_edited", - "message_deleted": "message_deleted", - "message_service": "chat_action", - "user_status": "user_joined", - "read_inbox": "message_read", - "read_outbox": "message_read", -} - _EVENT_TYPE_MAP = { "new_message": (events.NewMessage, {}), "message_edited": (events.MessageEdited, {}), @@ -82,6 +78,49 @@ def _bus_types(names: list[str]) -> set[str]: } +def _builder_for(event_type_name: str) -> Any | None: + mapping = _EVENT_TYPE_MAP.get(event_type_name) + if not mapping: + return None + event_cls, kwargs = mapping + if event_type_name == "new_message": + kwargs = {"incoming": True} + return event_cls(**kwargs) + + +def _builders(names: list[str]) -> list[tuple[str, Any]]: + """One Telethon event builder per v1 event name, in the job's order.""" + out: list[tuple[str, Any]] = [] + for name in names: + builder = _builder_for(name) + if builder is not None: + out.append((name, builder)) + return out + + +def _needs_resolving(ref: Any) -> bool: + if not isinstance(ref, str): + return False + try: + int(ref) + except ValueError: + return True + return False + + +def _chat_id_leaves(nodes: list[FilterNode | None]) -> list[FilterNode]: + leaves: list[FilterNode] = [] + stack = [node for node in nodes if node is not None] + while stack: + node = stack.pop() + if node.op is Op.LEAF: + if node.filter_name == "chat_id": + leaves.append(node) + else: + stack.extend(node.children) + return leaves + + class Gateway(BaseJob): """Generic pipeline job: filters -> processors -> actions.""" @@ -101,6 +140,7 @@ def __init__( self._stats: dict[str, int] = {"matched": 0, "skipped": 0, "errors": 0} async def setup(self) -> None: + await self._resolve_chat_refs() log.info( "[%s] events=%s filters=%s actions=%s", self.name, @@ -109,6 +149,28 @@ async def setup(self) -> None: [a.name for a in self._gw.actions], ) + async def _resolve_chat_refs(self) -> None: + """Turn every `chat_id: "@name"` into the marked id the filter compares. + + `filter_chat_id` compares `event.chat_id`, an int, and skips string + refs on the promise that the gateway resolves them first. Nothing + did, so a job filtered on `@channel` could never match. + """ + nodes = [self._gw.filters] + [action.filters for action in self._gw.actions] + for leaf in _chat_id_leaves(nodes): + refs = leaf.filter_value if isinstance(leaf.filter_value, list) else [leaf.filter_value] + resolved: list = [] + for ref in refs: + if not _needs_resolving(ref): + resolved.append(ref) + continue + try: + resolved.append(await self.client.resolve_chat(str(ref))) + except Exception as exc: + log.warning("[%s] cannot resolve chat %s: %s", self.name, ref, exc) + resolved.append(ref) + leaf.filter_value = resolved if isinstance(leaf.filter_value, list) else resolved[0] + async def run(self) -> None: if self._bus is not None: await self._run_on_bus() @@ -119,6 +181,7 @@ async def _run_on_bus(self) -> None: """Subscribe to the daemon's bus instead of the update loop (ROB-02).""" wanted = _bus_types(list(self._gw.events)) account = self._gw.account + builders = _builders(list(self._gw.events)) async def on_event(envelope, raw) -> None: if envelope.type not in wanted: @@ -129,7 +192,11 @@ async def on_event(envelope, raw) -> None: # A self-origin echo or a synthesised event: the filters read # the raw Telethon object, so there is nothing to evaluate. return - await self._handle(raw, _V1_TYPE_MAP.get(envelope.type, envelope.type)) + built = await self._build(builders, raw) + if built is None: + return + tg_event, event_type = built + await self._handle(tg_event, event_type) self._bus_handler = on_event self._bus.add_handler(on_event) @@ -138,19 +205,42 @@ async def on_event(envelope, raw) -> None: except asyncio.CancelledError: raise + async def _build(self, builders: list[tuple[str, Any]], update: Any) -> tuple[Any, str] | None: + """The high-level event Telethon would have handed a handler, or None. + + This mirrors `EventBuilderDict` in Telethon's `_dispatch_update`: build, + attach the update's entities and the client, then apply the builder's + own filter, which is where `incoming=True` drops outgoing messages. + """ + client = self.client.client + self_id = getattr(client, "_self_id", None) + for event_type, builder in builders: + event = builder.build(update, None, self_id) + if not event: + continue + if isinstance(event, events.common.EventCommon): + event.original_update = update + event._entities = getattr(update, "_entities", None) or {} + event._set_client(client) + if not builder.resolved: + await builder.resolve(client) + passed = builder.filter(event) + if inspect.isawaitable(passed): + passed = await passed + if passed: + return event, event_type + return None + async def _run_on_client(self) -> None: for event_type_name in self._gw.events: - mapping = _EVENT_TYPE_MAP.get(event_type_name) - if not mapping: + builder = _builder_for(event_type_name) + if builder is None: log.warning("[%s] unknown event type: %s", self.name, event_type_name) continue - event_cls, kwargs = mapping - if event_type_name == "new_message": - kwargs = {"incoming": True} et = event_type_name - @self.client.client.on(event_cls(**kwargs)) + @self.client.client.on(builder) async def handler(tg_event, _et=et): await self._handle(tg_event, _et) From eeeb0996370a70e20f82455422e067128ae09e2d Mon Sep 17 00:00:00 2001 From: Pouri Date: Sat, 3 Oct 2026 19:33:31 +0330 Subject: [PATCH 4/6] daemon: start the jobs in jobs.yaml at boot, and retry a chat ref that failed while offline reload_jobs() ran only on 'tlgr job reload', so every daemon restart left the gateway with no jobs until someone noticed. On the live install that was from 2026-09-30 to 2026-10-03, across seven restarts. run() now starts the jobs in a background task once accounts are connected, so readiness does not wait for a job's account, and shutdown cancels it. Because a job can now start while its account is still offline, a '@chat' ref that fails to resolve is retried from the event path, at most once a minute, instead of leaving the job unable to match for good. --- tests/test_daemon_jobs.py | 70 +++++++++++++++++++++++++++++++++++++++ tests/test_gateway_bus.py | 30 +++++++++++++++++ tlgr/daemon/app.py | 30 ++++++++++++++--- tlgr/gateway/engine.py | 22 ++++++++++-- 4 files changed, 146 insertions(+), 6 deletions(-) create mode 100644 tests/test_daemon_jobs.py diff --git a/tests/test_daemon_jobs.py b/tests/test_daemon_jobs.py new file mode 100644 index 0000000..073d4a4 --- /dev/null +++ b/tests/test_daemon_jobs.py @@ -0,0 +1,70 @@ +"""The daemon starts the jobs in `jobs.yaml` when it boots. + +Until 2.0.1 it did not: `reload_jobs()` ran only on `tlgr job reload`, so a +restarted daemon (launchd, an upgrade, a crash) ran no jobs at all and said +nothing about it. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +from fake_telethon import fake_client_factory + +JOBS = """\ +jobs: + - name: archive + account: {account} + actions: + - forward: + to: ["@archive"] + - name: parked + account: {account} + enabled: false + actions: + - reply: "parked" +""" + + +async def _until(predicate, timeout: float = 5.0) -> None: + deadline = asyncio.get_running_loop().time() + timeout + while not predicate(): + if asyncio.get_running_loop().time() > deadline: + raise AssertionError("condition never became true") + await asyncio.sleep(0.02) + + +class TestJobsAtBoot: + async def test_enabled_jobs_are_running_once_the_daemon_is_up( + self, tlgr_home: Path, stub_account: str, world + ): + from tlgr.daemon.app import Daemon + + (tlgr_home / "jobs.yaml").write_text(JOBS.format(account=stub_account)) + daemon = Daemon(tlgr_home, client_factory=fake_client_factory(world)) + runner = asyncio.create_task(daemon.run()) + try: + await _until(lambda: any(job.get("running") for job in daemon.list_jobs())) + jobs = {job["name"]: job for job in daemon.list_jobs()} + assert jobs["archive"]["running"] is True + assert "parked" not in jobs + finally: + daemon.request_shutdown() + await asyncio.wait_for(runner, timeout=10) + + async def test_a_broken_jobs_file_does_not_stop_the_daemon( + self, tlgr_home: Path, stub_account: str, world + ): + from tlgr.daemon.app import Daemon + + (tlgr_home / "jobs.yaml").write_text("jobs: [unclosed") + daemon = Daemon(tlgr_home, client_factory=fake_client_factory(world)) + runner = asyncio.create_task(daemon.run()) + try: + await _until(lambda: daemon.ready) + await _until(lambda: daemon._jobs_task is not None and daemon._jobs_task.done()) + assert daemon.list_jobs() == [] + finally: + daemon.request_shutdown() + await asyncio.wait_for(runner, timeout=10) diff --git a/tests/test_gateway_bus.py b/tests/test_gateway_bus.py index bc70dec..b3735f2 100644 --- a/tests/test_gateway_bus.py +++ b/tests/test_gateway_bus.py @@ -242,3 +242,33 @@ async def test_an_unresolvable_ref_is_logged_and_kept(self, bus, client, caplog) assert config.filters.filter_value == "@nope" assert "cannot resolve chat @nope" in caplog.text + + async def test_a_ref_that_failed_at_boot_is_retried_when_events_arrive(self, bus, client): + """An account still offline when the job starts must not leave it deaf for good.""" + answers = [ValueError("offline"), SOURCE_MARKED] + + async def resolve(ref): + answer = answers.pop(0) + if isinstance(answer, Exception): + raise answer + return answer + + client.resolve_chat = AsyncMock(side_effect=resolve) + config = _parse_job( + { + "name": "archive", + "account": "Neo", + "filters": {"chat_id": "@source"}, + "actions": [{"forward": {"to": [-1000000009999]}}], + } + ) + job = Gateway(config, client, bus=bus) + await _running(job, bus) + try: + job._next_resolve = 0.0 # the retry interval has passed + await bus.deliver(_channel_post()) + finally: + await job.stop() + + assert job._stats["matched"] == 1 + assert config.filters.filter_value == SOURCE_MARKED diff --git a/tlgr/daemon/app.py b/tlgr/daemon/app.py index 423c41b..bb6e68c 100644 --- a/tlgr/daemon/app.py +++ b/tlgr/daemon/app.py @@ -119,6 +119,7 @@ def __init__( self._start_time = time.time() self._runner: web.AppRunner | None = None self._idle_task: asyncio.Task[None] | None = None + self._jobs_task: asyncio.Task[None] | None = None self._token: str | None = None self.idle_timeout = effective_idle_timeout( self.config.daemon.idle_timeout, @@ -358,9 +359,29 @@ async def run(self) -> None: extra={"count": len(results)}, ) self._idle_task = asyncio.create_task(self._idle_monitor(), name="tlgr-idle") + self._jobs_task = asyncio.create_task(self._start_jobs(), name="tlgr-jobs") await self._shutdown_event.wait() await self.shutdown() + async def _start_jobs(self) -> None: + """Load `jobs.yaml` and start every enabled job, once accounts are up. + + Without this a restarted daemon ran no jobs until someone ran + `tlgr job reload`, which is how the gateway sat idle from a restart + on 2026-09-30 until it was noticed. It runs in the background because + each job may wait for its account to come online, and readiness must + not wait for that. + """ + try: + result = await self.reload_jobs() + except asyncio.CancelledError: + raise + except Exception: + log.exception("could not start jobs from jobs.yaml") + return + if result.get("added"): + log.info("started %d job(s) from jobs.yaml", len(result["added"])) + async def _idle_monitor(self) -> None: while not self._shutdown_event.is_set(): await asyncio.sleep(5) @@ -385,10 +406,11 @@ async def shutdown(self, *, drain: float | None = None) -> None: drain if drain is not None else self.config.daemon.drain_seconds ) - if self._idle_task is not None: - self._idle_task.cancel() - with contextlib.suppress(asyncio.CancelledError, Exception): - await self._idle_task + for task in (self._idle_task, self._jobs_task): + if task is not None: + task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await task # Wait for in-flight requests rather than cancelling them: a ten # minute scan that is killed at second 599 has cost the account the diff --git a/tlgr/gateway/engine.py b/tlgr/gateway/engine.py index 5b65c83..b6447f2 100644 --- a/tlgr/gateway/engine.py +++ b/tlgr/gateway/engine.py @@ -27,6 +27,7 @@ import asyncio import inspect import logging +import time from typing import Any from telethon import events @@ -40,6 +41,10 @@ log = logging.getLogger("tlgr.gateway") +#: How long a job waits before retrying a `@chat` ref it could not resolve, +#: typically because its account was still offline when the job started. +_RESOLVE_RETRY_SECONDS = 60.0 + class _GatewayJobConfig: """Minimal shim so Gateway can sit on top of BaseJob. @@ -138,6 +143,9 @@ def __init__( self._bus = bus self._bus_handler = None self._stats: dict[str, int] = {"matched": 0, "skipped": 0, "errors": 0} + self._refs_pending = False + self._next_resolve = 0.0 + self._unresolved_logged: set[str] = set() async def setup(self) -> None: await self._resolve_chat_refs() @@ -154,9 +162,13 @@ async def _resolve_chat_refs(self) -> None: `filter_chat_id` compares `event.chat_id`, an int, and skips string refs on the promise that the gateway resolves them first. Nothing - did, so a job filtered on `@channel` could never match. + did, so a job filtered on `@channel` could never match. A ref that + fails (the account is offline at boot, say) stays a string and is + retried from `_handle`, at most once a minute. """ nodes = [self._gw.filters] + [action.filters for action in self._gw.actions] + self._refs_pending = False + self._next_resolve = time.monotonic() + _RESOLVE_RETRY_SECONDS for leaf in _chat_id_leaves(nodes): refs = leaf.filter_value if isinstance(leaf.filter_value, list) else [leaf.filter_value] resolved: list = [] @@ -167,7 +179,10 @@ async def _resolve_chat_refs(self) -> None: try: resolved.append(await self.client.resolve_chat(str(ref))) except Exception as exc: - log.warning("[%s] cannot resolve chat %s: %s", self.name, ref, exc) + level = logging.DEBUG if ref in self._unresolved_logged else logging.WARNING + self._unresolved_logged.add(ref) + log.log(level, "[%s] cannot resolve chat %s: %s", self.name, ref, exc) + self._refs_pending = True resolved.append(ref) leaf.filter_value = resolved if isinstance(leaf.filter_value, list) else resolved[0] @@ -267,6 +282,9 @@ async def teardown(self) -> None: ) async def _handle(self, tg_event, event_type: str = "new_message") -> None: + if self._refs_pending and time.monotonic() >= self._next_resolve: + await self._resolve_chat_refs() + envelope = Event( source="telegram", raw=tg_event, From 1a66a25527ed96d75012f634079da884c9b023e0 Mon Sep 17 00:00:00 2001 From: Pouri Date: Sat, 3 Oct 2026 19:34:00 +0330 Subject: [PATCH 5/6] job list: always report enabled and running, so the table stops printing '-' for both JobState inherited omit_defaults, so an enabled job dropped 'enabled' (True is the default) and a stopped one dropped 'running' (False is the default). The table renders a missing field as '-', so an enabled, stopped job looked exactly like a disabled one. Same fix and reasoning as DaemonStatus. --- tests/test_ops_daemon.py | 11 +++++++++++ tlgr/models/daemon.py | 9 +++++++-- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/tests/test_ops_daemon.py b/tests/test_ops_daemon.py index 7f34452..f801351 100644 --- a/tests/test_ops_daemon.py +++ b/tests/test_ops_daemon.py @@ -520,6 +520,17 @@ async def test_an_unclaimed_prefix_is_not_found(self, live_daemon, client, in_th class TestJobs: + def test_a_job_row_always_says_whether_it_is_enabled_and_running(self): + """With `omit_defaults`, an enabled, stopped job printed `-` in both columns.""" + import msgspec + + from tlgr.models.daemon import JobState + + row = msgspec.to_builtins(JobState(name="archive", enabled=True, running=False)) + assert row["enabled"] is True + assert row["running"] is False + assert row["matched"] == 0 + def test_a_job_is_added_from_flags(self, tlgr_home): added = local( "job.add", diff --git a/tlgr/models/daemon.py b/tlgr/models/daemon.py index 8eeba03..20564d5 100644 --- a/tlgr/models/daemon.py +++ b/tlgr/models/daemon.py @@ -250,8 +250,13 @@ class DeadLetterResult(Model): dry_run: bool = False -class JobState(Model): - """One gateway job.""" +class JobState(Model, omit_defaults=False): + """One gateway job. + + `omit_defaults=False` because `enabled` and `running` are the answer: + with the default, an enabled job dropped `enabled` and a stopped one + dropped `running`, and `job list` printed `-` for both. + """ name: str account: str = "" From 644f5795f87817665a82d4f7b9f056eb5e223d25 Mon Sep 17 00:00:00 2001 From: Pouri Date: Sat, 3 Oct 2026 19:34:20 +0330 Subject: [PATCH 6/6] changelog: gateway jobs, update dispatch, MessageBox hook and job list fixes --- CHANGELOG.md | 29 +++++++++++++++++++++++++++++ 1 file changed, 29 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4820443..4806cff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -31,6 +31,35 @@ codes documented in `AGENT.md` are the public API. ### Fixed +- **Gateway jobs work again, and incoming updates reach the daemon at all.** + Four faults stacked up, and each one alone was enough to stop every job: + + - The daemon attached its update handler before the account's client + existed, so no incoming update reached the event bus. Jobs, webhooks and + `watch` saw the daemon's own sends and nothing else. Every job stopped + with `matched=0 skipped=0`. + - The bus hands a job the raw TL update, but filters and actions read a + Telethon event (`chat_id`, `is_private`, `reply()`). The job now builds + that event the way a Telethon handler would, including + `NewMessage(incoming=True)`, so an auto-reply never answers the + account's own messages. + - `filter_chat_id` skipped `@name` refs, expecting the gateway to resolve + them, and nothing did, so `chat_id: "@channel"` never matched. Jobs now + resolve them at start, and retry once a minute if the account was + offline. + - A restarted daemon started no jobs until `tlgr job reload`. It now + starts them at boot. + +- **Accounts no longer go "degraded" on every start** with `AttributeError: + 'MessageBox' object attribute 'apply_difference' is read-only`. Telethon's + `MessageBox` declares `__slots__`, so the too-long hook now installs on a + per-client subclass, and the account no longer burns a reconnect or runs + without the hook. + +- **`job list` reports `enabled` and `running` even when they are at their + defaults.** Before, an enabled job that was not running showed `-` in both + columns, the same as a disabled one. + - **`chat list` no longer loses a block of dialogs at a page boundary.** The dialog walk paired each row with its top message by message id alone, but only private chats share one id space: every channel and supergroup numbers