From d3a6a9f0d50ce9e5cb3697c5da812ffbe7976d24 Mon Sep 17 00:00:00 2001
From: TZZheng <tzzheng1996@gmail.com>
Date: Sun, 12 Jul 2026 15:52:35 -0500
Subject: [PATCH] fix(telegram): align media schema with send support (#745)

---
 src/lingtai/mcp_servers/telegram/manager.py |  7 +-
 tests/test_telegram_send_media_contract.py  | 88 +++++++++++++++++++++
 2 files changed, 93 insertions(+), 2 deletions(-)
 create mode 100644 tests/test_telegram_send_media_contract.py

diff --git a/src/lingtai/mcp_servers/telegram/manager.py b/src/lingtai/mcp_servers/telegram/manager.py
index 8aeef1d1d..2e0a0dbd2 100644
--- a/src/lingtai/mcp_servers/telegram/manager.py
+++ b/src/lingtai/mcp_servers/telegram/manager.py
@@ -347,6 +347,9 @@ def _transcribe_voice(audio_path: str, model_name: str = "base") -> dict:
         log.warning("Voice transcription failed: %s", e)
         return {"error": str(e)}
 
+SUPPORTED_SEND_MEDIA_TYPES = ("photo", "document")
+
+
 SCHEMA = {
     "type": "object",
     "properties": {
@@ -408,11 +411,11 @@ def _transcribe_voice(audio_path: str, model_name: str = "base") -> dict:
         "media": {
             "type": "object",
             "properties": {
-                "type": {"type": "string", "enum": ["photo", "document", "voice", "audio"]},
+                "type": {"type": "string", "enum": list(SUPPORTED_SEND_MEDIA_TYPES)},
                 "path": {"type": "string"},
             },
             "description": (
-                "Media attachment: {type: 'photo'|'document'|'voice'|'audio', path: '/path/to/file'}. "
+                "Media attachment: {type: 'photo'|'document', path: '/path/to/file'}. "
                 "For charts, HTML/SVG/PNG reports, CSVs, PDFs, and other generated artifacts that should arrive as an intact file, use type='document'. "
                 "Use type='photo' only for native inline photo previews; Telegram photo delivery can crop, compress, thumbnail, or otherwise display text-heavy charts poorly. "
                 "Do not paste local file paths in message text as a substitute for attaching the file."
diff --git a/tests/test_telegram_send_media_contract.py b/tests/test_telegram_send_media_contract.py
new file mode 100644
index 000000000..2d686887a
--- /dev/null
+++ b/tests/test_telegram_send_media_contract.py
@@ -0,0 +1,88 @@
+"""Contract tests for Telegram's advertised outbound media support."""
+
+from pathlib import Path
+
+import pytest
+
+from lingtai.mcp_servers.telegram.manager import (
+    SCHEMA,
+    SUPPORTED_SEND_MEDIA_TYPES,
+    TelegramManager,
+)
+from tests._notification_store_helpers import notification_store_for
+
+
+class _Account:
+    alias = "bot"
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, str]] = []
+
+    def send_photo(self, _chat_id, path, **_kwargs):
+        self.calls.append(("photo", path))
+        return {"message_id": 1}
+
+    def send_document(self, _chat_id, path, **_kwargs):
+        self.calls.append(("document", path))
+        return {"message_id": 2}
+
+
+class _Service:
+    def __init__(self) -> None:
+        self.account = _Account()
+
+    def get_account(self, alias):
+        assert alias == "bot"
+        return self.account
+
+
+def _manager(tmp_path: Path) -> tuple[TelegramManager, _Account]:
+    service = _Service()
+    manager = TelegramManager(
+        service,
+        working_dir=tmp_path,
+        on_inbound=lambda _: None,
+        notification_store=notification_store_for(tmp_path),
+    )
+    return manager, service.account
+
+
+def test_send_schema_advertises_exactly_runtime_supported_media_types():
+    advertised = SCHEMA["properties"]["media"]["properties"]["type"]["enum"]
+
+    assert advertised == list(SUPPORTED_SEND_MEDIA_TYPES) == ["photo", "document"]
+
+
+@pytest.mark.parametrize("media_type", SUPPORTED_SEND_MEDIA_TYPES)
+def test_each_advertised_media_type_dispatches(tmp_path: Path, media_type: str):
+    media_path = tmp_path / f"attachment.{media_type}"
+    media_path.write_bytes(b"content")
+    manager, account = _manager(tmp_path)
+
+    result = manager._send({
+        "account": "bot",
+        "chat_id": 123,
+        "media": {"type": media_type, "path": str(media_path)},
+    })
+
+    assert result == {
+        "status": "sent",
+        "message_id": f"bot:123:{1 if media_type == 'photo' else 2}",
+    }
+    assert account.calls == [(media_type, str(media_path))]
+
+
+@pytest.mark.parametrize("media_type", ["voice", "audio"])
+def test_unadvertised_media_types_remain_rejected(tmp_path: Path, media_type: str):
+    media_path = tmp_path / "unsupported-media"
+    media_path.write_bytes(b"content")
+    manager, account = _manager(tmp_path)
+
+    result = manager._send({
+        "account": "bot",
+        "chat_id": 123,
+        "media": {"type": media_type, "path": str(media_path)},
+    })
+
+    assert result == {"error": f"Unknown media type: {media_type}"}
+    assert account.calls == []