From a7f0abc8456e2cd420f74a42f81fdea7d50d4951 Mon Sep 17 00:00:00 2001 From: Teknium <127238744+teknium1@users.noreply.github.com> Date: Thu, 13 Aug 2026 00:20:26 -0700 Subject: [PATCH] fix(email): dispatch partial batches, seen-after-fetch UIDs, reconnect UID baseline restore MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-ups to the salvaged #80032 fatal-error escalation, closing the gaps its review thread identified plus a sibling of the same class: 1. Partial-batch loss: _check_inbox now dispatches whatever the fetch returned BEFORE escalating a failure — the early-return dropped already-fetched messages whose UIDs were marked seen. 2. Seen-after-fetch: UIDs enter _seen_uids only after their fetch returns a response, so a mid-batch connection failure leaves the remaining UIDs eligible for the next poll. Per-message processing moved to _parse_fetched_message behind a poison guard: a message that fails parsing/auth-verification is marked seen, logged with its UID, and skipped once — never an eternal crash loop. 3. Reconnect mail loss: connect(is_reconnect=True) restores the account's seen-UID baseline from a class-level snapshot instead of re-marking the entire mailbox seen — mail that arrived during an outage is now processed after the reconnect the escalation triggers. 7 new regression tests. --- plugins/platforms/email/adapter.py | 194 +++++++++++++++++++---------- tests/gateway/test_email.py | 190 ++++++++++++++++++++++++++++ 2 files changed, 319 insertions(+), 65 deletions(-) diff --git a/plugins/platforms/email/adapter.py b/plugins/platforms/email/adapter.py index a718443267..317eae72e8 100644 --- a/plugins/platforms/email/adapter.py +++ b/plugins/platforms/email/adapter.py @@ -507,6 +507,15 @@ def _extract_attachments( class EmailAdapter(BasePlatformAdapter): """Email gateway adapter using IMAP (receive) and SMTP (send).""" + # Per-account snapshot of seen UIDs, surviving adapter recreation. + # The gateway's reconnect watcher builds a FRESH adapter instance for + # each retry; without this, connect(is_reconnect=True) would re-mark the + # entire mailbox seen and silently skip every message that arrived + # during the outage. Keyed by account address (multiplex gateways can + # run several email accounts in one process). Same-process only by + # design — after a full restart the usual mark-all-seen baseline applies. + _seen_uids_snapshot: Dict[str, set] = {} + def __init__(self, config: PlatformConfig): super().__init__(config, Platform.EMAIL) @@ -675,16 +684,34 @@ class EmailAdapter(BasePlatformAdapter): imap = imaplib.IMAP4_SSL(self._imap_host, self._imap_port, timeout=30) imap.login(self._address, self._password) _send_imap_id(imap) - # Mark all existing messages as seen so we only process new ones imap.select("INBOX") - status, data = imap.uid("search", None, "ALL") - if status == "OK" and data and data[0]: - for uid in data[0].split(): - self._seen_uids.add(uid) - # Keep only the most recent UIDs to prevent unbounded growth - self._trim_seen_uids() - imap.logout() - logger.info("[Email] IMAP connection test passed. %d existing messages skipped.", len(self._seen_uids)) + snapshot = self._seen_uids_snapshot.get(self._address) + if is_reconnect and snapshot is not None: + # Reconnect within the same process: restore the previous + # adapter's seen-UID baseline instead of re-marking the whole + # mailbox. Mail that arrived during the outage stays UNSEEN + # relative to the baseline and is dispatched by the next poll + # instead of being silently skipped. + self._seen_uids = set(snapshot) + self._trim_seen_uids() + imap.logout() + logger.info( + "[Email] IMAP reconnect test passed. Restored %d seen UIDs; " + "messages received during the outage will be processed.", + len(self._seen_uids), + ) + else: + # First connect (or no snapshot): mark all existing messages as + # seen so we only process new ones. + status, data = imap.uid("search", None, "ALL") + if status == "OK" and data and data[0]: + for uid in data[0].split(): + self._seen_uids.add(uid) + # Keep only the most recent UIDs to prevent unbounded growth + self._trim_seen_uids() + imap.logout() + logger.info("[Email] IMAP connection test passed. %d existing messages skipped.", len(self._seen_uids)) + self._seen_uids_snapshot[self._address] = set(self._seen_uids) except Exception as e: logger.error("[Email] IMAP connection failed: %s", e) # Always set an explicit fatal code (OOF-156): returning False @@ -768,6 +795,12 @@ 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) + # Dispatch whatever the fetch managed to return BEFORE escalating a + # failure: on a mid-batch exception _fetch_new_messages returns the + # partial results, and dropping them here would lose those messages + # (their processing already marked them seen). + for msg_data in messages: + await self._dispatch_message(msg_data) 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 @@ -783,9 +816,6 @@ class EmailAdapter(BasePlatformAdapter): retryable=True, ) await self._notify_fatal_error() - return - for msg_data in messages: - await self._dispatch_message(msg_data) def _fetch_new_messages(self) -> List[Dict[str, Any]]: """Fetch new (unseen) messages from IMAP. Runs in executor thread.""" @@ -804,21 +834,26 @@ class EmailAdapter(BasePlatformAdapter): for uid in data[0].split(): if uid in self._seen_uids: continue + + status, msg_data = imap.uid("fetch", uid, "(RFC822)") + if status != "OK": + # Transient per-UID fetch refusal: leave the UID out of + # _seen_uids so the next poll retries it. + continue + + # IMAP fetch can return unexpected structures (e.g. a + # single bytes item instead of a list of tuples). Mark the + # UID seen once a response arrived (even a malformed one) + # so a garbage response is skipped once, not retried + # forever — but NOT before the fetch: a connection failure + # above must leave the remaining batch eligible for the + # next poll instead of permanently skipping it (#80032 + # review). self._seen_uids.add(uid) # Trim periodically to prevent unbounded memory growth if len(self._seen_uids) > self._seen_uids_max: self._trim_seen_uids() - status, msg_data = imap.uid("fetch", uid, "(RFC822)") - if status != "OK": - continue - - # IMAP fetch can return unexpected structures (e.g. a - # single bytes item instead of a list of tuples). Guard - # against IndexError / TypeError so one malformed response - # doesn't abort the batch — the UID is already in - # _seen_uids, so an abort would permanently skip the - # remaining messages in this batch. try: raw_email = msg_data[0][1] except (IndexError, TypeError): @@ -832,50 +867,22 @@ class EmailAdapter(BasePlatformAdapter): "[Email] Non-bytes IMAP payload for UID %s, skipping", uid ) continue - msg = email_lib.message_from_bytes(raw_email) - - sender_raw = msg.get("From", "") - sender_addr = _extract_email_address(sender_raw) - sender_name = _decode_header_value(sender_raw) - # Remove email from name if present - if "<" in sender_name: - sender_name = sender_name.split("<")[0].strip().strip('"') - - subject = _decode_header_value(msg.get("Subject", "(no subject)")) - message_id = msg.get("Message-ID", "") - in_reply_to = msg.get("In-Reply-To", "") - # Skip automated/noreply senders before any processing - msg_headers = dict(msg.items()) - if _is_automated_sender(sender_addr, msg_headers): - logger.debug("[Email] Skipping automated sender: %s", sender_addr) + # Per-message processing guard: one poison message + # (unparseable headers, pathological attachment, DNS + # hiccup in SPF/DKIM verification) must not abort the + # batch or escalate to a reconnect — it is already marked + # seen above, so log the UID and move on (#80032 review). + try: + parsed = self._parse_fetched_message(uid, raw_email) + except Exception as parse_exc: + logger.error( + "[Email] Failed to process message UID %s, skipping: %s", + uid, + parse_exc, + ) continue - - # Verify the From: domain is authenticated (SPF/DKIM/DMARC) - # while the raw message — and its trusted - # Authentication-Results header — is still in scope. The - # verdict is consumed at dispatch where authorization is - # decided. From: is attacker-controlled, so this is the only - # place a spoof can be caught (GHSA-rxqh-5572-8m77). - sender_authenticated, auth_reason = _verify_sender_authentication( - msg, sender_addr, authserv_id=self._authserv_id - ) - - body = _extract_text_body(msg) - attachments = _extract_attachments(msg, skip_attachments=self._skip_attachments) - - results.append({ - "uid": uid, - "sender_addr": sender_addr, - "sender_name": sender_name, - "subject": subject, - "message_id": message_id, - "in_reply_to": in_reply_to, - "body": body, - "attachments": attachments, - "date": msg.get("Date", ""), - "sender_authenticated": sender_authenticated, - "auth_reason": auth_reason, - }) + if parsed is not None: + results.append(parsed) finally: try: imap.logout() @@ -885,8 +892,65 @@ class EmailAdapter(BasePlatformAdapter): logger.error("[Email] IMAP fetch error: %s", e) self._last_fetch_failed = True self._last_fetch_error = str(e) + # Keep the reconnect snapshot current with every poll so a mid-outage + # adapter recreation restores an up-to-date baseline: stale snapshots + # would re-dispatch messages this instance already processed. + self._seen_uids_snapshot[self._address] = set(self._seen_uids) return results + def _parse_fetched_message(self, uid: bytes, raw_email: "bytes | bytearray") -> Optional[Dict[str, Any]]: + """Parse one fetched RFC822 payload into a dispatchable dict. + + Returns ``None`` for messages that should be silently skipped + (automated/noreply senders). Raises on pathological input — the + caller's per-message guard logs the UID and continues, so a poison + message never aborts the batch or escalates to a reconnect. + """ + msg = email_lib.message_from_bytes(raw_email) + + sender_raw = msg.get("From", "") + sender_addr = _extract_email_address(sender_raw) + sender_name = _decode_header_value(sender_raw) + # Remove email from name if present + if "<" in sender_name: + sender_name = sender_name.split("<")[0].strip().strip('"') + + subject = _decode_header_value(msg.get("Subject", "(no subject)")) + message_id = msg.get("Message-ID", "") + in_reply_to = msg.get("In-Reply-To", "") + # Skip automated/noreply senders before any processing + msg_headers = dict(msg.items()) + if _is_automated_sender(sender_addr, msg_headers): + logger.debug("[Email] Skipping automated sender: %s", sender_addr) + return None + + # Verify the From: domain is authenticated (SPF/DKIM/DMARC) + # while the raw message — and its trusted + # Authentication-Results header — is still in scope. The + # verdict is consumed at dispatch where authorization is + # decided. From: is attacker-controlled, so this is the only + # place a spoof can be caught (GHSA-rxqh-5572-8m77). + sender_authenticated, auth_reason = _verify_sender_authentication( + msg, sender_addr, authserv_id=self._authserv_id + ) + + body = _extract_text_body(msg) + attachments = _extract_attachments(msg, skip_attachments=self._skip_attachments) + + return { + "uid": uid, + "sender_addr": sender_addr, + "sender_name": sender_name, + "subject": subject, + "message_id": message_id, + "in_reply_to": in_reply_to, + "body": body, + "attachments": attachments, + "date": msg.get("Date", ""), + "sender_authenticated": sender_authenticated, + "auth_reason": auth_reason, + } + @staticmethod def _allow_all_senders() -> bool: """Return True when the operator opted into accepting any sender. diff --git a/tests/gateway/test_email.py b/tests/gateway/test_email.py index f1735f55a6..e0ea57ce3b 100644 --- a/tests/gateway/test_email.py +++ b/tests/gateway/test_email.py @@ -606,6 +606,196 @@ class TestPollLoop(unittest.TestCase): self.assertTrue(adapter.fatal_error_retryable) self.assertIn("read operation timed out", adapter.fatal_error_message) + def test_partial_batch_dispatched_before_escalation(self): + """A mid-batch IMAP failure must dispatch the messages already + fetched BEFORE escalating — dropping them would lose mail, since + their UIDs are marked seen (#80032 review).""" + import asyncio + adapter = self._make_adapter() + dispatched, notified = [], [] + + async def mock_dispatch(msg_data): + dispatched.append(msg_data) + + async def mock_fatal_handler(a): + notified.append(a) + + adapter._dispatch_message = mock_dispatch + adapter.set_fatal_error_handler(mock_fatal_handler) + + raw_email = MIMEText("Body", "plain", "utf-8") + raw_email["From"] = "sender@test.com" + raw_email["Subject"] = "First of batch" + raw_email["Message-ID"] = "" + + mock_imap = MagicMock() + fetches = [] + + def uid_handler(command, *args): + if command == "search": + return ("OK", [b"1 2"]) + if command == "fetch": + fetches.append(args) + if len(fetches) == 1: + return ("OK", [(b"1", raw_email.as_bytes())]) + raise OSError("connection dropped mid-batch") + return ("NO", []) + + mock_imap.uid.side_effect = uid_handler + + with patch("imaplib.IMAP4_SSL", return_value=mock_imap): + asyncio.run(adapter._check_inbox()) + + # The successfully fetched message was dispatched, not dropped. + self.assertEqual(len(dispatched), 1) + self.assertEqual(dispatched[0]["subject"], "First of batch") + # The failure still escalated through the fatal-error hook. + self.assertEqual(len(notified), 1) + self.assertEqual(adapter.fatal_error_code, "email_imap_fetch_failed") + + def test_mid_batch_failure_leaves_unfetched_uids_eligible(self): + """UIDs are marked seen only after their fetch returns — a + connection failure mid-batch must leave the remaining UIDs eligible + for the next poll instead of permanently skipping them.""" + adapter = self._make_adapter() + + raw_email = MIMEText("Body", "plain", "utf-8") + raw_email["From"] = "sender@test.com" + raw_email["Subject"] = "ok" + raw_email["Message-ID"] = "" + + mock_imap = MagicMock() + fetches = [] + + def uid_handler(command, *args): + if command == "search": + return ("OK", [b"1 2 3"]) + if command == "fetch": + fetches.append(args) + if len(fetches) == 1: + return ("OK", [(b"1", raw_email.as_bytes())]) + raise OSError("connection dropped") + return ("NO", []) + + mock_imap.uid.side_effect = uid_handler + + with patch("imaplib.IMAP4_SSL", return_value=mock_imap): + results = adapter._fetch_new_messages() + + self.assertEqual(len(results), 1) + self.assertIn(b"1", adapter._seen_uids) # fetched → seen + self.assertNotIn(b"2", adapter._seen_uids) # fetch raised → retry next poll + self.assertNotIn(b"3", adapter._seen_uids) # never reached → retry next poll + self.assertTrue(adapter._last_fetch_failed) + + def test_poison_message_skipped_once_without_escalation(self): + """A message whose processing raises is marked seen and skipped — + it must not abort the batch, escalate to a reconnect, or be + retried forever (#80032 review).""" + adapter = self._make_adapter() + + good_email = MIMEText("Body", "plain", "utf-8") + good_email["From"] = "sender@test.com" + good_email["Subject"] = "good" + good_email["Message-ID"] = "" + + mock_imap = MagicMock() + + def uid_handler(command, *args): + if command == "search": + return ("OK", [b"1 2"]) + if command == "fetch": + uid = args[0] + if uid == b"1": + return ("OK", [(b"1", b"poison")]) + return ("OK", [(b"2", good_email.as_bytes())]) + return ("NO", []) + + mock_imap.uid.side_effect = uid_handler + + with patch("imaplib.IMAP4_SSL", return_value=mock_imap), patch( + "plugins.platforms.email.adapter.EmailAdapter._parse_fetched_message", + side_effect=[ValueError("unparseable"), {"subject": "good"}], + ): + results = adapter._fetch_new_messages() + + # Poison message consumed (seen, skipped); good message survived. + self.assertEqual(len(results), 1) + self.assertIn(b"1", adapter._seen_uids) + self.assertIn(b"2", adapter._seen_uids) + self.assertFalse(adapter._last_fetch_failed) + + +class TestReconnectSeenUidsRestore(unittest.TestCase): + """connect(is_reconnect=True) must not re-mark the whole mailbox seen.""" + + 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_SMTP_HOST": "smtp.test.com", + }): + from plugins.platforms.email.adapter import EmailAdapter + adapter = EmailAdapter(PlatformConfig(enabled=True)) + return adapter + + def setUp(self): + from plugins.platforms.email.adapter import EmailAdapter + EmailAdapter._seen_uids_snapshot.clear() + + tearDown = setUp + + def _run_connect(self, adapter, mailbox_uids, *, is_reconnect): + import asyncio + + mock_imap = MagicMock() + + def uid_handler(command, *args): + if command == "search": + return ("OK", [mailbox_uids]) + return ("NO", []) + + mock_imap.uid.side_effect = uid_handler + smtp = MagicMock() + + with patch("imaplib.IMAP4_SSL", return_value=mock_imap), patch.object( + adapter, "_connect_smtp", return_value=smtp + ): + return asyncio.run(adapter.connect(is_reconnect=is_reconnect)) + + def test_reconnect_restores_snapshot_instead_of_marking_all_seen(self): + # First adapter connects with UIDs 1-2 in the mailbox. + first = self._make_adapter() + self.assertTrue(self._run_connect(first, b"1 2", is_reconnect=False)) + self.assertEqual(first._seen_uids, {b"1", b"2"}) + import asyncio + asyncio.run(first.disconnect()) + + # Outage: UID 3 arrives. The reconnect watcher builds a FRESH adapter + # and connects with is_reconnect=True. + second = self._make_adapter() + self.assertTrue(self._run_connect(second, b"1 2 3", is_reconnect=True)) + # Baseline restored from the snapshot — UID 3 stays eligible. + self.assertEqual(second._seen_uids, {b"1", b"2"}) + asyncio.run(second.disconnect()) + + def test_first_connect_still_marks_all_seen(self): + adapter = self._make_adapter() + self.assertTrue(self._run_connect(adapter, b"7 8 9", is_reconnect=False)) + self.assertEqual(adapter._seen_uids, {b"7", b"8", b"9"}) + import asyncio + asyncio.run(adapter.disconnect()) + + def test_reconnect_without_snapshot_falls_back_to_mark_all_seen(self): + # e.g. gateway restarted: no in-process snapshot exists. + adapter = self._make_adapter() + self.assertTrue(self._run_connect(adapter, b"4 5", is_reconnect=True)) + self.assertEqual(adapter._seen_uids, {b"4", b"5"}) + import asyncio + asyncio.run(adapter.disconnect()) + class TestSendEmailStandalone(unittest.TestCase): """Test the standalone _send_email function in send_message_tool."""