fix: require anchors for Telegram DM topic deliveries

(cherry picked from commit 6daafb3fd48f8ea6b092fa10e85ad589ca9e501c)

gateway/delivery.py

@@ -259,16 +259,33 @@ class DeliveryRouter:
         
         send_metadata = dict(metadata or {})
         if target.thread_id:
+            has_explicit_direct_topic = (
+                "direct_messages_topic_id" in send_metadata
+                or "telegram_direct_messages_topic_id" in send_metadata
+            )
             if (
                 target.platform == Platform.TELEGRAM
                 and _looks_like_telegram_private_chat_id(target.chat_id)
                 and "thread_id" not in send_metadata
                 and "message_thread_id" not in send_metadata
-                and "direct_messages_topic_id" not in send_metadata
+                and not has_explicit_direct_topic
-                and "telegram_direct_messages_topic_id" not in send_metadata
             ):
-                send_metadata["telegram_direct_messages_topic_id"] = target.thread_id
+                # Telegram has two similar-but-not-equivalent private topic modes:
-            elif "thread_id" not in send_metadata:
+                # true Bot API Direct Messages topics use direct_messages_topic_id,
+                # while Hermes-created private DM lanes only route reliably with
+                # message_thread_id plus a reply anchor to a message in that lane.
+                # DeliveryRouter often handles proactive/cron sends, so an anchor
+                # may not exist. Refuse the send rather than reporting success for
+                # a message that lands in General/All Messages or is invisible.
+                reply_anchor = send_metadata.get("telegram_reply_to_message_id")
+                if reply_anchor is None:
+                    raise RuntimeError(
+                        "Telegram private DM topic delivery requires telegram_reply_to_message_id; "
+                        "send to the bare chat or provide a reply anchor"
+                    )
+                send_metadata["thread_id"] = target.thread_id
+                send_metadata["telegram_dm_topic_reply_fallback"] = True
+            elif "thread_id" not in send_metadata and "message_thread_id" not in send_metadata and not has_explicit_direct_topic:
                 send_metadata["thread_id"] = target.thread_id
         result = await adapter.send(target.chat_id, content, metadata=send_metadata or None)
         if getattr(result, "success", True) is False:

tests/gateway/test_delivery.py

@@ -135,25 +135,60 @@ class RecordingAdapter:
 
 
 @pytest.mark.asyncio
-async def test_explicit_telegram_private_thread_uses_direct_messages_topic_id(tmp_path, monkeypatch):
+async def test_explicit_telegram_private_thread_requires_reply_anchor(tmp_path, monkeypatch):
     monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
     adapter = RecordingAdapter()
     router = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: adapter})
     target = DeliveryTarget.parse("telegram:722341991:32344")
 
-    await router._deliver_to_platform(target, "hello", metadata=None)
+    with pytest.raises(RuntimeError, match="requires telegram_reply_to_message_id"):
+        await router._deliver_to_platform(target, "hello", metadata=None)
+
+    assert adapter.calls == []
+
+
+@pytest.mark.asyncio
+async def test_explicit_telegram_private_thread_uses_reply_fallback_with_anchor(tmp_path, monkeypatch):
+    monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
+    adapter = RecordingAdapter()
+    router = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: adapter})
+    target = DeliveryTarget.parse("telegram:722341991:32344")
+
+    await router._deliver_to_platform(
+        target,
+        "hello",
+        metadata={"telegram_reply_to_message_id": "9001"},
+    )
 
     assert adapter.calls == [
         {
             "chat_id": "722341991",
             "content": "hello",
             "metadata": {
-                "telegram_direct_messages_topic_id": "32344",
+                "telegram_reply_to_message_id": "9001",
+                "thread_id": "32344",
+                "telegram_dm_topic_reply_fallback": True,
             },
         }
     ]
 
 
+@pytest.mark.asyncio
+async def test_explicit_telegram_direct_messages_topic_metadata_is_respected(tmp_path, monkeypatch):
+    monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
+    adapter = RecordingAdapter()
+    router = DeliveryRouter(GatewayConfig(), adapters={Platform.TELEGRAM: adapter})
+    target = DeliveryTarget.parse("telegram:722341991:32344")
+
+    await router._deliver_to_platform(
+        target,
+        "hello",
+        metadata={"telegram_direct_messages_topic_id": "32344"},
+    )
+
+    assert adapter.calls[0]["metadata"] == {"telegram_direct_messages_topic_id": "32344"}
+
+
 @pytest.mark.asyncio
 async def test_explicit_telegram_group_thread_does_not_mark_dm_fallback(tmp_path, monkeypatch):
     monkeypatch.setattr("gateway.delivery.get_hermes_home", lambda: tmp_path)
@@ -178,4 +213,4 @@ async def test_platform_send_failure_raises_for_delivery_result(tmp_path, monkey
     target = DeliveryTarget.parse("telegram:722341991:32344")
 
     with pytest.raises(RuntimeError, match="route failed"):
-        await router._deliver_to_platform(target, "hello", metadata=None)
+        await router._deliver_to_platform(target, "hello", metadata={"telegram_reply_to_message_id": "9001"})