fix(gateway): add IPv4 fallback for IMAP connections on unreachable IPv6

plugins/platforms/email/adapter.py

@@ -121,6 +121,16 @@ def _get_socket(self, host, port, timeout):  # type: ignore[override]
             server_hostname=getattr(self, "_host", host),
         )
 
+
+IMAP_CONNECT_TIMEOUT = 30
+
+
+class _IPv4IMAP4_SSL(imaplib.IMAP4_SSL):
+    def _create_socket(self, timeout):  # type: ignore[override]
+        raw_sock = _create_ipv4_connection(self.host, self.port, timeout)
+        return self.ssl_context.wrap_socket(raw_sock, server_hostname=self.host)
+
+
 # Supported image extensions for inline detection
 _IMAGE_EXTS = {".jpg", ".jpeg", ".png", ".gif", ".webp"}
 
@@ -404,6 +414,23 @@ def _connect(*, ipv4_only: bool = False) -> smtplib.SMTP:
             # Retry with IPv4 only.
             return _connect(ipv4_only=True)
 
+    def _connect_imap(self) -> imaplib.IMAP4_SSL:
+        """Create an IMAP4_SSL connection, falling back to IPv4-only on failure.
+
+        Mirrors :meth:`_connect_smtp`: when the host resolves to an IPv6
+        address that is unreachable, the default connection hangs until socket
+        timeout. We retry through an IPv4-only socket path. TLS verification
+        errors are not retried.
+        """
+        host = self._imap_host
+        port = self._imap_port
+        try:
+            return imaplib.IMAP4_SSL(host, port, timeout=IMAP_CONNECT_TIMEOUT)
+        except (socket.timeout, TimeoutError, ConnectionError, OSError) as exc:
+            if isinstance(exc, ssl.SSLError):
+                raise
+            return _IPv4IMAP4_SSL(host, port, timeout=IMAP_CONNECT_TIMEOUT)
+
     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
@@ -438,7 +465,7 @@ async def connect(self) -> bool:
 
         try:
             # Test IMAP connection
-            imap = imaplib.IMAP4_SSL(self._imap_host, self._imap_port, timeout=30)
+            imap = self._connect_imap()
             imap.login(self._address, self._password)
             _send_imap_id(imap)
             # Mark all existing messages as seen so we only process new ones
@@ -507,7 +534,7 @@ def _fetch_new_messages(self) -> List[Dict[str, Any]]:
         """Fetch new (unseen) messages from IMAP. Runs in executor thread."""
         results = []
         try:
-            imap = imaplib.IMAP4_SSL(self._imap_host, self._imap_port, timeout=30)
+            imap = self._connect_imap()
             try:
                 imap.login(self._address, self._password)
                 _send_imap_id(imap)

tests/gateway/test_email.py

@@ -1392,6 +1392,88 @@ def test_ipv4_connection_does_not_mutate_global_resolver(self):
         self.assertIs(_socket.getaddrinfo, original_getaddrinfo)
 
 
+class TestConnectImap(unittest.TestCase):
+    """Test _connect_imap() helper: IPv6 fallback for IMAP4_SSL connections."""
+
+    def _make_adapter(self):
+        from gateway.config import PlatformConfig
+        with patch.dict(os.environ, {
+            "EMAIL_ADDRESS": "hermes@test.com",
+            "EMAIL_PASSWORD": "secret",
+            "EMAIL_IMAP_HOST": "imap.test.com",
+            "EMAIL_IMAP_PORT": "993",
+            "EMAIL_SMTP_HOST": "smtp.test.com",
+        }):
+            from plugins.platforms.email.adapter import EmailAdapter
+            return EmailAdapter(PlatformConfig(enabled=True))
+
+    def test_default_connection_used_when_reachable(self):
+        """When IMAP4_SSL connects normally, no IPv4 fallback is attempted."""
+        from plugins.platforms.email import adapter as email_mod
+
+        adapter = self._make_adapter()
+
+        with patch("imaplib.IMAP4_SSL") as mock_imap_ssl, \
+             patch.object(email_mod, "_IPv4IMAP4_SSL") as mock_ipv4_imap_ssl:
+            mock_server = MagicMock()
+            mock_imap_ssl.return_value = mock_server
+
+            result = adapter._connect_imap()
+
+            mock_imap_ssl.assert_called_once_with("imap.test.com", 993, timeout=30)
+            mock_ipv4_imap_ssl.assert_not_called()
+            self.assertIs(result, mock_server)
+
+    def test_ipv6_timeout_falls_back_to_ipv4(self):
+        """When the default connection times out, retry with IPv4-only IMAP."""
+        import socket as _socket
+        from plugins.platforms.email import adapter as email_mod
+
+        adapter = self._make_adapter()
+
+        with patch("imaplib.IMAP4_SSL", side_effect=_socket.timeout("timed out")), \
+             patch.object(email_mod, "_IPv4IMAP4_SSL") as mock_ipv4_imap_ssl:
+            mock_server = MagicMock()
+            mock_ipv4_imap_ssl.return_value = mock_server
+
+            result = adapter._connect_imap()
+
+            self.assertIs(result, mock_server)
+            mock_ipv4_imap_ssl.assert_called_once_with("imap.test.com", 993, timeout=30)
+
+    def test_tls_verification_error_does_not_retry_ipv4(self):
+        """Certificate failures are security errors, not IPv6 reachability failures."""
+        import ssl as _ssl
+        from plugins.platforms.email import adapter as email_mod
+
+        adapter = self._make_adapter()
+
+        with patch("imaplib.IMAP4_SSL", side_effect=_ssl.SSLError("cert verify failed")), \
+             patch.object(email_mod, "_IPv4IMAP4_SSL") as mock_ipv4_imap_ssl:
+            with self.assertRaises(_ssl.SSLError):
+                adapter._connect_imap()
+
+            mock_ipv4_imap_ssl.assert_not_called()
+
+    def test_connect_uses_connect_imap(self):
+        """connect() must go through _connect_imap(), not call IMAP4_SSL directly."""
+        import asyncio
+
+        adapter = self._make_adapter()
+
+        mock_imap = MagicMock()
+        mock_imap.uid.return_value = ("OK", [b""])
+        with patch.object(adapter, "_connect_imap", return_value=mock_imap) as mock_ci, \
+             patch.object(adapter, "_connect_smtp") as mock_cs:
+            mock_smtp = MagicMock()
+            mock_cs.return_value = mock_smtp
+
+            result = asyncio.run(adapter.connect())
+
+            self.assertTrue(result)
+            mock_ci.assert_called()
+
+
 class TestConnectionConfigResolution(unittest.TestCase):
     """Host/address resolution and pre-connect validation (#49736)."""