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

gateway/run.py

@@ -3654,6 +3654,23 @@ async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> Non
         If the error is retryable (e.g. network blip, DNS failure), queue the
         platform for background reconnection instead of giving up permanently.
         """
+        # Snapshot the current owner of this platform slot before doing
+        # anything else. If it's neither this adapter nor empty, a different
+        # adapter has already taken over (e.g. this is a delayed notification
+        # from a background retry chain that raced with, and lost to, a
+        # reconnect that already succeeded). Acting on a stale notification
+        # would overwrite an already-healthy platform's runtime status and
+        # incorrectly re-queue it for reconnection, so bail out before any of
+        # that happens.
+        existing = self.adapters.get(adapter.platform)
+        if existing is not None and existing is not adapter:
+            logger.debug(
+                "Ignoring stale fatal error from a superseded %s adapter instance: %s",
+                adapter.platform.value,
+                adapter.fatal_error_code or "unknown",
+            )
+            return
+
         logger.error(
             "Fatal %s adapter error (%s): %s",
             adapter.platform.value,
@@ -3677,13 +3694,15 @@ async def _handle_adapter_fatal_error(self, adapter: BasePlatformAdapter) -> Non
             error_message=adapter.fatal_error_message,
         )
 
-        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 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
         )
         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 @@ async def _handle_polling_network_error(self, error: Exception) -> None:
                 )
                 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,96 @@ 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
+
+
+@pytest.mark.asyncio
+async def test_stale_fatal_notification_from_superseded_adapter_is_ignored(monkeypatch, tmp_path):
+    """
+    A delayed fatal-error notification from an adapter instance that has
+    since been replaced by a different, already-installed adapter (e.g. a
+    background retry chain on the old instance finally giving up after a
+    reconnect on a new instance already succeeded) must be ignored: it must
+    not disconnect the new adapter, must not re-queue an already-healthy
+    platform for reconnection, and must not shut the gateway down.
+    """
+    config = GatewayConfig(
+        platforms={
+            Platform.WHATSAPP: PlatformConfig(enabled=True, token="token")
+        },
+        sessions_dir=tmp_path / "sessions",
+    )
+    runner = GatewayRunner(config)
+
+    old_adapter = _RuntimeRetryableAdapter()
+    old_adapter._set_fatal_error(
+        "whatsapp_bridge_exited",
+        "stale failure from a superseded adapter instance",
+        retryable=True,
+    )
+
+    new_adapter = _RuntimeRetryableAdapter()
+    new_adapter.disconnect = AsyncMock()
+    runner.adapters = {Platform.WHATSAPP: new_adapter}
+    runner.delivery_router.adapters = runner.adapters
+    runner.stop = AsyncMock()
+
+    await runner._handle_adapter_fatal_error(old_adapter)
+
+    new_adapter.disconnect.assert_not_awaited()
+    assert runner.adapters[Platform.WHATSAPP] is new_adapter
+    assert Platform.WHATSAPP not in runner._failed_platforms
+    runner.stop.assert_not_awaited()

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():
     """