fix(telegram): close reconnect races that leave adapter half-destroyed

_handle_polling_network_error's chained retry never updated
self._polling_error_task, so the reentrancy guard shared with the
heartbeat loop and the pending-updates probe went stale mid-recovery,
letting more than one recovery attempt run concurrently against the
same adapter. Combined with a TOCTOU window in
_handle_adapter_fatal_error (the adapter was only removed from
self.adapters in a finally block after awaiting disconnect()), two
concurrent fatal notifications for the same adapter could both pass
the "still installed" check and call disconnect() twice, which is
where the reported "'NoneType' object has no attribute 'updater'"
originates once self._app is cleared by the first call.

- Reassign the chained retry task to self._polling_error_task so the
  guard reflects an in-flight recovery.
- Capture self._app in a local variable across the stop/start_polling
  sequence instead of re-reading self._app between awaits.
- Claim (pop) the adapter from self.adapters before awaiting
  disconnect() in _handle_adapter_fatal_error, not after, closing the
  TOCTOU window for a concurrent notification on the same adapter.

gateway/run.py

@@ -3679,11 +3679,14 @@ class GatewayRunner(GatewayAuthorizationMixin, GatewayKanbanWatchersMixin, Gatew
 
         existing = self.adapters.get(adapter.platform)
         if existing is adapter:
-            try:
-                await adapter.disconnect()
-            finally:
-                self.adapters.pop(adapter.platform, None)
-                self.delivery_router.adapters = self.adapters
+            # Claim this adapter for teardown before awaiting disconnect() —
+            # a second fatal-error notification for the same adapter (e.g.
+            # from a concurrent recovery path) would otherwise still see
+            # itself as "existing" during the await below and disconnect()
+            # the same object twice.
+            self.adapters.pop(adapter.platform, None)
+            self.delivery_router.adapters = self.adapters
+            await adapter.disconnect()
 
         # Queue retryable failures for background reconnection
         if adapter.fatal_error_retryable:

plugins/platforms/telegram/adapter.py

@@ -1773,16 +1773,24 @@ class TelegramAdapter(BasePlatformAdapter):
         )
         await asyncio.sleep(delay)
 
+        # Capture a stable local reference: self._app can be reassigned to None
+        # by a concurrent disconnect() while we're suspended across the awaits
+        # below, and re-reading self._app after that point would silently swap
+        # in None mid-sequence instead of failing fast in one place.
+        app = self._app
+
         try:
-            if self._app and self._app.updater and self._app.updater.running:
-                await self._app.updater.stop()
+            if app and app.updater and app.updater.running:
+                await app.updater.stop()
         except Exception:
             pass
 
         await self._drain_polling_connections()
 
         try:
