Gateway transient network failure resilience and recovery

Fixes on record

Merged to main. A stamp counts the issues a fix is linked to.

• #10947 fix(telegram): send resilience — retry transient tool sends + increase cold-boot budget (#9101, #5770)

  merged to main

  files: gateway/platforms/telegram.py, tests/tools/test_send_message_tool.py, tools/send_message_tool.py

  Show the diff

  • gateway/platforms/telegram.py

    @@ -665,14 +665,14 @@ def _env_float(name: str, default: float) -> float:
                     from telegram.error import NetworkError, TimedOut
                 except ImportError:
                     NetworkError = TimedOut = OSError  # type: ignore[misc,assignment]
    -            _max_connect = 3
    +            _max_connect = 8
                 for _attempt in range(_max_connect):
                     try:
                         await self._app.initialize()
                         break
                     except (NetworkError, TimedOut, OSError) as init_err:
                         if _attempt < _max_connect - 1:
    -                        wait = 2 ** _attempt
    +                        wait = min(2 ** _attempt, 15)
                             logger.warning(
                                 "[%s] Connect attempt %d/%d failed: %s — retrying in %ds",
                                 self.name, _attempt + 1, _max_connect, init_err, wait,

  • tests/tools/test_send_message_tool.py

    @@ -816,6 +816,23 @@ def test_html_parse_failure_falls_back_to_plain(self, monkeypatch):
             second_call = bot.send_message.await_args_list[1].kwargs
             assert second_call["parse_mode"] is None
     
    +    def test_transient_bad_gateway_retries_text_send(self, monkeypatch):
    +        bot = self._make_bot()
    +        bot.send_message = AsyncMock(
    +            side_effect=[
    +                Exception("502 Bad Gateway"),
    +                SimpleNamespace(message_id=2),
    +            ]
    +        )
    +        _install_telegram_mock(monkeypatch, bot)
    +
    +        with patch("asyncio.sleep", new=AsyncMock()) as sleep_mock:
    +            result = asyncio.run(_send_telegram("tok", "123", "hello"))
    +
    +        assert result["success"] is True
    +        assert bot.send_message.await_count == 2
    +        sleep_mock.assert_awaited_once()
    +
     
     # ---------------------------------------------------------------------------
     # Tests for Discord thread_id support

  • tools/send_message_tool.py

    @@ -5,6 +5,7 @@
     human-friendly channel names to IDs. Works in both CLI and gateway contexts.
     """
     
    +import asyncio
     import json
     import logging
     import os
    @@ -48,6 +49,49 @@ def _error(message: str) -> dict:
         return {"error": _sanitize_error_text(message)}
     
     
    +def _telegram_retry_delay(exc: Exception, attempt: int) -> float | None:
    +    retry_after = getattr(exc, "retry_after", None)
    +    if retry_after is not None:
    +        try:
    +            return max(float(retry_after), 0.0)
    +        except (TypeError, ValueError):
    +            return 1.0
    +
    +    text = str(exc).lower()
    +    if "timed out" in text or "timeout" in text:
    +        return None
    +    if (
    +        "bad gateway" in text
    +        or "502" in text
    +        or "too many requests" in text
    +        or "429" in text
    +        or "service unavailable" in text
    +        or "503" in text
    +        or "gateway timeout" in text
    +        or "504" in text
    +    ):
    +        return float(2 ** attempt)
    +    return None
    +
    +
    +async def _send_telegram_message_with_retry(bot, *, attempts: int = 3, **kwargs):
    +    for attempt in range(attempts):
    +        try:
    +            return await bot.send_message(**kwargs)
    +        except Exception as exc:
    +            delay = _telegram_retry_delay(exc, attempt)
    +            if delay is None or attempt >= attempts - 1:
    +                raise
    +            logger.warning(
    +                "Transient Telegram send failure (attempt %d/%d), retrying in %.1fs: %s",
    +                attempt + 1,
    +                attempts,
    +                delay,
    +                _sanitize_error_text(exc),
    +            )
    +            await asyncio.sleep(delay)
    +
    +
     SEND_MESSAGE_SCHEMA = {
         "name": "send_message",
         "description": (
    @@ -530,7 +574,8 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No
     
             if formatted.strip():
                 try:
    -                last_msg = await bot.send_message(
    +                last_msg = await _send_telegram_message_with_retry(
    +                    bot,
                         chat_id=int_chat_id, text=formatted,
                         parse_mode=send_parse_mode, **thread_kwargs
                     )
    @@ -550,7 +595,8 @@ async def _send_telegram(token, chat_id, message, media_files=None, thread_id=No
                                 plain = message
                         else:
                             plain = message
    -                    last_msg = await bot.send_message(
    +                    last_msg = await _send_telegram_message_with_retry(
    +                        bot,
                             chat_id=int_chat_id, text=plain,
                             parse_mode=None, **thread_kwargs
                         )

• #35387 fix(gateway): never auto-pause platforms on transient network/DNS failures

  merged to main

  files: gateway/run.py, tests/gateway/test_platform_reconnect.py

  Show the diff

  • gateway/run.py

    @@ -1734,9 +1734,9 @@ def __init__(self, config: Optional[GatewayConfig] = None):
             # as a fallback when a fresh config read transiently returns an empty
             # model (e.g. an mtime-keyed config-cache miss during a post-interrupt
             # recovery turn). Without this, the agent is built with model="" and
    -        # every API call fails HTTP 400 "No models provided" — the session
    -        # goes silent until the user manually re-sends. See #35314. The "*"
    -        # key holds a process-wide last-known-good for first-seen sessions.
    +        # every API call fails HTTP 400 "No models provided" — the session goes
    +        # silent until the user manually re-sends. See #35314. ``"*"`` holds a
    +        # process-wide last-known-good for sessions seen for the first time.
             self._last_resolved_model: Dict[str, str] = {}
             # Overflow buffer for explicit /queue commands.  The adapter-level
             # _pending_messages dict is a single slot per session (designed for
    @@ -2502,8 +2502,8 @@ def _resolve_session_agent_runtime(
             # successfully resolved for this session (or, failing that, the most
             # recent one resolved process-wide). Building an agent with model=""
             # makes every API call fail HTTP 400 "No models provided" and the
    -        # session goes silent until the user manually re-sends. getattr guards
    -        # against bare test runners built via object.__new__.
    +        # session goes silent until the user manually re-sends. ``getattr``
    +        # guards against bare test runners built via ``object.__new__``.
             _last_good = getattr(self, "_last_resolved_model", None)
             if _last_good is not None:
                 if not model:
    @@ -2818,10 +2818,12 @@ def _pause_failed_platform(self, platform, *, reason: str = "") -> None:
             """Mark a queued platform as paused — keep it in ``_failed_platforms``
             but stop the reconnect watcher from hammering it.
     
    -        Used by the circuit breaker after ``_PAUSE_AFTER_FAILURES`` consecutive
    -        retryable failures, and by ``/platform pause <name>`` for manual
    -        intervention.  Paused platforms are surfaced in ``/platform list``
    -        and resumed with ``/platform resume <name>``.
    +        Used by ``/platform pause <name>`` for manual operator intervention.
    +        Paused platforms are surfaced in ``/platform list`` and resumed with
    +        ``/platform resume <name>``.  Note: the reconnect watcher does NOT
    +        auto-pause — retryable (network/DNS) failures keep retrying at the
    +        backoff cap indefinitely so a transient outage self-heals without
    +        manual intervention.
             """
             info = getattr(self, "_failed_platforms", {}).get(platform)
             if info is None:
    @@ -5899,15 +5901,17 @@ async def _platform_reconnect_watcher(self) -> None:
             """Background task that periodically retries connecting failed platforms.
     
             Uses exponential backoff: 30s → 60s → 120s → 240s → 300s (cap).
    -        Retryable failures keep retrying at the backoff cap indefinitely
    -        — but if a platform fails ``_PAUSE_AFTER_FAILURES`` times in a row
    -        without ever succeeding, it is *paused*: kept in the retry queue
    -        but no longer hammered.  The user surfaces it with ``/platform list``
    -        and resumes it with ``/platform resume <name>``.  Non-retryable
    -        failures (bad auth, etc.) still drop out of the queue immediately.
    +        Retryable failures (network/DNS blips) keep retrying at the backoff
    +        cap indefinitely — they self-heal once connectivity returns, so a
    +        transient outage never requires manual intervention. Non-retryable
    +        failures (bad auth, etc.) drop out of the queue immediately. The
    +        circuit breaker (``_pause_failed_platform`` / ``/platform pause``)
    +        remains available for manual operator control via ``/platform list``
    +        and ``/platform resume <name>``, but is no longer triggered
    +        automatically — auto-pausing a recovered platform was the cause of
    +        bots silently staying dead after a transient DNS failure.
             """
             _BACKOFF_CAP = 300  # 5 minutes max between retries
    -        _PAUSE_AFTER_FAILURES = 10  # circuit-breaker threshold
     
             await asyncio.sleep(10)  # initial delay — let startup finish
             while self._running:
    @@ -6002,14 +6006,14 @@ async def _platform_reconnect_watcher(self) -> None:
                                 "Reconnect %s failed, next retry in %ds",
                                 platform.value, backoff,
                             )
    -                        if attempt >= _PAUSE_AFTER_FAILURES:
    -                            self._pause_failed_platform(
    -                                platform,
    -                                reason=(
    -                                    adapter.fatal_error_message
    -                                    or "failed to reconnect"
    -                                ),
    -                            )
    +                        # Retryable failures (network/DNS blips) keep retrying
    +                        # at the backoff cap indefinitely — they self-heal once
    +                        # connectivity returns. We do NOT auto-pause them: a
    +                        # transient outage must never require manual `/platform
    +                        # resume` to recover. Non-retryable failures (bad auth,
    +                        # etc.) already drop out of the queue via the
    +                        # `not fatal_error_retryable` branch above, so anything
    +                        # reaching here is by definition retryable.
                     except Exception as e:
                         self._update_platform_runtime_status(
                             platform.value,
    @@ -6024,8 +6028,9 @@ async def _platform_reconnect_watcher(self) -> None:
                             "Reconnect %s error: %s, next retry in %ds",
                             platform.value, e, backoff,
                         )
    -                    if attempt >= _PAUSE_AFTER_FAILURES:
    -                        self._pause_failed_platform(platform, reason=str(e))
    +                    # A raised exception during reconnect (connect timeout, DNS
    +                    # resolution failure, etc.) is inherently transient — keep
    +                    # retrying at the backoff cap rather than auto-pausing.
     
                 # Check every 10 seconds for platforms that need reconnection
                 for _ in range(10):

  • tests/gateway/test_platform_reconnect.py

    @@ -294,19 +294,20 @@ async def fake_sleep(n):
             assert runner._failed_platforms[Platform.TELEGRAM]["attempts"] == 2
     
         @pytest.mark.asyncio
    -    async def test_reconnect_pauses_after_circuit_breaker_threshold(self):
    -        """After enough consecutive retryable failures, the watcher should
    -        *pause* the platform (keep it in the queue but stop hammering it),
    -        not drop it. The user resumes via /platform resume.
    +    async def test_reconnect_never_auto_pauses_retryable_failures(self):
    +        """Retryable failures (network/DNS) must keep retrying indefinitely —
    +        the watcher must NOT auto-pause them. Auto-pausing a transiently-failed
    +        platform left bots silently dead after a DNS blip (#35284). The pause
    +        circuit breaker remains available for manual /platform pause only.
             """
             runner = _make_runner()
     
             platform_config = PlatformConfig(enabled=True, token="test")
    -        # 9 prior attempts — the next failure will be the 10th and should
    -        # trip the circuit breaker.
    +        # Far past the old circuit-breaker threshold (10): even after many
    +        # consecutive retryable failures the platform must stay unpaused.
             runner._failed_platforms[Platform.TELEGRAM] = {
                 "config": platform_config,
    -            "attempts": 9,
    +            "attempts": 25,
                 "next_retry": time.monotonic() - 1,
             }
     
    @@ -332,12 +333,15 @@ async def fake_sleep(n):
     
                 await run_one_iteration()
     
    -        # Platform stays in queue — paused, not dropped
    +        # Platform stays in queue and keeps retrying — never auto-paused.
             assert Platform.TELEGRAM in runner._failed_platforms
             info = runner._failed_platforms[Platform.TELEGRAM]
    -        assert info["paused"] is True
    -        assert info["attempts"] == 10
    -        assert "pause_reason" in info
    +        assert info.get("paused") is not True
    +        assert "pause_reason" not in info
    +        assert info["attempts"] == 26
    +        # next_retry is pushed out by the backoff (capped at 300s), not inf.
    +        assert info["next_retry"] != float("inf")
    +        assert info["next_retry"] > time.monotonic()
     
         @pytest.mark.asyncio
         async def test_reconnect_skips_paused_platforms(self):

• #50368 fix(email): resolve IMAP/SMTP host from config.extra and validate before connecting (#49736)

  merged to main

  files: plugins/platforms/email/adapter.py, scripts/release.py, tests/gateway/test_email.py

  Show the diff

  • plugins/platforms/email/adapter.py

    @@ -159,14 +159,16 @@ def _is_automated_sender(address: str, headers: dict) -> bool:
         return False
         
     def check_email_requirements() -> bool:
    -    """Check if email platform dependencies are available."""
    -    addr = os.getenv("EMAIL_ADDRESS")
    -    pwd = os.getenv("EMAIL_PASSWORD")
    -    imap = os.getenv("EMAIL_IMAP_HOST")
    -    smtp = os.getenv("EMAIL_SMTP_HOST")
    -    if not all([addr, pwd, imap, smtp]):
    -        return False
    -    return True
    +    """Check if email platform settings are available and non-blank.
    +
    +    Treats blank/whitespace-only values as missing so an abandoned setup that
    +    left empty ``EMAIL_*`` keys in ``.env`` does not enable the platform (#40715).
    +    """
    +    addr = os.getenv("EMAIL_ADDRESS", "").strip()
    +    pwd = os.getenv("EMAIL_PASSWORD", "").strip()
    +    imap = os.getenv("EMAIL_IMAP_HOST", "").strip()
    +    smtp = os.getenv("EMAIL_SMTP_HOST", "").strip()
    +    return all([addr, pwd, imap, smtp])
     
     
     def _decode_header_value(raw: str) -> str:
    @@ -307,11 +309,20 @@ class EmailAdapter(BasePlatformAdapter):
         def __init__(self, config: PlatformConfig):
             super().__init__(config, Platform.EMAIL)
     
    -        self._address = os.getenv("EMAIL_ADDRESS", "")
    +        # Resolve connection settings from the env vars first, then fall back to
    +        # PlatformConfig.extra (address/imap_host/smtp_host) — the canonical dict
    +        # gateway.config populates and that the "connected" check, the
    +        # send-helper, and `hermes config show` already read. Without the
    +        # fallback a config.yaml-only setup left these empty. Host/address values
    +        # are stripped: a stray space or newline made IMAP4_SSL raise the
    +        # misleading ``[Errno 8] nodename nor servname`` (an unresolvable name)
    +        # instead of an obvious "host not set" error.
    +        extra = config.extra or {}
    +        self._address = (os.getenv("EMAIL_ADDRESS", "") or extra.get("address", "")).strip()
             self._password = [redacted:secret-assignment]"EMAIL_PASSWORD", "")
    -        self._imap_host = os.getenv("EMAIL_IMAP_HOST", "")
    +        self._imap_host = (os.getenv("EMAIL_IMAP_HOST", "") or extra.get("imap_host", "")).strip()
             self._imap_port = env_int("EMAIL_IMAP_PORT", 993)
    -        self._smtp_host = os.getenv("EMAIL_SMTP_HOST", "")
    +        self._smtp_host = (os.getenv("EMAIL_SMTP_HOST", "") or extra.get("smtp_host", "")).strip()
             self._smtp_port = env_int("EMAIL_SMTP_PORT", 587)
             self._poll_interval = env_int("EMAIL_POLL_INTERVAL", 15)
     
    @@ -319,7 +330,6 @@ def __init__(self, config: PlatformConfig):
             #   platforms:
             #     email:
             #       skip_attachments: true
    -        extra = config.extra or {}
             self._skip_attachments = extra.get("skip_attachments", False)
     
             # Track message IDs we've already processed to avoid duplicates
    @@ -396,6 +406,36 @@ def _connect(*, ipv4_only: bool = False) -> smtplib.SMTP:
     
         async def connect(self) -> bool:
             """Connect to the IMAP server and start polling for new messages."""
    +        # Validate up front so a missing host surfaces as an actionable config
    +        # error instead of IMAP4_SSL("") raising the cryptic
    +        # ``[Errno 8] nodename nor servname provided, or not known``.
    +        missing = [
    +            name
    +            for name, value in (
    +                ("EMAIL_ADDRESS", self._address),
    +                ("EMAIL_PASSWORD", self._password),
    +                ("EMAIL_IMAP_HOST", self._imap_host),
    +                ("EMAIL_SMTP_HOST", self._smtp_host),
    +            )
    +            if not value
    +        ]
    +        if missing:
    +            message = (
    +                "Not configured — missing "
    +                + ", ".join(missing)
    +                + ". Set it via `hermes gateway setup` (env) or platforms.email "
    +                "in config.yaml."
    +            )
    +            logger.error("[Email] %s", message)
    +            # Mark non-retryable so the gateway does NOT keep reconnecting against
    +            # an empty host. A blank-but-present env var (e.g. ``EMAIL_IMAP_HOST=``)
    +            # used to slip past the startup gate and drive an indefinite retry
    +            # loop that leaked memory until the host OOM-killed (#40715).
    +            self._set_fatal_error(
    +                "email_missing_configuration", message, retryable=False
    +            )
    +            return False
    +
             try:
                 # Test IMAP connection
                 imap = imaplib.IMAP4_SSL(self._imap_host, self._imap_port, timeout=30)

  • scripts/release.py

    @@ -120,6 +120,7 @@
         "[redacted:email]": "synapsesx",
         "[redacted:email]": "itsflownium",
         "[redacted:email]": "dirtyren",
    +    "[redacted:email]": "devorun",
         "[redacted:email]": "x7peeps",
         "[redacted:email]": "andressommerhoff",
         "[redacted:email]": "x7peeps",

  • tests/gateway/test_email.py

    @@ -1392,5 +1392,95 @@ def test_ipv4_connection_does_not_mutate_global_resolver(self):
             self.assertIs(_socket.getaddrinfo, original_getaddrinfo)
     
     
    +class TestConnectionConfigResolution(unittest.TestCase):
    +    """Host/address resolution and pre-connect validation (#49736)."""
    +
    +    def test_host_and_address_whitespace_stripped(self):
    +        """A stray space/newline must not reach IMAP4_SSL as part of the host.
    +
    +        Whitespace in the host produced the misleading
    +        ``[Errno 8] nodename nor servname`` (unresolvable name) instead of a
    +        successful connection.
    +        """
    +        from gateway.config import PlatformConfig
    +        from plugins.platforms.email.adapter import EmailAdapter
    +        with patch.dict(os.environ, {
    +            "EMAIL_ADDRESS": "  [redacted:email]\n",
    +            "EMAIL_PASSWORD": "secret",
    +            "EMAIL_IMAP_HOST": " imap.test.com ",
    +            "EMAIL_SMTP_HOST": "smtp.test.com\n",
    +        }, clear=False):
    +            adapter = EmailAdapter(PlatformConfig(enabled=True))
    +        self.assertEqual(adapter._imap_host, "imap.test.com")
    +        self.assertEqual(adapter._smtp_host, "smtp.test.com")
    +        self.assertEqual(adapter._address, "[redacted:email]")
    +
    +    def test_falls_back_to_platform_config_extra(self):
    +        """When env vars are absent, settings come from PlatformConfig.extra —
    +        the same dict gateway.config populates and `hermes config show` reads."""
    +        from gateway.config import PlatformConfig
    +        from plugins.platforms.email.adapter import EmailAdapter
    +        cfg = PlatformConfig(enabled=True)
    +        cfg.extra.update({
    +            "address": "[redacted:email]",
    +            "imap_host": "imap.test.com",
    +            "smtp_host": "smtp.test.com",
    +        })
    +        with patch.dict(os.environ, {
    +            "EMAIL_ADDRESS": "", "EMAIL_IMAP_HOST": "", "EMAIL_SMTP_HOST": "",
    +            "EMAIL_PASSWORD": "secret",
    +        }, clear=False):
    +            adapter = EmailAdapter(cfg)
    +        self.assertEqual(adapter._imap_host, "imap.test.com")
    +        self.assertEqual(adapter._smtp_host, "smtp.test.com")
    +        self.assertEqual(adapter._address, "[redacted:email]")
    +
    +    def test_connect_aborts_without_attempting_imap_when_host_missing(self):
    +        """A missing host returns False without the cryptic DNS error, and marks
    +        the failure non-retryable so the gateway stops reconnecting (#40715)."""
    +        import asyncio
    +        from gateway.config import PlatformConfig
    +        from plugins.platforms.email.adapter import EmailAdapter
    +        with patch.dict(os.environ, {
    +            "EMAIL_ADDRESS": "[redacted:email]",
    +            "EMAIL_PASSWORD": "secret",
    +            "EMAIL_IMAP_HOST": "",
    +            "EMAIL_SMTP_HOST": "smtp.test.com",
    +        }, clear=False):
    +            adapter = EmailAdapter(PlatformConfig(enabled=True))
    +
    +        with patch("imaplib.IMAP4_SSL") as mock_imap:
    +            result = asyncio.run(adapter.connect())
    +
    +        self.assertFalse(result)
    +        mock_imap.assert_not_called()
    +        # The OOM fix (#40715): a blank host must NOT leave the platform in the
    +        # retryable reconnect loop — it is a permanent config error.
    +        self.assertTrue(adapter.has_fatal_error)
    +        self.assertEqual(adapter.fatal_error_code, "email_missing_configuration")
    +        self.assertFalse(adapter.fatal_error_retryable)
    +        self.assertIn("EMAIL_IMAP_HOST", adapter.fatal_error_message or "")
    +
    +    def test_blank_present_env_vars_are_not_required(self):
    +        """Blank/whitespace EMAIL_* values must read as missing (#40715) — an
    +        abandoned setup with empty keys must not enable the platform."""
    +        from plugins.platforms.email.adapter import check_email_requirements
    +        for blank in ("", "   ", "\n"):
    +            with patch.dict(os.environ, {
    +                "EMAIL_ADDRESS": blank, "EMAIL_PASSWORD": blank,
    +                "EMAIL_IMAP_HOST": blank, "EMAIL_SMTP_HOST": blank,
    +            }, clear=False):
    +                self.assertFalse(check_email_requirements())
    +
    +    def test_all_settings_present_satisfies_requirements(self):
    +        """The connected check passes only when all four settings are non-blank."""
    +        from plugins.platforms.email.adapter import check_email_requirements
    +        with patch.dict(os.environ, {
    +            "EMAIL_ADDRESS": "[redacted:email]", "EMAIL_PASSWORD": "secret",
    +            "EMAIL_IMAP_HOST": "imap.test.com", "EMAIL_SMTP_HOST": "smtp.test.com",
    +        }, clear=False):
    +            self.assertTrue(check_email_requirements())
    +
    +
     if __name__ == "__main__":
         unittest.main()