fix(telegram): bound the 3 sibling updater.stop() calls with the same CLOSE-WAIT timeout

The salvaged fix (#58272) guarded the primary network-error reconnect path.
Issue #58270's Scope section names three more unguarded await updater.stop()
sites that can hang identically on a CLOSE-WAIT socket:

- conflict handler (before the retry back-off sleep)
- conflict-retries-exhausted teardown (before the fatal notify)
- disconnect() teardown (would hang gateway shutdown/restart)

Each is now wrapped in asyncio.wait_for(..., _UPDATER_STOP_TIMEOUT) with a
warning on timeout, matching the primary path, so no reconnect/teardown ladder
can wedge on a dead socket.

Also hoist the shared 15.0s bound to a single module constant
_UPDATER_STOP_TIMEOUT (self-documenting + DRY across all 4 sites), and update
the CLOSE-WAIT regression test to patch that constant instead of monkeypatching
asyncio.wait_for process-wide.

Fatal-notify idempotency: bounding the conflict-exhausted teardown stop() adds
an await AFTER _set_fatal_error, which yields the loop and lets a concurrent
retry task (scheduled by an earlier conflict, already suspended past the entry
guard) reach the fatal branch too — double-firing the fatal handler (surfaced
as a Python 3.11 CI failure in test_polling_conflict_becomes_fatal_after_retries).
Snapshot the pre-transition fatal state and only notify on the first transition.

plugins/platforms/telegram/adapter.py

@@ -397,6 +397,14 @@ def _rich_normalize_linebreaks(text: str) -> str:
     return ''.join(out)
 
 
+# Watchdog bound for `await updater.stop()`. When the underlying TCP socket is
+# in CLOSE-WAIT the PTB polling task is blocked on epoll on the dead socket and
+# never wakes, so an unguarded stop() hangs indefinitely and wedges the whole
+# reconnect/teardown ladder. This is an internal safety bound (not a user knob),
+# applied identically at every stop() site so no path can hang on a dead socket.
+_UPDATER_STOP_TIMEOUT = 15.0
+
+
 class TelegramAdapter(BasePlatformAdapter):
     """
     Telegram bot adapter.
@@ -2001,7 +2009,7 @@ class TelegramAdapter(BasePlatformAdapter):
                     # the gateway silently drops messages for hours.
                     # Bounding stop() lets the reconnect ladder always advance.
                     # Refs: NousResearch/hermes-agent#58270
-                    await asyncio.wait_for(app.updater.stop(), timeout=15.0)
+                    await asyncio.wait_for(app.updater.stop(), timeout=_UPDATER_STOP_TIMEOUT)
                 except asyncio.TimeoutError:
                     logger.warning(
                         "[%s] updater.stop() timed out during network-error "
@@ -2382,10 +2390,19 @@ class TelegramAdapter(BasePlatformAdapter):
             )
             # Stop the local updater cleanly before sleeping.  If it's already
             # stopped (e.g. PTB raised before updater.running was set) this is
-            # a no-op.
+            # a no-op.  Bounded with a timeout for the same reason as the
+            # network-error path: a CLOSE-WAIT socket can wedge stop() on epoll
+            # forever, which would stall the conflict-retry ladder.
             try:
                 if self._app and self._app.updater and self._app.updater.running:
-                    await self._app.updater.stop()
+                    try:
+                        await asyncio.wait_for(self._app.updater.stop(), timeout=_UPDATER_STOP_TIMEOUT)
+                    except asyncio.TimeoutError:
+                        logger.warning(
+                            "[%s] updater.stop() timed out during conflict "
+                            "retry (likely CLOSE-WAIT socket); continuing",
+                            self.name,
+                        )
             except Exception:
                 pass
 
@@ -2453,16 +2470,33 @@ class TelegramAdapter(BasePlatformAdapter):
             "[%s] %s Original error: %s",
             self.name, message, error,
         )
+        # Snapshot whether we are the call that actually transitions to fatal.
+        # A concurrent retry task scheduled by an earlier conflict may already
+        # be suspended past the entry guard; once _set_fatal_error flips the
+        # flag, adding an await below (the bounded stop()) yields the loop and
+        # lets that task reach this branch too — double-notifying the fatal
+        # handler.  Only the first transition notifies.
+        _already_fatal = (
+            self.has_fatal_error
+            and self.fatal_error_code == "telegram_polling_conflict"
+        )
         self._set_fatal_error("telegram_polling_conflict", message, retryable=False)
         try:
             if self._app and self._app.updater:
-                await self._app.updater.stop()
+                await asyncio.wait_for(self._app.updater.stop(), timeout=_UPDATER_STOP_TIMEOUT)
+        except asyncio.TimeoutError:
+            logger.warning(
+                "[%s] updater.stop() timed out after exhausting conflict "
+                "retries (likely CLOSE-WAIT socket); proceeding to fatal notify",
+                self.name,
+            )
         except Exception as stop_error:
             logger.warning(
                 "[%s] Failed stopping Telegram updater after exhausting conflict retries: %s",
                 self.name, stop_error, exc_info=True,
             )
-        await self._notify_fatal_error()
+        if not _already_fatal:
+            await self._notify_fatal_error()
 
     async def _create_dm_topic(
         self,
@@ -3352,9 +3386,20 @@ class TelegramAdapter(BasePlatformAdapter):
 
         if self._app:
             try:
-                # Only stop the updater if it's running
+                # Only stop the updater if it's running.  Bounded with a
+                # timeout: a CLOSE-WAIT socket can wedge stop() on epoll
+                # indefinitely, which would hang disconnect() (and any
+                # gateway shutdown/restart waiting on it) forever.  On timeout
+                # we fall through to app.stop()/shutdown() to force teardown.
                 if self._app.updater and self._app.updater.running:
-                    await self._app.updater.stop()
+                    try:
+                        await asyncio.wait_for(self._app.updater.stop(), timeout=_UPDATER_STOP_TIMEOUT)
+                    except asyncio.TimeoutError:
+                        logger.warning(
+                            "[%s] updater.stop() timed out during disconnect "
+                            "(likely CLOSE-WAIT socket); forcing app shutdown",
+                            self.name,
+                        )
                 if self._app.running:
                     await self._app.stop()
                 await self._app.shutdown()

tests/gateway/test_telegram_network_reconnect.py

@@ -856,15 +856,12 @@ async def test_handle_polling_network_error_updater_stop_timeout():
 
     app.updater.start_polling = AsyncMock(side_effect=_fake_start_polling)
 
-    # Patch the timeout constant so the test completes fast.
+    # Shrink the stop() watchdog bound so the test completes fast instead of
+    # waiting the full _UPDATER_STOP_TIMEOUT. Patching the named constant is
+    # cleaner than monkeypatching asyncio.wait_for process-wide.
     import plugins.platforms.telegram.adapter as _mod
-    original_wait_for = asyncio.wait_for
 
-    async def _fast_wait_for(coro, timeout):
-        # Substitute a 0.05s timeout so the test doesn't take 15s.
-        return await original_wait_for(coro, 0.05)
-
-    with patch("asyncio.wait_for", side_effect=_fast_wait_for):
+    with patch.object(_mod, "_UPDATER_STOP_TIMEOUT", 0.05):
         await adapter._handle_polling_network_error(OSError("CLOSE-WAIT test"))
 
     # The reconnect ladder must have advanced past the hung stop().