fix(imap-proxy): persist IMAP flags and mark mail as read (#1090)

Fix IMAP flag persistence so read/unread state survives reconnects, and align SEARCH/FETCH behavior with persisted flags.

Co-authored-by: bounce12340 <bounce12340@users.noreply.github.com>

smtp_proxy_server/.env.example

@@ -7,3 +7,4 @@ imap_port=11143
 # imap_tls_key=/path/to/key.pem
 # imap_cache_size=500
 # imap_http_timeout=30.0
+# imap_flag_db_path=data/imap_flags.db

smtp_proxy_server/config.py

@@ -21,6 +21,7 @@ class Settings(BaseSettings):
     imap_tls_key: str = ""
     imap_cache_size: int = 500
     imap_http_timeout: float = 30.0
+    imap_flag_db_path: str = "data/imap_flags.db"
 
     model_config = SettingsConfigDict(env_file=".env")
 

smtp_proxy_server/docker-compose.yaml

@@ -12,3 +12,6 @@ services:
       - proxy_url=https://temp-email-api.xxx.xxx
       - port=8025
       - imap_port=11143
+    volumes:
+      # Persists IMAP flags (e.g. \Seen) across container restarts
+      - ./data:/app/data

smtp_proxy_server/flag_store.py

@@ -0,0 +1,178 @@
+import json
+import os
+import sqlite3
+import threading
+
+
+def _encode_flags(flags: set[str]) -> str:
+    # JSON, not a comma-joined string: commas are legal inside IMAP keywords,
+    # so "foo,bar" as a single keyword must not be split into two on read.
+    return json.dumps(sorted(flags))
+
+
+def _decode_flags(text: str) -> set[str]:
+    if not text:
+        return set()
+    try:
+        parsed = json.loads(text)
+        if isinstance(parsed, list):
+            return {str(flag) for flag in parsed}
+    except ValueError:
+        pass
+    # Rows written before the switch to JSON were comma-joined.
+    return set(text.split(","))
+
+
+class FlagStore:
+    """Persists IMAP message flags (e.g. \\Seen) to a local SQLite file.
+
+    SimpleMailbox previously kept flags only in an in-memory dict that was
+    recreated from scratch for every IMAP connection (see SimpleRealm.requestAvatar
+    in imap_server.py), so a client's STORE command (e.g. marking a message as
+    read) was silently lost as soon as the session ended. This store gives
+    flags a durable home keyed by (address, mailbox, uid) so they survive
+    reconnects.
+    """
+
+    _UPSERT_SQL = """
+        INSERT INTO imap_flags (address, mailbox, uid, flags)
+        VALUES (?, ?, ?, ?)
+        ON CONFLICT(address, mailbox, uid) DO UPDATE SET flags = excluded.flags
+        """
+
+    def __init__(self, db_path: str):
+        self._db_path = db_path
+        self._lock = threading.Lock()
+        self._init_db()
+
+    def _connect(self) -> sqlite3.Connection:
+        # isolation_level=None: transactions are managed explicitly below, so
+        # sqlite3 must not inject its own implicit BEGIN.
+        return sqlite3.connect(self._db_path, timeout=10, isolation_level=None)
+
+    def _init_db(self):
+        dirname = os.path.dirname(self._db_path)
+        if dirname:
+            os.makedirs(dirname, exist_ok=True)
+        with self._lock:
+            conn = self._connect()
+            try:
+                conn.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS imap_flags (
+                        address TEXT NOT NULL,
+                        mailbox TEXT NOT NULL,
+                        uid INTEGER NOT NULL,
+                        flags TEXT NOT NULL,
+                        PRIMARY KEY (address, mailbox, uid)
+                    )
+                    """
+                )
+            finally:
+                conn.close()
+
+    def get_all(self, address: str, mailbox: str) -> dict[int, set[str]]:
+        """Return {uid: flags} for every UID with stored flags in a mailbox."""
+        with self._lock:
+            conn = self._connect()
+            try:
+                rows = conn.execute(
+                    "SELECT uid, flags FROM imap_flags WHERE address = ? AND mailbox = ?",
+                    (address, mailbox),
+                ).fetchall()
+            finally:
+                conn.close()
+        return {uid: _decode_flags(flags) for uid, flags in rows}
+
+    def update_flags(
+        self,
+        address: str,
+        mailbox: str,
+        uids: list[int],
+        flags: set[str],
+        mode: int,
+    ) -> dict[int, set[str]]:
+        """Atomically apply an IMAP STORE to the given UIDs.
+
+        mode follows twisted's convention: 1 adds (+FLAGS), -1 removes
+        (-FLAGS), 0 replaces (FLAGS). Returns {uid: resulting flags}.
+
+        The read-modify-write runs inside one BEGIN IMMEDIATE transaction:
+        two concurrent sessions that loaded the same stale in-memory flags
+        cannot overwrite each other's STORE, because each session's update is
+        recomputed from the row current at its own commit.
+        """
+        if not uids:
+            return {}
+        flags = set(flags)
+        result: dict[int, set[str]] = {}
+        with self._lock:
+            conn = self._connect()
+            try:
+                conn.execute("BEGIN IMMEDIATE")
+                try:
+                    placeholders = ",".join("?" for _ in uids)
+                    rows = conn.execute(
+                        "SELECT uid, flags FROM imap_flags"
+                        " WHERE address = ? AND mailbox = ?"
+                        f" AND uid IN ({placeholders})",
+                        (address, mailbox, *uids),
+                    ).fetchall()
+                    current = {uid: _decode_flags(text) for uid, text in rows}
+                    upserts = []
+                    for uid in uids:
+                        old = current.get(uid, set())
+                        if mode == 1:
+                            new = old | flags
+                        elif mode == -1:
+                            new = old - flags
+                        else:
+                            new = set(flags)
+                        result[uid] = new
+                        upserts.append((address, mailbox, uid, _encode_flags(new)))
+                    conn.executemany(self._UPSERT_SQL, upserts)
+                    conn.execute("COMMIT")
+                except BaseException:
+                    conn.execute("ROLLBACK")
+                    raise
+            finally:
+                conn.close()
+        return result
+
+    def set_flags_bulk(
+        self, address: str, mailbox: str, uid_flags: dict[int, set[str]]
+    ) -> None:
+        """Upsert flags for multiple UIDs in a single transaction."""
+        if not uid_flags:
+            return
+        rows = [
+            (address, mailbox, uid, _encode_flags(flags))
+            for uid, flags in uid_flags.items()
+        ]
+        with self._lock:
+            conn = self._connect()
+            try:
+                conn.execute("BEGIN IMMEDIATE")
+                try:
+                    conn.executemany(self._UPSERT_SQL, rows)
+                    conn.execute("COMMIT")
+                except BaseException:
+                    conn.execute("ROLLBACK")
+                    raise
+            finally:
+                conn.close()
+
+
+_default_store: FlagStore | None = None
+_default_store_lock = threading.Lock()
+
+
+def get_flag_store() -> FlagStore:
+    """Return the process-wide FlagStore, created lazily from settings."""
+    global _default_store
+    if _default_store is None:
+        with _default_store_lock:
+            if _default_store is None:
+                from config import settings
+                _default_store = FlagStore(settings.imap_flag_db_path)
+    return _default_store

smtp_proxy_server/imap_mailbox.py

@@ -3,11 +3,12 @@ import logging
 import time
 from collections import OrderedDict
 
-from twisted.internet import defer
+from twisted.internet import defer, threads
 from twisted.mail import imap4
 from zope.interface import implementer
 
 from config import settings
+from flag_store import FlagStore, get_flag_store
 from imap_http_client import BackendClient
 from imap_message import SimpleMessage
 from parse_email import generate_email_model, parse_email, clean_raw_headers, fix_mojibake
@@ -51,9 +52,12 @@ class MessageCache:
 @implementer(imap4.IMailboxInfo, imap4.IMailbox, imap4.ISearchableMailbox)
 class SimpleMailbox:
 
-    def __init__(self, name: str, client: BackendClient):
+    def __init__(self, name: str, client: BackendClient, address: str,
+                 flag_store: FlagStore = None):
         self.name = name
         self._client = client
+        self._address = address
+        self._flag_store = flag_store if flag_store is not None else get_flag_store()
         self.listeners = []
         self.addListener = self.listeners.append
         self.removeListener = self.listeners.remove
@@ -76,7 +80,10 @@ class SimpleMailbox:
         return 0
 
     def getUnseenCount(self):
-        return 0
+        return sum(
+            1 for u in self._uid_index
+            if r"\Seen" not in self._flags.get(u, set())
+        )
 
     def isWriteable(self):
         return 1
@@ -123,6 +130,7 @@ class SimpleMailbox:
         if count == 0:
             self._uid_index = []
             self._uid_index_built = True
+            self._flags = {}
             return
 
         uid_set = set()
@@ -146,6 +154,11 @@ class SimpleMailbox:
 
         self._uid_index = sorted(uid_set)
         self._uid_index_built = True
+        # Load persisted flags (e.g. \Seen set by a previous IMAP session) so
+        # STORE results survive reconnects instead of resetting every session.
+        self._flags = yield threads.deferToThread(
+            self._flag_store.get_all, self._address, self.name
+        )
         _logger.info(
             "UID index built for %s: %d UIDs, range=%s..%s",
             self.name, len(self._uid_index),
@@ -254,9 +267,10 @@ class SimpleMailbox:
                     else:
                         continue
 
-                    if uid_val not in self._flags:
-                        self._flags[uid_val] = {r"\Seen"}
-                    flags = self._flags[uid_val]
+                    # Flags default to unseen (empty set) unless a previous
+                    # STORE persisted them via self._flag_store; self._flags
+                    # is refreshed from that store in _build_uid_index().
+                    flags = self._flags.get(uid_val, set())
                     msg = SimpleMessage(
                         uid_val, email_model, flags=flags, raw=raw,
                         created_at=item.get("created_at"),
@@ -309,18 +323,20 @@ class SimpleMailbox:
             return {}
 
         target_uids = self._resolve_message_set(messages, uid)
+        if not target_uids:
+            return {}
+
+        # Apply the delta inside the store's own transaction instead of
+        # computing new flag sets from this session's in-memory copy: another
+        # concurrent session may have STOREd since we loaded, and basing the
+        # write on stale state would silently drop its change.
+        updated_flags = yield threads.deferToThread(
+            self._flag_store.update_flags,
+            self._address, self.name, target_uids, set(flags), mode,
+        )
+
         result = {}
-
-        for u in target_uids:
-            current_flags = self._flags.get(u, set())
-
-            if mode == 1:    # +FLAGS
-                current_flags = current_flags | set(flags)
-            elif mode == -1:  # -FLAGS
-                current_flags = current_flags - set(flags)
-            elif mode == 0:   # FLAGS (replace)
-                current_flags = set(flags)
-
+        for u, current_flags in updated_flags.items():
             self._flags[u] = current_flags
             seq = self._uid_to_seq(u)
             if seq is not None:
@@ -328,28 +344,58 @@ class SimpleMailbox:
 
         return result
 
+    # SEARCH keys we can answer from persisted flag state, mapped to the flag
+    # and whether it must be present. Anything else falls through to matching
+    # every message, which is how this mailbox answered every query before
+    # flags were persisted.
+    _FLAG_SEARCH_KEYS = {
+        "SEEN": ("\\Seen", True),
+        "UNSEEN": ("\\Seen", False),
+        "DELETED": ("\\Deleted", True),
+        "UNDELETED": ("\\Deleted", False),
+        "FLAGGED": ("\\Flagged", True),
+        "UNFLAGGED": ("\\Flagged", False),
+        "ANSWERED": ("\\Answered", True),
+        "UNANSWERED": ("\\Answered", False),
+        "DRAFT": ("\\Draft", True),
+        "UNDRAFT": ("\\Draft", False),
+    }
+
     @defer.inlineCallbacks
     def search(self, query, uid):
         if not self._uid_index_built:
             yield self._build_uid_index()
 
-        results = []
-
+        predicates = []
         for term in query:
-            if isinstance(term, str) and term.upper() == "ALL":
-                if uid:
-                    results = list(self._uid_index)
-                else:
-                    results = list(range(1, len(self._uid_index) + 1))
-                break
+            if isinstance(term, bytes):
+                term = term.decode("ascii", "ignore")
+            if not isinstance(term, str):
+                continue
+            predicate = self._FLAG_SEARCH_KEYS.get(term.upper())
+            if predicate is not None:
+                predicates.append(predicate)
 
-        if not results:
-            if uid:
-                results = list(self._uid_index)
-            else:
-                results = list(range(1, len(self._uid_index) + 1))
+        matched = [
+            u for u in self._uid_index
+            if all(
+                (flag in self._flags.get(u, set())) is present
+                for flag, present in predicates
+            )
+        ]
 
-        return results
+        _logger.info(
+            "SEARCH: uid=%s predicates=%d matched=%d/%d",
+            uid, len(predicates), len(matched), len(self._uid_index),
+        )
+
+        if uid:
+            return matched
+
+        return [
+            seq for seq in (self._uid_to_seq(u) for u in matched)
+            if seq is not None
+        ]
 
     def getUIDNext(self):
         if self._uid_index:

smtp_proxy_server/imap_server.py

@@ -48,6 +48,49 @@ class SimpleIMAPServer(imap4.IMAP4Server):
             return real_write_seq(data)
         self.transport.writeSequence = logging_write_seq
 
+    @staticmethod
+    def _fetch_implies_seen(query):
+        """Whether a FETCH query marks the messages it returns as read.
+
+        Per RFC 3501, ``BODY[...]`` sets ``\\Seen`` while ``BODY.PEEK[...]``
+        does not. ``RFC822`` and ``RFC822.TEXT`` are defined as equivalents of
+        ``BODY[]`` / ``BODY[TEXT]`` and so also set it; ``RFC822.HEADER`` is
+        equivalent to ``BODY.PEEK[HEADER]`` and does not.
+        """
+        for part in query:
+            part_type = getattr(part, "type", None)
+            if part_type == "body" and not getattr(part, "peek", False):
+                return True
+            if part_type in ("rfc822", "rfc822text"):
+                return True
+        return False
+
+    def do_FETCH(self, tag, messages, query, uid=0):
+        """Set ``\\Seen`` for non-peek fetches before delivering the response.
+
+        Twisted hands the parsed query to its own response builder and never
+        passes it to the mailbox, so the mailbox alone cannot tell a ``BODY[]``
+        from a ``BODY.PEEK[]``. Flag the messages here, before the FETCH
+        response is generated, so the flags the client receives are current.
+        """
+        if not query or not self._fetch_implies_seen(query):
+            return imap4.IMAP4Server.do_FETCH(self, tag, messages, query, uid)
+
+        def _fetch(_):
+            return imap4.IMAP4Server.do_FETCH(self, tag, messages, query, uid)
+
+        def _store_failed(failure):
+            # A failure to persist \Seen must not cost the client its mail.
+            _logger.warning("FETCH: could not set \\Seen: %s", failure.value)
+            return None
+
+        d = defer.maybeDeferred(
+            self.mbox.store, messages, ["\\Seen"], 1, uid
+        )
+        d.addErrback(_store_failed)
+        d.addCallback(_fetch)
+        return d
+
     def _cbSelectWork(self, mbox, cmdName, tag):
         """Override to add UIDNEXT in SELECT response (RFC 3501)."""
         if mbox is None:
@@ -83,7 +126,7 @@ class Account(imap4.MemoryAccount):
     def _emptyMailbox(self, name, id):
         """Return a dummy mailbox for CREATE requests (e.g. Gmail creating Drafts)."""
         _logger.debug("Accepting CREATE request for %s", name)
-        return SimpleMailbox(name, self._client)
+        return SimpleMailbox(name, self._client, self.name)
 
     def create(self, pathspec):
         """Accept CREATE silently without actually creating mailboxes."""
@@ -111,8 +154,8 @@ class SimpleRealm:
 
         client = BackendClient(password)
 
-        inbox = SimpleMailbox("INBOX", client)
-        sent = SimpleMailbox("SENT", client)
+        inbox = SimpleMailbox("INBOX", client, username)
+        sent = SimpleMailbox("SENT", client, username)
 
         account = Account(username)
         account._client = client

smtp_proxy_server/test_flag_persistence.py

@@ -0,0 +1,288 @@
+"""Regression test for GH issue #1074: IMAP STORE could not mark mail as read.
+
+Root cause (see 1074-report.md for the full investigation):
+- smtp_proxy_server/imap_server.py SimpleRealm.requestAvatar() builds a brand
+  new SimpleMailbox (with an empty in-memory `_flags` dict) on every IMAP
+  login/connection.
+- smtp_proxy_server/imap_mailbox.py SimpleMailbox.store() only ever wrote
+  flag changes into that in-memory dict, and
+  SimpleMailbox._fetch_and_cache_messages() unconditionally defaulted every
+  message's flags to `{\\Seen}`.
+- Net effect: every message always looked "already read", and a client's
+  STORE +FLAGS (\\Seen) command appeared to succeed but was silently
+  discarded the moment the IMAP session ended (e.g. Thunderbird reconnecting
+  to poll), which matches the reported "cannot mark mail as read".
+
+This test drives SimpleMailbox directly (no Docker / no live worker
+backend) against a FakeBackendClient, and verifies that:
+1. A never-touched message starts unseen (not hardcoded \\Seen).
+2. STORE +FLAGS (\\Seen) is visible to a brand new SimpleMailbox instance
+   pointed at the same flag_store.FlagStore, simulating a client
+   disconnect + reconnect.
+"""
+import os
+import shutil
+import tempfile
+import unittest
+
+from twisted.internet import defer
+from twisted.mail import imap4
+from twisted.python.failure import Failure
+
+import imap_mailbox
+from flag_store import FlagStore
+from imap_mailbox import SimpleMailbox
+from imap_server import SimpleIMAPServer
+
+
+def _sync_defer_to_thread(f, *args, **kwargs):
+    """Stand-in for twisted.internet.threads.deferToThread.
+
+    imap_mailbox.py offloads FlagStore's blocking sqlite calls to a thread
+    via deferToThread. Tests don't run a reactor to service that thread
+    pool, so this executes the callable inline and wraps the result as an
+    already-fired Deferred instead.
+    """
+    try:
+        result = f(*args, **kwargs)
+    except Exception:
+        return defer.fail()
+    return defer.succeed(result)
+
+
+def _run(d):
+    """Extract the result of a Deferred that fires synchronously."""
+    box = []
+    d.addBoth(box.append)
+    assert box, "Deferred did not fire synchronously"
+    value = box[0]
+    if isinstance(value, Failure):
+        value.raiseException()
+    return value
+
+
+class FakeBackendClient:
+    """Minimal stand-in for imap_http_client.BackendClient.
+
+    Serves a fixed in-memory list of message rows so tests don't need a
+    live worker backend or Docker.
+    """
+
+    def __init__(self, messages):
+        self._messages = messages
+
+    def get_message_count(self, mailbox_name):
+        return defer.succeed(len(self._messages))
+
+    def get_messages(self, mailbox_name, limit, offset):
+        page = self._messages[offset:offset + limit]
+        count = len(self._messages) if offset == 0 else None
+        return defer.succeed((page, count))
+
+
+class _MailboxTestBase(unittest.TestCase):
+    """Shared fixture: a temp-dir FlagStore and an inline deferToThread."""
+
+    def setUp(self):
+        self._tmpdir = tempfile.mkdtemp()
+        self._db_path = os.path.join(self._tmpdir, "flags.db")
+        self._orig_deferToThread = imap_mailbox.threads.deferToThread
+        imap_mailbox.threads.deferToThread = _sync_defer_to_thread
+
+    def tearDown(self):
+        imap_mailbox.threads.deferToThread = self._orig_deferToThread
+        shutil.rmtree(self._tmpdir, ignore_errors=True)
+
+    def _make_mailbox(self, messages, address="user@example.com"):
+        flag_store = FlagStore(self._db_path)
+        client = FakeBackendClient(messages)
+        return SimpleMailbox("INBOX", client, address, flag_store=flag_store)
+
+
+class ImapMarkAsSeenPersistenceTest(_MailboxTestBase):
+
+    def test_new_message_starts_unseen(self):
+        mbox = self._make_mailbox([{"id": 1, "raw": ""}])
+        _run(mbox._build_uid_index())
+        self.assertEqual(mbox.getUnseenCount(), 1)
+
+    def test_store_seen_survives_reconnect(self):
+        messages = [{"id": 1, "raw": ""}, {"id": 2, "raw": ""}]
+
+        # --- Session 1: client connects, selects INBOX, sees 2 unseen ---
+        mbox1 = self._make_mailbox(messages)
+        _run(mbox1._build_uid_index())
+        self.assertEqual(mbox1.getUnseenCount(), 2)
+
+        # Client sends: UID STORE 1 +FLAGS (\Seen)
+        message_set = imap4.MessageSet(1, 1)
+        result = _run(mbox1.store(message_set, [r"\Seen"], mode=1, uid=True))
+        self.assertIn(1, [seq for seq in result])
+        self.assertEqual(mbox1.getUnseenCount(), 1)
+
+        # --- Session 2: client disconnects and reconnects (new SimpleMailbox
+        # instance, exactly like SimpleRealm.requestAvatar creates per login) ---
+        mbox2 = self._make_mailbox(messages)
+        _run(mbox2._build_uid_index())
+
+        self.assertEqual(
+            mbox2.getUnseenCount(), 1,
+            "flag set via STORE in a previous session was lost on reconnect",
+        )
+        self.assertIn(r"\Seen", mbox2._flags.get(1, set()))
+        self.assertNotIn(r"\Seen", mbox2._flags.get(2, set()))
+
+    def test_store_minus_flags_removes_seen(self):
+        messages = [{"id": 1, "raw": ""}]
+        mbox = self._make_mailbox(messages)
+        _run(mbox._build_uid_index())
+
+        message_set = imap4.MessageSet(1, 1)
+        _run(mbox.store(message_set, [r"\Seen"], mode=1, uid=True))
+        self.assertEqual(mbox.getUnseenCount(), 0)
+
+        _run(mbox.store(message_set, [r"\Seen"], mode=-1, uid=True))
+        self.assertEqual(mbox.getUnseenCount(), 1)
+
+        # The removal must also land in SQLite, not just this instance's
+        # in-memory dict: a fresh mailbox (= reconnect) must see it unseen.
+        mbox2 = self._make_mailbox(messages)
+        _run(mbox2._build_uid_index())
+        self.assertEqual(
+            mbox2.getUnseenCount(), 1,
+            "-FLAGS (\\Seen) was not persisted across reconnect",
+        )
+        self.assertNotIn(r"\Seen", mbox2._flags.get(1, set()))
+
+    def test_keyword_containing_comma_round_trips(self):
+        # Commas are legal in IMAP keywords; "foo,bar" is ONE keyword and
+        # must not come back from storage split into "foo" and "bar".
+        messages = [{"id": 1, "raw": ""}]
+        mbox = self._make_mailbox(messages)
+        _run(mbox._build_uid_index())
+
+        message_set = imap4.MessageSet(1, 1)
+        _run(mbox.store(message_set, ["foo,bar"], mode=1, uid=True))
+
+        mbox2 = self._make_mailbox(messages)
+        _run(mbox2._build_uid_index())
+        self.assertEqual(mbox2._flags.get(1), {"foo,bar"})
+
+    def test_concurrent_sessions_do_not_clobber_each_other(self):
+        # Two sessions load the same (empty) flags, then each STOREs a
+        # different flag on the same message. The second write must not be
+        # computed from its stale in-memory copy, or the first flag is lost.
+        messages = [{"id": 1, "raw": ""}]
+        mbox1 = self._make_mailbox(messages)
+        mbox2 = self._make_mailbox(messages)
+        _run(mbox1._build_uid_index())
+        _run(mbox2._build_uid_index())
+
+        message_set = imap4.MessageSet(1, 1)
+        _run(mbox1.store(message_set, [r"\Seen"], mode=1, uid=True))
+        _run(mbox2.store(message_set, [r"\Flagged"], mode=1, uid=True))
+
+        mbox3 = self._make_mailbox(messages)
+        _run(mbox3._build_uid_index())
+        self.assertEqual(
+            mbox3._flags.get(1), {r"\Seen", r"\Flagged"},
+            "a session's +FLAGS was lost to a concurrent session's stale write",
+        )
+
+
+class ImapSearchFlagTest(_MailboxTestBase):
+    """SEARCH must answer flag keys from persisted state.
+
+    SimpleMailbox declares ISearchableMailbox, so IMAP4Server.do_SEARCH hands
+    the whole query to it and never runs its own search_UNSEEN/search_SEEN
+    helpers. Before this, every SEARCH fell through to "return everything",
+    so `SEARCH UNSEEN` listed mail the user had already read.
+    """
+
+    def _seen_mailbox(self):
+        """Two messages, UID 1 marked \\Seen, UID 2 left unread."""
+        messages = [{"id": 1, "raw": ""}, {"id": 2, "raw": ""}]
+        mbox = self._make_mailbox(messages)
+        _run(mbox._build_uid_index())
+        _run(mbox.store(imap4.MessageSet(1, 1), [r"\Seen"], mode=1, uid=True))
+        return mbox
+
+    def test_search_unseen_excludes_seen_messages(self):
+        mbox = self._seen_mailbox()
+        self.assertEqual(_run(mbox.search(["UNSEEN"], uid=True)), [2])
+
+    def test_search_seen_returns_only_seen_messages(self):
+        mbox = self._seen_mailbox()
+        self.assertEqual(_run(mbox.search(["SEEN"], uid=True)), [1])
+
+    def test_search_unseen_returns_sequence_numbers_when_not_uid(self):
+        mbox = self._seen_mailbox()
+        self.assertEqual(_run(mbox.search(["UNSEEN"], uid=False)), [2])
+
+    def test_search_all_still_returns_everything(self):
+        mbox = self._seen_mailbox()
+        self.assertEqual(_run(mbox.search(["ALL"], uid=True)), [1, 2])
+
+    def test_search_accepts_bytes_terms(self):
+        mbox = self._seen_mailbox()
+        self.assertEqual(_run(mbox.search([b"UNSEEN"], uid=True)), [2])
+
+    def test_unsupported_search_key_still_matches_everything(self):
+        # Keys this mailbox cannot evaluate keep the previous lenient
+        # behaviour rather than silently returning an empty result.
+        mbox = self._seen_mailbox()
+        self.assertEqual(_run(mbox.search(["SINCE"], uid=True)), [1, 2])
+
+    def test_combined_keys_are_conjunctive(self):
+        mbox = self._seen_mailbox()
+        _run(mbox.store(imap4.MessageSet(2, 2), [r"\Flagged"], mode=1, uid=True))
+        self.assertEqual(_run(mbox.search(["UNSEEN", "FLAGGED"], uid=True)), [2])
+        self.assertEqual(_run(mbox.search(["SEEN", "FLAGGED"], uid=True)), [])
+
+
+class FetchImpliesSeenTest(unittest.TestCase):
+    """Only non-peek body fetches may set \\Seen (RFC 3501 §6.4.5).
+
+    Twisted passes the parsed FETCH query to its own response builder and
+    never to the mailbox, so SimpleMailbox alone cannot distinguish BODY[]
+    from BODY.PEEK[]. SimpleIMAPServer.do_FETCH makes that call instead.
+    """
+
+    def _implies_seen(self, command):
+        parser = imap4._FetchParser()
+        parser.parseString(command)
+        return SimpleIMAPServer._fetch_implies_seen(parser.result)
+
+    def test_body_sets_seen(self):
+        self.assertTrue(self._implies_seen(b"BODY[]"))
+
+    def test_body_section_sets_seen(self):
+        # BODY[HEADER] is a non-peek fetch and does set \Seen, unlike the
+        # RFC822.HEADER shorthand below.
+        self.assertTrue(self._implies_seen(b"BODY[HEADER]"))
+
+    def test_body_peek_does_not_set_seen(self):
+        self.assertFalse(self._implies_seen(b"BODY.PEEK[]"))
+
+    def test_rfc822_sets_seen(self):
+        self.assertTrue(self._implies_seen(b"RFC822"))
+
+    def test_rfc822_text_sets_seen(self):
+        self.assertTrue(self._implies_seen(b"RFC822.TEXT"))
+
+    def test_rfc822_header_does_not_set_seen(self):
+        # Defined as equivalent to BODY.PEEK[HEADER].
+        self.assertFalse(self._implies_seen(b"RFC822.HEADER"))
+
+    def test_metadata_only_fetch_does_not_set_seen(self):
+        self.assertFalse(self._implies_seen(b"FLAGS"))
+        self.assertFalse(self._implies_seen(b"UID"))
+        self.assertFalse(self._implies_seen(b"RFC822.SIZE"))
+
+    def test_mixed_query_sets_seen_if_any_part_does(self):
+        self.assertTrue(self._implies_seen(b"(FLAGS BODY[])"))
+        self.assertFalse(self._implies_seen(b"(FLAGS BODY.PEEK[])"))
+
+
+if __name__ == "__main__":
+    unittest.main()