Merge PR #1796: fix(telegram): avoid media filename collisions

fix(telegram): avoid media filename collisions

nanobot/channels/telegram.py

@@ -534,7 +534,8 @@ class TelegramChannel(BaseChannel):
                 getattr(media_file, "file_name", None),
             )
             media_dir = get_media_dir("telegram")
-            file_path = media_dir / f"{media_file.file_id[:16]}{ext}"
+            unique_id = getattr(media_file, "file_unique_id", media_file.file_id)
+            file_path = media_dir / f"{unique_id}{ext}"
             await file.download_to_drive(str(file_path))
             path_str = str(file_path)
             if media_type in ("voice", "audio"):

tests/test_telegram_channel.py

@@ -446,6 +446,56 @@ async def test_download_message_media_returns_path_when_download_succeeds(
     assert "[image:" in parts[0]
 
 
+@pytest.mark.asyncio
+async def test_download_message_media_uses_file_unique_id_when_available(
+    monkeypatch, tmp_path
+) -> None:
+    media_dir = tmp_path / "media" / "telegram"
+    media_dir.mkdir(parents=True)
+    monkeypatch.setattr(
+        "nanobot.channels.telegram.get_media_dir",
+        lambda channel=None: media_dir if channel else tmp_path / "media",
+    )
+
+    downloaded: dict[str, str] = {}
+
+    async def _download_to_drive(path: str) -> None:
+        downloaded["path"] = path
+
+    channel = TelegramChannel(
+        TelegramConfig(enabled=True, token="123:abc", allow_from=["*"]),
+        MessageBus(),
+    )
+    app = _FakeApp(lambda: None)
+    app.bot.get_file = AsyncMock(
+        return_value=SimpleNamespace(download_to_drive=_download_to_drive)
+    )
+    channel._app = app
+
+    msg = SimpleNamespace(
+        photo=[
+            SimpleNamespace(
+                file_id="file-id-that-should-not-be-used",
+                file_unique_id="stable-unique-id",
+                mime_type="image/jpeg",
+                file_name=None,
+            )
+        ],
+        voice=None,
+        audio=None,
+        document=None,
+        video=None,
+        video_note=None,
+        animation=None,
+    )
+
+    paths, parts = await channel._download_message_media(msg)
+
+    assert downloaded["path"].endswith("stable-unique-id.jpg")
+    assert paths == [str(media_dir / "stable-unique-id.jpg")]
+    assert parts == [f"[image: {media_dir / 'stable-unique-id.jpg'}]"]
+
+
 @pytest.mark.asyncio
 async def test_on_message_attaches_reply_to_media_when_available(monkeypatch, tmp_path) -> None:
     """When user replies to a message with media, that media is downloaded and attached to the turn."""