From 30da57511991e8975bf66f66cd585ffc103ac92f Mon Sep 17 00:00:00 2001 From: alarcritty Date: Tue, 23 Jun 2026 00:29:03 +0530 Subject: [PATCH 1/4] fix(connectors): harden IMAP sync header decoding and reconnect Two reliability fixes in the shared IMAP sync used by the Gmail, Outlook, and generic IMAP connectors: - Decode every header (From, To, Subject, In-Reply-To, Message-ID) to a plain string. Real mail with raw 8-bit header bytes makes imaplib return an email.header.Header object, which crashed json.dumps in the knowledge store ("Object of type Header is not JSON serializable") and aborted the sync. The decoder also tolerates bogus charsets like unknown-8bit. - Reconnect and retry on a dropped TLS connection mid-backfill (for example "[SSL: BAD_LENGTH]") instead of aborting the whole run, so a large inbox finishes without manual retries. --- src/openjarvis/connectors/gmail_imap.py | 97 +++++++++++++++++++------ 1 file changed, 76 insertions(+), 21 deletions(-) diff --git a/src/openjarvis/connectors/gmail_imap.py b/src/openjarvis/connectors/gmail_imap.py index 61836bae7..4538db1fb 100644 --- a/src/openjarvis/connectors/gmail_imap.py +++ b/src/openjarvis/connectors/gmail_imap.py @@ -14,6 +14,7 @@ from datetime import datetime from email.header import decode_header from email.utils import parsedate_to_datetime +from imaplib import IMAP4 from typing import Iterator, List, Optional from openjarvis.connectors._stubs import BaseConnector, Document, SyncStatus @@ -27,17 +28,26 @@ _DEFAULT_CREDENTIALS_PATH = str(DEFAULT_CONFIG_DIR / "connectors" / "gmail_imap.json") -def _decode_subject(raw: str) -> str: - """Decode a possibly-encoded email subject header.""" +def _decode_header(raw: object) -> str: + """Decode an email header (str or Header) to a safe plain string. + + Real mail can carry raw 8-bit bytes in headers, which makes ``msg.get()`` + return an ``email.header.Header`` (not JSON serializable) and yields bogus + charsets like ``unknown-8bit``. Coerce to a string, replacing what can't be + decoded. + """ if not raw: return "" - decoded_parts = decode_header(raw) - return "".join( - part.decode(enc or "utf-8", errors="replace") - if isinstance(part, bytes) - else part - for part, enc in decoded_parts - ) + out = [] + for part, enc in decode_header(raw if isinstance(raw, str) else str(raw)): + if isinstance(part, bytes): + try: + out.append(part.decode(enc or "utf-8", errors="replace")) + except LookupError: + out.append(part.decode("utf-8", errors="replace")) + else: + out.append(part) + return "".join(out) def _extract_text_body(msg: email_lib.message.Message) -> str: @@ -139,6 +149,14 @@ def handle_callback(self, code: str) -> None: {"email": "", "password": code.strip()}, ) + def _open_imap(self) -> "imaplib.IMAP4_SSL": + """Open an authenticated, INBOX-selected IMAP connection.""" + em, pw = self._resolve_credentials() + imap = imaplib.IMAP4_SSL(self._imap_host) + imap.login(em, pw) + imap.select("INBOX", readonly=True) + return imap + def sync( self, *, @@ -149,15 +167,12 @@ def sync( if not em or not pw: return - imap = imaplib.IMAP4_SSL(self._imap_host) try: - imap.login(em, pw) - except imaplib.IMAP4.error as exc: + imap = self._open_imap() + except (IMAP4.error, OSError) as exc: logger.error("IMAP login failed: %s", exc) return - imap.select("INBOX", readonly=True) - # Always SEARCH ALL. IMAP has no native cursor that survives a # server restart, so applying the SyncEngine's ``since`` filter # during a partial backfill would silently skip the older @@ -178,22 +193,59 @@ def sync( ordered = ordered[: self._max_messages] synced = 0 - for mid in ordered: + # A dropped TLS connection mid-backfill (e.g. "[SSL: BAD_LENGTH]") is + # transient. Reconnect and retry the message instead of aborting the + # whole sync, so a large inbox finishes without manual "Retry Sync". + # OSError covers ssl.SSLError and socket errors; IMAP4.abort is raised + # when the server drops the connection. + max_reconnects = 5 + consecutive_reconnects = 0 + index = 0 + while index < len(ordered): + mid = ordered[index] try: _, msg_data = imap.fetch(mid, "(RFC822)") raw = msg_data[0][1] msg = email_lib.message_from_bytes(raw) + except (IMAP4.abort, OSError) as exc: + if consecutive_reconnects >= max_reconnects: + logger.error( + "IMAP sync stopping after %d reconnect attempts: %s", + consecutive_reconnects, + exc, + ) + break + consecutive_reconnects += 1 + logger.warning( + "IMAP connection dropped (%s); reconnecting %d/%d", + exc, + consecutive_reconnects, + max_reconnects, + ) + try: + imap.logout() + except Exception: + pass + try: + imap = self._open_imap() + except (IMAP4.error, OSError) as reconnect_exc: + logger.error("IMAP reconnect failed: %s", reconnect_exc) + break + continue except Exception: + index += 1 continue - subject = _decode_subject(msg.get("Subject", "")) - sender = msg.get("From", "") - to = msg.get("To", "") + subject = _decode_header(msg.get("Subject", "")) + sender = _decode_header(msg.get("From", "")) + to = _decode_header(msg.get("To", "")) body = _extract_text_body(msg) timestamp = _parse_date(msg) - message_id = msg.get("Message-ID", mid.decode()) + message_id = str(msg.get("Message-ID", mid.decode())) synced += 1 + index += 1 + consecutive_reconnects = 0 yield Document( doc_id=f"gmail:{message_id}", source="gmail", @@ -203,14 +255,17 @@ def sync( author=sender, participants=[a.strip() for a in (to or "").split(",") if a.strip()], timestamp=timestamp, - thread_id=msg.get("In-Reply-To", ""), + thread_id=_decode_header(msg.get("In-Reply-To", "")), url="https://mail.google.com/mail/u/0/#inbox", metadata={ "message_id": message_id, }, ) - imap.logout() + try: + imap.logout() + except Exception: + pass self._items_synced = synced def sync_status(self) -> SyncStatus: From 6a78e13f88aae51ad4ecd256c41057718399bcfc Mon Sep 17 00:00:00 2001 From: alarcritty Date: Tue, 23 Jun 2026 00:29:16 +0530 Subject: [PATCH 2/4] feat(connectors): add generic IMAP connector for any provider (#518) Add a single configuration-light IMAP connector that works with any email provider instead of a hardcoded subclass per provider. The IMAP host is resolved from the email domain (a small table of common providers, falling back to imap.), so most users only supply an email address and an app password. Registered in connectors/__init__.py so it appears in the Data Sources API and UI. Tests cover registration, host resolution, credential round-trip, document output, raw 8-bit headers, and reconnect-on-drop. Refs #518. --- src/openjarvis/connectors/__init__.py | 5 + src/openjarvis/connectors/imap.py | 98 ++++++++++++++++ tests/connectors/test_imap.py | 159 ++++++++++++++++++++++++++ 3 files changed, 262 insertions(+) create mode 100644 src/openjarvis/connectors/imap.py create mode 100644 tests/connectors/test_imap.py diff --git a/src/openjarvis/connectors/__init__.py b/src/openjarvis/connectors/__init__.py index 6aa3ad2cd..8e400e841 100644 --- a/src/openjarvis/connectors/__init__.py +++ b/src/openjarvis/connectors/__init__.py @@ -73,6 +73,11 @@ except ImportError: pass +try: + import openjarvis.connectors.imap # noqa: F401 +except ImportError: + pass + try: import openjarvis.connectors.gcalendar # noqa: F401 except ImportError: diff --git a/src/openjarvis/connectors/imap.py b/src/openjarvis/connectors/imap.py new file mode 100644 index 000000000..17b9721fb --- /dev/null +++ b/src/openjarvis/connectors/imap.py @@ -0,0 +1,98 @@ +"""Generic IMAP email connector for any provider.""" + +from __future__ import annotations + +from datetime import datetime +from typing import Iterator, Optional + +from openjarvis.connectors._stubs import Document +from openjarvis.connectors.gmail_imap import GmailIMAPConnector +from openjarvis.connectors.oauth import load_tokens, save_tokens +from openjarvis.core.config import DEFAULT_CONFIG_DIR +from openjarvis.core.registry import ConnectorRegistry + +_DEFAULT_CREDENTIALS_PATH = str(DEFAULT_CONFIG_DIR / "connectors" / "imap.json") + +_PROVIDER_HOSTS = { + "gmail.com": "imap.gmail.com", + "googlemail.com": "imap.gmail.com", + "outlook.com": "outlook.office365.com", + "hotmail.com": "outlook.office365.com", + "live.com": "outlook.office365.com", + "icloud.com": "imap.mail.me.com", + "me.com": "imap.mail.me.com", + "yahoo.com": "imap.mail.yahoo.com", + "fastmail.com": "imap.fastmail.com", + "aol.com": "imap.aol.com", + "zoho.com": "imap.zoho.com", + "gmx.com": "imap.gmx.com", +} + + +def resolve_imap_host(email_address: str) -> str: + """Resolve the IMAP host for an email address from its domain.""" + domain = email_address.rsplit("@", 1)[-1].lower() if "@" in email_address else "" + if not domain: + return "" + return _PROVIDER_HOSTS.get(domain, f"imap.{domain}") + + +@ConnectorRegistry.register("imap") +class IMAPConnector(GmailIMAPConnector): + """Generic IMAP connector for any email provider.""" + + connector_id = "imap" + display_name = "Email (IMAP)" + _default_imap_host = "" + + def __init__( + self, + email_address: str = "", + app_password: str = "", + credentials_path: str = "", + *, + imap_host: str = "", + max_messages: Optional[int] = None, + ) -> None: + super().__init__( + email_address, + app_password, + credentials_path or _DEFAULT_CREDENTIALS_PATH, + imap_host=imap_host, + max_messages=max_messages, + ) + + def auth_url(self) -> str: + return "" + + def handle_callback(self, code: str) -> None: + # code format: "email:password" + if ":" in code: + email_addr, password = code.split(":", 1) + email_addr, password = email_addr.strip(), password.strip() + else: + email_addr, password = "", code.strip() + save_tokens( + self._credentials_path, + { + "email": email_addr, + "password": password, + "imap_host": resolve_imap_host(email_addr), + }, + ) + + def sync( + self, + *, + since: Optional[datetime] = None, + cursor: Optional[str] = None, + ) -> Iterator[Document]: + if not self._imap_host: + tokens = load_tokens(self._credentials_path) or {} + self._imap_host = tokens.get("imap_host", "") or resolve_imap_host( + tokens.get("email", "") + ) + for doc in super().sync(since=since, cursor=cursor): + doc.source = self.connector_id + doc.doc_id = doc.doc_id.replace("gmail:", f"{self.connector_id}:", 1) + yield doc diff --git a/tests/connectors/test_imap.py b/tests/connectors/test_imap.py new file mode 100644 index 000000000..edf38bac5 --- /dev/null +++ b/tests/connectors/test_imap.py @@ -0,0 +1,159 @@ +"""Tests for IMAPConnector — generic IMAP connector with domain-resolved hosts.""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +from openjarvis.connectors.imap import IMAPConnector, resolve_imap_host +from openjarvis.connectors.oauth import load_tokens +from openjarvis.core.registry import ConnectorRegistry + + +def test_imap_registered() -> None: + ConnectorRegistry.register_value("imap", IMAPConnector) + assert ConnectorRegistry.contains("imap") + cls = ConnectorRegistry.get("imap") + assert cls.connector_id == "imap" + assert cls.display_name == "Email (IMAP)" + + +def test_resolve_known_provider_host() -> None: + assert resolve_imap_host("user@gmail.com") == "imap.gmail.com" + assert resolve_imap_host("user@outlook.com") == "outlook.office365.com" + assert resolve_imap_host("user@fastmail.com") == "imap.fastmail.com" + + +def test_resolve_unknown_domain_falls_back() -> None: + assert resolve_imap_host("user@example.org") == "imap.example.org" + assert resolve_imap_host("not-an-email") == "" + + +def test_imap_handle_callback(tmp_path: Path) -> None: + creds_path = str(tmp_path / "imap.json") + conn = IMAPConnector(credentials_path=creds_path) + conn.handle_callback("user@example.org:mypassword123") + tokens = load_tokens(creds_path) + assert tokens is not None + assert tokens["email"] == "user@example.org" + assert tokens["password"] == "mypassword123" + assert tokens["imap_host"] == "imap.example.org" + + +def test_imap_is_connected(tmp_path: Path) -> None: + creds_path = str(tmp_path / "imap.json") + conn = IMAPConnector(credentials_path=creds_path) + assert conn.is_connected() is False + conn.handle_callback("user@example.org:pass") + assert conn.is_connected() is True + + +def test_imap_sync_uses_resolved_host_and_source(tmp_path: Path) -> None: + creds_path = str(tmp_path / "imap.json") + conn = IMAPConnector(credentials_path=creds_path) + conn.handle_callback("user@example.org:pass") + + mock_imap = MagicMock() + mock_imap.login.return_value = ("OK", []) + mock_imap.select.return_value = ("OK", []) + mock_imap.search.return_value = ("OK", [b"1"]) + + raw_email = ( + b"From: sender@test.com\r\n" + b"To: user@example.org\r\n" + b"Subject: Test Email\r\n" + b"Date: Mon, 01 Jan 2024 00:00:00 +0000\r\n" + b"Message-ID: \r\n" + b"\r\n" + b"Hello from IMAP test" + ) + mock_imap.fetch.return_value = ("OK", [(b"1", raw_email)]) + mock_imap.logout.return_value = ("OK", []) + + with patch("openjarvis.connectors.gmail_imap.imaplib") as mock_imaplib: + mock_imaplib.IMAP4_SSL.return_value = mock_imap + mock_imaplib.IMAP4 = type(mock_imap) + docs = list(conn.sync()) + + mock_imaplib.IMAP4_SSL.assert_called_once_with("imap.example.org") + assert len(docs) == 1 + assert docs[0].source == "imap" + assert docs[0].doc_id.startswith("imap:") + assert docs[0].title == "Test Email" + + +def test_imap_sync_handles_raw_8bit_headers(tmp_path: Path) -> None: + import json + + creds_path = str(tmp_path / "imap.json") + conn = IMAPConnector(credentials_path=creds_path) + conn.handle_callback("user@example.org:pass") + + mock_imap = MagicMock() + mock_imap.login.return_value = ("OK", []) + mock_imap.select.return_value = ("OK", []) + mock_imap.search.return_value = ("OK", [b"1"]) + # Raw 8-bit (non-RFC2047) bytes in From/To/Subject -> Header objects. + raw_email = ( + b"From: Jos\xe9 \r\n" + b"To: caf\xe9 \r\n" + b"Subject: caf\xe9 meeting\r\n" + b"Date: Mon, 01 Jan 2024 00:00:00 +0000\r\n" + b"Message-ID: \r\n" + b"\r\n" + b"body" + ) + mock_imap.fetch.return_value = ("OK", [(b"1", raw_email)]) + mock_imap.logout.return_value = ("OK", []) + + with patch("openjarvis.connectors.gmail_imap.imaplib") as mock_imaplib: + mock_imaplib.IMAP4_SSL.return_value = mock_imap + mock_imaplib.IMAP4 = type(mock_imap) + docs = list(conn.sync()) + + assert len(docs) == 1 + d = docs[0] + assert isinstance(d.author, str) + assert isinstance(d.title, str) + assert all(isinstance(p, str) for p in d.participants) + # The fields must be JSON-serializable (the bug was a Header reaching json.dumps). + json.dumps( + { + "author": d.author, + "title": d.title, + "participants": d.participants, + **d.metadata, + } + ) + + +def test_imap_sync_reconnects_on_dropped_connection(tmp_path: Path) -> None: + creds_path = str(tmp_path / "imap.json") + conn = IMAPConnector(credentials_path=creds_path) + conn.handle_callback("user@example.org:pass") + + raw_email = ( + b"From: a@test.com\r\nTo: user@example.org\r\nSubject: Hi\r\n" + b"Date: Mon, 01 Jan 2024 00:00:00 +0000\r\n" + b"Message-ID: \r\n\r\nbody" + ) + + mock_imap = MagicMock() + mock_imap.login.return_value = ("OK", []) + mock_imap.select.return_value = ("OK", []) + mock_imap.search.return_value = ("OK", [b"1"]) + # First fetch drops the TLS connection; after reconnect it succeeds. + mock_imap.fetch.side_effect = [ + OSError("[SSL: BAD_LENGTH] bad length"), + ("OK", [(b"1", raw_email)]), + ] + mock_imap.logout.return_value = ("OK", []) + + with patch("openjarvis.connectors.gmail_imap.imaplib") as mock_imaplib: + mock_imaplib.IMAP4_SSL.return_value = mock_imap + docs = list(conn.sync()) + + # Reconnected (opened the connection twice) and still indexed the message. + assert mock_imaplib.IMAP4_SSL.call_count >= 2 + assert len(docs) == 1 + assert docs[0].doc_id.startswith("imap:") From 1b9009b772522607d6672bd6813a75083f8d4648 Mon Sep 17 00:00:00 2001 From: alarcritty Date: Tue, 23 Jun 2026 00:29:16 +0530 Subject: [PATCH 3/4] feat(web): add Email (IMAP) data source card (#518) Add a catalog entry for the generic IMAP connector so the Data Sources page renders an "Email (IMAP)" card with email and app-password inputs, alongside the existing Gmail and Outlook cards. Refs #518. --- frontend/src/types/connectors.ts | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/frontend/src/types/connectors.ts b/frontend/src/types/connectors.ts index 760913963..73807f993 100644 --- a/frontend/src/types/connectors.ts +++ b/frontend/src/types/connectors.ts @@ -120,6 +120,29 @@ export const SOURCE_CATALOG: ConnectorMeta[] = [ { name: 'password', placeholder: 'App password (xxxx xxxx xxxx xxxx)', type: 'password' }, ], }, + { + connector_id: 'imap', + display_name: 'Email (IMAP)', + auth_type: 'oauth', + category: 'communication', + icon: 'Mail', + color: 'text-amber-400', + description: 'Any IMAP mailbox', + unitLabel: 'emails', + steps: [ + { + label: 'Generate an app password with your email provider (IMAP must be enabled), then enter your email address and the app password below. The server is detected from your email domain.', + }, + ], + troubleshooting: [ + 'Most providers require an app password rather than your normal password.', + 'Connects over TLS on port 993; providers that only offer STARTTLS on port 143 are not supported yet.', + ], + inputFields: [ + { name: 'email', placeholder: 'you@example.com', type: 'text' }, + { name: 'password', placeholder: 'App password', type: 'password' }, + ], + }, { connector_id: 'slack', display_name: 'Slack', From 96ad5fd7ae81735c744a08514d86fc07f76f2492 Mon Sep 17 00:00:00 2001 From: alarcritty Date: Tue, 23 Jun 2026 00:29:32 +0530 Subject: [PATCH 4/4] feat(server): include knowledge_search in the default chat agent tools Add knowledge_search to the default tool set and inject a KnowledgeStore when building the chat and channel agents, so non-streaming agent requests can search ingested data. The streaming web chat path bypasses the agent, so this does not yet surface in the streaming UI. --- src/openjarvis/cli/serve.py | 30 ++++++++++++++++++++++++++---- 1 file changed, 26 insertions(+), 4 deletions(-) diff --git a/src/openjarvis/cli/serve.py b/src/openjarvis/cli/serve.py index ed95c35bd..8fd8808bc 100644 --- a/src/openjarvis/cli/serve.py +++ b/src/openjarvis/cli/serve.py @@ -295,7 +295,12 @@ def serve( from openjarvis.core.registry import ToolRegistry from openjarvis.tools._stubs import BaseTool - _DEFAULT_TOOLS = {"think", "calculator", "web_search"} + _DEFAULT_TOOLS = { + "think", + "calculator", + "web_search", + "knowledge_search", + } configured = config.agent.tools if configured: if isinstance(configured, list): @@ -319,7 +324,12 @@ def serve( if isinstance(tool_cls, type) and issubclass( tool_cls, BaseTool ): - tools.append(tool_cls()) + if name == "knowledge_search": + from openjarvis.connectors.store import KnowledgeStore + + tools.append(tool_cls(store=KnowledgeStore())) + else: + tools.append(tool_cls()) elif isinstance(tool_cls, BaseTool): tools.append(tool_cls) @@ -399,7 +409,12 @@ def serve( from openjarvis.core.registry import ToolRegistry from openjarvis.tools._stubs import BaseTool - _DEFAULT_TOOLS = {"think", "calculator", "web_search"} + _DEFAULT_TOOLS = { + "think", + "calculator", + "web_search", + "knowledge_search", + } configured = config.agent.tools if configured: if isinstance(configured, list): @@ -422,7 +437,14 @@ def serve( continue _tcls = ToolRegistry.get(_tname) if isinstance(_tcls, type) and issubclass(_tcls, BaseTool): - _channel_tools.append(_tcls()) + if _tname == "knowledge_search": + from openjarvis.connectors.store import ( + KnowledgeStore, + ) + + _channel_tools.append(_tcls(store=KnowledgeStore())) + else: + _channel_tools.append(_tcls()) elif isinstance(_tcls, BaseTool): _channel_tools.append(_tcls)