hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

fix(email): dispatch partial batches, seen-after-fetch UIDs, reconnect UID baseline restore

Browse Source 
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.
This commit is contained in:
Teknium
2026-08-13 00:20:26 -07:00
parent 9b8da52f41
commit a7f0abc845
2 changed files with 319 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files

194
plugins/platforms/email/adapter.py
View File

@@ -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.

190
tests/gateway/test_email.py
View File

@@ -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"] = "<batch1@test.com>"
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"] = "<ok@test.com>"
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"] = "<good@test.com>"
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."""