From 0c56527d124d2356f622f2e3f651417a9c76349f Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Robert=20Vind-Gardo=C8=99?=
 <41092915+robyroro@users.noreply.github.com>
Date: Wed, 14 Oct 2026 11:01:42 +0300
Subject: [PATCH] fix(email): preserve mail when bulk Trash moves fail (#6633)

* fix(email): preserve mail when bulk Trash moves fail

* fix(email): selective UID EXPUNGE and safe bulk delete cache invalidation

* fix(email): eliminate mailbox-wide EXPUNGE TOCTOU race

- Strictly prohibit mailbox-wide EXPUNGE in bulk email deletion fallback
- Restrict cleanup exclusively to selective RFC 4315 UID EXPUNGE <set>
- Track and preserve pre-existing \Deleted flags if selective expunge fails
- Retain local index records and preserve source emails when cleanup fails
- Ensure consistent partial-failure error semantics across all failure modes
- Add comprehensive hermetic tests for concurrent clients, missing UIDPLUS,
  disconnects after COPY/STORE/EXPUNGE, and retry safety

---------

Co-authored-by: PR #6166 Reviewer <reviewer@odysseus-review.local>
---
 routes/email/email_routes.py           | 107 +++++-
 tests/test_email_bulk_delete_safety.py | 509 +++++++++++++++++++++++++
 2 files changed, 597 insertions(+), 19 deletions(-)
 create mode 100644 tests/test_email_bulk_delete_safety.py

diff --git a/routes/email/email_routes.py b/routes/email/email_routes.py
index fd6bd2747..172790a6c 100644
--- a/routes/email/email_routes.py
+++ b/routes/email/email_routes.py
@@ -81,6 +81,7 @@ _SERVER_OWNED_OAUTH_FIELDS = {
     "oauth_refresh_token",
     "oauth_token_expiry",
 }
+_CACHE_HOOKS: dict = {"invalidate_list": None}
 
 
 def _normalized_mail_host(value) -> str:
@@ -572,6 +573,23 @@ def _uid_exists(conn, uid: str, *, strict: bool = False) -> bool:
         return False
 
 
+def _uid_has_deleted_flag(conn, uid: str) -> bool:
+    r"""Return True if message already carries \Deleted flag."""
+    try:
+        status, data = conn.uid("FETCH", _uid_bytes(uid), "(FLAGS)")
+        if status == "OK" and data:
+            for part in data:
+                if not part:
+                    continue
+                meta = part[0] if isinstance(part, tuple) else part
+                meta_b = meta if isinstance(meta, bytes) else str(meta).encode()
+                if rb"\Deleted" in meta_b:
+                    return True
+    except Exception:
+        pass
+    return False
+
+
 def _resolve_current_email_uid(conn, uid: str, message_id: str | None = None) -> str:
     """Resolve a stale cached UID by the message's stable RFC Message-ID."""
     uid = str(uid or "").strip()
@@ -1831,6 +1849,12 @@ def setup_email_routes():
         delete, move) so the UI doesn't show stale read/unread counts for up to
         the 8s TTL after a manual flag change. With no args, clears everything.
         """
+        hook = _CACHE_HOOKS.get("invalidate_list")
+        if hook:
+            try:
+                hook(account_id, folder)
+            except Exception:
+                pass
         if account_id is None and folder is None:
             _LIST_CACHE.clear()
             return
@@ -4764,6 +4788,7 @@ def setup_email_routes():
             return {"success": False, "error": "No email UIDs provided", "deleted_uids": [], "failed_uids": []}
         deleted_uids = []
         failed_uids = []
+        trash_folder = None
         try:
             with _imap(account_id, owner=owner) as conn:
                 select_status, _ = conn.select(_q(folder), readonly=False)
@@ -4779,7 +4804,8 @@ def setup_email_routes():
                             trash_folder = "Trash"
                     except Exception:
                         pass
-                pending_expunge = False
+                pending_expunge = []
+                preexisting_deleted = set()
                 for uid in uids:
                     try:
                         if not _uid_exists(conn, uid):
@@ -4789,32 +4815,75 @@ def setup_email_routes():
                         if status != "OK":
                             copy_status, _ = conn.uid("COPY", _uid_bytes(uid), _q(trash_folder))
                             if copy_status != "OK":
-                                # Keep deletion reliable even when this IMAP
-                                # server has no writable Trash folder.
-                                store_status, _ = conn.uid("STORE", _uid_bytes(uid), "+FLAGS", "\\Deleted")
-                                if store_status != "OK":
-                                    failed_uids.append(uid)
-                                    continue
-                                pending_expunge = True
-                            else:
-                                store_status, _ = conn.uid("STORE", _uid_bytes(uid), "+FLAGS", "\\Deleted")
-                                if store_status != "OK":
-                                    failed_uids.append(uid)
-                                    continue
-                                pending_expunge = True
+                                # Bulk Delete means move to Trash. Never remove
+                                # the original without a recoverable Trash copy.
+                                failed_uids.append(uid)
+                                continue
+                            if _uid_has_deleted_flag(conn, uid):
+                                preexisting_deleted.add(uid)
+                            store_status, _ = conn.uid("STORE", _uid_bytes(uid), "+FLAGS", "\\Deleted")
+                            if store_status != "OK":
+                                failed_uids.append(uid)
+                                continue
+                            # Keep the local index until the server confirms
+                            # that the copied originals have been removed.
+                            pending_expunge.append(uid)
+                            continue
                         deleted_uids.append(uid)
                         _email_index_delete(owner, account_id, folder, uid)
                     except Exception:
                         failed_uids.append(uid)
                         logger.debug("Bulk email delete failed for uid=%s", uid, exc_info=True)
                 if pending_expunge:
-                    conn.expunge()
-            if deleted_uids:
-                _invalidate_list_cache(account_id, folder)
-            return {"success": True, "deleted_uids": deleted_uids, "failed_uids": failed_uids}
+                    expunge_ok = False
+                    # Attempt selective RFC 4315 UID EXPUNGE so unrelated messages
+                    # carrying \Deleted in the selected mailbox are never expunged.
+                    # Mailbox-wide EXPUNGE is strictly avoided to eliminate TOCTOU races
+                    # where concurrent clients mark unrelated messages \Deleted.
+                    try:
+                        uid_set_bytes = _uid_bytes(",".join(pending_expunge))
+                        uid_exp_status, _ = conn.uid("EXPUNGE", uid_set_bytes)
+                        if uid_exp_status == "OK":
+                            expunge_ok = True
+                    except Exception:
+                        logger.debug("Selective UID EXPUNGE failed", exc_info=True)
+                        expunge_ok = False
+
+                    if expunge_ok:
+                        for uid in pending_expunge:
+                            deleted_uids.append(uid)
+                            _email_index_delete(owner, account_id, folder, uid)
+                    else:
+                        # Prioritize preservation over cleanup: unflag \Deleted from our
+                        # copied messages so future EXPUNGE operations do not destroy them.
+                        # Do not remove a pre-existing \Deleted flag present before this operation.
+                        for uid in pending_expunge:
+                            if uid not in preexisting_deleted:
+                                try:
+                                    conn.uid("STORE", _uid_bytes(uid), "-FLAGS", "\\Deleted")
+                                except Exception:
+                                    pass
+                        failed_uids.extend(pending_expunge)
+            if not deleted_uids:
+                return {"success": False, "error": "Failed to delete requested emails", "deleted_uids": [], "failed_uids": failed_uids}
+            res = {"success": True, "deleted_uids": deleted_uids, "failed_uids": failed_uids}
+            if failed_uids:
+                res["error"] = "Mail operation partially failed"
+            return res
         except Exception as e:
             logger.error(f"Failed to bulk delete emails: {e}")
-            return {"success": False, "error": "Mail operation failed", "deleted_uids": deleted_uids, "failed_uids": [u for u in uids if u not in deleted_uids]}
+            failed_all = [u for u in uids if u not in deleted_uids]
+            return {
+                "success": bool(deleted_uids),
+                "error": "Mail operation failed" if not deleted_uids else "Mail operation partially failed",
+                "deleted_uids": deleted_uids,
+                "failed_uids": failed_all,
+            }
+        finally:
+            if deleted_uids:
+                _invalidate_list_cache(account_id, folder)
+                if trash_folder:
+                    _invalidate_list_cache(account_id, trash_folder)
 
     @router.delete("/odysseus/reminders")
     async def delete_odysseus_reminder_emails(
diff --git a/tests/test_email_bulk_delete_safety.py b/tests/test_email_bulk_delete_safety.py
new file mode 100644
index 000000000..b903c8e97
--- /dev/null
+++ b/tests/test_email_bulk_delete_safety.py
@@ -0,0 +1,509 @@
+"""Bulk Delete must retain messages that could not be moved to Trash."""
+
+from contextlib import contextmanager
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+
+class Mailbox:
+    def __init__(self):
+        self.inbox = {"101", "102", "103"}
+        self.trash = set()
+        self.deleted = set()
+        self.calls = []
+        self.move_status = "NO"
+        self.copy_status = {}
+        self.store_status = {}
+        self.expunge_status = "OK"
+        self.uid_expunge_status = None
+        self.select_status = "OK"
+
+    def select(self, folder, readonly=False):
+        return self.select_status, [b"3"]
+
+    def uid(self, command, raw_uid, *args):
+        uid = raw_uid.decode() if isinstance(raw_uid, bytes) else (str(raw_uid) if raw_uid is not None else "")
+        self.calls.append((command, uid))
+        if command == "FETCH":
+            if uid not in self.inbox:
+                return "OK", []
+            if args and "(FLAGS)" in args:
+                flags = rb"(\Deleted)" if uid in self.deleted else rb"()"
+                return "OK", [(f"1 (UID {uid} FLAGS ".encode() + flags + b")")]
+            return "OK", [f"1 (UID {uid})".encode()]
+        if command == "SEARCH":
+            if args and "DELETED" in args:
+                return "OK", [b" ".join(u.encode() for u in sorted(self.deleted))]
+            searched_uid = args[-1].split()[-1] if args else ""
+            return "OK", [searched_uid.encode() if searched_uid in self.inbox else b""]
+        if command == "MOVE":
+            status = self.move_status.get(uid, "NO") if isinstance(self.move_status, dict) else self.move_status
+            if isinstance(status, Exception):
+                raise status
+            if status == "OK":
+                self.trash.add(uid)
+                self.inbox.remove(uid)
+            return status, []
+        if command == "COPY":
+            status = self.copy_status.get(uid, "OK")
+            if isinstance(status, Exception):
+                raise status
+            if status == "OK":
+                self.trash.add(uid)
+            return status, []
+        if command == "STORE":
+            status = self.store_status.get(uid, "OK")
+            if isinstance(status, Exception):
+                raise status
+            if status == "OK":
+                if args and args[0] == "-FLAGS":
+                    self.deleted.discard(uid)
+                else:
+                    self.deleted.add(uid)
+            return status, []
+        if command == "EXPUNGE":
+            status = self.uid_expunge_status if self.uid_expunge_status is not None else self.expunge_status
+            if isinstance(status, Exception):
+                raise status
+            if status == "OK":
+                expunged = {u for u in uid.split(",") if u in self.deleted}
+                self.inbox.difference_update(expunged)
+                self.deleted.difference_update(expunged)
+            return status, []
+        raise AssertionError(f"Unexpected IMAP command: {command}")
+
+    def expunge(self):
+        self.calls.append(("EXPUNGE", None))
+        if isinstance(self.expunge_status, Exception):
+            raise self.expunge_status
+        if self.expunge_status == "OK":
+            self.inbox.difference_update(self.deleted)
+            self.deleted.clear()
+        return self.expunge_status, []
+
+
+@pytest.fixture
+def bulk_delete(monkeypatch):
+    from routes.email import email_routes as routes
+
+    mailbox = Mailbox()
+    index_delete = Mock()
+    invalidate_cache = Mock()
+
+    @contextmanager
+    def imap(account_id=None, owner=""):
+        assert owner == "alice"
+        yield mailbox
+
+    monkeypatch.setattr(routes, "_start_poller", lambda: None)
+    monkeypatch.setattr(routes, "_POOL_HOOKS", {})
+    monkeypatch.setattr(routes, "_imap", imap)
+    monkeypatch.setattr(routes, "_resolve_mail_folder", lambda *args: "Trash")
+    monkeypatch.setattr(routes, "_list_imap_folders", lambda conn: ({}, ["INBOX", "Trash"]))
+    monkeypatch.setattr(routes, "_email_index_delete", index_delete)
+    monkeypatch.setitem(routes._CACHE_HOOKS, "invalidate_list", invalidate_cache)
+    app = FastAPI()
+    app.include_router(routes.setup_email_routes())
+    app.dependency_overrides[routes.require_owner] = lambda: "alice"
+
+    with TestClient(app) as client:
+        def delete(uids):
+            response = client.post("/api/email/delete-bulk", json={"uids": uids})
+            assert response.status_code == 200
+            return response.json()
+
+        yield SimpleNamespace(
+            mailbox=mailbox,
+            delete=delete,
+            index_delete=index_delete,
+            invalidate_cache=invalidate_cache,
+            routes=routes,
+        )
+
+
+@pytest.mark.parametrize("copy_status", ["NO", "BAD"])
+def test_failed_trash_copy_preserves_original(bulk_delete, copy_status):
+    bulk_delete.mailbox.copy_status["101"] = copy_status
+
+    result = bulk_delete.delete(["101"])
+
+    assert "101" in bulk_delete.mailbox.inbox
+    assert bulk_delete.mailbox.deleted == set()
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101"]
+    assert ("STORE", "101") not in bulk_delete.mailbox.calls
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_mixed_batch_only_deletes_messages_copied_to_trash(bulk_delete):
+    bulk_delete.mailbox.copy_status["102"] = "NO"
+
+    result = bulk_delete.delete(["101", "102", "103"])
+
+    assert bulk_delete.mailbox.inbox == {"102"}
+    assert bulk_delete.mailbox.trash == {"101", "103"}
+    assert result["deleted_uids"] == ["101", "103"]
+    assert result["failed_uids"] == ["102"]
+    assert [call.args[-1] for call in bulk_delete.index_delete.call_args_list] == ["101", "103"]
+
+
+@pytest.mark.parametrize("expunge_status", ["NO", "BAD", RuntimeError("Disconnected")])
+def test_failed_expunge_keeps_index_and_reports_failure(bulk_delete, expunge_status):
+    bulk_delete.mailbox.expunge_status = expunge_status
+
+    result = bulk_delete.delete(["101", "102"])
+
+    assert bulk_delete.mailbox.inbox == {"101", "102", "103"}
+    assert bulk_delete.mailbox.trash == {"101", "102"}
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101", "102"]
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_store_failure_retains_original_and_index(bulk_delete):
+    bulk_delete.mailbox.store_status["101"] = "NO"
+
+    result = bulk_delete.delete(["101"])
+
+    assert "101" in bulk_delete.mailbox.inbox
+    assert bulk_delete.mailbox.trash == {"101"}
+    assert result["failed_uids"] == ["101"]
+    assert result["deleted_uids"] == []
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_native_move_needs_no_copy_or_expunge(bulk_delete):
+    bulk_delete.mailbox.move_status = "OK"
+
+    result = bulk_delete.delete(["101", "102"])
+
+    assert result["deleted_uids"] == ["101", "102"]
+    assert result["failed_uids"] == []
+    assert bulk_delete.mailbox.inbox == {"103"}
+    assert bulk_delete.mailbox.trash == {"101", "102"}
+    assert not any(command in {"COPY", "STORE", "EXPUNGE"} for command, _ in bulk_delete.mailbox.calls)
+
+
+def test_copy_fallback_expunge_runs_once_for_the_batch(bulk_delete):
+    result = bulk_delete.delete(["101", "102"])
+
+    assert result["deleted_uids"] == ["101", "102"]
+    assert result["failed_uids"] == []
+    assert bulk_delete.mailbox.inbox == {"103"}
+    assert bulk_delete.mailbox.trash == {"101", "102"}
+    assert sum(1 for command, _ in bulk_delete.mailbox.calls if command == "EXPUNGE") == 1
+    assert bulk_delete.index_delete.call_count == 2
+
+
+def test_failed_folder_selection_never_changes_messages(bulk_delete):
+    bulk_delete.mailbox.select_status = "NO"
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is False
+    assert result["failed_uids"] == ["101"]
+    assert bulk_delete.mailbox.calls == []
+    bulk_delete.index_delete.assert_not_called()
+
+
+@pytest.mark.parametrize("expunge_status", ["NO", RuntimeError("Disconnected")])
+def test_completed_native_move_is_retained_when_fallback_expunge_fails(bulk_delete, expunge_status):
+    bulk_delete.mailbox.move_status = {"101": "OK"}
+    bulk_delete.mailbox.expunge_status = expunge_status
+
+    result = bulk_delete.delete(["101", "102"])
+
+    assert result["deleted_uids"] == ["101"]
+    assert result["failed_uids"] == ["102"]
+    assert bulk_delete.mailbox.inbox == {"102", "103"}
+    assert bulk_delete.mailbox.trash == {"101", "102"}
+    bulk_delete.index_delete.assert_called_once_with("alice", None, "INBOX", "101")
+
+
+def test_absent_uid_is_reported_failed_without_deleting_another_message(bulk_delete):
+    result = bulk_delete.delete(["999"])
+
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["999"]
+    assert bulk_delete.mailbox.inbox == {"101", "102", "103"}
+    assert bulk_delete.mailbox.trash == set()
+    assert not any(command in {"MOVE", "COPY", "STORE", "EXPUNGE"} for command, _ in bulk_delete.mailbox.calls)
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_duplicate_and_invalid_uids_do_not_repeat_mail_operations(bulk_delete):
+    result = bulk_delete.delete(["101", "101", "", None, "invalid"])
+
+    assert result["deleted_uids"] == ["101"]
+    assert result["failed_uids"] == []
+    assert bulk_delete.mailbox.calls.count(("COPY", "101")) == 1
+    bulk_delete.index_delete.assert_called_once_with("alice", None, "INBOX", "101")
+
+
+def test_bulk_delete_invalidates_source_and_trash_caches(bulk_delete):
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is True
+    assert result["deleted_uids"] == ["101"]
+    bulk_delete.invalidate_cache.assert_any_call(None, "INBOX")
+    bulk_delete.invalidate_cache.assert_any_call(None, "Trash")
+
+
+def test_partial_failure_with_exception_invalidates_caches(bulk_delete):
+    bulk_delete.mailbox.move_status = {"101": "OK"}
+    bulk_delete.mailbox.uid_expunge_status = RuntimeError("Socket reset during expunge")
+    bulk_delete.mailbox.expunge_status = RuntimeError("Socket reset during expunge")
+
+    result = bulk_delete.delete(["101", "102"])
+
+    assert result["success"] is True
+    assert result["deleted_uids"] == ["101"]
+    assert result["failed_uids"] == ["102"]
+    assert result["error"] == "Mail operation partially failed"
+    bulk_delete.invalidate_cache.assert_any_call(None, "INBOX")
+    bulk_delete.invalidate_cache.assert_any_call(None, "Trash")
+
+
+def test_total_failure_returns_success_false(bulk_delete):
+    bulk_delete.mailbox.copy_status["101"] = "NO"
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is False
+    assert result["error"] == "Failed to delete requested emails"
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101"]
+
+
+def test_uid_expunge_preserves_unrelated_deleted_message(bulk_delete):
+    # Unrelated message 999 is already marked \Deleted before bulk delete
+    bulk_delete.mailbox.inbox.add("999")
+    bulk_delete.mailbox.deleted.add("999")
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is True
+    assert result["deleted_uids"] == ["101"]
+    # 101 was expunged, 999 remains in inbox and remains marked deleted
+    assert "101" not in bulk_delete.mailbox.inbox
+    assert "999" in bulk_delete.mailbox.inbox
+    assert bulk_delete.mailbox.deleted == {"999"}
+    assert ("EXPUNGE", "101") in bulk_delete.mailbox.calls
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+
+
+def test_server_lacking_uidplus_never_calls_mailbox_expunge(bulk_delete):
+    # Server does not support UIDPLUS (UID EXPUNGE returns BAD)
+    # Even if no other deleted messages exist, mailbox-wide EXPUNGE is strictly forbidden.
+    bulk_delete.mailbox.uid_expunge_status = "BAD"
+    assert bulk_delete.mailbox.deleted == set()
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is False
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101"]
+    assert "101" in bulk_delete.mailbox.inbox
+    assert "101" not in bulk_delete.mailbox.deleted
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_concurrent_client_marks_unrelated_deleted_message_during_operation(bulk_delete):
+    # Another client marks unrelated UID 999 \Deleted before/during the operation
+    bulk_delete.mailbox.inbox.add("999")
+    bulk_delete.mailbox.deleted.add("999")
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is True
+    assert result["deleted_uids"] == ["101"]
+    # 101 was expunged, unrelated 999 is NEVER expunged
+    assert "101" not in bulk_delete.mailbox.inbox
+    assert "999" in bulk_delete.mailbox.inbox
+    assert bulk_delete.mailbox.deleted == {"999"}
+    assert ("EXPUNGE", "101") in bulk_delete.mailbox.calls
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+
+
+def test_preexisting_deleted_flag_preserved_when_expunge_fails(bulk_delete):
+    # Message 101 already had \Deleted flag before the bulk operation
+    bulk_delete.mailbox.deleted.add("101")
+    bulk_delete.mailbox.uid_expunge_status = "NO"
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is False
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101"]
+    assert "101" in bulk_delete.mailbox.inbox
+    # Pre-existing \Deleted flag must NOT be removed
+    assert "101" in bulk_delete.mailbox.deleted
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_preexisting_deleted_flag_expunged_on_success(bulk_delete):
+    # Message 101 already had \Deleted flag before the bulk operation
+    bulk_delete.mailbox.deleted.add("101")
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is True
+    assert result["deleted_uids"] == ["101"]
+    assert "101" not in bulk_delete.mailbox.inbox
+    assert "101" not in bulk_delete.mailbox.deleted
+    assert "101" in bulk_delete.mailbox.trash
+    bulk_delete.index_delete.assert_called_once_with("alice", None, "INBOX", "101")
+
+
+@pytest.mark.parametrize("status", ["NO", "BAD"])
+def test_uid_expunge_status_failures(bulk_delete, status):
+    bulk_delete.mailbox.uid_expunge_status = status
+
+    result = bulk_delete.delete(["101", "102"])
+
+    assert result["success"] is False
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101", "102"]
+    assert bulk_delete.mailbox.inbox == {"101", "102", "103"}
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_uid_expunge_connection_exception_handled_conservatively(bulk_delete):
+    bulk_delete.mailbox.uid_expunge_status = ConnectionResetError("Connection lost during selective expunge")
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is False
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101"]
+    assert "101" in bulk_delete.mailbox.inbox
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_disconnect_after_copy_before_store_preserves_message(bulk_delete):
+    # COPY succeeds, but STORE raises a disconnect
+    bulk_delete.mailbox.store_status["101"] = ConnectionResetError("Disconnected")
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is False
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101"]
+    assert "101" in bulk_delete.mailbox.inbox
+    assert "101" in bulk_delete.mailbox.trash
+    assert bulk_delete.mailbox.deleted == set()
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_disconnect_after_store_preserves_messages_and_index(bulk_delete):
+    # COPY and STORE succeed, but UID EXPUNGE disconnects
+    bulk_delete.mailbox.uid_expunge_status = ConnectionResetError("Disconnected during expunge")
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is False
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101"]
+    assert "101" in bulk_delete.mailbox.inbox
+    assert "101" in bulk_delete.mailbox.trash
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_disconnect_with_partial_move_success(bulk_delete):
+    # 101 succeeds via MOVE, 102 falls back to COPY+STORE and UID EXPUNGE disconnects
+    bulk_delete.mailbox.move_status = {"101": "OK", "102": "NO"}
+    bulk_delete.mailbox.uid_expunge_status = ConnectionResetError("Disconnected during expunge")
+
+    result = bulk_delete.delete(["101", "102"])
+
+    assert result["success"] is True
+    assert result["deleted_uids"] == ["101"]
+    assert result["failed_uids"] == ["102"]
+    assert "101" not in bulk_delete.mailbox.inbox
+    assert "102" in bulk_delete.mailbox.inbox
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_called_once_with("alice", None, "INBOX", "101")
+    bulk_delete.invalidate_cache.assert_any_call(None, "INBOX")
+    bulk_delete.invalidate_cache.assert_any_call(None, "Trash")
+
+
+def test_lost_acknowledgement_retains_local_index_and_permits_safe_retry(bulk_delete):
+    # Server succeeds in moving, but connection drops during acknowledgement
+    bulk_delete.mailbox.uid_expunge_status = OSError("Connection dropped")
+    bulk_delete.mailbox.expunge_status = OSError("Connection dropped")
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is False
+    assert result["deleted_uids"] == []
+    assert result["failed_uids"] == ["101"]
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+    bulk_delete.index_delete.assert_not_called()
+
+    # Retry when server is reconnected and 101 was already moved (absent from source)
+    bulk_delete.mailbox.inbox.discard("101")
+    bulk_delete.mailbox.uid_expunge_status = "OK"
+    bulk_delete.mailbox.expunge_status = "OK"
+
+    retry_result = bulk_delete.delete(["101"])
+    assert retry_result["deleted_uids"] == []
+    assert retry_result["failed_uids"] == ["101"]
+    bulk_delete.index_delete.assert_not_called()
+
+
+def test_repeated_requests_after_partial_failure(bulk_delete):
+    # Initial request: 101 moves, 102 fails copy
+    bulk_delete.mailbox.move_status = {"101": "OK", "102": "NO"}
+    bulk_delete.mailbox.copy_status["102"] = "NO"
+
+    res1 = bulk_delete.delete(["101", "102"])
+    assert res1["deleted_uids"] == ["101"]
+    assert res1["failed_uids"] == ["102"]
+
+    # Retry with failed UID 102 now succeeding via copy
+    bulk_delete.mailbox.copy_status["102"] = "OK"
+    res2 = bulk_delete.delete(["102"])
+    assert res2["deleted_uids"] == ["102"]
+    assert res2["failed_uids"] == []
+
+    assert bulk_delete.mailbox.inbox == {"103"}
+    assert bulk_delete.mailbox.trash == {"101", "102"}
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+
+
+def test_unrelated_messages_never_expunged(bulk_delete):
+    # Mailbox contains unrelated 888 (with \Deleted) and 999 (without \Deleted)
+    bulk_delete.mailbox.inbox.update({"888", "999"})
+    bulk_delete.mailbox.deleted.add("888")
+
+    result = bulk_delete.delete(["101"])
+
+    assert result["success"] is True
+    assert result["deleted_uids"] == ["101"]
+    assert "101" not in bulk_delete.mailbox.inbox
+    assert "101" in bulk_delete.mailbox.trash
+    # 888 and 999 are completely untouched in inbox
+    assert "888" in bulk_delete.mailbox.inbox
+    assert "999" in bulk_delete.mailbox.inbox
+    assert "888" in bulk_delete.mailbox.deleted
+    assert "888" not in bulk_delete.mailbox.trash
+    assert "999" not in bulk_delete.mailbox.trash
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls
+
+
+def test_standard_expunge_never_invoked_by_fallback(bulk_delete):
+    # Perform operations under multiple fallback conditions; standard expunge is never called
+    bulk_delete.delete(["101"])
+    assert ("EXPUNGE", None) not in bulk_delete.mailbox.calls