mirrors/hermes-agent
1
0
Fork 0
mirror of https://github.com/NousResearch/hermes-agent.git synced 2026-07-09 13:21:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

Merge pull request #58165 from kshitijk4poor/fix/52060-cron-dm-topic-routing

Browse source 
fix(cron): route Telegram forum topics via message_thread_id, support channel DM topics (#52060)
This commit is contained in:
kshitij 2026-07-04 14:38:51 +05:30 • committed by GitHub
commit 5e94bc12d3
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 423 additions and 33 deletions
Download patch file Download diff file Expand all files Collapse all files

96
cron/scheduler.py
View file

@ -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:<positive_chat_id>:<numeric_thread_id>``) — 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:<positive_chat_id>:<numeric_thread_id>`` 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:

360
tests/cron/test_scheduler.py
View file

@ -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):