fix(telegram): heartbeat loop exits cleanly when bot has no get_me

CI shard test_telegram_conflict.py timed out (140s) because the new
_polling_heartbeat_loop, started by connect(), busy-spun under those
tests: they monkeypatch asyncio.sleep to instant and pass a bot double
with no get_me(), so the probe raised AttributeError (swallowed) and the
loop re-entered immediately with no real pacing, starving the event loop.

Guard the loop to return when bot.get_me is not callable — a real PTB Bot
always exposes it, so this only triggers on a torn-down app or a test
double, where there is nothing to probe. Also cancel the heartbeat task in
the conflict tests that call connect() without disconnect(), matching the
production disconnect() teardown.

Verified: test_telegram_conflict.py now runs in ~4.5s; the 22
heartbeat/reconnect tests still pass; E2E confirms a hanging get_me still
fires the reconnect ladder while a missing get_me exits without spinning.

plugins/platforms/telegram/adapter.py

@@ -1745,9 +1745,15 @@ class TelegramAdapter(BasePlatformAdapter):
                 await asyncio.sleep(HEARTBEAT_INTERVAL)
                 if self.has_fatal_error:
                     return
-                if not (self._app and self._app.bot):
+                bot = self._app.bot if self._app else None
+                if bot is None:
                     continue
-                await asyncio.wait_for(self._app.bot.get_me(), PROBE_TIMEOUT)
+                # A real PTB Bot always exposes get_me(); if it's absent the
+                # app isn't a live polling client (e.g. torn down or a test
+                # double), so there is nothing to probe — exit rather than spin.
+                if not callable(getattr(bot, "get_me", None)):
+                    return
+                await asyncio.wait_for(bot.get_me(), PROBE_TIMEOUT)
             except asyncio.CancelledError:
                 return
             except (asyncio.TimeoutError, OSError) as probe_err:

tests/gateway/test_telegram_conflict.py

@@ -47,6 +47,25 @@ def _no_auto_discovery(monkeypatch):
     monkeypatch.setattr("plugins.platforms.telegram.adapter.HTTPXRequest", lambda **kwargs: MagicMock())
 
 
+async def _cancel_heartbeat(adapter):
+    """Cancel the lifetime heartbeat task connect() starts in polling mode.
+
+    These tests call the real connect() but never disconnect(), so the
+    _polling_heartbeat_loop task would otherwise outlive the test. With
+    asyncio.sleep monkeypatched to instant, leaving it running busy-spins the
+    event loop and starves the test (CI per-file timeout). disconnect() does
+    this in production; tests that only connect() must do it themselves.
+    """
+    task = getattr(adapter, "_polling_heartbeat_task", None)
+    if task and not task.done():
+        task.cancel()
+        try:
+            await task
+        except (asyncio.CancelledError, Exception):
+            pass
+    adapter._polling_heartbeat_task = None
+
+
 @pytest.mark.asyncio
 async def test_connect_rejects_same_host_token_lock(monkeypatch):
     adapter = TelegramAdapter(PlatformConfig(enabled=True, token="secret-token"))
@@ -127,6 +146,11 @@ async def test_polling_conflict_retries_before_fatal(monkeypatch):
     assert adapter.has_fatal_error is False, "First conflict should not be fatal"
     assert adapter._polling_conflict_count == 0, "Count should reset after successful retry"
 
+    # connect() now starts a lifetime _polling_heartbeat_loop task. With
+    # asyncio.sleep mocked to instant above, it must not be left running or it
+    # busy-spins on the event loop and starves the test. Cancel it explicitly.
+    await _cancel_heartbeat(adapter)
+
 
 @pytest.mark.asyncio
 async def test_polling_conflict_becomes_fatal_after_retries(monkeypatch):
@@ -205,6 +229,7 @@ async def test_polling_conflict_becomes_fatal_after_retries(monkeypatch):
     )
     assert adapter.has_fatal_error is True
     fatal_handler.assert_awaited_once()
+    await _cancel_heartbeat(adapter)
 
 
 @pytest.mark.asyncio
@@ -285,6 +310,7 @@ async def test_connect_clears_webhook_before_polling(monkeypatch):
 
     assert ok is True
     bot.delete_webhook.assert_awaited_once_with(drop_pending_updates=False)
+    await _cancel_heartbeat(adapter)
 
 
 @pytest.mark.asyncio
@@ -398,3 +424,4 @@ async def test_polling_conflict_reschedule_uses_running_loop(monkeypatch):
         await adapter._polling_error_task
     except (asyncio.CancelledError, Exception):
         pass
+    await _cancel_heartbeat(adapter)