diff --git a/cron/scheduler.py b/cron/scheduler.py index 6c26efb8624..39f4ed82c00 100644 --- a/cron/scheduler.py +++ b/cron/scheduler.py @@ -31,7 +31,7 @@ except ImportError: except ImportError: msvcrt = None from pathlib import Path -from typing import List, Optional +from typing import Any, List, Optional # Add parent directory to path for imports BEFORE repo-level imports. # Without this, standalone invocations (e.g. after `hermes update` reloads @@ -1217,6 +1217,62 @@ def _confirm_adapter_delivery(send_result) -> bool: return bool(getattr(send_result, "success")) +def _is_channel_dm_topic( + runtime_adapter: Any, + chat_id: Any, + loop: Any, + job_id: str, +) -> bool: + """Decide whether an (already-ambiguous) Telegram topic target is a genuine + Bot API *channel* Direct-Messages topic (route via + ``direct_messages_topic_id``) rather than a forum-style topic in a private + chat (route via ``message_thread_id``). + + Callers gate this on the ambiguous shape first + (``telegram::``) — that shape is + identical for both cases, so shape alone cannot decide (this was the #52060 + regression). The real signal is the chat *type*: a genuine channel DM topic + lives on a ``channel`` chat. Probe the live adapter's ``get_chat_info`` once + and only return True when the chat is a channel. + + Fails SAFE to ``message_thread_id`` (returns False) for adapters without a + probe, or any probe error/timeout — that is the pre-#22773 behaviour and the + correct default for the common forum-topic case. + """ + # Resolve on the CLASS, not the instance (general pitfall #11): a MagicMock + # instance auto-creates a truthy ``get_chat_info`` attribute, so an + # instance-level probe would misclassify test doubles. Real adapters expose + # the coroutine on the class regardless. + get_chat_info = getattr(type(runtime_adapter), "get_chat_info", None) + if not callable(get_chat_info): + return False + try: + from agent.async_utils import safe_schedule_threadsafe + + future = safe_schedule_threadsafe( + get_chat_info(runtime_adapter, str(chat_id)), loop, # type: ignore[arg-type] + ) + if future is None: + return False + # Lighter than a send (metadata-only Bot API call), so a shorter bound + # than the 30s/60s send waits elsewhere in this file is intentional. + info = future.result(timeout=10) + except Exception: + logger.debug( + "Job '%s': get_chat_info probe failed for chat=%s — " + "defaulting to message_thread_id routing", + job_id, chat_id, exc_info=True, + ) + return False + is_channel = isinstance(info, dict) and str(info.get("type") or "").lower() == "channel" + if is_channel: + logger.info( + "Job '%s': chat=%s is a channel — routing via direct_messages_topic_id", + job_id, chat_id, + ) + return is_channel + + def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Optional[str]: """ Deliver job output to the configured target(s) (origin chat, specific platform, etc.). @@ -1440,14 +1496,16 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option opened_thread_id = new_thread_id if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)(): - # Telegram three-mode topic routing (#22773): a private chat - # (positive chat_id) with a NUMERIC topic id is a Bot API Direct - # Messages topic and must be addressed via ``direct_messages_topic_id`` - # — a bare ``message_thread_id`` is rejected/mis-routed by Bot API - # 10.0 and lands in General. Forum/supergroup targets (negative - # chat_id) and named DM-topic lanes keep the default thread_id - # handling. Compute the routed metadata ONCE so both the text send - # (via DeliveryRouter) and the media send use the same routing. + # Telegram topic routing (#22773, regression fixed #52060): a + # ``telegram::`` cron target is + # ambiguous — a forum-style topic in a private chat and a genuine + # Bot API channel Direct-Messages topic share the same shape and + # need OPPOSITE routing. Disambiguate at delivery time via + # ``_is_channel_dm_topic`` (see its docstring for the full + # rationale); ``thread_id`` goes in ``route_metadata`` so the + # anchorless cron send bypasses the DeliveryRouter's private-chat + # reply-anchor requirement. Compute the routed metadata ONCE so both + # the text send (via DeliveryRouter) and the media send agree. from gateway.delivery import ( DeliveryRouter, DeliveryTarget, @@ -1455,14 +1513,18 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option looks_like_telegram_private_chat_id, ) - is_private_dm_topic = ( + is_ambiguous_telegram_topic = ( platform == Platform.TELEGRAM and thread_id is not None and looks_like_telegram_private_chat_id(str(chat_id)) and _looks_like_int(str(thread_id)) ) - if is_private_dm_topic: - # Routed via direct_messages_topic_id (mode 2), no bare thread_id. + route_via_dm_topic = is_ambiguous_telegram_topic and _is_channel_dm_topic( + runtime_adapter, chat_id, loop, job["id"], + ) + if route_via_dm_topic: + # Genuine Bot API channel Direct-Messages topic (#22773 mode 2): + # routed via direct_messages_topic_id, no bare thread_id. route_thread_id = None route_metadata = { "direct_messages_topic_id": str(thread_id), @@ -1472,8 +1534,18 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option # the same DM topic instead of the General lane (#22773). media_metadata = {"direct_messages_topic_id": str(thread_id)} else: + # Forum-style topic (private chat / supergroup) or non-topic + # target: route via message_thread_id (#52060). Put thread_id in + # *route_metadata* (not just the DeliveryTarget) deliberately — + # the DeliveryRouter's private-chat topic detection + # (gateway/delivery.py) demands a reply anchor when thread_id is + # absent from metadata; cron deliveries have no inbound reply + # anchor, so the metadata key bypasses that check and lets the + # adapter route via a plain message_thread_id. route_thread_id = str(thread_id) if thread_id is not None else None route_metadata = {"job_id": job["id"]} + if route_thread_id: + route_metadata["thread_id"] = route_thread_id media_metadata = {"thread_id": thread_id} if thread_id else None try: diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py index 54445e7054c..104147cf471 100644 --- a/tests/cron/test_scheduler.py +++ b/tests/cron/test_scheduler.py @@ -3296,35 +3296,40 @@ class TestDeliverResultTimeoutCancelsFuture: standalone_send.assert_awaited_once() assert result is None, f"standalone should have delivered, got: {result!r}" - def test_live_adapter_private_dm_topic_routes_via_direct_messages_topic_id(self): - """#22773: a cron target to a PRIVATE Telegram chat with a numeric topic - id must be routed via ``direct_messages_topic_id`` (Bot API DM topics), - NOT a bare ``message_thread_id`` (which Bot API 10.0 rejects / mis-routes - to General). The cron live-adapter path routes through the gateway - DeliveryRouter, which applies the same three-mode routing as live - messages. + def test_live_adapter_forum_topic_in_private_chat_routes_via_message_thread_id(self): + """#52060: a cron target to a PRIVATE Telegram chat with a numeric topic + id is a normal forum-style topic — it must route via ``message_thread_id``, + NOT ``direct_messages_topic_id``. The #22773 heuristic inferred a Bot API + channel DM topic from positive chat_id + numeric thread and nulled + ``message_thread_id``, so deliveries landed in General. We now probe the + live adapter's ``get_chat_info``; a non-channel chat routes via + ``message_thread_id``. """ from gateway.config import Platform from gateway.platforms.base import SendResult from concurrent.futures import Future send_result = SendResult(success=True, message_id="42") - adapter = MagicMock() + + class _ForumAdapter(MagicMock): + async def get_chat_info(self, chat_id): + return {"name": "Proyectos", "type": "forum", "is_forum": True} + + adapter = _ForumAdapter() adapter.send = AsyncMock(return_value=send_result) pconfig = MagicMock() pconfig.enabled = True mock_cfg = MagicMock() mock_cfg.platforms = {Platform.TELEGRAM: pconfig} - # DeliveryRouter consults the silence-narration config flag. mock_cfg.filter_silence_narration = False loop = MagicMock() loop.is_running.return_value = True job = { - "id": "dm-topic-job", - "deliver": "telegram:226252250:7072", # private chat + numeric topic + "id": "forum-topic-job", + "deliver": "telegram:226252250:7072", # private chat + numeric forum topic } def fake_run_coro(coro, _loop): @@ -3352,16 +3357,256 @@ class TestDeliverResultTimeoutCancelsFuture: sent_metadata = adapter.send.call_args[1]["metadata"] assert sent_chat_id == "226252250" assert sent_text == "Hello world" - # The topic must be addressed via direct_messages_topic_id, and a bare - # message_thread_id must NOT be set (that is the Bot API 10.0 bug). + # Forum topics route via message_thread_id (thread_id in metadata), NOT + # direct_messages_topic_id. + assert not sent_metadata.get("direct_messages_topic_id") + assert str(sent_metadata.get("thread_id")) == "7072" + + def test_live_adapter_ambiguous_topic_probe_failure_falls_back_to_message_thread_id(self): + """Fail SAFE: when the ``get_chat_info`` probe cannot resolve the chat + type (adapter with no usable probe / raising probe), an ambiguous + private-chat topic target defaults to ``message_thread_id`` — the common + forum-topic case and pre-#22773 behaviour, never the DM-topic route. + """ + from gateway.config import Platform + from gateway.platforms.base import SendResult + from concurrent.futures import Future + + send_result = SendResult(success=True, message_id="42") + + # Plain MagicMock: its auto-created get_chat_info returns a MagicMock, + # not an awaitable, so the scheduled coroutine raises and the probe + # fails closed to message_thread_id. + adapter = MagicMock() + adapter.send = AsyncMock(return_value=send_result) + + pconfig = MagicMock() + pconfig.enabled = True + mock_cfg = MagicMock() + mock_cfg.platforms = {Platform.TELEGRAM: pconfig} + mock_cfg.filter_silence_narration = False + + loop = MagicMock() + loop.is_running.return_value = True + + job = { + "id": "probe-fail-job", + "deliver": "telegram:226252250:7072", + } + + def fake_run_coro(coro, _loop): + import asyncio as _asyncio + future = Future() + try: + future.set_result(_asyncio.run(coro)) + except BaseException as _e: # noqa: BLE001 + future.set_exception(_e) + return future + + with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro): + result = _deliver_result( + job, + "Hello world", + adapters={Platform.TELEGRAM: adapter}, + loop=loop, + ) + + assert result is None, f"expected clean delivery, got: {result!r}" + adapter.send.assert_called_once() + sent_metadata = adapter.send.call_args[1]["metadata"] + assert not sent_metadata.get("direct_messages_topic_id") + assert str(sent_metadata.get("thread_id")) == "7072" + + def test_live_adapter_probe_returns_none_falls_back_to_message_thread_id(self): + """Fail SAFE when the probe yields a non-dict result. A relay/proxy + adapter (or a future ``get_chat_info`` variant) may return ``None`` + rather than a dict; the ``isinstance(info, dict)`` guard must still route + via ``message_thread_id``, distinct from the raising-probe path. + + (The real Telegram adapter returns a dict on every path — a + ``type="dm"`` dict with an ``error`` key on failure, covered separately + by ``..._adapter_error_dict_falls_back...`` — never ``None``. This test + locks the non-dict defensive branch for other adapters.)""" + from gateway.config import Platform + from gateway.platforms.base import SendResult + from concurrent.futures import Future + + send_result = SendResult(success=True, message_id="42") + + class _NoneProbeAdapter(MagicMock): + async def get_chat_info(self, chat_id): + return None + + adapter = _NoneProbeAdapter() + adapter.send = AsyncMock(return_value=send_result) + + pconfig = MagicMock() + pconfig.enabled = True + mock_cfg = MagicMock() + mock_cfg.platforms = {Platform.TELEGRAM: pconfig} + mock_cfg.filter_silence_narration = False + + loop = MagicMock() + loop.is_running.return_value = True + + job = { + "id": "none-probe-job", + "deliver": "telegram:226252250:7072", + } + + def fake_run_coro(coro, _loop): + import asyncio as _asyncio + future = Future() + try: + future.set_result(_asyncio.run(coro)) + except BaseException as _e: # noqa: BLE001 + future.set_exception(_e) + return future + + with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro): + result = _deliver_result( + job, + "Hello world", + adapters={Platform.TELEGRAM: adapter}, + loop=loop, + ) + + assert result is None, f"expected clean delivery, got: {result!r}" + adapter.send.assert_called_once() + sent_metadata = adapter.send.call_args[1]["metadata"] + assert not sent_metadata.get("direct_messages_topic_id") + assert str(sent_metadata.get("thread_id")) == "7072" + + def test_live_adapter_error_dict_falls_back_to_message_thread_id(self): + """Fail SAFE on the REAL Telegram adapter error contract: on a failed + ``get_chat.get_chat`` the adapter returns ``{"type": "dm", "error": ...}`` + (plugins/platforms/telegram/adapter.py::get_chat_info), NOT ``None`` and + NOT a raise. A ``type="dm"`` (or bot-missing ``{"type": "dm"}``) result + must route via ``message_thread_id`` — only a genuine ``type="channel"`` + gets ``direct_messages_topic_id``. This locks the exact dict shape + production emits so a forum-topic cron never mis-routes to General.""" + from gateway.config import Platform + from gateway.platforms.base import SendResult + from concurrent.futures import Future + + send_result = SendResult(success=True, message_id="42") + + class _ErrorDictAdapter(MagicMock): + async def get_chat_info(self, chat_id): + # Mirrors the real adapter's except-branch return shape. + return {"name": str(chat_id), "type": "dm", "error": "Chat not found"} + + adapter = _ErrorDictAdapter() + adapter.send = AsyncMock(return_value=send_result) + + pconfig = MagicMock() + pconfig.enabled = True + mock_cfg = MagicMock() + mock_cfg.platforms = {Platform.TELEGRAM: pconfig} + mock_cfg.filter_silence_narration = False + + loop = MagicMock() + loop.is_running.return_value = True + + job = { + "id": "error-dict-job", + "deliver": "telegram:226252250:7072", + } + + def fake_run_coro(coro, _loop): + import asyncio as _asyncio + future = Future() + try: + future.set_result(_asyncio.run(coro)) + except BaseException as _e: # noqa: BLE001 + future.set_exception(_e) + return future + + with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro): + result = _deliver_result( + job, + "Hello world", + adapters={Platform.TELEGRAM: adapter}, + loop=loop, + ) + + assert result is None, f"expected clean delivery, got: {result!r}" + adapter.send.assert_called_once() + sent_metadata = adapter.send.call_args[1]["metadata"] + assert not sent_metadata.get("direct_messages_topic_id") + assert str(sent_metadata.get("thread_id")) == "7072" + + def test_live_adapter_channel_dm_topic_routes_via_direct_messages_topic_id(self): + """#22773 (done right): a genuine Bot API 10.0 *channel* Direct-Messages + topic must be routed via ``direct_messages_topic_id`` (a bare + ``message_thread_id`` is rejected / mis-routed there). We recognise it + from the real runtime signal — ``get_chat_info`` reports the chat as a + ``channel`` — not from a positive-chat-id + numeric-thread guess. + """ + from gateway.config import Platform + from gateway.platforms.base import SendResult + from concurrent.futures import Future + + send_result = SendResult(success=True, message_id="42") + + class _ChannelAdapter(MagicMock): + async def get_chat_info(self, chat_id): + return {"name": "My Channel", "type": "channel", "is_forum": False} + + adapter = _ChannelAdapter() + adapter.send = AsyncMock(return_value=send_result) + + pconfig = MagicMock() + pconfig.enabled = True + mock_cfg = MagicMock() + mock_cfg.platforms = {Platform.TELEGRAM: pconfig} + mock_cfg.filter_silence_narration = False + + loop = MagicMock() + loop.is_running.return_value = True + + job = { + "id": "channel-dm-topic-job", + "deliver": "telegram:226252250:7072", + } + + def fake_run_coro(coro, _loop): + import asyncio as _asyncio + future = Future() + try: + future.set_result(_asyncio.run(coro)) + except BaseException as _e: # noqa: BLE001 + future.set_exception(_e) + return future + + with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro): + result = _deliver_result( + job, + "Hello world", + adapters={Platform.TELEGRAM: adapter}, + loop=loop, + ) + + assert result is None, f"expected clean delivery, got: {result!r}" + adapter.send.assert_called_once() + sent_metadata = adapter.send.call_args[1]["metadata"] + # Genuine channel DM topic routes via direct_messages_topic_id, no bare + # message_thread_id. assert str(sent_metadata.get("direct_messages_topic_id")) == "7072" assert not sent_metadata.get("message_thread_id") - def test_live_adapter_private_dm_topic_media_routes_via_direct_messages_topic_id(self, tmp_path, monkeypatch): - """#22773 (media): MEDIA attachments to a private DM topic must also be - routed via ``direct_messages_topic_id``, not a bare ``message_thread_id`` - — the media path previously used the bare thread_id and landed - attachments in the General lane.""" + def test_live_adapter_forum_topic_media_routes_via_message_thread_id(self, tmp_path, monkeypatch): + """#52060 (media): MEDIA attachments to a forum-style topic in a private + chat must also route via ``thread_id`` (message_thread_id), not + ``direct_messages_topic_id``.""" from gateway.config import Platform from gateway.platforms.base import SendResult from concurrent.futures import Future @@ -3376,7 +3621,14 @@ class TestDeliverResultTimeoutCancelsFuture: ) media_path = media_file.resolve() - adapter = AsyncMock() + probe_calls = {"n": 0} + + class _ForumAdapter(AsyncMock): + async def get_chat_info(self, chat_id): + probe_calls["n"] += 1 + return {"name": "Proyectos", "type": "forum", "is_forum": True} + + adapter = _ForumAdapter() adapter.send.return_value = SendResult(success=True, message_id="1") adapter.send_image_file.return_value = SendResult(success=True, message_id="2") @@ -3390,8 +3642,74 @@ class TestDeliverResultTimeoutCancelsFuture: loop.is_running.return_value = True job = { - "id": "dm-topic-media-job", - "deliver": "telegram:226252250:7072", # private chat + numeric topic + "id": "forum-topic-media-job", + "deliver": "telegram:226252250:7072", # private chat + numeric forum topic + } + + def fake_run_coro(coro, _loop): + import asyncio as _asyncio + future = Future() + try: + future.set_result(_asyncio.run(coro)) + except BaseException as _e: # noqa: BLE001 + future.set_exception(_e) + return future + + with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \ + patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \ + patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro): + _deliver_result( + job, + f"Chart attached\nMEDIA:{media_path}", + adapters={Platform.TELEGRAM: adapter}, + loop=loop, + ) + + adapter.send_image_file.assert_called_once() + media_metadata = adapter.send_image_file.call_args[1]["metadata"] + assert str(media_metadata.get("thread_id")) == "7072" + assert not media_metadata.get("direct_messages_topic_id") + # Probe exactly once and reuse the result for BOTH the text and media + # sends — never re-probe per send (the "compute ONCE" contract). + assert probe_calls["n"] == 1 + + def test_live_adapter_channel_dm_topic_media_routes_via_direct_messages_topic_id(self, tmp_path, monkeypatch): + """#22773 (media, done right): MEDIA attachments to a genuine channel DM + topic must route via ``direct_messages_topic_id``.""" + from gateway.config import Platform + from gateway.platforms.base import SendResult + from concurrent.futures import Future + + media_root = tmp_path / "media-cache" + media_file = media_root / "chart.png" + media_file.parent.mkdir(parents=True, exist_ok=True) + media_file.write_bytes(b"media") + monkeypatch.setattr( + "gateway.platforms.base.MEDIA_DELIVERY_SAFE_ROOTS", + (media_root,), + ) + media_path = media_file.resolve() + + class _ChannelAdapter(AsyncMock): + async def get_chat_info(self, chat_id): + return {"name": "My Channel", "type": "channel", "is_forum": False} + + adapter = _ChannelAdapter() + adapter.send.return_value = SendResult(success=True, message_id="1") + adapter.send_image_file.return_value = SendResult(success=True, message_id="2") + + pconfig = MagicMock() + pconfig.enabled = True + mock_cfg = MagicMock() + mock_cfg.platforms = {Platform.TELEGRAM: pconfig} + mock_cfg.filter_silence_narration = False + + loop = MagicMock() + loop.is_running.return_value = True + + job = { + "id": "channel-dm-topic-media-job", + "deliver": "telegram:226252250:7072", } def fake_run_coro(coro, _loop):