fix: auto-create Telegram DM topics for delivery

(cherry picked from commit 5cde0614e8)

tests/gateway/test_delivery.py

@@ -128,11 +128,16 @@ class TestPlatformNameCaseInsensitivity:
 class RecordingAdapter:
     def __init__(self):
         self.calls = []
+        self.ensure_dm_topic_calls = []
 
     async def send(self, chat_id, content, metadata=None):
         self.calls.append({"chat_id": chat_id, "content": content, "metadata": metadata})
         return {"success": True}
 
+    async def ensure_dm_topic(self, chat_id, topic_name):
+        self.ensure_dm_topic_calls.append({"chat_id": chat_id, "topic_name": topic_name})
+        return "38049"
+
 
 @pytest.mark.asyncio
 async def test_explicit_telegram_private_thread_requires_reply_anchor(tmp_path, monkeypatch):
@@ -147,6 +152,30 @@ async def test_explicit_telegram_private_thread_requires_reply_anchor(tmp_path,
     assert adapter.calls == []
 
 
+@pytest.mark.asyncio
+async def test_named_telegram_private_topic_is_created_before_delivery(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:Hermes API Test")
+
+    await router._deliver_to_platform(target, "hello", metadata=None)
+
+    assert adapter.ensure_dm_topic_calls == [
+        {"chat_id": "722341991", "topic_name": "Hermes API Test"}
+    ]
+    assert adapter.calls == [
+        {
+            "chat_id": "722341991",
+            "content": "hello",
+            "metadata": {
+                "thread_id": "38049",
+                "telegram_dm_topic_created_for_send": True,
+            },
+        }
+    ]
+
+
 @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)

tests/gateway/test_dm_topics.py

@@ -205,6 +205,28 @@ async def test_create_dm_topic_returns_none_without_bot():
     assert result is None
 
 
+@pytest.mark.asyncio
+async def test_ensure_dm_topic_creates_on_demand_and_persists():
+    """Named delivery targets should create missing private DM topics on demand."""
+    adapter = _make_adapter()
+    adapter._bot = AsyncMock()
+    adapter._bot.create_forum_topic.return_value = SimpleNamespace(message_thread_id=444)
+    adapter._persist_dm_topic_thread_id = MagicMock()
+
+    result = await adapter.ensure_dm_topic("111", "On Demand")
+
+    assert result == "444"
+    adapter._bot.create_forum_topic.assert_called_once_with(
+        chat_id=111,
+        name="On Demand",
+    )
+    assert adapter._dm_topics["111:On Demand"] == 444
+    assert adapter._dm_topics_config == [
+        {"chat_id": 111, "topics": [{"name": "On Demand", "thread_id": 444}]}
+    ]
+    adapter._persist_dm_topic_thread_id.assert_called_once_with(111, "On Demand", 444)
+
+
 # ── _persist_dm_topic_thread_id ──
 
 

tests/gateway/test_telegram_thread_fallback.py

@@ -597,6 +597,33 @@ async def test_send_uses_reply_fallback_for_hermes_dm_topics():
     assert "direct_messages_topic_id" not in call_log[0]
 
 
+@pytest.mark.asyncio
+async def test_send_created_private_topic_uses_message_thread_without_anchor():
+    """Topics created via createForumTopic are addressable by message_thread_id directly."""
+    adapter = _make_adapter()
+    call_log = []
+
+    async def mock_send_message(**kwargs):
+        call_log.append(kwargs)
+        return SimpleNamespace(message_id=781)
+
+    adapter._bot = SimpleNamespace(send_message=mock_send_message)
+
+    result = await adapter.send(
+        chat_id="123",
+        content="created topic message",
+        metadata={
+            "thread_id": "38049",
+            "telegram_dm_topic_created_for_send": True,
+        },
+    )
+
+    assert result.success is True
+    assert call_log[0]["reply_to_message_id"] is None
+    assert call_log[0]["message_thread_id"] == 38049
+    assert "direct_messages_topic_id" not in call_log[0]
+
+
 @pytest.mark.asyncio
 async def test_send_uses_metadata_reply_fallback_for_streaming_dm_topics():
     """Metadata-only sends still stay in Hermes-created Telegram DM topics."""