fix(email): surface IMAP fetch failures through the fatal-error hook (#80016)

_fetch_new_messages() wrapped the whole IMAP connect/login/select/search/
fetch sequence in a bare except that logged and returned an empty list —
indistinguishable from a genuinely empty inbox. The adapter never invoked
its fatal-error handler, so the gateway's reconnect/backoff/status
machinery never learned the mailbox was unreachable; outages lasted until
a manual restart.

Track fetch failure on the adapter and, when the poll loop observes it,
set a retryable fatal error (email_imap_fetch_failed) and notify the
gateway handler so the platform enters the reconnect queue just like a
startup connection failure.

plugins/platforms/email/adapter.py

@@ -566,6 +566,11 @@ class EmailAdapter(BasePlatformAdapter):
         self._seen_uids_max: int = 2000   # cap to prevent unbounded memory growth
         self._poll_task: Optional[asyncio.Task] = None
 
+        # Track the last IMAP fetch attempt so the poll loop can distinguish
+        # "checked, nothing new" from "the check itself failed" (#80016).
+        self._last_fetch_failed: bool = False
+        self._last_fetch_error: str = ""
+
         # Map chat_id (sender email) -> last subject + message-id for threading
         self._thread_context: Dict[str, Dict[str, str]] = {}
 
@@ -763,6 +768,22 @@ class EmailAdapter(BasePlatformAdapter):
         # Run IMAP operations in a thread to avoid blocking the event loop
         loop = asyncio.get_running_loop()
         messages = await loop.run_in_executor(None, self._fetch_new_messages)
+        if self._last_fetch_failed:
+            # The IMAP check itself failed (connect/login/select/search/fetch),
+            # not just an empty inbox. Surface it through the fatal-error hook
+            # so the gateway's existing reconnect/backoff/status machinery
+            # re-establishes the mailbox instead of silently treating every
+            # failed check as "nothing new" (#80016). The handler runs in a
+            # detached task (gateway/run.py), so awaiting it from our own poll
+            # task is safe even though teardown cancels this task.
+            self._last_fetch_failed = False
+            self._set_fatal_error(
+                "email_imap_fetch_failed",
+                self._last_fetch_error or "IMAP fetch failed",
+                retryable=True,
+            )
+            await self._notify_fatal_error()
+            return
         for msg_data in messages:
             await self._dispatch_message(msg_data)
 
@@ -862,6 +883,8 @@ class EmailAdapter(BasePlatformAdapter):
                     pass
         except Exception as e:
             logger.error("[Email] IMAP fetch error: %s", e)
+            self._last_fetch_failed = True
+            self._last_fetch_error = str(e)
         return results
 
     @staticmethod

tests/gateway/test_email.py

@@ -581,6 +581,31 @@ class TestPollLoop(unittest.TestCase):
         self.assertEqual(len(dispatched), 1)
         self.assertEqual(dispatched[0]["subject"], "Inbox Test")
 
+    def test_check_inbox_notifies_fatal_error_on_fetch_failure(self):
+        """A failed IMAP check must surface through the fatal-error hook so
+        the gateway's reconnect/backoff machinery learns email is unhealthy
+        instead of silently treating the failed check as an empty inbox
+        (#80016)."""
+        import asyncio
+        adapter = self._make_adapter()
+        notified = []
+
+        async def mock_fatal_handler(adapter):
+            notified.append(adapter)
+
+        adapter.set_fatal_error_handler(mock_fatal_handler)
+
+        mock_imap = MagicMock()
+        mock_imap.login.side_effect = Exception("read operation timed out")
+
+        with patch("imaplib.IMAP4_SSL", return_value=mock_imap):
+            asyncio.run(adapter._check_inbox())
+
+        self.assertEqual(len(notified), 1)
+        self.assertEqual(adapter.fatal_error_code, "email_imap_fetch_failed")
+        self.assertTrue(adapter.fatal_error_retryable)
+        self.assertIn("read operation timed out", adapter.fatal_error_message)
+
 
 class TestSendEmailStandalone(unittest.TestCase):
     """Test the standalone _send_email function in send_message_tool."""