-            await self._app.updater.start_polling(
+            if not app:
+                raise RuntimeError("Telegram application was torn down during reconnect")
+            await app.updater.start_polling(
                 allowed_updates=Update.ALL_TYPES,
                 drop_pending_updates=False,
                 error_callback=self._polling_error_callback_ref,
@@ -1824,6 +1832,12 @@ class TelegramAdapter(BasePlatformAdapter):
                 )
                 self._background_tasks.add(task)
                 task.add_done_callback(self._background_tasks.discard)
+                # This chained retry IS the in-flight recovery attempt — it
+                # must replace the reentrancy guard, otherwise the heartbeat
+                # loop, the pending-updates probe, and the PTB error callback
+                # all see _polling_error_task as "done" and can each start a
+                # second, concurrent recovery for the same outage.
+                self._polling_error_task = task
 
     async def _polling_heartbeat_loop(self) -> None:
         """Detect dead Telegram TCP sockets (CLOSE-WAIT) by periodic probing.

tests/gateway/test_runner_fatal_adapter.py

@@ -1,3 +1,4 @@
+import asyncio
 from unittest.mock import AsyncMock
 
 import pytest
@@ -98,3 +99,57 @@ async def test_runner_queues_retryable_runtime_fatal_for_reconnection(monkeypatc
     assert runner._exit_with_failure is False
     assert Platform.WHATSAPP in runner._failed_platforms
     assert runner._failed_platforms[Platform.WHATSAPP]["attempts"] == 0
+
+
+@pytest.mark.asyncio
+async def test_concurrent_fatal_notifications_disconnect_same_adapter_once(monkeypatch, tmp_path):
+    """
+    Two fatal-error notifications for the same still-installed adapter (e.g.
+    from two concurrent recovery paths racing on the same underlying outage)
+    must result in exactly one disconnect() call.
+
+    Regression test for the TOCTOU race in _handle_adapter_fatal_error: the
+    old code only removed the adapter from self.adapters in a `finally` block
+    *after* awaiting disconnect(), so a second concurrent call could still see
+    itself as "existing" and disconnect() the same object twice — the
+    concrete origin of the "'NoneType' object has no attribute 'updater'"
+    crash when the adapter's own teardown code re-reads self._app afterwards.
+    """
+    config = GatewayConfig(
+        platforms={
+            Platform.WHATSAPP: PlatformConfig(enabled=True, token="token")
+        },
+        sessions_dir=tmp_path / "sessions",
+    )
+    runner = GatewayRunner(config)
+    adapter = _RuntimeRetryableAdapter()
+    adapter._set_fatal_error(
+        "whatsapp_bridge_exited",
+        "WhatsApp bridge process exited unexpectedly (code 1).",
+        retryable=True,
+    )
+
+    runner.adapters = {Platform.WHATSAPP: adapter}
+    runner.delivery_router.adapters = runner.adapters
+    runner.stop = AsyncMock()
+
+    disconnect_calls = 0
+    release_second_call = asyncio.Event()
+
+    async def slow_disconnect():
+        nonlocal disconnect_calls
+        disconnect_calls += 1
+        # Yield control so the second concurrent notification can run its
+        # "existing is adapter" check before this call finishes tearing down.
+        release_second_call.set()
+        await asyncio.sleep(0)
+        adapter._mark_disconnected()
+
+    monkeypatch.setattr(adapter, "disconnect", slow_disconnect)
+
+    await asyncio.gather(
+        runner._handle_adapter_fatal_error(adapter),
+        runner._handle_adapter_fatal_error(adapter),
+    )
+
+    assert disconnect_calls == 1

tests/gateway/test_telegram_network_reconnect.py

@@ -117,6 +117,43 @@ async def test_reconnect_does_not_self_schedule_when_fatal_error_set():
     )
 
 
+@pytest.mark.asyncio
+async def test_reconnect_chained_retry_updates_polling_error_task():
+    """
+    When start_polling() fails and the handler self-schedules a retry, that
+    retry task must become the new `_polling_error_task` — otherwise the
+    reentrancy guard used by the heartbeat loop, the pending-updates probe,
+    and the PTB error callback goes stale while a recovery is still in
+    flight, letting a second concurrent recovery start for the same outage.
+
+    Regression test for the race behind the "half-destroyed adapter" bug
+    (gateway reports connected but silently stops processing messages).
+    """
+    adapter = _make_adapter()
+    adapter._polling_network_error_count = 1
+
+    mock_updater = MagicMock()
+    mock_updater.running = True
+    mock_updater.stop = AsyncMock()
+    mock_updater.start_polling = AsyncMock(side_effect=Exception("Timed out"))
+
+    mock_app = MagicMock()
+    mock_app.updater = mock_updater
+    adapter._app = mock_app
+
+    with patch("asyncio.sleep", new_callable=AsyncMock):
+        await adapter._handle_polling_network_error(Exception("Bad Gateway"))
+
+    assert adapter._polling_error_task is not None
+    assert not adapter._polling_error_task.done()
+
+    adapter._polling_error_task.cancel()
+    try:
+        await adapter._polling_error_task
+    except (asyncio.CancelledError, Exception):
+        pass
+
+
 @pytest.mark.asyncio
 async def test_reconnect_success_resets_error_count():
     """